沉淀：认证队列、固化成口径草稿、发布后补比；账本统计补上对账与认证队列的转化（口径账本做法 8、10，拍板 L9）

为什么：同一类问题被反复问、Agent 反复写出同一个形状的 SQL，就该沉淀成一份口径；以前 from-run 保留 Agent，沉淀出的模板没有固定 SQL，以后永远比对不上。第 2 波的统计在 E2 反馈、E3 认证队列上照实写了「目前没有数据」，这一波补上能补的。

做了什么（新增 ledger/certify.py）：
- GET /api/ledger/certify-queue（搭建者）：确定性重复（问数据运行里被报告引用、没对上口径的查询，按数据源 + normalize_sql 文本分组，≥3 次运行且来自 ≥2 段对话；跨运行变化的字面量是参数候选，展示的形状不带取值）、问题聚类（app/chat/questions 全站唯一的那份）、被反复用的答案（导出 ≥2 次用第 2 波的 report.exported 审计、做成过应用看蓝图来源、在画布里打开或提取过看工作流 origin）。每项给代表问题、轮次、相关口径、是否已固化。
- POST /api/ledger/certify-queue/dismiss：「不是口径」写审计 ledger.certify.dismiss，之后不再进队。
- 固化成口径草稿（from-run 加 freeze、peers 后转调 certify.freeze）：报告引用的查询变成固定 SQL 的取数节点，跨轮变化的字面量换成 '{{ input.pN }}'（缺省值取代表运行的），单行结果的数值列成口径卡指标，报告核对规则按受管要求写、契约 report_from + metrics_from + required（多行加 cells）；过 validate_graph 和受管门禁。新工作流 origin.sources 记来源问题。
- 发布之后补比：api/workflows.commit_publish 提交后调 certify.after_publish，来源运行按新口径补比一次（新的 trigger=certified，界面写「按之后发布的口径回看」，时间线和账本接口同一句）。
- 统计：E2 加对过账的答案数、参数经人核对的、指标由人选的次数（反馈照实写第 4 波）；E3 认证队列的转化（固化、发布、不是口径）。

测试：tests/test_ledger_certify.py（参数推断与模板、三种信号进队、门槛、不是口径、固化草稿过受管门禁、发布后来源运行补比成等价已证明并出队、再问标等价已证明、统计）。

backend/app/api/copilot.py

@@ -2591,6 +2591,10 @@ class FromRunIn(BaseModel):
     #: 问数据回答上点「做成应用」时带上：这次运行出自哪段对话的哪一轮，记进新工作流的来源（workflows.origin）
     conversation_id: str | None = None
     turn_id: str | None = None
+    #: 固化成口径草稿（next-round 02 做法 8，app/ledger/certify.freeze）：报告引用的查询变成固定 SQL 的取数节点，跨轮变化
+    #: 的字面量成为参数（不保留 Agent）。peers 是同一组（同一个 SQL 形状）的别的运行，认哪些字面量位置变了
+    freeze: bool = False
+    peers: list[str] = Field(default_factory=list, max_length=50)
 
 
 @router.post("/from-run")
@@ -2612,6 +2616,15 @@ async def extract_template(
 
     # 对话只是记进来源：不能用的不让记（D-d），不存在的照旧记下 id
     await _ensure_conversation_access(session, payload.conversation_id)
+    if payload.freeze:
+        # 固化成口径草稿：取数是固定的 SQL（以后比对认得出等价），不是保留 Agent 的执行路径
+        from app.ledger.certify import CertifyRefused, freeze
+
+        try:
+            return await freeze(session, payload.run_id, peers=payload.peers, name=payload.name,
+                                conversation_id=payload.conversation_id, turn_id=payload.turn_id)
+        except CertifyRefused as e:
+            raise HTTPException(e.status_code, e.message) from e
     try:
         path = await extract_run_path(session, payload.run_id)
     except PathRefused as e:

backend/app/api/ledger.py

@@ -161,3 +161,30 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]:
     except ledger_reconcile.ReconcileRefused as e:
         raise CodedHTTPException(e.status_code, e.message, e.code) from e
     return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)}
+
+
+@router.get("/certify-queue")
+async def certify_queue(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """可沉淀的问题（认证队列，app/ledger/certify.py）：确定性重复的 SQL 形状、问题聚类、被反复用的答案。每项给代表问题、
+    轮次、（形状的）参数候选和相关口径；点过「不是口径」的不再列。固化走 POST /api/copilot/from-run（freeze: true）。"""
+    from app.api.deps import current_principal
+    from app.ledger import certify
+
+    return await certify.queue(session, principal=current_principal())
+
+
+class DismissIn(BaseModel):
+    item_id: str = Field(min_length=1, max_length=64)
+    kind: str | None = Field(default=None, max_length=20)
+    note: str | None = Field(default=None, max_length=500)
+
+
+@router.post("/certify-queue/dismiss")
+async def certify_dismiss(payload: DismissIn, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """「不是口径」：记一条审计（ledger.certify.dismiss），这一项之后不再进队。"""
+    from app.ledger import certify
+
+    if payload.kind is not None and payload.kind not in certify.KINDS:
+        raise HTTPException(422, "可沉淀的问题只有三种：确定性重复、问题聚类、被反复用的答案")
+    await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note)
+    return {"ok": True, "item_id": payload.item_id}

backend/app/api/workflows.py

@@ -564,6 +564,10 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo
     workflow.published_version = snapshot.version
     workflow.published_by = actor
     await session.commit()
+    # 口径账本（next-round 02 做法 8）：从问数据固化出的口径发布了，来源运行在后台按它补比一次（不挡发布）
+    from app.ledger.certify import after_publish
+
+    after_publish(workflow.id, workflow.origin)
     return {
         "ok": True,
         "level": level,

backend/app/ledger/certify.py

@@ -0,0 +1,550 @@
+"""沉淀：可沉淀的问题（认证队列）和「固化成口径草稿」（docs/design/next-round/02-口径账本.md 做法 8，拍板 L9）。
+
+同一类问题被反复问、Agent 反复写出同一个形状的 SQL，就该沉淀成一份口径：以后再问，比对器认得出「等价已证明」，还能按口径
+正式出具。以前的 from-run 提取保留 Agent，沉淀出的模板没有固定 SQL，以后永远比对不上；聚类又混着 history。
+
+**认证队列**（GET /api/ledger/certify-queue，搭建者及以上）三种信号：
+- 确定性重复（shape）：问数据的探索运行里执行过、被报告引用（比对工件里标签记着的查询）、又没对上任何口径（none / related）
+  的查询，按（数据源、normalize_sql 的规范文本）分组；出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。
+  跨运行变化的字面量位置是参数候选，不变的是口径常量。展示的是规范文本（字面量都换成了「?」，不带取值）；
+- 问题聚类（cluster）：app/chat/questions 全站唯一的聚类（只取问题原文、不含 history、排除系统派生的运行），≥ REPEAT_RUNS
+  个问题的类；
+- 被反复用的答案（reused）：导出过 ≥ 2 次（report.exported 审计，第 2 波）、做成过应用（蓝图的来源是这次运行）、在画布里
+  打开过或提取过（工作流的 origin.run_id 是这次运行）的轮次。
+每项给代表问题、轮次链接、（shape 的）SQL 形状和参数候选、有没有相关口径（可以对账的那份）、是不是已经固化过。点「不是口径」
+写一条审计（ledger.certify.dismiss，subject 是这一项的 id），之后不再进队。门槛同 ledger/stats（E3 的重复度）。
+
+**固化成口径草稿**（freeze，from-run 加 freeze 选项后转调它）：代表运行里报告引用了的查询，各变成一个调用工具节点——SQL 用
+规范化后的文本，跨轮变化的字面量换成 '{{ input.p1 }}'（同组的别的运行 peers 用来认哪些位置变了），入口字段的缺省值取
+代表运行里的值；只有一行的结果每个数值列建一项口径卡指标（cell(nodes.取数, 0, '列')）；报告节点保留，出口契约写
+report_from + metrics_from（有多行结果的另加 cells: true），过 validate_graph。新工作流记 origin.sources：
+[{run_id, conversation_id, turn_id, question}]，这就是「口径 → 来源问题」的记录。之后由人在画布上审改、按受管级别发布——
+提取器是作者，不是发布者。
+
+**发布之后补比**（after_publish）：工作流发布时，origin.sources 里的来源运行在后台按现在的口径补比一次（trigger=certified，
+标「按之后发布的口径回看」），E1 也就有了回溯数据。
+"""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Any
+
+from sqlalchemy import select
+
+from app.db.base import SessionLocal
+from app.db.models import Run
+
+logger = logging.getLogger(__name__)
+
+#: 门槛：同一个形状出现在几次运行里、来自几段对话；被导出几次算反复用（同 ledger/stats 的 E3）
+REPEAT_RUNS = 3
+REPEAT_CONVERSATIONS = 2
+EXPORTS = 2
+#: 看最近多少次问数据运行、每项最多列几轮
+RUN_LIMIT = 500
+TURNS_LIMIT = 10
+QUESTION_LIMIT = 500
+#: 「不是口径」的审计
+DISMISS_ACTION = "ledger.certify.dismiss"
+SUBJECT_KIND = "ledger_certify"
+KINDS = ("shape", "cluster", "reused")
+
+#: 固化被拒的说法
+NOT_FOUND = "运行记录不存在，可能已被删除"
+NOT_MATCHED = "这次运行还没有比对口径，比对完才能固化（打开它的口径账本会按需比对）"
+NOT_SUCCEEDED = "这次运行没有成功结束，提取出的查询不完整"
+NO_QUERIES = "这次运行的报告没有引用任何查询结果，没有可固化的取数"
+UNPARSED = "这次的 SQL 无法规范化（{reason}），不能固化成固定的取数"
+SNAPSHOT = "查询快照读不出来（文件缺失或与哈希不一致），不能固化"
+#: 新工作流的说明、报告指令
+DRAFT_NAME = "{question}（固化的口径草稿）"
+DRAFT_DESCRIPTION = ("从问数据的回答固化：报告引用了的 {n} 条查询变成固定的取数，跨轮变化的 {p} 个字面量成为入口参数。"
+                     "经人工审核、按受管级别发布后，再问同样的问题就能认出等价、按口径出具")
+REPORT_INSTRUCTIONS = "回答：{question}"
+
+
+class CertifyRefused(Exception):
+    def __init__(self, status_code: int, message: str) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+
+
+def item_id(kind: str, key: str) -> str:
+    return f"{kind}:{hashlib.sha256(key.encode()).hexdigest()[:16]}"
+
+
+# ==========================================================================
+# SQL 的形状和参数
+# ==========================================================================
+
+
+@dataclass(frozen=True)
+class Shape:
+    """一条查询的形状：数据源、规范文本（字面量换成「?」）、字面量（SQL 写法，字符串带引号）、规范化失败的原因。"""
+
+    source: str
+    text: str
+    literals: tuple[str, ...]
+    problem: str | None = None
+
+
+def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None:
+    from app.data.sql_semantics import normalize_sql
+
+    if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or not snap["sql"].strip():
+        return None
+    source = str(snap.get("source") or "")
+    norm = normalize_sql(snap["sql"], kinds.get(source))
+    return Shape(source, norm.text, norm.literals, norm.parse_error)
+
+
+def varying(literals: Sequence[Sequence[str]]) -> list[int]:
+    """几次运行同一个形状的字面量：哪几个位置跨运行变过（参数候选），不变的是口径常量。长度不一致的那几份不算。"""
+    rows = [list(row) for row in literals]
+    if not rows:
+        return []
+    width = len(rows[0])
+    same = [r for r in rows if len(r) == width]
+    return [i for i in range(width) if len({r[i] for r in same}) > 1]
+
+
+def _unquote(literal: str) -> str | None:
+    """'2026-09-01' → 2026-09-01；不是普通单引号字符串（N'…'、x'…'、$$…$$）的为 None。"""
+    if len(literal) >= 2 and literal[0] == "'" and literal[-1] == "'":
+        return literal[1:-1].replace("''", "'")
+    return None
+
+
+def _number(literal: str) -> int | float | None:
+    try:
+        value = float(literal)
+    except ValueError:
+        return None
+    return int(value) if value.is_integer() and "." not in literal and "e" not in literal.lower() else value
+
+
+def template_of(shape: Shape, params: Sequence[int], dialect: str | None, *,
+                start: int = 1) -> tuple[str, list[dict[str, Any]]]:
+    """规范文本 + 参数位置 → (口径的取数 SQL 模板, 入口字段)。参数位置的字面量换成 '{{ input.pN }}'（字符串）或
+    {{ input.pN }}（数，N 从 start 起），缺省值是这一次的值；别的位置照原样写回字面量。规范文本按方言重新切词，按「?」的
+    先后对上字面量。"""
+    from app.data.sql_semantics import PLACEHOLDER, _read, _tokens
+
+    tokens = [t for t in _tokens(shape.text, _read(dialect)) if shape.text[t.start:t.end + 1] == PLACEHOLDER]
+    if len(tokens) != len(shape.literals):
+        raise CertifyRefused(409, UNPARSED.format(reason="字面量对不上"))
+    out: list[str] = []
+    fields: list[dict[str, Any]] = []
+    pos = 0
+    wanted = set(params)
+    for i, (token, literal) in enumerate(zip(tokens, shape.literals)):
+        out.append(shape.text[pos:token.start])
+        pos = token.end + 1
+        text, number = _unquote(literal), _number(literal)
+        if i in wanted and (text is not None or number is not None):
+            name = f"p{start + len(fields)}"
+            if text is not None:
+                out.append(f"'{{{{ input.{name} }}}}'")
+                fields.append({"name": name, "type": "string", "required": True, "default": text})
+            else:
+                out.append(f"{{{{ input.{name} }}}}")
+                fields.append({"name": name, "type": "number", "required": True, "default": number})
+        else:
+            out.append(literal)
+    out.append(shape.text[pos:])
+    return "".join(out), fields
+
+
+# ==========================================================================
+# 认证队列
+# ==========================================================================
+
+
+@dataclass
+class _Member:
+    run_id: str
+    artifact: str
+    literals: tuple[str, ...]
+    conversation_id: str | None
+    turn_id: str | None
+    question: str
+    related: list[dict[str, Any]] = field(default_factory=list)
+
+
+def referenced_queries(content: dict[str, Any]) -> list[str]:
+    """比对工件里被报告引用了的查询（标签记着的：单元格的 query、指标的 queries），按先后去重。"""
+    out: list[str] = []
+    for report in (content.get("reports") or {}).values():
+        for label in (report.get("labels") or {}).values():
+            for artifact in [label.get("query"), *(label.get("queries") or [])]:
+                if isinstance(artifact, str) and artifact and artifact not in out:
+                    out.append(artifact)
+    return out
+
+
+async def _turns(session: Any, runs: list[Run]) -> dict[str, dict[str, Any]]:
+    """运行 → {conversation_id, turn_id, question}：运行来源里封存的（origin_ref）；老运行退回 conversation_turns.run_id。"""
+    from app.db.models import ConversationTurn
+
+    out: dict[str, dict[str, Any]] = {}
+    ids = [r.id for r in runs]
+    if ids:
+        for turn in (await session.execute(select(ConversationTurn).where(ConversationTurn.run_id.in_(ids)))).scalars():
+            out.setdefault(turn.run_id, {"conversation_id": turn.conversation_id, "turn_id": turn.id,
+                                         "question": turn.question or ""})
+    for run in runs:
+        origin = run.origin if isinstance(run.origin, dict) else {}
+        found = out.setdefault(run.id, {"conversation_id": None, "turn_id": None, "question": ""})
+        if origin.get("conversation_id"):
+            found["conversation_id"] = str(origin["conversation_id"])
+            found["turn_id"] = str(origin.get("turn_id") or found.get("turn_id") or "") or None
+        if not found["question"] and isinstance(run.input, dict) and isinstance(run.input.get("question"), str):
+            found["question"] = run.input["question"]
+    return out
+
+
+async def _dismissed(session: Any) -> set[str]:
+    from app.db.models import AuditEvent
+
+    return {sid for (sid,) in (await session.execute(select(AuditEvent.subject_id).where(
+        AuditEvent.action == DISMISS_ACTION, AuditEvent.subject_kind == SUBJECT_KIND))).all() if sid}
+
+
+async def _frozen(session: Any) -> dict[str, list[dict[str, Any]]]:
+    """来源运行 → 从它固化出的工作流 [{workflow_id, name, status, published_version}]（origin.sources）。"""
+    from app.db.models import Workflow
+
+    out: dict[str, list[dict[str, Any]]] = {}
+    for wf in (await session.execute(select(Workflow).where(Workflow.origin.is_not(None)))).scalars():
+        for src in (wf.origin or {}).get("sources") or [] if isinstance(wf.origin, dict) else []:
+            if isinstance(src, dict) and src.get("run_id"):
+                out.setdefault(str(src["run_id"]), []).append(
+                    {"workflow_id": wf.id, "name": wf.name, "status": wf.status,
+                     "published_version": wf.published_version})
+    return out
+
+
+def _turn_view(member: Any) -> dict[str, Any]:
+    return {"run_id": member.run_id, "conversation_id": member.conversation_id, "turn_id": member.turn_id,
+            "question": member.question}
+
+
+async def queue(session: Any, *, principal: Any) -> dict[str, Any]:
+    """认证队列（见模块说明）：{items: [...], threshold, considered}。只读。"""
+    from app.chat.questions import cluster_questions, embedder, recent_questions
+    from app.core.artifact_store import load
+    from app.db.models import Artifact, DataSource
+    from app.engine.origin import counted_runs
+    from app.ledger import store
+
+    dismissed = await _dismissed(session)
+    frozen = await _frozen(session)
+    kinds = {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()}
+    runs = list((await session.execute(select(Run).where(
+        Run.origin_kind == "chat", Run.status == "succeeded", Run.run_class != "formal", counted_runs())
+        .order_by(Run.created_at.desc()).limit(RUN_LIMIT))).scalars())
+    turns = await _turns(session, runs)
+    by_run = {r.id: r for r in runs}
+    # 每次运行最新的一份比对（store.latest 按事件认这一次封存的；这里只要内容，按工件引用的时间取最新）
+    latest: dict[str, tuple[Any, str]] = {}
+    if runs:
+        for artifact, run_id, at in (await session.execute(select(Artifact.id, Artifact.run_id, Artifact.created_at)
+                                                           .where(Artifact.kind == store.KIND,
+                                                                  Artifact.run_id.in_(list(by_run))))).all():
+            if run_id not in latest or (at and latest[run_id][0] and at > latest[run_id][0]):
+                latest[run_id] = (at, artifact)
+
+    def read() -> dict[tuple[str, str], list[_Member]]:
+        groups: dict[tuple[str, str], list[_Member]] = {}
+        for run_id, (_, artifact) in latest.items():
+            try:
+                content = load(artifact)
+            except ValueError:
+                continue
+            if not isinstance(content, dict) or content.get("format") != store.FORMAT:
+                continue
+            statuses = {q.get("artifact"): q for q in content.get("queries") or []}
+            for query in referenced_queries(content):
+                row = statuses.get(query) or {}
+                if row.get("status") not in ("none", "related"):
+                    continue
+                try:
+                    shape = shape_of(load(query), kinds)
+                except ValueError:
+                    shape = None
+                if shape is None or shape.problem:
+                    continue
+                info = turns.get(run_id) or {}
+                related = [{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level")}
+                           for c in row.get("candidates") or [] if c.get("status") == "related"]
+                groups.setdefault((shape.source, shape.text), []).append(_Member(
+                    run_id, query, shape.literals, info.get("conversation_id"), info.get("turn_id"),
+                    str(info.get("question") or ""), related))
+        return groups
+
+    groups = await asyncio.to_thread(read)
+    items: list[dict[str, Any]] = []
+    for (source, text), members in groups.items():
+        once = list({m.run_id: m for m in members}.values())
+        conversations = {m.conversation_id or f"run:{m.run_id}" for m in once}
+        if len(once) < REPEAT_RUNS or len(conversations) < REPEAT_CONVERSATIONS:
+            continue
+        iid = item_id("shape", f"{source}\n{text}")
+        if iid in dismissed:
+            continue
+        params = varying([m.literals for m in once])
+        related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values())
+        rep = once[0]
+        items.append({
+            "id": iid, "kind": "shape", "source": source, "shape": text, "runs": len(once),
+            "conversations": len(conversations), "question": rep.question,
+            "turns": [_turn_view(m) for m in once[:TURNS_LIMIT]],
+            "params": [{"index": i, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})}
+                       for i in params],
+            "constants": len(rep.literals) - len(params), "related": related,
+            "representative": {"run_id": rep.run_id, "query": rep.artifact},
+            "peers": [m.run_id for m in once[1:]],
+            "frozen_into": [w for m in once for w in frozen.get(m.run_id, [])],
+        })
+    # 问题聚类
+    questions = await recent_questions(session, principal, limit=QUESTION_LIMIT)
+    for cluster in await cluster_questions(questions) if questions else []:
+        if cluster.count < REPEAT_RUNS:
+            continue
+        seed = cluster.members[0]
+        iid = item_id("cluster", " ".join(seed.text.split()))
+        if iid in dismissed:
+            continue
+        items.append({"id": iid, "kind": "cluster", "question": seed.text, "count": cluster.count,
+                      "turns": [{"run_id": q.run_id, "turn_id": q.turn_id, "question": q.text}
+                                for q in cluster.members[:TURNS_LIMIT]],
+                      "frozen_into": [w for q in cluster.members if q.run_id for w in frozen.get(q.run_id, [])]})
+    items += await _reused(session, runs, turns, dismissed, frozen)
+    return {"items": items, "considered": {"runs": len(runs), "matched": len(latest), "questions": len(questions)},
+            "threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS, "exports": EXPORTS},
+            "embedder": embedder()}
+
+
+async def _reused(session: Any, runs: list[Run], turns: dict[str, dict[str, Any]], dismissed: set[str],
+                  frozen: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]:
+    """被反复用的答案：导出过 ≥ EXPORTS 次、做成过应用、在画布里打开过或提取过（见模块说明）。"""
+    from sqlalchemy import func
+
+    from app.db.models import AuditEvent, Workflow
+    from app.db.models_blueprints import Blueprint
+
+    ids = [r.id for r in runs]
+    if not ids:
+        return []
+    exports = {sid: int(n) for sid, n in (await session.execute(
+        select(AuditEvent.subject_id, func.count()).where(AuditEvent.action == "report.exported",
+                                                           AuditEvent.subject_kind == "run",
+                                                           AuditEvent.subject_id.in_(ids))
+        .group_by(AuditEvent.subject_id))).all()}
+    apps: set[str] = set()
+    for source in (await session.execute(select(Blueprint.source))).scalars():
+        if isinstance(source, dict) and source.get("kind") == "run" and source.get("id") in ids:
+            apps.add(str(source["id"]))
+    canvas: dict[str, list[str]] = {}
+    for wf_id, origin, tags in (await session.execute(select(Workflow.id, Workflow.origin, Workflow.tags)
+                                                      .where(Workflow.origin.is_not(None)))).all():
+        if isinstance(origin, dict) and origin.get("run_id") in ids and not origin.get("sources"):
+            canvas.setdefault(str(origin["run_id"]), []).append("extracted" if "extracted" in (tags or []) else "canvas")
+    out: list[dict[str, Any]] = []
+    for run in runs:
+        uses = {"exports": exports.get(run.id, 0), "app": run.id in apps,
+                "extracted": "extracted" in canvas.get(run.id, []), "canvas": "canvas" in canvas.get(run.id, [])}
+        if not (uses["exports"] >= EXPORTS or uses["app"] or uses["extracted"] or uses["canvas"]):
+            continue
+        iid = item_id("reused", run.id)
+        if iid in dismissed:
+            continue
+        info = turns.get(run.id) or {}
+        out.append({"id": iid, "kind": "reused", "question": info.get("question") or "", "uses": uses,
+                    "turns": [{"run_id": run.id, "conversation_id": info.get("conversation_id"),
+                               "turn_id": info.get("turn_id"), "question": info.get("question") or ""}],
+                    "representative": {"run_id": run.id}, "frozen_into": frozen.get(run.id, [])})
+    return out
+
+
+async def dismiss(session: Any, iid: str, *, kind: str | None = None, note: str | None = None) -> None:
+    """「不是口径」：写一条审计，之后这一项不再进队。"""
+    from app.auth.audit import record
+
+    await record(session, DISMISS_ACTION, SUBJECT_KIND, iid,
+                 data={k: v for k, v in {"kind": kind, "note": note}.items() if v})
+    await session.commit()
+
+
+# ==========================================================================
+# 固化成口径草稿
+# ==========================================================================
+
+
+async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: str = "",
+                 conversation_id: str | None = None, turn_id: str | None = None) -> dict[str, Any]:
+    """把一次问数据运行固化成口径草稿（见模块说明），落库成新的草稿工作流，返回 {workflow_id, name, nodes, edges, params,
+    sources, ...}。peers 是同一组（同一个 SQL 形状）的别的运行：认哪些字面量位置跨轮变了。不能固化抛 CertifyRefused。"""
+    from app.api.deps import can_read_run, current_principal
+    from app.api.evidence import _reports, _sealed
+    from app.core.artifact_store import graph_hash, load
+    from app.db.models import DataSource, Workflow, WorkflowVersion
+    from app.engine.governance import REPORT_POLICY
+    from app.engine.layout import auto_layout
+    from app.engine.schema import GraphSpec
+    from app.ledger import store
+    from app.tools.datasource import QUERY_PREFIX
+
+    run = await session.get(Run, run_id)
+    principal = current_principal()
+    if run is None or not await can_read_run(principal, run):
+        raise CertifyRefused(404, NOT_FOUND)
+    if run.status != "succeeded":
+        raise CertifyRefused(409, NOT_SUCCEEDED)
+    recorded = await store.latest(run_id)
+    if recorded is None or recorded.content is None:
+        raise CertifyRefused(409, NOT_MATCHED)
+    queries = referenced_queries(recorded.content)
+    if not queries:
+        raise CertifyRefused(409, NO_QUERIES)
+    kinds = {n: k for n, k in (await session.execute(select(DataSource.name, DataSource.kind))).all()}
+
+    # 同组的别的运行：它们封存的查询里同形状的那一条的字面量
+    peer_literals: dict[tuple[str, str], list[tuple[str, ...]]] = {}
+    peer_sources: list[dict[str, Any]] = []
+    peer_runs = [r for r in [await session.get(Run, p) for p in dict.fromkeys(peers) if p != run_id] if r is not None]
+    readable = [r for r in peer_runs if await can_read_run(principal, r)]
+    for peer in readable:
+        sealed = await _sealed(peer.id)
+        for artifact in sealed.queries:
+            try:
+                shape = shape_of(load(artifact), kinds)
+            except ValueError:
+                continue
+            if shape is not None and not shape.problem:
+                peer_literals.setdefault((shape.source, shape.text), []).append(shape.literals)
+    sealed = await _sealed(run_id)
+    shapes: list[Shape] = []
+    snaps: list[dict[str, Any]] = []
+    for artifact in queries:
+        try:
+            snap = load(artifact)
+        except ValueError:
+            raise CertifyRefused(409, SNAPSHOT) from None
+        shape = shape_of(snap, kinds)
+        if shape is None:
+            raise CertifyRefused(409, SNAPSHOT)
+        if shape.problem:
+            raise CertifyRefused(409, UNPARSED.format(reason=shape.problem))
+        shapes.append(shape)
+        snaps.append(snap)
+
+    def nid(i: int) -> str:
+        return "fetch" if len(shapes) == 1 else f"fetch_{i + 1}"
+
+    nodes: list[dict[str, Any]] = []
+    fields: list[dict[str, Any]] = []
+    metrics: list[dict[str, Any]] = []
+    multi = False
+    for i, (shape, snap) in enumerate(zip(shapes, snaps)):
+        params = varying([shape.literals, *peer_literals.get((shape.source, shape.text), [])])
+        template, new_fields = template_of(shape, params, kinds.get(shape.source), start=len(fields) + 1)
+        fields += new_fields
+        nodes.append({"id": nid(i), "type": "tool", "position": {"x": 0, "y": 0},
+                      "data": {"label": f"取数 {i + 1}" if len(shapes) > 1 else "取数",
+                               "config": {"tool": f"{QUERY_PREFIX}{shape.source}", "args": {"sql": template}}}})
+        rows = snap.get("rows") if isinstance(snap.get("rows"), list) else []
+        columns = [str(c) for c in snap.get("columns") or []]
+        if len(rows) == 1 and isinstance(rows[0], (list, tuple)):
+            for col, value in zip(columns, rows[0]):
+                if isinstance(value, (int, float)) and not isinstance(value, bool):
+                    metrics.append({"id": f"m{len(metrics) + 1}", "name": col,
+                                    "expression": f"cell(nodes.{nid(i)}, 0, {col!r})"})
+        else:
+            multi = True
+    question = ((await _turns(session, [run])).get(run.id) or {}).get("question") or ""
+    reports = _reports(sealed)
+    report_cfg: dict[str, Any] = {}
+    if reports:
+        found = next((n for n in (run.graph or {}).get("nodes") or [] if n.get("id") == reports[0].node_id), None)
+        cfg = ((found or {}).get("data") or {}).get("config") or {}
+        report_cfg = {k: cfg[k] for k in ("provider", "model", "entities") if k in cfg}
+    report_cfg["instructions"] = REPORT_INSTRUCTIONS.format(question=question or "（问题）")
+    # 报告的核对规则一律按受管级别的要求写（engine/governance.REPORT_POLICY，和发布门禁同一份）：草稿要能按受管发布
+    report_cfg.update({key: value for key, value, _ in REPORT_POLICY})
+    contract: dict[str, Any] = {"report_from": "write"}
+    tail: list[dict[str, Any]] = []
+    if metrics:
+        tail.append({"id": "card", "type": "metrics", "position": {"x": 0, "y": 0},
+                     "data": {"label": "口径卡", "config": {"caliber": (question or "口径")[:40], "caliber_version": "v1",
+                                                         "metrics": metrics}}})
+        report_cfg["metrics_from"] = ["card"]
+        # 受管出具要写必需指标（不然「不予出具」这一档触发不了）：固化出的每个指标都是这个口径要给的数
+        contract.update({"metrics_from": "card", "required": [m["id"] for m in metrics]})
+    if multi or not metrics:
+        contract["cells"] = True
+    chain = [{"id": "start", "type": "input", "position": {"x": 0, "y": 0},
+              "data": {"label": "入口", "config": {"fields": fields}}},
+             *nodes, *tail,
+             {"id": "write", "type": "report", "position": {"x": 0, "y": 0},
+              "data": {"label": "报告", "config": report_cfg}},
+             {"id": "out", "type": "output", "position": {"x": 0, "y": 0},
+              "data": {"label": "出口", "config": {"fields": [{"name": "报告", "value": "{{ nodes.write.text }}"}],
+                                                  "contract": contract}}}]
+    edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(chain, chain[1:])]
+    graph = auto_layout(GraphSpec.model_validate({"nodes": chain, "edges": edges})).model_dump(mode="json")
+
+    info = (await _turns(session, [run])).get(run.id) or {}
+    sources = [{"run_id": run.id, "conversation_id": conversation_id or info.get("conversation_id"),
+                "turn_id": turn_id or info.get("turn_id"), "question": question}]
+    for peer, peer_info in (await _turns(session, readable)).items():
+        sources.append({"run_id": peer, "conversation_id": peer_info.get("conversation_id"),
+                        "turn_id": peer_info.get("turn_id"), "question": peer_info.get("question") or ""})
+    workflow = Workflow(
+        name=(name or DRAFT_NAME.format(question=(question or run.workflow_name or "问数据")[:30]))[:200],
+        description=DRAFT_DESCRIPTION.format(n=len(shapes), p=len(fields)), graph=graph, tags=["certified"],
+        status="draft",
+        origin={"run_id": run.id, **({"conversation_id": sources[0]["conversation_id"]} if sources[0]["conversation_id"]
+                                     else {}), **({"turn_id": sources[0]["turn_id"]} if sources[0]["turn_id"] else {}),
+                "freeze": True, "sources": sources})
+    session.add(workflow)
+    await session.flush()
+    session.add(WorkflowVersion(workflow_id=workflow.id, version=1, graph=graph, graph_hash=graph_hash(graph),
+                                note=f"从运行 {run.id} 固化"))
+    await session.commit()
+    return {"workflow_id": workflow.id, "name": workflow.name, "nodes": len(chain), "edges": len(edges),
+            "dropped_nodes": 0, "source_run": run.id, "frozen": True, "params": [f["name"] for f in fields],
+            "metrics": [m["id"] for m in metrics], "sources": sources}
+
+
+# ==========================================================================
+# 发布之后补比
+# ==========================================================================
+
+_background: set[asyncio.Task[Any]] = set()
+
+
+def after_publish(workflow_id: str, origin: Any) -> int:
+    """工作流发布了：从问数据固化来的（origin.sources），来源运行在后台按现在的口径补比一次（trigger=certified）。返回排上了
+    几个。不挡发布，失败只记日志（ledger/match 的 schedule 自己兜着）。"""
+    from app.ledger import match as ledger_match
+
+    sources = origin.get("sources") if isinstance(origin, dict) else None
+    count = 0
+    for src in sources or []:
+        if isinstance(src, dict) and src.get("run_id"):
+            try:
+                if ledger_match.schedule(str(src["run_id"]), trigger="certified"):
+                    count += 1
+            except Exception:  # noqa: BLE001 - 补比是辅助步骤
+                logger.exception("工作流 %s 发布后补比来源运行出错", workflow_id)
+    return count
+
+
+__all__ = ["DISMISS_ACTION", "KINDS", "REPEAT_CONVERSATIONS", "REPEAT_RUNS", "CertifyRefused", "Shape", "after_publish",
+           "dismiss", "freeze", "item_id", "queue", "referenced_queries", "shape_of", "template_of", "varying"]

backend/app/ledger/labels.py

@@ -30,6 +30,8 @@ HOWS = ("independent", "composed")
 #: 界面上标签要写明的两句：比对在封存之后做，不在封存范围内；事后补比的是按现在的口径回看
 POST_SEAL_NOTE = "封存后比对，不在封存范围内"
 RETRO_NOTE = "按现在的口径回看"
+#: 固化出的口径发布之后对来源运行补比的（store 的 trigger=certified）
+CERTIFIED_NOTE = "按之后发布的口径回看"
 #: 发布级别的叫法（同 blueprints/publish_plan.LEVEL_TEXT、前端 lib/terms.ts）
 LEVEL_TEXT = {"published": "已发布", "governed": "受管"}
 MASKED = "已遮罩"
@@ -356,6 +358,6 @@ def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown:
     return out
 
 
-__all__ = ["HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key",
+__all__ = ["CERTIFIED_NOTE", "HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key",
            "card_expressions", "cell_label", "diff_view", "label_doc", "metric_label", "related_view", "view_label",
            "whole_of"]

backend/app/ledger/reconcile.py

@@ -65,7 +65,7 @@ PAIRS_NONE = "认不出这次的哪一列对应口径的哪个指标，请选择
 PAIR_COLUMN = "这次的结果里没有列「{column}」"
 PAIR_MASKED = "列「{column}」设置了遮罩，不能对账"
 PAIR_METRIC = "口径卡「{card}」上没有指标「{metric}」"
-PAIRS_SHAPE = "要对照的指标写成 [{ours: 这次的列, metric: 口径卡上的指标 id}]"
+PAIRS_SHAPE = "要对照的指标需要写成一组「这次的列 → 口径卡上的指标」的对照"
 ROWS_NONE = "报告里没有引用这条查询的格，不知道要对哪几行"
 ROWS_NO_KEY = "这次的结果有多行，但没有可以用来对齐的维度列（不是数的列）"
 ROWS_MASKED_KEY = "用来对齐的维度列「{column}」设置了遮罩，不能对账"

backend/app/ledger/stats.py

@@ -10,11 +10,13 @@
 - E1 等价命中：比对过的查询里 proven / composed / related / none / skipped 各几条（skipped 单独列，不混进 none），有候选口径
   的查询里各自的比率；证明等价的绑定是在哪一级配上的；没证明的原因（不可比、参数绑不上、配上了但不等价）；不等价的差异类型
   分布（time / filter / aggregate / join / other / parse，每份口径的差异每类只计一次）；逐个数的标签各几个、整题等价几次。
-- E2 对账：对账运行（origin=reconcile）发起、完成、失败各几次；相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的
-  反馈要等对账面板（第 3 波）记成对账运行上的封存后事件，这之前照实写没有数据。
+- E2 对账：对账运行（origin=reconcile，第 3 波起由 ledger/reconcile 发起）发起、完成、失败各几次，对过账的答案几个，其中参数
+  经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的反馈要等对账面板
+  （第 4 波）记成对账运行上的封存后事件，这之前照实写没有数据。
 - E3 重复度：问数据运行里执行过的查询按（数据源、normalize_sql 文本）分组，出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS
   段对话的算重复的形状，给出这样的形状几个、有重复形状的运行占几成；问题聚类（app/chat/questions：全站唯一的聚类，只取问题
-  原文、不含 history、排除系统派生的运行）里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化（固化、发布、驳回）在第 3 波。
+  原文、不含 history、排除系统派生的运行）里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化（第 3 波，ledger/certify）：固化成
+  口径草稿几份（origin.freeze 的工作流）、其中发布了几份、点了「不是口径」几项。
 - E4 召回：每个口径版本被用了多少次（比对里等价已证明的运行数、这一版的正式运行数）。每次发布召回了几组、其中数变了的比例要等
   召回（第 4 波），这之前照实写没有数据，只给召回运行的次数。
 """
@@ -40,8 +42,7 @@ ASPECT_GROUP = {"time": "time", "time_range": "time", "filter": "filter", "aggre
 DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse")
 STATUSES = ("proven", "composed", "related", "none", "skipped")
 #: 没有数据时的说法（这一块的功能还没做）
-NOT_YET_RECONCILE = "对账面板的反馈在第 3 波接上，目前没有数据"
-NOT_YET_CERTIFY = "认证队列在第 3 波接上，目前没有数据"
+NOT_YET_RECONCILE = "对账面板的反馈在第 4 波接上，目前没有数据"
 NOT_YET_RECALL = "召回在第 4 波接上，目前没有数据"
 #: 聚类最多取几个问题（同治理接口的上限）
 QUESTION_LIMIT = 500
@@ -154,6 +155,8 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
         if scope.get("until") is not None:
             stmt = stmt.where(Run.created_at < scope["until"])
     by_status = {s: int(n) for s, n in (await session.execute(stmt.group_by(Run.status))).all()}
+    origins = [o for (o,) in (await session.execute(stmt.with_only_columns(Run.origin))).all() if isinstance(o, dict)]
+    answers = {o.get("source_run") for o in origins if o.get("source_run")}
     explained = fallback = 0
     for content in contents:
         for query in content.get("queries") or []:
@@ -162,7 +165,10 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
                     explained += 1
                     fallback += bool(cand["diff"].get("fallback"))
     return {"launched": sum(by_status.values()), "succeeded": by_status.get("succeeded", 0),
-            "failed": by_status.get("failed", 0), "explained": explained, "fallback": fallback,
+            "failed": by_status.get("failed", 0), "answers": len(answers),
+            "inputs_checked": sum(1 for o in origins if o.get("inputs_from") == "checked"),
+            "pairs_chosen": sum(1 for o in origins if o.get("pairs_chosen")),
+            "explained": explained, "fallback": fallback,
             "fallback_rate": _ratio(fallback, explained),
             "feedback": {"available": False, "note": NOT_YET_RECONCILE}}
 
@@ -233,10 +239,37 @@ async def _e3(session: Any, *, principal: Any, **scope: Any) -> dict[str, Any]:
         "threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS},
         "questions": len(questions), "clusters": len(clusters), "questions_in_repeated_clusters": clustered,
         "cluster_share": _ratio(clustered, len(questions)), "embedder": embedder(),
-        "certify": {"available": False, "note": NOT_YET_CERTIFY},
+        "certify": await _certify(session, **scope),
     }
 
 
+async def _certify(session: Any, **scope: Any) -> dict[str, Any]:
+    """认证队列的转化：固化成口径草稿几份、其中发布了几份、点了「不是口径」几项（按固化、点选的时间筛窗口）。"""
+    from app.db.models import AuditEvent, Workflow
+    from app.ledger.certify import DISMISS_ACTION
+
+    since, until = scope.get("since"), scope.get("until")
+    frozen = published = 0
+    for created, origin, current in (await session.execute(
+            select(Workflow.created_at, Workflow.origin, Workflow.published_version)
+            .where(Workflow.origin.is_not(None)))).all():
+        if not isinstance(origin, dict) or not origin.get("freeze"):
+            continue
+        if (since is not None and created is not None and created < since) or \
+                (until is not None and created is not None and created >= until):
+            continue
+        frozen += 1
+        published += current is not None
+    stmt = select(func.count()).select_from(AuditEvent).where(AuditEvent.action == DISMISS_ACTION)
+    if since is not None:
+        stmt = stmt.where(AuditEvent.at >= since)
+    if until is not None:
+        stmt = stmt.where(AuditEvent.at < until)
+    dismissed = int(await session.scalar(stmt) or 0)
+    return {"available": True, "frozen": frozen, "published": published, "dismissed": dismissed,
+            "publish_rate": _ratio(published, frozen)}
+
+
 async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dict[str, Any]:
     from app.db.models import Run
     from app.engine.origin import counted_runs

backend/app/ledger/store.py

@@ -32,8 +32,8 @@ FORMAT = "agentlab.caliber_match/1"
 #: 2（第 3 波）：认出助手选用的口径（查询 composed、标签 how=composed），指标的标签记它取自哪几条查询（queries，认证队列
 #: 认「被报告引用的查询」用）
 ALGO = "2"
-#: 触发方式（见模块说明）
-TRIGGERS = ("sealed", "backfill", "on_demand")
+#: 触发方式（见模块说明）。certified（第 3 波）：从问数据固化出的口径发布之后，对来源运行补比的（ledger/certify）
+TRIGGERS = ("sealed", "backfill", "on_demand", "certified")
 
 
 @dataclass(frozen=True)

backend/app/ledger/view.py

@@ -98,7 +98,8 @@ def _stamp(recorded: store.Recorded) -> dict[str, Any]:
         if isinstance(recorded.ts, (int, float)) else None
     return {"artifact": recorded.artifact, "algo": recorded.algo, "trigger": recorded.trigger, "retro": recorded.retro,
             "matched_at": at, "post_seal": True, "note": ledger_labels.POST_SEAL_NOTE,
-            **({"retro_note": ledger_labels.RETRO_NOTE} if recorded.retro else {})}
+            **({"retro_note": ledger_labels.CERTIFIED_NOTE if recorded.trigger == "certified"
+                else ledger_labels.RETRO_NOTE} if recorded.retro else {})}
 
 
 async def run_view(run_id: str) -> dict[str, Any]:

backend/tests/test_ledger_certify.py

@@ -0,0 +1,224 @@
+"""沉淀：可沉淀的问题（认证队列）和「固化成口径草稿」（next-round 02 做法 8，app/ledger/certify.py）。
+
+真的跑三次问数据（两段对话、同一个 SQL 形状、日期不同，虚构店铺库），等后台比对做完：
+- 队列：确定性重复进队（3 次运行、2 段对话），变了的字面量是参数候选、不变的是常量，形状里不带取值；同一个问题问了 3 次
+  进问题聚类；导出过两次的、做成过应用的答案进「被反复用的答案」；点「不是口径」写审计后不再进队；
+- 固化：报告引用的查询变成固定 SQL 的取数节点，跨轮变化的日期成为入口参数（缺省值取代表运行的），常量 'paid' 照写；单行
+  结果的数值列成口径卡指标；出口契约 report_from + metrics_from；过 validate_graph 和受管级别的发布门禁；origin.sources 记
+  三个来源问题；
+- 按受管级别发布之后：来源运行在后台按它补比（trigger=certified），查询变成等价已证明、出了队；再问一次同样的形状，标
+  「等价已证明」；
+- 统计 E3：认证队列的转化（固化、发布、不是口径）有数了。
+数据是虚构店铺库。
+"""
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.db.base import SessionLocal
+from app.db.models import Conversation, ConversationTurn, Run, Workflow
+from app.engine.runner import run_manager
+from app.engine.schema import GraphSpec, validate_graph
+from app.ledger import certify
+from app.ledger import match as ledger_match
+from app.ledger import store
+from app.main import app
+from tests import ledger_samples as S
+
+QUESTION = "九月第一周卖了多少"
+WEEKS = [("2026-09-01", "2026-09-08"), ("2026-09-02", "2026-09-09"), ("2026-09-03", "2026-09-10")]
+
+
+def weekly(start: str, end: str) -> str:
+    return (f"SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{start}' "
+            f"AND paid_at < '{end}'")
+
+
+@pytest.fixture(autouse=True)
+async def engine_up(tmp_path):
+    await run_manager.setup()
+    await S.fresh_source(tmp_path / "shop.db")
+    yield
+    await ledger_match.drain()
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def conversation() -> str:
+    async with SessionLocal() as session:
+        conv = Conversation(title="问销售", kind="chat")
+        session.add(conv)
+        await session.commit()
+        return conv.id
+
+
+async def ask(monkeypatch, conversation_id: str, sql: str, question: str = QUESTION) -> Run:
+    """问数据的一轮：建轮次、跑一次问数据运行（来源记着对话和轮次）、挂回轮次、等比对做完。"""
+    async with SessionLocal() as session:
+        turn = ConversationTurn(conversation_id=conversation_id, question=question)
+        session.add(turn)
+        await session.commit()
+        turn_id = turn.id
+    S.script(monkeypatch, "这一周销售额 [[v:Q1.r0.gmv]]。")
+    run = await run_manager.start(graph=S.explore_graph(sql), input_payload={"question": question}, origin_kind="chat",
+                                  origin={"conversation_id": conversation_id, "turn_id": turn_id})
+    row = await S.wait(run.id)
+    assert row.status == "succeeded", row.error
+    async with SessionLocal() as session:
+        (await session.get(ConversationTurn, turn_id)).run_id = row.id
+        await session.commit()
+    await ledger_match.drain()
+    return row
+
+
+async def three_asks(monkeypatch) -> list[Run]:
+    a, b = await conversation(), await conversation()
+    return [await ask(monkeypatch, conv, weekly(*week)) for conv, week in zip((a, a, b), WEEKS)]
+
+
+def shape_items(body: dict) -> list[dict]:
+    return [i for i in body["items"] if i["kind"] == "shape" and i["source"] == S.CURRENT["source"]]
+
+
+# --------------------------------------------------------------------------
+# 队列
+# --------------------------------------------------------------------------
+
+
+def test_varying_literals_are_parameters_and_the_template_puts_them_back():
+    shape = certify.Shape("shop", "SELECT SUM(amount) AS gmv FROM orders WHERE status = ? AND paid_at >= ? AND x = ?",
+                          ("'paid'", "'2026-09-01'", "3"))
+    assert certify.varying([shape.literals, ("'paid'", "'2026-09-02'", "4")]) == [1, 2]
+    template, fields = certify.template_of(shape, [1, 2], "sqlite", start=3)
+    assert template == ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.p3 }}' "
+                        "AND x = {{ input.p4 }}")
+    assert fields == [{"name": "p3", "type": "string", "required": True, "default": "2026-09-01"},
+                      {"name": "p4", "type": "number", "required": True, "default": 3}]
+
+
+async def test_repeated_shapes_questions_and_reused_answers_queue_up(client, monkeypatch):
+    runs = await three_asks(monkeypatch)
+    body = (await client.get("/api/ledger/certify-queue")).json()
+    [item] = shape_items(body)
+    assert item["runs"] == 3 and item["conversations"] == 2 and item["question"] == QUESTION
+    assert item["params"] == [{"index": 1, "distinct": 3}, {"index": 2, "distinct": 3}] and item["constants"] == 1
+    assert "2026-09" not in item["shape"] and "'paid'" not in item["shape"] and "?" in item["shape"]
+    assert {t["run_id"] for t in item["turns"]} == {r.id for r in runs}
+    assert all(t["turn_id"] and t["conversation_id"] for t in item["turns"])
+    assert item["representative"]["run_id"] in {r.id for r in runs} and len(item["peers"]) == 2
+    assert item["related"] == [] and item["frozen_into"] == []
+    clusters = [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == QUESTION]
+    assert clusters and clusters[0]["count"] >= 3
+
+    # 被反复用的答案：导出过两次的、做成过应用的
+    from app.db.models_blueprints import Blueprint
+
+    async with SessionLocal() as session:
+        from app.auth.audit import record
+
+        for _ in range(2):
+            await record(session, "report.exported", "run", runs[0].id, data={"format": "html"})
+        session.add(Blueprint(name="周销售应用", source={"kind": "run", "id": runs[1].id}, spec={}))
+        await session.commit()
+    body = (await client.get("/api/ledger/certify-queue")).json()
+    reused = {i["representative"]["run_id"]: i for i in body["items"] if i["kind"] == "reused"}
+    assert reused[runs[0].id]["uses"]["exports"] == 2 and reused[runs[1].id]["uses"]["app"] is True
+    assert runs[2].id not in reused
+
+    # 「不是口径」：记审计，之后不再进队
+    res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": item["id"], "kind": "shape"})
+    assert res.status_code == 200
+    assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []
+    stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"]
+    assert stats["available"] is True and stats["dismissed"] >= 1
+
+
+async def test_two_runs_or_one_conversation_are_not_enough(client, monkeypatch):
+    conv = await conversation()
+    for week in WEEKS:
+        await ask(monkeypatch, conv, weekly(*week))
+    assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [], "只来自一段对话"
+
+
+# --------------------------------------------------------------------------
+# 固化、发布、再问
+# --------------------------------------------------------------------------
+
+
+async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(client, monkeypatch):
+    from app.engine.governance import publish_issues
+
+    runs = await three_asks(monkeypatch)
+    [item] = shape_items((await client.get("/api/ledger/certify-queue")).json())
+    rep = item["representative"]["run_id"]
+    res = await client.post("/api/copilot/from-run", json={"run_id": rep, "freeze": True, "peers": item["peers"]})
+    assert res.status_code == 200, res.text
+    made = res.json()
+    assert made["frozen"] is True and made["params"] == ["p1", "p2"] and made["metrics"] == ["m1"]
+    async with SessionLocal() as session:
+        wf = await session.get(Workflow, made["workflow_id"])
+    assert wf.status == "draft" and "certified" in wf.tags
+    nodes = {n["id"]: n for n in wf.graph["nodes"]}
+    assert set(nodes) == {"start", "fetch", "card", "write", "out"}
+    rep_run = next(r for r in runs if r.id == rep)
+    start, end = WEEKS[[r.id for r in runs].index(rep)]
+    assert nodes["start"]["data"]["config"]["fields"] == [
+        {"name": "p1", "type": "string", "required": True, "default": start},
+        {"name": "p2", "type": "string", "required": True, "default": end}]
+    sql = nodes["fetch"]["data"]["config"]["args"]["sql"]
+    assert "'{{ input.p1 }}'" in sql and "'{{ input.p2 }}'" in sql and "'paid'" in sql
+    assert nodes["fetch"]["data"]["config"]["tool"] == S.tool()
+    assert nodes["card"]["data"]["config"]["metrics"] == [
+        {"id": "m1", "name": "gmv", "expression": "cell(nodes.fetch, 0, 'gmv')"}]
+    assert nodes["out"]["data"]["config"]["contract"] == {"report_from": "write", "metrics_from": "card",
+                                                          "required": ["m1"]}
+    assert {k: nodes["write"]["data"]["config"].get(k) for k in ("numbers", "on_violation", "claims")} == {
+        "numbers": "strict", "on_violation": "fail", "claims": "require_citation"}
+    assert nodes["write"]["data"]["config"]["instructions"] == f"回答：{QUESTION}"
+    spec = GraphSpec.model_validate(wf.graph)
+    assert not [i for i in validate_graph(spec).issues if i.level == "error"]
+    assert not [i for i in publish_issues(spec, level="governed") if i.level == "error"], \
+        [i.message for i in publish_issues(spec, level="governed") if i.level == "error"]
+    assert [s["run_id"] for s in wf.origin["sources"]][0] == rep and len(wf.origin["sources"]) == 3
+    assert all(s["question"] == QUESTION for s in wf.origin["sources"])
+    assert wf.origin["run_id"] == rep_run.id and wf.origin["freeze"] is True
+
+    # 队列上标着已经固化过
+    [again] = shape_items((await client.get("/api/ledger/certify-queue")).json())
+    assert {w["workflow_id"] for w in again["frozen_into"]} == {wf.id}
+
+    # 按受管级别发布：来源运行在后台补比，变成等价已证明、出了队
+    res = await client.post(f"/api/workflows/{wf.id}/publish", json={"level": "governed"})
+    assert res.status_code == 200, res.text
+    await ledger_match.drain()
+    for run in runs:
+        recorded = await store.latest(run.id)
+        assert recorded.trigger == "certified" and recorded.retro is True
+        [query] = recorded.content["queries"]
+        assert query["status"] == "proven" and query["candidates"][0]["caliber"] == f"{wf.id}@1"
+    body = (await client.get(f"/api/runs/{runs[0].id}/ledger")).json()
+    assert body["retro_note"] == "按之后发布的口径回看"
+    assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []
+
+    # 再问一次同样的形状：标等价已证明
+    later = await ask(monkeypatch, await conversation(), weekly("2026-09-05", "2026-09-12"))
+    [label] = (await store.latest(later.id)).content["reports"]["write"]["labels"].values()
+    assert label["kind"] == "caliber" and label["how"] == "independent" and label["caliber"] == f"{wf.id}@1"
+    assert label["binding"] == {"p1": "2026-09-05", "p2": "2026-09-12"}
+
+    stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"]
+    assert stats["frozen"] >= 1 and stats["published"] >= 1
+
+
+async def test_freezing_needs_a_matched_successful_run(client, monkeypatch):
+    res = await client.post("/api/copilot/from-run", json={"run_id": "nope", "freeze": True})
+    assert res.status_code == 404
+    row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共几家店，没有引用。")
+    res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True})
+    assert res.status_code == 409 and res.json()["detail"] == certify.NO_QUERIES

frontend/src/lib/terms.ts

@@ -4522,6 +4522,8 @@ export const LEDGER_TEXT = {
   /** 副标题：比对在封存之后；事后补比、打开时按需比的另写一句 */
   sub: '封存后比对，不在封存范围内',
   retro: '按现在的口径回看',
+  /** 从问数据固化出的口径发布之后，对来源运行补比的那一次（服务端 trigger=certified） */
+  certified: '按之后发布的口径回看',
   /** 标签的四种、认法的两种（第 4 波的证据面板、轮次摘要用同一套） */
   kind: { caliber: '来自口径', related: '相关口径', reasoning: '这次的推理', unknown: '没有比对' },
   how: { independent: '等价已证明', composed: '助手选用' },

frontend/src/run/decode.ts

@@ -2411,7 +2411,8 @@ export function decodeRun(events: RunEvent[], final?: RunFinal, opts?: DecodeOpt
           id: `lm-${seq}`, seq, kind: 'note', status: 'done', code: 'ledger_matched',
           level: typeof counts.skipped === 'number' && counts.skipped > 0 ? 'warn' : 'info',
           title: LEDGER_TEXT.timeline(counts),
-          sub: [LEDGER_TEXT.sub, retro ? LEDGER_TEXT.retro : ''].filter(Boolean).join(' · '),
+          sub: [LEDGER_TEXT.sub, d.trigger === 'certified' ? LEDGER_TEXT.certified : retro ? LEDGER_TEXT.retro : '']
+            .filter(Boolean).join(' · '),
         })
         break
       }

scripts/check-decode.mjs

@@ -1850,6 +1850,9 @@ await section('口径账本：封存后比对已发布口径（ledger.matched）
   const late = run({ counts: { proven: 0, related: 0, none: 1, skipped: 1 }, trigger: 'backfill' }).find((s) => s.code === 'ledger_matched')
   check('事后补比的写明按现在的口径回看；没比完的是提醒', late?.title === '比对已发布口径：1 条没有相关的已发布口径，1 条没有比完'
     && late?.sub === '封存后比对，不在封存范围内 · 按现在的口径回看' && late?.level === 'warn', `${late?.title} / ${late?.sub} / ${late?.level}`)
+  const certified = run({ counts: { proven: 1 }, trigger: 'certified' }).find((s) => s.code === 'ledger_matched')
+  check('固化出的口径发布后补比的写明按之后发布的口径回看', certified?.sub === '封存后比对，不在封存范围内 · 按之后发布的口径回看',
+    certified?.sub)
   const odd = run({ counts: 'x' }).find((s) => s.code === 'ledger_matched')
   check('缺字段：只说比对过，不出 undefined、不瞎填数', odd?.title === '比对已发布口径' && !/undefined|NaN/.test(`${odd?.title}${odd?.sub}`),
     odd?.title)