增加口径可调用 descriptor：入口 Schema、依赖与出具策略

backend/app/api/ledger.py

@@ -16,6 +16,8 @@
   （app/ledger/certify.py）。
 - GET /api/ledger/recall/{workflow_id}/{version}、POST …/continue（router，搭建者，第 4 波）：口径升到这一版之后的召回批次——
   每组的结论、「另有 M 组未重算」、手动接着算（app/ledger/recall.py）。版本用量（workflow-versions）另加问数回答、导出、召回。
+- GET /api/ledger/calibers、GET /api/ledger/calibers/{workflow_id}/{version}/{node_id}（router，搭建者）：口径可调用描述
+  （app/ledger/descriptor.py）：入口字段的 JSON Schema、依赖、能不能组合和出具；不返回 SQL 和参数值。
 这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。
 """
 from __future__ import annotations
@@ -256,3 +258,28 @@ async def certify_dismiss(payload: DismissIn, session: AsyncSession = Depends(ge
     await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note,
                           principal=current_principal())
     return {"ok": True, "item_id": payload.item_id}
+
+
+@router.get("/calibers")
+async def calibers(workflow_id: str | None = None, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """口径可调用描述（app/ledger/descriptor.py）：每个工作流最新「已发布且未退役」版本里的源头口径卡一张一条，含入口字段的
+    JSON Schema、指标、依赖的表和发布时钉住的数据目录/结构版本、能不能组合、出具策略。workflow_id 可只列一个工作流。
+
+    不返回取数 SQL、不返回参数值（SQL 里可能写着遮罩值）。issuance 只描述这一版的策略（级别、有没有人工节点或写外部系统），
+    不代表这次运行一定能出具：全局自动出具开关、每日上限这类运行时条件不在这里判。"""
+    from app.ledger.descriptor import list_calibers as list_descriptors
+
+    return {"calibers": await list_descriptors(session, workflow_id=workflow_id)}
+
+
+@router.get("/calibers/{workflow_id}/{version}/{node_id}")
+async def caliber_descriptor(workflow_id: str, version: int, node_id: str,
+                             session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """一张源头口径卡的描述（形状同列表里的一条）。工作流不存在、这一版没发布过或已退役、节点不是源头口径卡（转抄卡也算）
+    回 404。同样不返回 SQL、参数值；issuance 只描述版本策略，不代表这次运行一定能出具。"""
+    from app.ledger.descriptor import caliber_descriptor as descriptor_of
+
+    made = await descriptor_of(session, workflow_id, version, node_id)
+    if made is None:
+        raise HTTPException(404, "没有这份口径：工作流不存在、这一版没发布过或已退役，或者节点不是源头口径卡")
+    return made

backend/app/ledger/descriptor.py

@@ -0,0 +1,128 @@
+"""口径可调用描述（descriptor）：给目录页、应用表单、外部集成一份机器可读的稳定描述（next-round 02 的后续）。
+
+口径已经能按固定版本实例化（look/calibers 的查看结果、ledger/compose 的 use_caliber），但入口字段、依赖、能不能出具这些
+契约只写在给模型看的文字里。这里把它们写成结构化的数据：
+
+- 只列每个工作流最新「已发布且未退役」版本里的**源头口径卡**（metrics 节点、caliber_from 为空），和 look/calibers、
+  ledger/derive 的认法一致：转抄卡不列，源头才能复用；
+- 入口字段用 ledger/derive.entry_fields，按 apps/forms.input_schema 转成 JSON Schema（和表单、use_caliber 校验同一个）；
+- 依赖只从图里的取数节点认数据源和表名，**不返回 SQL、不返回参数值**（SQL 里可能写着遮罩值，组合时由服务端照抄）；
+- 出具只描述这一版的策略：级别、图里有没有人工节点或写外部系统的节点。全局的自动出具开关、每日上限是运行时条件，这里
+  不查——descriptor 说的是「这一版允许什么」，不代表这次运行一定能出具。
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.apps import versions as app_versions
+from app.apps.backfill import graph_side_effect
+from app.apps.forms import input_schema
+from app.apps.tree import node_map
+from app.db.models import Workflow
+from app.engine.evidence import sql_tables
+from app.ledger.derive import entry_fields
+from app.look.calibers import card_inputs
+from app.tools.datasource import QUERY_PREFIX
+
+#: 指标只写口径卡配置里真有的这几个键
+METRIC_KEYS = ("id", "name", "unit", "decimals", "format")
+
+
+async def list_calibers(session: AsyncSession, *, workflow_id: str | None = None) -> list[dict[str, Any]]:
+    """每个工作流最新「已发布且未退役」版本里的源头口径卡，一张卡一条描述。给了 workflow_id 只列那个工作流。
+
+    同一版图里的 graph_side_effect 按 (workflow_id, version) 只算一次：它要读子图、跑发布检查，比其余字段慢得多。"""
+    query = select(Workflow).order_by(Workflow.name, Workflow.id)
+    if workflow_id:
+        query = query.where(Workflow.id == workflow_id)
+    side_cache: dict[tuple[str, int], str | None] = {}
+    out: list[dict[str, Any]] = []
+    for workflow in (await session.execute(query)).scalars().all():
+        row = await app_versions.latest_published(session, workflow.id)
+        if row is None:
+            continue
+        for node_id, node in node_map(row.graph or {}).items():
+            if node["type"] != "metrics" or node["config"].get("caliber_from") not in (None, "", {}):
+                continue
+            made = await _describe(session, workflow, row, node_id, side_cache)
+            if made is not None:
+                out.append(made)
+    return out
+
+
+async def caliber_descriptor(session: AsyncSession, workflow_id: str, version: int,
+                             node_id: str) -> dict[str, Any] | None:
+    """一个工作流版本里一张源头口径卡的描述；工作流不存在、版本没发布过或已退役、节点不是源头口径卡时返回 None。"""
+    workflow = await session.get(Workflow, str(workflow_id or ""))
+    if workflow is None:
+        return None
+    row = await app_versions.version_row(session, workflow.id, int(version))
+    if row is None or not app_versions.is_published(row, current=workflow.published_version):
+        return None
+    return await _describe(session, workflow, row, str(node_id), {})
+
+
+async def _describe(session: AsyncSession, workflow: Workflow, row: Any, node_id: str,
+                    side_cache: dict[tuple[str, int], str | None]) -> dict[str, Any] | None:
+    """一张卡的描述；节点不是源头口径卡时返回 None。"""
+    graph = row.graph or {}
+    nodes = node_map(graph)
+    card = nodes.get(node_id)
+    if card is None or card["type"] != "metrics" or card["config"].get("caliber_from") not in (None, "", {}):
+        return None
+    cfg = card["config"]
+    level = str(row.level or "")
+    _, reason = card_inputs(graph, node_id)
+    key = (workflow.id, int(row.version))
+    if key not in side_cache:
+        side_cache[key] = await graph_side_effect(session, graph, level=level or None)
+    side_effect = side_cache[key]
+    sources, tables = _fetches(nodes)
+    reasons: list[str] = []
+    if level != "governed":
+        reasons.append("这一版是已发布级别，只支持一键出具")
+    if side_effect is not None:
+        reasons.append(f"这一版{side_effect}，不能替人重跑")
+    return {
+        "workflow_id": workflow.id,
+        "workflow_name": workflow.name,
+        "version": int(row.version),
+        "level": level,
+        "graph_hash": row.graph_hash,
+        "card": {"node_id": node_id, "title": card["title"], "caliber": cfg.get("caliber"),
+                 "caliber_version": cfg.get("caliber_version")},
+        "metrics": [{k: m[k] for k in METRIC_KEYS if k in m}
+                    for m in cfg.get("metrics") or []
+                    if isinstance(m, dict) and m.get("id")],
+        "entry_schema": input_schema(list(entry_fields(graph))),
+        "composable": reason is None,
+        "compose_reason": reason,
+        "dependencies": {"sources": sorted(sources), "tables": sorted(tables),
+                         "catalog_versions": row.catalog_versions, "schema_versions": row.schema_versions},
+        "issuance": {"level": level, "auto_issue": level == "governed" and side_effect is None,
+                     "requires_independent_proof": True, "side_effects": side_effect, "reasons": reasons},
+    }
+
+
+def _fetches(nodes: dict[str, dict[str, Any]]) -> tuple[set[str], set[str]]:
+    """图里调用数据库查询工具、而且写了 SQL 的节点：数据源名、SQL 里 FROM / JOIN 的表名（去库名前缀）。只认表名，不读 SQL。"""
+    sources: set[str] = set()
+    tables: set[str] = set()
+    for node in nodes.values():
+        if node["type"] != "tool":
+            continue
+        tool = str(node["config"].get("tool") or "")
+        args = node["config"].get("args")
+        sql = args.get("sql") if isinstance(args, dict) else None
+        # 工具名带模板的不是固定的数据源，认不出「哪一版库」
+        if not tool.startswith(QUERY_PREFIX) or "{{" in tool or not isinstance(sql, str) or not sql.strip():
+            continue
+        sources.add(tool[len(QUERY_PREFIX):])
+        tables.update(t.rsplit(".", 1)[-1] for t in sql_tables(sql))
+    return sources, tables
+
+
+__all__ = ["caliber_descriptor", "list_calibers"]

backend/tests/test_ledger_descriptor.py

@@ -0,0 +1,107 @@
+"""口径可调用描述（app/ledger/descriptor.py、GET /api/ledger/calibers）：目录页、应用表单、外部集成用的机器可读契约。
+
+- 只列最新「已发布且未退役」版本里的源头口径卡：入口字段成 JSON Schema、指标照配置、依赖只报数据源和表名；
+- 能不能组合按 look.calibers.card_inputs（和对账、use_caliber 同一个认法）；
+- 出具只描述版本策略：受管级别、图里没有人工节点或写外部系统的节点才自动出具，原因照写；
+- 转抄别处口径卡的卡不列；响应里没有 SQL、没有参数值。
+数据是虚构店铺库。
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.main import app
+from tests import ledger_samples as S
+
+
+@pytest.fixture(autouse=True)
+async def shop(tmp_path):
+    await S.fresh_source(tmp_path / "shop.db")
+    yield
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def descriptors(client: AsyncClient, wf: str) -> list[dict]:
+    body = (await client.get("/api/ledger/calibers", params={"workflow_id": wf})).json()
+    assert set(body) == {"calibers"}, body
+    return body["calibers"]
+
+
+def with_human() -> dict:
+    graph = S.caliber_graph()
+    graph["nodes"].append(S.node("approve", "human", assignee="运营"))
+    return graph
+
+
+def with_copy() -> dict:
+    graph = S.caliber_graph()
+    graph["nodes"].append(S.node(
+        "copy", "metrics", caliber="抄来的周销售", caliber_version="v1",
+        caliber_from={"workflow_id": "elsewhere", "workflow_version": 1, "node_id": "card"},
+        metrics=[{"id": "gmv", "name": "销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}]))
+    return graph
+
+
+async def test_a_source_caliber_is_described_as_a_callable_contract(client):
+    wf = await S.publish()
+    [made] = await descriptors(client, wf)
+    assert (made["workflow_id"], made["workflow_name"], made["version"]) == (wf, "周销售", 1)
+    assert made["level"] == "governed" and made["graph_hash"]
+    assert made["card"] == {"node_id": "card", "title": "card", "caliber": "周销售", "caliber_version": "v1"}
+    assert made["metrics"] == [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands"}]
+    schema = made["entry_schema"]
+    assert schema["format"] == "agentlab.form/1" and schema["type"] == "object"
+    assert schema["required"] == ["start", "end"]
+    assert schema["properties"]["store"]["default"] == "all"
+    assert made["composable"] is True and made["compose_reason"] is None
+    assert made["dependencies"]["sources"] == [S.CURRENT["source"]]
+    assert made["dependencies"]["tables"] == ["orders"]
+    assert made["dependencies"]["catalog_versions"] is None and made["dependencies"]["schema_versions"] is None
+    issuance = made["issuance"]
+    assert issuance["level"] == "governed" and issuance["auto_issue"] is True
+    assert issuance["requires_independent_proof"] is True
+    assert issuance["side_effects"] is None and issuance["reasons"] == []
+
+    # 单条接口交同一份描述；转抄卡、没发布过的版本、不存在的节点都是 404
+    single = (await client.get(f"/api/ledger/calibers/{wf}/1/card")).json()
+    assert single == made
+    for path in (f"/api/ledger/calibers/{wf}/1/nope", f"/api/ledger/calibers/{wf}/9/card",
+                 "/api/ledger/calibers/nope/1/card"):
+        assert (await client.get(path)).status_code == 404, path
+
+    # 契约里没有取数 SQL、没有参数值
+    text = json.dumps((await client.get("/api/ledger/calibers", params={"workflow_id": wf})).json())
+    assert "SELECT" not in text and '"sql"' not in text
+    assert "SUM(" not in text and "paid" not in text
+
+
+async def test_a_human_node_turns_auto_issue_off_with_a_reason(client):
+    wf = await S.publish(name="审批周销售", graph=with_human(), level="governed")
+    [made] = await descriptors(client, wf)
+    assert made["level"] == "governed"
+    assert made["issuance"]["auto_issue"] is False
+    assert "approve" in made["issuance"]["side_effects"]
+    assert any("人工节点" in r for r in made["issuance"]["reasons"])
+    assert made["issuance"]["requires_independent_proof"] is True
+
+
+async def test_a_published_level_version_only_supports_one_click_issuance(client):
+    wf = await S.publish(name="已发布周销售", graph=S.caliber_graph(), level="published")
+    [made] = await descriptors(client, wf)
+    assert made["level"] == "published" and made["issuance"]["auto_issue"] is False
+    assert any("已发布级别" in r and "一键出具" in r for r in made["issuance"]["reasons"])
+
+
+async def test_a_card_that_copies_another_caliber_is_not_listed(client):
+    wf = await S.publish(graph=with_copy())
+    listed = await descriptors(client, wf)
+    assert [d["card"]["node_id"] for d in listed] == ["card"]
+    assert (await client.get(f"/api/ledger/calibers/{wf}/1/copy")).status_code == 404