记录口径调用审计：compose 与正式出具各记一条参数指纹

backend/app/api/copilot.py

@@ -2076,6 +2076,14 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None, analysis_
     spent = usage.meter()
     async for op in usage.metered(steps(), spent):
         yield op
+    if applier.caliber_uses:
+        # 调用审计（caliber.use）：这一轮选用了哪几版口径、带什么参数。只记指纹，不落参数值；失败不影响这一轮交付
+        from app.ledger.usage import audit_composed_uses
+
+        try:
+            await audit_composed_uses(applier.caliber_uses, conversation_id=payload.conversation_id)
+        except Exception:  # noqa: BLE001 - 审计是附加记录，不能挡住已经生成的图
+            logger.warning("记录口径调用审计失败", exc_info=True)
     yield {"op": "usage", **spent.summary()}
 
 

backend/app/api/copilot_gen.py

@@ -544,6 +544,9 @@ class GraphApplier:
         self.composer = composer
         #: 选用口径实例化出来的段：[(第一个取数节点, 全部节点 id)]，接入口用
         self.composed: list[tuple[str, list[str]]] = []
+        #: 这一轮选用过的口径（调用审计 caliber.use 用）：[{workflow_id, version, card, metrics, inputs}]。
+        #: inputs 只在内存里，写审计时只留指纹，不落参数值
+        self.caliber_uses: list[dict[str, Any]] = []
         #: 已经落地的 use_caliber（ledger/compose.signature），接着写时重发的同一个不再落地
         self._composed_ops: set[str] = set()
         #: 系统接上的连线（入口 → 取数）：模型再发一条一模一样的不重复加
@@ -643,6 +646,12 @@ class GraphApplier:
         if composed.op_id not in self._landed:
             self._landed.append(composed.op_id)
         self._wire()
+        self.caliber_uses.append({
+            "workflow_id": composed.option.workflow_id, "version": int(composed.option.version),
+            "card": composed.option.node_id,
+            "metrics": [str(m.get("id")) for m in composed.metrics if m.get("id")],
+            "inputs": dict(composed.inputs or {}),
+        })
         op["result"] = composed.result()
         return Applied(True, True)
 

backend/app/ledger/issue.py

@@ -392,6 +392,16 @@ async def issue(run_id: str, workflow_id: str, version: int, *, principal: Any,
             except BaseException:
                 await quota.give_back(QUOTA, day=day)
                 raise
+            try:
+                from app.ledger import usage as ledger_usage
+
+                await ledger_usage.record_use(
+                    session, workflow_id=workflow_id, version=int(version), how="issued", params=plan.inputs,
+                    run_id=run_id, actor=actor, auto=auto)
+                await session.commit()
+            except Exception:  # noqa: BLE001 - 审计是附加记录，正式运行已经发起，不能因此回滚
+                logger.warning("记录口径调用审计失败（来源运行 %s，口径 %s@%s）", run_id, workflow_id, version,
+                               exc_info=True)
             return launched, True
 
 

backend/app/ledger/usage.py

@@ -0,0 +1,87 @@
+"""口径调用审计（next-round 02 的后续）：谁在什么时候、用哪一版口径、带什么参数。
+
+`caliber_match` 是封存后的等价证明，不是调用日志；一次 `use_caliber` 组合或一次按口径正式出具，在这里各记一条
+`AuditEvent`（action=caliber.use）。和账本其他部分一样不另建表：audit_events 已经按 subject 建了索引，按
+subject_kind=caliber、subject_id=「工作流@版本」查得到。
+
+**只记参数指纹，不记参数值。** 参数可能是遮罩值、手机号这类不该再落一份的东西；记 hash 和字段名，排查「同一次调用
+用了哪组参数」够用，要复核参数去运行输入和比对工件里看（那里本来就有遮罩处理）。指标只记 id，不记算出来的值。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Iterable, Mapping
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+ACTION = "caliber.use"
+SUBJECT_KIND = "caliber"
+HOWS = ("composed", "issued")
+
+logger = logging.getLogger(__name__)
+
+
+def params_fingerprint(values: Mapping[str, Any] | None) -> tuple[str, list[str]]:
+    """(参数指纹, 字段名)。空值不参与指纹：没填和填空串是同一组参数；字段名排序，保证同一组参数写法不同也相同。"""
+    from app.core.artifact_store import canonical_json, content_hash
+
+    clean = {str(k): v for k, v in (values or {}).items() if v is not None and v != ""}
+    return content_hash(canonical_json(clean)), sorted(clean)
+
+
+async def record_use(session: AsyncSession, *, workflow_id: str, version: int, how: str,
+                     card: str | None = None, metrics: Iterable[str] = (), params: Mapping[str, Any] | None = None,
+                     run_id: str | None = None, conversation_id: str | None = None,
+                     actor: str | None = None, auto: bool | None = None) -> None:
+    """记一条调用。不提交：和调用方自己的事务一起提交。how 只能是 composed / issued。"""
+    from app.apps.audit import record
+
+    if how not in HOWS:
+        raise ValueError(f"不认识的调用方式：{how}")
+    digest, fields = params_fingerprint(params)
+    data: dict[str, Any] = {
+        "workflow_id": workflow_id,
+        "version": int(version),
+        "how": how,
+        "params_hash": digest,
+        "param_fields": fields,
+        "metrics": [str(m) for m in dict.fromkeys(metrics)],
+    }
+    if card:
+        data["card"] = card
+    if run_id:
+        data["run_id"] = run_id
+    if conversation_id:
+        data["conversation_id"] = conversation_id
+    if auto is not None:
+        data["auto"] = bool(auto)
+    await record(session, action=ACTION, subject_kind=SUBJECT_KIND,
+                 subject_id=f"{workflow_id}@{int(version)}", actor=actor, data=data)
+
+
+async def audit_composed_uses(uses: Iterable[Mapping[str, Any]], *, conversation_id: str | None = None,
+                              actor: str | None = None) -> int:
+    """一轮生成里所有 use_caliber 落成审计，返回记了几条。取数组合发生在建图阶段，没有运行可挂，只记对话。"""
+    uses = list(uses)
+    if not uses:
+        return 0
+    from app.db.base import SessionLocal
+
+    async with SessionLocal() as session:
+        for use in uses:
+            await record_use(
+                session,
+                workflow_id=str(use.get("workflow_id") or ""),
+                version=int(use.get("version") or 0),
+                how="composed",
+                card=str(use.get("card") or "") or None,
+                metrics=use.get("metrics") or (),
+                params=use.get("inputs") if isinstance(use.get("inputs"), Mapping) else None,
+                conversation_id=conversation_id,
+                actor=actor,
+            )
+        await session.commit()
+    return len(uses)
+
+
+__all__ = ["ACTION", "HOWS", "SUBJECT_KIND", "audit_composed_uses", "params_fingerprint", "record_use"]

backend/tests/test_ledger_issue.py

@@ -21,7 +21,7 @@ from sqlalchemy import select
 
 from app.api.deps import Principal
 from app.db.base import SessionLocal
-from app.db.models import Run, Setting, WorkflowVersion
+from app.db.models import AuditEvent, Run, Setting, WorkflowVersion
 from app.engine.runner import run_manager
 from app.ledger import issue as ledger_issue
 from app.ledger import match as ledger_match
@@ -130,6 +130,13 @@ async def test_an_independent_equivalent_answer_to_a_governed_caliber_is_issued_
     origin = formal.origin
     assert origin["source_run"] == row.id and origin["caliber"] == f"{wf}@1" and origin["how"] == "independent"
     assert origin["auto"] is True and origin["binding_hash"] and origin["match"]
+    async with SessionLocal() as session:
+        rows = (await session.execute(select(AuditEvent).where(
+            AuditEvent.action == "caliber.use", AuditEvent.subject_id == f"{wf}@1"))).scalars().all()
+    [use] = [r for r in rows if r.data.get("run_id") == row.id]
+    assert use.data["how"] == "issued" and use.data["auto"] is True
+    assert use.data["param_fields"] == ["end", "start", "store"] and use.data["params_hash"]
+    assert "2026-09-01" not in str(use.data)
 
     body = (await client.get(f"/api/runs/{row.id}/ledger")).json()
     [action] = [a for a in body["actions"] if a["kind"] == "issue"]

backend/tests/test_ledger_usage.py

@@ -0,0 +1,55 @@
+"""口径调用审计（app/ledger/usage.py）：谁在什么时候用了哪一版、带什么参数。
+
+- 参数只记指纹和字段名，不记值；
+- composed（问数据里 use_caliber）和 issued（按口径正式出具）各记一条 caliber.use；
+- 同一组参数写法不同（键顺序、空值）指纹相同；参数不同指纹不同。
+"""
+from __future__ import annotations
+
+import uuid
+
+from sqlalchemy import select
+
+from app.db.base import SessionLocal
+from app.db.models import AuditEvent
+from app.ledger.usage import audit_composed_uses, params_fingerprint, record_use
+
+
+def test_params_fingerprint_ignores_order_and_blanks_but_not_values():
+    first, fields = params_fingerprint({"start": "2026-09-01", "end": "2026-09-08", "store": ""})
+    second, _ = params_fingerprint({"end": "2026-09-08", "start": "2026-09-01", "store": None})
+    assert first == second and fields == ["end", "start"]
+    other, _ = params_fingerprint({"start": "2026-09-01", "end": "2026-09-07"})
+    assert other != first
+
+
+async def test_record_use_keeps_a_fingerprint_not_the_values():
+    wf = f"usage{uuid.uuid4().hex[:10]}"
+    async with SessionLocal() as session:
+        await record_use(session, workflow_id=wf, version=3, how="issued", card="card", metrics=["gmv", "gmv"],
+                         params={"start": "2026-09-01", "end": "2026-09-08"}, run_id="run-1", auto=True,
+                         actor="王敏")
+        await session.commit()
+        [event] = (await session.execute(select(AuditEvent).where(
+            AuditEvent.action == "caliber.use", AuditEvent.subject_id == f"{wf}@3"))).scalars().all()
+    assert event.actor == "王敏" and event.subject_kind == "caliber"
+    assert event.data["how"] == "issued" and event.data["metrics"] == ["gmv"]
+    assert event.data["run_id"] == "run-1" and event.data["auto"] is True
+    assert event.data["param_fields"] == ["end", "start"] and event.data["params_hash"]
+    assert "2026-09-01" not in str(event.data) and "2026-09-08" not in str(event.data)
+
+
+async def test_composed_uses_are_written_as_one_audit_each():
+    wf = f"usage{uuid.uuid4().hex[:10]}"
+    count = await audit_composed_uses([
+        {"workflow_id": wf, "version": 1, "card": "card", "metrics": ["gmv"],
+         "inputs": {"start": "2026-09-01", "end": "2026-09-08"}},
+        {"workflow_id": wf, "version": 2, "card": "card2", "metrics": ["orders"], "inputs": {}},
+    ], conversation_id="conv-1")
+    assert count == 2
+    async with SessionLocal() as session:
+        rows = list((await session.execute(select(AuditEvent).where(
+            AuditEvent.action == "caliber.use", AuditEvent.subject_kind == "caliber")
+            .where(AuditEvent.subject_id.in_([f"{wf}@1", f"{wf}@2"])))).scalars())
+    assert {r.subject_id for r in rows} == {f"{wf}@1", f"{wf}@2"}
+    assert all(r.data["how"] == "composed" and r.data["conversation_id"] == "conv-1" for r in rows)