出口契约：分析说法出口独立复核、指标可比性、引用模式的血缘口子。引用模式下 verify_doc 按文档记下的那一项再核一遍分析说法：处置取文档和报告节点产出里更严的那个，strict 有违规记一条缺口（降档），报告节点报上来的同一条不重复收，check 只写进出具声明（analysis_claims，counted 为 false），要求 strict 却没有核对记录的照实记缺口，老文档不核。口径卡指标上的可比性（MetricComparability）：报告引用了的（引用模式按文档，别的按数字回指）和契约必需的指标两期不可比记缺口，有条件可比的只写进出具声明（comparability）。血缘（D15）：收齐报告引用的指标（[[m:]] 片段和句末依据里的 m:），按卡上记的输入认追不到原始数据的，受管级别的正式运行记缺口，其余写进出具声明的 untraced（counted 为 false）；_untraced_gaps 拆出可传子集的 _untraced。现有受管正式运行的两条用例（裁判判档）档位不变、缺口里多一条血缘，断言不涉及缺口列表，不用改

backend/app/engine/nodes/io.py

@@ -236,6 +236,15 @@ def _apply_contract(
         )
         unmatched = trace.unmatched if trace else []
 
+    # 口径卡指标上的可比性（口径卡跨两份同形查询算环比、同比时自动查的）：报告引用了的（引用模式按文档，别的按数字回指上
+    # 的指标）、契约必需的指标两期不可比，记缺口降档；有条件可比的写进出具声明作标注
+    required = {str(r) for r in contract.get("required") or []}
+    matched_ids = {str(h.get("metric")) for h in (trace.matched if trace else []) if h.get("metric")}
+    compared_gaps, comparability = _comparability(
+        [*(cited.get("cited_metrics") or []),
+         *(m for m in metrics if str(m.get("id")) in required | (set() if citations else matched_ids))])
+    gaps.extend(compared_gaps)
+
     # 同一条查询算出几个指标、报告又引用了它的格，只记一句（按出问题的那条查询归并，列出受影响的引用）
     titles = {n.id: n.title for n in ctx.run.spec.nodes}
     aliases: dict[str, str] = cited.get("aliases") or {}
@@ -271,8 +280,9 @@ def _apply_contract(
         if isinstance(payload, dict) and payload.get("exhausted"):
             gaps.append(f"协作团队「{titles.get(node_id, node_id)}」用完 {payload.get('rounds', '?')} "
                         "轮仍未完成，交付的是成员最后的回复")
-    # 上游节点自己报上来的缺口（双盲复算比不上、反方审稿的反驳没回应……）：只会降档，不会升档
-    gaps.extend(_upstream_issuance_gaps(state, ctx))
+    # 上游节点自己报上来的缺口（双盲复算比不上、反方审稿的反驳没回应……）：只会降档，不会升档。报告节点报的分析说法那一条，
+    # 引用模式下出口自己复核过、记了自己的缺口的不再重复收
+    gaps.extend(_upstream_issuance_gaps(state, ctx, skip=cited.get("analysis_skip")))
 
     tier = decide_tier(
         missing_required=missing_required,
@@ -290,7 +300,7 @@ def _apply_contract(
         return _declare_citations(tier, cited, calibers=calibers, metrics=metrics, gaps=gaps,
                                   missing_required=missing_required, missing_expected=missing_expected,
                                   report_from=str(report_from), declared_at=datetime.now(timezone.utc),
-                                  ctx=ctx, sql_checks=sql_checks)
+                                  ctx=ctx, sql_checks=sql_checks, comparability=comparability)
 
     # 逐个数字的出处。以前只有一个计数，界面标不出「这个数来自哪个指标」
     matched = [
@@ -311,6 +321,8 @@ def _apply_contract(
         "gaps": gaps,
         "declared_at": datetime.now(timezone.utc).isoformat(),
         **({"sql_checks": sql_checks} if sql_checks else {}),
+        # 指标的可比性（有条件可比的标注、不可比的同时在缺口里）：只在有的时候带这个键
+        **({"comparability": comparability} if comparability else {}),
     }
     ctx.emit(
         EventType.ISSUANCE,
@@ -343,8 +355,9 @@ def _shown_text(fields: dict[str, Any]) -> str:
     return "\n".join(parts)
 
 
-def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]:
-    """指标出具的缺口：输入没追到原始数据的指标，每个一句，说清停在哪个节点（engine/lineage 的认法）。"""
+def _untraced(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[dict[str, Any]]:
+    """输入没追到原始数据的指标（engine/lineage 的认法）：[{metric, name, stops: [{path, node_id}], gap}]。gap 是缺口那一句，
+    说清停在哪个节点。metrics 可以是契约收来的全部指标（指标出具），也可以只是报告引用的那几个（引用模式，D15）。"""
     from app.engine.lineage import untraced
 
     titles = {n.id: n.title for n in ctx.run.spec.nodes}
@@ -357,11 +370,45 @@ def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]
         where = "、".join(dict.fromkeys(
             f"{s.get('path')}（停在「{titles.get(at, at or '未知来源')}」）"
             for s in stops[:3] for at in [str(s.get('node_id') or '').split('/', 1)[0]]))
-        out.append(f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据：{where}")
+        out.append({"metric": m.get("id"), "name": m.get("name") or m.get("id"),
+                    "stops": [{"path": s.get("path"), "node_id": s.get("node_id")} for s in stops[:3]],
+                    "gap": f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据：{where}"})
     return out
 
 
-def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]:
+def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str]:
+    """指标出具的缺口：输入没追到原始数据的指标，每个一句（_untraced）。"""
+    return [u["gap"] for u in _untraced(metrics, ctx)]
+
+
+def _comparability(metrics: list[dict[str, Any]]) -> tuple[list[str], list[dict[str, Any]]]:
+    """口径卡指标上的可比性（contract.MetricComparability，口径卡跨两份同形查询算环比、同比时自动做的检查）：
+    (缺口, 出具声明里的那一项)。metrics 是要看的指标（报告引用了的、契约必需的，调用方挑好）：两期不可比的记缺口（降档，
+    同 sql_check_failed），有条件可比的只标注。没有可比性记录的指标不在里面；同一条指标只看一次。"""
+    gaps: list[str] = []
+    declared: list[dict[str, Any]] = []
+    seen: set[int] = set()
+    for m in metrics:
+        mid = str(m.get("id") or "") if isinstance(m, dict) else ""
+        found = m.get("comparability") if mid else None
+        if not isinstance(found, dict) or id(m) in seen:
+            continue
+        seen.add(id(m))
+        verdict = found.get("verdict")
+        if verdict not in ("conditional", "incomparable"):
+            continue
+        name = m.get("name") or mid
+        phrase = str(found.get("phrase") or "").strip()
+        declared.append({"metric": mid, "name": name, "verdict": verdict, "phrase": phrase,
+                         "reasons": [str(r) for r in found.get("reasons") or [] if r],
+                         **({"artifact": found["artifact"]} if found.get("artifact") else {})})
+        if verdict == "incomparable":
+            gaps.append(f"指标「{name}」{phrase or '两期不可比'}")
+    return gaps, declared
+
+
+def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext,
+                            skip: dict[str, set[str]] | None = None) -> list[str]:
     """上游节点产出里的 issuance_gaps（字符串列表），每条前面加上是哪个节点说的。
 
     这是给「把关类」节点留的通用口子：节点觉得这次的结论不该完整出具（双盲复算两路对不上、
@@ -372,6 +419,9 @@ def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]:
 
     只认这个成果节点的上游：另一条支路上的节点管不到这份成果。不是字符串、空白的条目跳过——
     这个键谁都能写（数据整形的 JSON 模板也行），写坏了只当没写，不能让出具崩掉。
+
+    skip：{节点 id: 不再收的那几条}。引用模式的出口自己复核过报告文档的分析说法、记了自己的缺口时，报告节点报上来的同一件事
+    不再重复收一条（_analysis_claims）。
     """
     from app.engine.labels import type_label
 
@@ -388,7 +438,9 @@ def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]:
             continue
         kind = type_label(node.type)
         prefix = f"{kind}：" if node.title == kind else f"{kind}「{node.title}」："
-        out.extend(prefix + gap.strip() for gap in reported if isinstance(gap, str) and gap.strip())
+        dropped = (skip or {}).get(node.id) or set()
+        out.extend(prefix + gap.strip() for gap in reported
+                   if isinstance(gap, str) and gap.strip() and gap not in dropped)
     return out
 
 
@@ -418,12 +470,14 @@ def _check_citations(
     governed：受管级别的正式运行。可疑实体（名字哪里都找不到）只在这时按缺口降档，别处只标注。
     """
     from app.core.artifact_store import load
+    from app.engine import analysis_claims as said
     from app.engine.evidence import iter_units, ledger_enabled, resolve_ref, verify_doc
     from app.engine.nodes.report import report_catalog, report_entities
 
     out: dict[str, Any] = {"matched": [], "unmatched": [], "unresolved": [], "gaps": [],
                            "doc_artifact": None, "stats": None, "claims_policy": None, "uncited": None,
-                           "unsupported": None, "claims": None, "entities": None, "queries": {}, "aliases": {}}
+                           "unsupported": None, "claims": None, "entities": None, "queries": {}, "aliases": {},
+                           "analysis": None, "analysis_skip": {}, "cited_metrics": [], "untraced": None}
     gaps = out["gaps"]
     spec = ctx.run.spec
     node = spec.node_map().get(report_from)
@@ -458,7 +512,11 @@ def _check_citations(
     broken: list[str] = []
     flagged: set[str] = set()
     suspicious: dict[str, list[dict[str, Any]]] = {"unknown_entity": [], "unverified_entity": []}
+    said_found: list[dict[str, Any]] = []       # 分析说法的违规（只有记了核对这一项的文档才有，_analysis_claims）
     for v in checked["violations"]:
+        if v["code"] in said.CODES:
+            said_found.append(v)
+            continue
         where = {k: v[k] for k in ("segment", "unit") if v.get(k)}
         if v.get("span"):
             where.update(start=v["span"][0], end=v["span"][1])
@@ -519,6 +577,16 @@ def _check_citations(
 
     _claims(out, contract, payload, evidence_on=ledger_enabled(ctx.run), governed=governed,
             uncited=checked.get("uncited") or [], doc=doc, title=node.title)
+    _analysis_claims(out, doc, payload, said_found, title=node.title, node_id=str(report_from))
+
+    # 报告引用的指标（[[m:]] 片段和句末依据里的 m:）：可比性按它们看（_apply_contract）；输入追不到原始数据的，受管级别的
+    # 正式运行记缺口（降档，和证据面板「追不到原始数据」的说法一致：代码节点自称取数也不能完整出具），其余写进出具声明
+    out["cited_metrics"] = _cited_metrics(doc, catalog, state)
+    if stray := _untraced(out["cited_metrics"], ctx):
+        if governed:
+            gaps.extend(u["gap"] for u in stray)
+        out["untraced"] = {"metrics": [{k: u[k] for k in ("metric", "name", "stops")} for u in stray[:_LISTED]],
+                           "counted": governed}
 
     mine = next((r for r in reports if r["node"].id == report_from), None)
     edited = mine["edited"] if mine else []
@@ -858,6 +926,70 @@ def _claims(out: dict[str, Any], contract: dict[str, Any], payload: dict[str, An
     }
 
 
+#: 报告节点 analysis_claims 的三种处置，从松到严
+_ANALYSIS_ORDER = ("off", "check", "strict")
+
+
+def _analysis_claims(out: dict[str, Any], doc: dict[str, Any], payload: Any, found: list[dict[str, Any]], *,
+                     title: str, node_id: str) -> None:
+    """分析说法核对的出口独立复核（00-总方案 §3.5、D9）：found 是 verify_doc 这次重新核出的说法违规（只有文档记了
+    analysis_claims 那一项才核，老文档复核结果不变）。
+
+    处置取报告文档记下的和报告节点产出里写的更严的那个：strict 有违规就记一条缺口（降档），报告节点报上来的同一条
+    （产出的 analysis_gap）不再重复收；check 只写进出具声明作标注、不影响档位。要求 strict、文档里却没有核对记录的
+    （不该发生），照实记一条缺口：出口替不了报告节点去核。结果放进 out["analysis"]（出具声明的 analysis_claims），
+    strict 和有违规时才有。"""
+    from app.engine import analysis_claims as said
+
+    found_mode = said.recorded(doc)
+    written = payload.get("analysis_claims") if isinstance(payload, dict) else None
+    modes = [m for m in (found_mode[0] if found_mode else None, written) if m in _ANALYSIS_ORDER]
+    mode = max(modes, key=_ANALYSIS_ORDER.index) if modes else None
+    if mode in (None, "off"):
+        return
+    if found_mode is None:
+        if mode == "strict":
+            out["gaps"].append(f"报告「{title}」要求严格核对分析说法，但文档中没有分析说法的核对记录，分析说法未核对")
+        return
+    counted = mode == "strict"
+    if found and counted:
+        out["gaps"].append(f"报告「{title}」有 {len(found)} 处分析说法未通过核对（{said.summary(found)}）："
+                           "判断要挂对应的分析结论并照判定写，原因要有知识库原话或运行输入作依据")
+        if isinstance(payload, dict) and isinstance(payload.get("analysis_gap"), str):
+            out["analysis_skip"] = {node_id: {payload["analysis_gap"]}}
+    if found or counted:
+        keep = ("code", "text", "claim", "op", "analyses", "unit", "segment", "span", "message", "context")
+        out["analysis"] = {"mode": mode, "count": len(found), "counts": said.counts(found), "counted": counted,
+                           "violations": [{k: v[k] for k in keep if v.get(k) not in (None, "", [])}
+                                          for v in found[:_LISTED]]}
+
+
+def _cited_metrics(doc: dict[str, Any], catalog: dict[str, Any], state: GraphState) -> list[dict[str, Any]]:
+    """报告引用了的指标（[[m:]] 片段和句末 [[see:m:…]]）：口径卡产出里的那一条（带 inputs、comparability），按引用的先后、
+    同一张卡的同一个指标只算一次。目录里认不出、卡的产出里找不到的跳过。"""
+    from app.engine.evidence import iter_units
+
+    nodes = state.get("nodes") or {}
+    out: list[dict[str, Any]] = []
+    seen: set[tuple[str, str]] = set()
+    for _, unit in iter_units(doc):
+        cites = [s.get("cite") for s in unit.get("segments") or [] if s.get("ref") and not s.get("auto")]
+        for cite in [*cites, *(unit.get("see") or [])]:
+            if not isinstance(cite, dict) or cite.get("kind") != "metric" or cite.get("status") != "resolved":
+                continue
+            entry = catalog.get(cite.get("alias")) or {}
+            card, mid = str(entry.get("node_id") or ""), str((entry.get("locator") or {}).get("metric") or "")
+            if entry.get("kind") != "metric" or not card or not mid or (card, mid) in seen:
+                continue
+            seen.add((card, mid))
+            payload = nodes.get(card)
+            metric = next((m for m in (payload.get("metrics") or []) if isinstance(m, dict) and str(m.get("id")) == mid),
+                          None) if isinstance(payload, dict) else None
+            if metric is not None:
+                out.append(metric)
+    return out
+
+
 def _cited_no_data(doc: dict[str, Any], catalog: dict[str, Any]) -> list[str]:
     """报告直接引用的格里，落在「查询区间里这张表没有数据」的列上的（查询快照的 no_data，data/coverage.py）：按列一句缺口
     「报告引用的 Q1「上周到访人数」：所依据的查询区间里这张表没有数据（…）」。读不出来的快照不算。"""
@@ -933,6 +1065,7 @@ def _declare_citations(
     tier: str, cited: dict[str, Any], *, calibers: list[dict[str, str]], metrics: list[dict[str, Any]],
     gaps: list[str], missing_required: list[str], missing_expected: list[str], report_from: str,
     declared_at: Any, ctx: NodeContext, sql_checks: list[dict[str, Any]] | None = None,
+    comparability: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     """引用模式的 _issuance。旧字段一个不少（老前端照常显示），另加 mode、report、unresolved。"""
     # 节点 id 和报告文档、report.checked 事件上记的同一个（这次运行里的，子工作流里的带前缀）
@@ -957,6 +1090,10 @@ def _declare_citations(
         **({"entities": cited["entities"]} if cited.get("entities") else {}),
         # 没通过 SQL 检查的查询（结构化）：报告页据此在显眼处说明降档原因。没有就不加这个键
         **({"sql_checks": sql_checks} if sql_checks else {}),
+        # 分析说法核对（strict，或者有违规时）、指标的可比性、报告引用的指标里追不到原始数据的：同样只在有的时候带
+        **({"analysis_claims": cited["analysis"]} if cited.get("analysis") else {}),
+        **({"comparability": comparability} if comparability else {}),
+        **({"untraced": cited["untraced"]} if cited.get("untraced") else {}),
     }
     entities = cited.get("entities")
     ctx.emit(
@@ -983,6 +1120,9 @@ def _declare_citations(
         **({"entities": {"unknown": (cited["stats"] or {}).get("unknown_entities", len(entities["unknown"])),
                          "unverified": (cited["stats"] or {}).get("unverified_entities", len(entities["unverified"])),
                          "counted": entities["counted"]}} if entities else {}),
+        # 分析说法：处置、几处、记没记缺口（名单只在出具声明里）
+        **({"analysis_claims": {k: cited["analysis"][k] for k in ("mode", "count", "counted")}}
+           if cited.get("analysis") else {}),
     )
     return issuance
 

backend/tests/test_contract_analysis_claims.py

@@ -0,0 +1,301 @@
+"""出口契约（引用模式）的分析说法复核、口径卡指标的可比性、报告引用的指标的血缘（00-总方案 §3.4、§3.5、§3.9，D9、D15）。
+
+- 分析说法：出口按报告文档记下的那一项独立再核一遍（verify_doc）。strict 有违规记一条缺口（降档），报告节点报上来的同一条
+  不重复收；check 只写进出具声明（analysis_claims，counted 为 false），不影响档位；没记这一项的老文档不核
+- 可比性：报告引用了的、契约必需的指标两期不可比记缺口（降档），有条件可比的只写进出具声明（comparability）
+- 血缘：报告引用的指标（[[m:]] 片段、句末 [[see:m:…]]）输入追不到原始数据的，受管级别的正式运行记缺口，其余写进出具声明的
+  untraced（counted 为 false）
+"""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+import uuid
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+from sqlalchemy import select
+
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion
+from app.engine import analysis_claims as ac
+from app.engine import compiler
+from app.engine.nodes.io import _analysis_claims, _comparability
+from app.engine.runner import run_manager
+from app.engine.schema import NodeType
+from app.main import app
+
+METRICS = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"},
+           {"id": "wow", "name": "环比增幅", "unit": "%", "decimals": 1, "format": "plain", "expression": "vars.kpi.wow"}]
+SOURCE = "本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv]]"
+PLAIN = "本周销售额 [[m:gmv]]。[[see:m:gmv]]"
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def weekly(*, report=None, contract=None, fetch=None, metrics=METRICS) -> dict:
+    """取数 → 口径卡 → 报告撰写 → 出口（引用模式契约）。缺省取数是数据整形（追不到原始数据）。"""
+    nodes = [
+        node("start", "input", fields=[{"name": "week", "default": "2026-W37"}]),
+        fetch or node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5, 'wow': -3.1}",
+                      assign_to="kpi"),
+        node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=metrics),
+        node("write", "report", instructions="写周报", **(report or {})),
+        node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
+             contract={"report_from": "write", "metrics_from": ["caliber"], "required": ["gmv"], "strict": True,
+                       **(contract or {})}),
+    ]
+    ids = [n["id"] for n in nodes]
+    return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(ids, ids[1:])]}
+
+
+def writes(monkeypatch, text: str) -> None:
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
+
+
+async def wait(run_id: str) -> Run:
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            return row
+    raise AssertionError(f"run {run_id} 没跑完：{row.status}")
+
+
+async def issued(graph, *, run_class="exploratory") -> dict:
+    run = await run_manager.start(graph=graph, input_payload={}, run_class=run_class)
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    return row.output["_issuance"]
+
+
+async def governed(client, graph) -> dict:
+    """把这张图发布成受管级别，从发布版本发起正式运行。"""
+    async with SessionLocal() as session:
+        wf = Workflow(name=f"分析-{uuid.uuid4().hex[:6]}", graph=graph, status="governed", published_version=1)
+        session.add(wf)
+        await session.flush()
+        session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph))
+        await session.commit()
+        wf_id = wf.id
+    res = await client.post("/api/runs", json={"workflow_id": wf_id, "run_class": "formal"})
+    assert res.status_code == 201, res.text
+    row = await wait(res.json()["id"])
+    assert row.status == "succeeded", row.error
+    return row.output["_issuance"]
+
+
+async def issuance_event(run_id: str) -> dict:
+    async with SessionLocal() as session:
+        [event] = (await session.execute(select(RunEvent).where(
+            RunEvent.run_id == run_id, RunEvent.type == "issuance"))).scalars()
+    return event.data
+
+
+def said_gaps(issuance: dict) -> list[str]:
+    return [g for g in issuance["gaps"] if "分析说法" in g]
+
+
+# --------------------------------------------------------------------------
+# 分析说法：出口独立复核
+# --------------------------------------------------------------------------
+
+
+async def test_strict_leftovers_degrade_with_one_gap_from_the_exit(monkeypatch):
+    writes(monkeypatch, SOURCE)
+    issuance = await issued(weekly(report={"analysis_claims": "strict"}))
+    assert issuance["tier"] == "degraded"
+    [gap] = said_gaps(issuance)
+    assert gap.startswith("报告「write」有 1 处分析说法未通过核对（1 处判断没挂分析结论）"), "报告节点那一条不重复收"
+    declared = issuance["analysis_claims"]
+    assert declared["mode"] == "strict" and declared["counted"] is True and declared["count"] == 1
+    [v] = declared["violations"]
+    assert v["code"] == ac.UNCITED and v["text"] == "主要来自" and v["claim"] == "source" and v["span"]
+
+
+async def test_check_leftovers_are_declared_but_do_not_change_the_tier(monkeypatch):
+    writes(monkeypatch, SOURCE)
+    issuance = await issued(weekly())
+    assert issuance["tier"] == "formal" and said_gaps(issuance) == []
+    declared = issuance["analysis_claims"]
+    assert declared["mode"] == "check" and declared["counted"] is False and declared["counts"] == {ac.UNCITED: 1}
+
+
+async def test_a_clean_report_declares_nothing_under_check_and_a_clean_slate_under_strict(monkeypatch):
+    writes(monkeypatch, PLAIN)
+    assert "analysis_claims" not in await issued(weekly())
+    issuance = await issued(weekly(report={"analysis_claims": "strict"}))
+    assert issuance["tier"] == "formal" and issuance["analysis_claims"] == {
+        "mode": "strict", "count": 0, "counts": {}, "counted": True, "violations": []}
+
+
+async def test_off_and_old_documents_are_not_checked(monkeypatch):
+    writes(monkeypatch, SOURCE)
+    issuance = await issued(weekly(report={"analysis_claims": "off"}))
+    assert "analysis_claims" not in issuance and said_gaps(issuance) == [] and issuance["tier"] == "formal"
+
+
+async def test_a_narrative_contract_still_takes_the_report_gap(monkeypatch):
+    writes(monkeypatch, SOURCE)
+    graph = weekly(report={"analysis_claims": "strict"})
+    contract = graph["nodes"][-1]["data"]["config"]["contract"]
+    del contract["report_from"]
+    contract["narrative"] = "{{ nodes.write.text }}"
+    issuance = await issued(graph)
+    [gap] = said_gaps(issuance)
+    assert gap.startswith("报告撰写「write」：报告有 1 处分析说法未通过核对") and issuance["tier"] == "degraded"
+
+
+def test_the_exit_policy_takes_the_stricter_of_the_doc_and_the_node():
+    def run(doc, payload, found=()):
+        out = {"gaps": [], "analysis": None, "analysis_skip": {}}
+        _analysis_claims(out, doc, payload, list(found), title="周报", node_id="write")
+        return out
+
+    violation = {"code": ac.UNCITED, "text": "主要来自", "claim": "source", "op": "decompose", "message": "…"}
+    # 老文档（没有这一项）、节点也没写：不核
+    assert run({}, {"claims": "off"}) == {"gaps": [], "analysis": None, "analysis_skip": {}}
+    # 节点要求 strict，文档里却没有核对记录：照实记缺口
+    assert run({}, {"analysis_claims": "strict"})["gaps"] == [
+        "报告「周报」要求严格核对分析说法，但文档中没有分析说法的核对记录，分析说法未核对"]
+    assert run({}, {"analysis_claims": "check"})["gaps"] == []
+    # 文档记的是 check、节点写的是 strict：按更严的
+    out = run({ac.DOC_FIELD: ac.record("check")}, {"analysis_claims": "strict", "analysis_gap": "报告有 1 处…"},
+              [violation])
+    assert len(out["gaps"]) == 1 and out["analysis"]["counted"] is True
+    assert out["analysis_skip"] == {"write": {"报告有 1 处…"}}
+
+
+# --------------------------------------------------------------------------
+# 可比性
+# --------------------------------------------------------------------------
+
+
+def test_comparability_gaps_only_for_incomparable_metrics():
+    metrics = [
+        {"id": "wow", "name": "环比增幅", "comparability": {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据",
+                                                       "reasons": ["本期只有 3 天数据"], "artifact": "c" * 64}},
+        {"id": "aov", "name": "客单价", "comparability": {"verdict": "conditional",
+                                                      "phrase": "两期有条件可比：两段的周末天数不同：4 天对 6 天"}},
+        {"id": "gmv", "name": "销售额", "comparability": {"verdict": "comparable", "phrase": "两期可比"}},
+        {"id": "orders", "name": "订单数"},
+    ]
+    gaps, declared = _comparability(metrics + metrics[:1])
+    assert gaps == ["指标「环比增幅」两期不可比：本期只有 3 天数据"]
+    assert [(d["metric"], d["verdict"]) for d in declared] == [("wow", "incomparable"), ("aov", "conditional")]
+    assert declared[0]["reasons"] == ["本期只有 3 天数据"] and declared[0]["artifact"] == "c" * 64
+
+
+@pytest.fixture
+def comparability(monkeypatch):
+    """口径卡给指标记上可比性（口径卡跨两份同形查询算环比时自动做的检查由入口包实现；这里直接记上）。"""
+    marks: dict[str, dict] = {}
+    original = compiler.RUNNERS[NodeType.METRICS]
+
+    async def card(state, ctx):
+        updates = await original(state, ctx)
+        for metric in updates["nodes"][ctx.node.id]["metrics"]:
+            if metric["id"] in marks:
+                metric["comparability"] = marks[metric["id"]]
+        return updates
+
+    monkeypatch.setitem(compiler.RUNNERS, NodeType.METRICS, card)
+    return marks
+
+
+async def test_a_cited_incomparable_metric_degrades_and_a_conditional_one_is_noted(monkeypatch, comparability):
+    comparability["wow"] = {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据",
+                            "reasons": ["本期只有 3 天数据"]}
+    writes(monkeypatch, "本周销售额 [[m:gmv]]，环比 [[m:wow]]。[[see:m:gmv,m:wow]]")
+    issuance = await issued(weekly())
+    assert issuance["tier"] == "degraded" and "指标「环比增幅」两期不可比：本期只有 3 天数据" in issuance["gaps"]
+    assert [c["metric"] for c in issuance["comparability"]] == ["wow"]
+    comparability["wow"] = {"verdict": "conditional", "phrase": "两期有条件可比：两段的周末天数不同"}
+    issuance = await issued(weekly())
+    assert issuance["tier"] == "formal" and issuance["comparability"][0]["verdict"] == "conditional"
+
+
+async def test_an_incomparable_metric_nobody_cites_or_requires_is_left_alone(monkeypatch, comparability):
+    comparability["wow"] = {"verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据"}
+    writes(monkeypatch, PLAIN)
+    issuance = await issued(weekly())
+    assert issuance["tier"] == "formal" and "comparability" not in issuance
+    # 契约必需的指标不引用也算
+    issuance = await issued(weekly(contract={"required": ["gmv", "wow"]}))
+    assert "指标「环比增幅」两期不可比：本期只有 3 天数据" in issuance["gaps"]
+
+
+# --------------------------------------------------------------------------
+# 血缘：报告引用的指标追没追到原始数据
+# --------------------------------------------------------------------------
+
+
+async def test_a_cited_metric_that_stops_at_a_transform_is_only_noted_outside_governed_runs(monkeypatch):
+    writes(monkeypatch, PLAIN)
+    issuance = await issued(weekly(), run_class="formal")
+    assert issuance["tier"] == "formal" and not any("没追到原始数据" in g for g in issuance["gaps"])
+    untraced = issuance["untraced"]
+    assert untraced["counted"] is False and [m["metric"] for m in untraced["metrics"]] == ["gmv"]
+    assert untraced["metrics"][0]["stops"][0]["node_id"] == "fetch"
+
+
+async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, monkeypatch):
+    writes(monkeypatch, PLAIN)
+    issuance = await governed(client, weekly())
+    assert issuance["tier"] == "degraded"
+    [gap] = [g for g in issuance["gaps"] if "没追到原始数据" in g]
+    assert gap.startswith("指标「销售额」的输入没追到原始数据：vars.kpi.gmv（停在「fetch」）")
+    assert issuance["untraced"]["counted"] is True
+    # 没被引用的指标（环比增幅同样停在整形节点）不算：只看报告引用了的
+    assert not any("环比增幅" in g for g in issuance["gaps"])
+
+
+@pytest.fixture
+async def shop(tmp_path):
+    path = tmp_path / "shop.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)])
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
+        if row is None:
+            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            session.add(row)
+        else:
+            row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield
+    await engines.invalidate(source_id)
+
+
+async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop):
+    writes(monkeypatch, PLAIN)
+    fetch = node("fetch", "tool", tool="db_query__shop", args={"sql": "SELECT SUM(amount) AS gmv FROM orders"})
+    graph = weekly(fetch=fetch, metrics=[{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1,
+                                          "expression": "cell(nodes.fetch, 0, 'gmv')"}])
+    issuance = await governed(client, graph)
+    assert "untraced" not in issuance and not any("没追到原始数据" in g for g in issuance["gaps"])