diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py
index 52afad10..44cd639d 100644
--- a/backend/app/api/evidence.py
+++ b/backend/app/api/evidence.py
@@ -21,6 +21,9 @@ artifacts 表可以事后插行，run.output 是可以改写的一列，封存
 options.mask_columns 里的列换成「已遮罩」——在有身份体系之前，这只减少暴露，不是
 安全边界：知道工件 id 的人照样能取到整份快照。
 
+期 4 的核查状态同样只读封存范围内的东西：指标步骤带双盲复算的结果（_dual_check_of，读双盲复算节点的产出和台账
+条目），句子带反方审稿的判断（_challenge_of，读反方审稿节点的产出和审稿记录）。它们是核查，不是证据：没有就不带这两个键。
+
 结论句的判定（四期）是模型给的，不是证据：正式运行的判定在报告文档里、随文档封存；探索运行
 按需裁判（POST …/evidence/judge），判定作为 evidence.judged 事件追加在封存之后，只填补没判过的
 句子。交给裁判的摘录同样只取封存范围内的工件。
@@ -75,7 +78,7 @@ from app.data.provenance_types import (
 from app.data.table_versions import SnapshotManifestMismatch, SnapshotMissing
 from app.db.base import SessionLocal
 from app.db.models import DataSource, Run, RunEvent, Workflow
-from app.engine import direct_select
+from app.engine import direct_select, recompute
 from app.engine.evidence import (
     ENTITY_KINDS,
     GUESS_NOTE,
@@ -103,6 +106,7 @@ from app.engine.evidence import (
 )
 from app.engine.judge import FIELDS_MAX, SETTLED, UNSUPPORTED, VERDICTS, candidates, outdated
 from app.engine.merge_query import MERGE_SOURCE, trace_cell
+from app.engine.schema import GraphSpec, challenge_report, dual_check_card
 from app.engine.toolcalls import QUERY_PREFIX
 
 router = APIRouter(prefix="/api/runs", tags=["evidence"])
@@ -726,8 +730,11 @@ async def evidence_segment(run_id: str, segment_id: str, report: str | None = No
     start, end = unit.get("span") or [0, 0]
     chain = await _chain(seg, doc, sealed, chosen.node_id)
     covered = sealed.trusted and all(step.get("sealed") for step in chain if "sealed" in step)
-    masked = list(dict.fromkeys(c for step in chain if step.get("step") == "query" for c in step.get("masked") or []))
+    # 遮罩列：查询步骤里遮着的，加上指标步骤里取自遮罩列的输入（按位置取的那一格可能没有对应的查询步骤）
+    masked = list(dict.fromkeys(c for step in chain if step.get("step") in ("query", "input")
+                                for c in step.get("masked") or []))
     verdict = _effective_verdicts(chosen, sealed)[0].get(str(unit.get("id")))
+    challenged = _challenge_of(sealed, chosen, str(unit.get("id")))
     return {
         "report": {"node_id": chosen.node_id, "doc_artifact": chosen.doc_artifact},
         "segment": {"id": seg["id"], "text": seg.get("text"), "kind": seg.get("kind"), "state": seg.get("state"),
@@ -736,6 +743,8 @@ async def evidence_segment(run_id: str, segment_id: str, report: str | None = No
         "unit": {"id": unit.get("id"), "kind": unit.get("kind"), "text": markdown[start:end],
                  "span": unit.get("span"), "cites": unit.get("cites") or [],
                  **({"verdict": verdict} if verdict is not None else {}),
+                 # 期 4：反方审稿对这一句的判断，没审到这一句的没有这个键
+                 **({"challenge": challenged} if challenged is not None else {}),
                  "on_demand": _on_demand(sealed, doc, str(unit.get("id")), verdict)},
         "block": {"id": block.get("id"), "type": block.get("type")},
         "chain": chain,
@@ -855,10 +864,15 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
         "source": await _caliber_source((card or {}).get("source")),
         "caliber_upgrade": _caliber_upgrade(sealed.upgrades.get(str(entry.get("node_id") or ""))),
     }
+    # 期 4：这个指标做没做过双盲复算、结果如何。没有双盲复算复核过这份指标集时没有这个键，老运行一字不差
+    if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id):
+        step["dual_check"] = checked
     catalog = doc.get("catalog") or {}
     inputs: list[dict[str, Any]] = []
     #: 输入指到的查询快照 → 被引用的格；同一份快照只给一个查询步骤，格合在一起高亮
     wanted: dict[str, list[tuple[Any, Any]]] = {}
+    hide = _InputMasks(sealed, masks, str(entry.get("node_id") or ""), artifact)
+    withheld: set[str] = set()
     for item in (metric or {}).get("inputs") or []:
         one = {"step": "input", **item}
         artifact_of = item.get("artifact")
@@ -867,11 +881,99 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
             if cell := _input_cell(item, catalog):
                 one["cell"] = cell
             wanted.setdefault(str(artifact_of), []).extend(_located(item.get("locator") or {}))
+        if columns := await hide.columns(item):
+            one = _withheld(one, columns)
+            withheld.add(str(item.get("path")))
         inputs.append(one)
+    if withheld:
+        # 代入式里那几处同样写「已遮罩」：不然标签里遮着、代入式里亮着，同一个值换个地方照样看得到
+        step["substituted"] = recompute.substituted_without(metric or {}, withheld, MASKED)
     queries = [s for a, cells in wanted.items() for s in await _query_steps(a, cells, doc, sealed, masks)]
     return [step, *inputs, *queries]
 
 
+def _withheld(step: dict[str, Any], columns: list[str]) -> dict[str, Any]:
+    """取自遮罩列的输入：值写「已遮罩」（本来就没有值的照旧是空的，缺输入还得看得出来），模型报的值、快照里的值一并
+    拿掉，masked 写明是哪几列。"""
+    out = {k: v for k, v in step.items() if k not in ("model_value", "snapshot_value")}
+    if out.get("value") is not None:
+        out["value"] = MASKED
+    out["masked"] = columns
+    return out
+
+
+class _InputMasks:
+    """一个指标的输入里哪些取自遮罩列。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询
+    按它合并的那几个数据源算；快照读不出来时只能按数据源现在设的算。
+
+    cell() 的格、Agent 交来的字段按定位里的列认。vars.res.rows[0][2] 这种按位置取自查询结果的标量卡上没记定位：顺着
+    取值链找到那份结果（recompute.positional_cell，推演用同一个认法），结果里带着快照的 id 和数据源。取值链从写变量的
+    节点在口径卡算这一次之前的产出起走（_card_outputs）。认不出来的不遮——卡上本来就只记着路径和值。
+    """
+
+    def __init__(self, sealed: _Sealed, masks: _Masks, card_node: str, artifact: str) -> None:
+        self._sealed, self._masks = sealed, masks
+        self._card, self._artifact = card_node, artifact
+        self._outputs: dict[str, str] | None = None
+        self._memo: dict[tuple[str | None, str | None], set[str]] = {}
+
+    async def columns(self, item: dict[str, Any]) -> list[str]:
+        """这个输入用到的遮罩列（列名照原样，按出现的先后）；没用到的交回空列表。"""
+        if item.get("via") in ("tool_cell", "agent_field") and item.get("artifact"):
+            artifact, source = str(item["artifact"]), None
+            used = [c for _, c in _located(item.get("locator") or {}) if isinstance(c, str)]
+        else:
+            found = recompute.positional_cell(str(item.get("path") or ""), item.get("node_id"), self._output)
+            if found is None:
+                return []
+            artifact, source, used = found["artifact"], found["source"], [found["column"]]
+        if not used:
+            return []
+        hidden = await self._hidden(artifact, source)
+        return list(dict.fromkeys(c for c in used if c.lower() in hidden))
+
+    def _output(self, node_id: Any) -> Any:
+        if self._outputs is None:
+            self._outputs = _card_outputs(self._sealed, self._card, self._artifact)
+        artifact = self._outputs.get(str(node_id or ""))
+        content, ok = _fetch(artifact) if artifact else (None, None)
+        return content if ok is True else None
+
+    async def _hidden(self, artifact: str | None, source: str | None) -> set[str]:
+        key = (artifact, source)
+        if key not in self._memo:
+            snap, ok = _fetch(artifact) if artifact else (None, None)
+            snap = snap if ok is True and isinstance(snap, dict) else None
+            if snap is not None and snap.get("source") == MERGE_SOURCE:
+                hidden, _ = await _merge_masks(snap, self._masks)
+            else:
+                info = self._sealed.queries.get(artifact) if artifact else None
+                current, _ = await self._masks.of((info or {}).get("source") or source or (snap or {}).get("source"))
+                hidden = set(current) | (_recorded_masks(snap) if snap is not None else set())
+            self._memo[key] = hidden
+        return self._memo[key]
+
+
+def _card_outputs(sealed: _Sealed, card_node: str, artifact: str) -> dict[str, str]:
+    """口径卡算出这份指标集的那一次执行开始之前，每个节点最后一次的产出工件：卡上 vars.*、nodes.* 取到的就是它们。
+    认不出是哪一次执行（台账里没有这份指标集）时，取封存范围内每个节点最后一次的产出。"""
+    finished = next((seq for seq, etype, node_id, data in sealed.events
+                     if etype == "node.finished" and node_id == card_node and isinstance(data, dict)
+                     and any(isinstance(e, dict) and e.get("kind") == "metric_set" and e.get("artifact") == artifact
+                             for e in data.get("evidence") or [])), None)
+    if finished is None:
+        return dict(sealed.outputs)
+    started = max((seq for seq, etype, node_id, _ in sealed.events
+                   if etype == "node.started" and node_id == card_node and seq < finished), default=finished)
+    out: dict[str, str] = {}
+    for seq, etype, node_id, data in sealed.events:
+        if seq >= started:
+            break
+        if etype == "node.finished" and node_id and isinstance(data, dict) and data.get("artifact"):
+            out[node_id] = str(data["artifact"])
+    return out
+
+
 def _located(locator: dict[str, Any]) -> list[tuple[Any, Any]]:
     """输入的定位 → 被引用的格。单个格 {row, column}；数组字段是一段行 {rows:[a,b], column | columns}。"""
     if isinstance(locator.get("row"), int):
@@ -899,6 +1001,144 @@ def _caliber_upgrade(event: dict[str, Any] | None) -> dict[str, Any] | None:
     return {k: event.get(k) for k in ("workflow_id", "caliber_node", "pinned", "latest", "policy", "policy_label")}
 
 
+#: 双盲复算节点产出里一个指标的结果（engine/nodes/dual_check.py 的 SAME / DIFFERENT / UNCHECKED），原样交给界面
+_DUAL_RESULTS = ("agreed", "disagreed", "unchecked")
+
+
+def _spec_of(sealed: _Sealed) -> GraphSpec | None:
+    """这次运行的图（认双盲复算复算哪张卡、反方审稿审哪份报告，和执行时同一个认法）。图读不懂的是 None。"""
+    try:
+        return GraphSpec.model_validate(sealed.graph)
+    except Exception:  # noqa: BLE001 - 图读不懂只是认不出来，证据照常给
+        return None
+
+
+def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: Any) -> dict[str, Any] | None:
+    """这个指标的双盲复算状态：{status, node_id, choice?, blind_rendered?, reason?}。没有双盲复算复核过这份指标集的是 None。
+
+    认的是复核了口径卡的哪一次结果：台账条目（kind dual_check）记着它复算的指标集工件，和指标步骤的工件对上才算——口径卡
+    在循环里算了几次时，报告引用的那一次才是。直通了的（设置里关了、口径卡上没有来自查询的输入）没有台账条目：按画布上
+    它复算的那张卡、和它跑完之前那张卡最后一次的指标集认。几个双盲复算复核了同一份指标集时取最后跑完的那个。
+
+    - agreed / disagreed / unchecked：节点产出里这个指标的结果。不一致的带审核人的选择（a / b / neither，没选的不带）；
+      一致、不一致的带盲算值（渲染好的）；无法复算的带原因
+    - skipped：节点直通，reason 同节点产出（disabled / no_inputs）；或者节点跑了、这个指标不在复算范围里（没选它、值不是
+      数、没有来自查询的输入），reason 为 not_targeted
+    产出读不出来（文件不在、与哈希不一致）时按台账条目说状态和选择，不带盲算值和原因。
+    """
+    if not card_node or not artifact or metric_id is None:
+        return None
+    checkers = set(sealed.nodes_of("dual_check"))
+    if not checkers:
+        return None
+    spec: GraphSpec | None = None
+    #: 口径卡节点 → 到这一刻为止它最后一次的指标集
+    latest: dict[str, str] = {}
+    found: dict[str, Any] | None = None
+    for _, etype, node_id, data in sealed.events:
+        if etype != "node.finished" or not node_id or not isinstance(data, dict):
+            continue
+        ledger = [e for e in data.get("evidence") or [] if isinstance(e, dict)]
+        for e in ledger:
+            if e.get("kind") == "metric_set" and e.get("artifact"):
+                latest[node_id] = str(e["artifact"])
+        if node_id not in checkers:
+            continue
+        output, ok = _fetch(str(data.get("artifact") or ""))
+        output = output if ok is True and isinstance(output, dict) else {}
+        entry = next((e for e in ledger if e.get("kind") == "dual_check"), None)
+        if entry is not None:
+            card, checked = str(entry.get("card") or ""), str(entry.get("metric_set") or "")
+        elif output.get("status") == "skipped":
+            spec = spec or _spec_of(sealed)
+            node = spec.node_map().get(node_id) if spec else None
+            card = (dual_check_card(spec, node)[0] or "") if spec and node else ""
+            checked = latest.get(card, "")
+        else:
+            continue
+        if card != card_node or checked != artifact:
+            continue
+        found = _dual_status(node_id, output, entry, str(metric_id))
+    return found
+
+
+def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | None, metric_id: str) -> dict[str, Any]:
+    """一次双盲复算里这个指标的状态（_dual_check_of 的答复）。"""
+    if output.get("status") == "skipped":
+        return {"status": "skipped", "node_id": node_id, "reason": str(output.get("reason") or "")}
+    row = next((m for m in output.get("metrics") or [] if isinstance(m, dict) and str(m.get("metric_id")) == metric_id),
+               None)
+    if row is None and not output and entry is not None:
+        # 产出读不出来：退回台账条目，只有状态和选择
+        brief = next((m for m in entry.get("metrics") or [] if isinstance(m, dict) and str(m.get("id")) == metric_id),
+                     None)
+        row = {"status": brief.get("status"), **({"choice": brief["choice"]} if "choice" in brief else {})} \
+            if brief is not None else None
+    if row is None:
+        return {"status": "skipped", "node_id": node_id, "reason": "not_targeted"}
+    status = row.get("status") if row.get("status") in _DUAL_RESULTS else "unchecked"
+    out: dict[str, Any] = {"status": status, "node_id": node_id}
+    if status == "disagreed" and row.get("choice") in ("a", "b", "neither"):
+        out["choice"] = row["choice"]
+    rendered = row.get("blind_rendered")
+    if status != "unchecked" and isinstance(rendered, str) and rendered:
+        out["blind_rendered"] = rendered
+    if status == "unchecked" and isinstance(row.get("reason"), str) and row["reason"]:
+        out["reason"] = row["reason"]
+    return out
+
+
+def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, Any] | None:
+    """这一句的反方审稿判断：{verdict, node_id, artifact, on_refuted, report_label}。没审到这一句的是 None。
+
+    只认审的就是正文这一份报告文档的：审稿记录（challenge_artifact）里记着它审的报告节点和文档工件，和这份对上才算——
+    报告在审稿之后又重写了（循环、接着跑），同一个句子编号在新的一稿里说的可能是另一句话。记录没存上（产出里的
+    challenge_artifact 为空）或读不出来时，按画布上它审的报告、和它跑完之前那份报告最后一次自查记下的文档认，判断取
+    产出里列的那几句（最多 10 句，被推翻、被削弱的排在前面），artifact 照产出写（没存上的为空）。几个反方审稿审了同一
+    份文档时取最后跑完、审到了这一句的那个。on_refuted 是实际生效的处置（degrade / flag），report_label 是报告节点的名字。
+    """
+    reviewers = set(sealed.nodes_of("challenge"))
+    if not reviewers or not unit_id:
+        return None
+    spec: GraphSpec | None = None
+    #: 报告节点 → 到这一刻为止它最后一次自查记下的文档
+    drafts: dict[str, str] = {}
+    found: dict[str, Any] | None = None
+    for _, etype, node_id, data in sealed.events:
+        if not node_id or not isinstance(data, dict):
+            continue
+        if etype == "report.checked" and data.get("doc_artifact"):
+            drafts[node_id] = str(data["doc_artifact"])
+            continue
+        if etype != "node.finished" or node_id not in reviewers:
+            continue
+        output, ok = _fetch(str(data.get("artifact") or ""))
+        if ok is not True or not isinstance(output, dict) or not isinstance(output.get("counts"), dict):
+            continue    # 直通了（关闭、没有可审的句子）或者产出读不出来
+        artifact = output.get("challenge_artifact") if isinstance(output.get("challenge_artifact"), str) else None
+        record, saved = _fetch(artifact) if artifact else (None, None)
+        if saved is True and isinstance(record, dict):
+            if record.get("report") != report.node_id or record.get("doc_artifact") != report.doc_artifact:
+                continue
+            claims = record.get("claims")
+        else:
+            spec = spec or _spec_of(sealed)
+            node = spec.node_map().get(node_id) if spec else None
+            target = challenge_report(spec, node)[0] if spec and node else None
+            if target != report.node_id or drafts.get(target) != report.doc_artifact:
+                continue
+            claims = output.get("items")
+        verdict = next((c.get("verdict") for c in claims or [] if isinstance(c, dict) and c.get("unit") == unit_id), None)
+        if not isinstance(verdict, str) or not verdict:
+            continue
+        spec = spec or _spec_of(sealed)
+        writer = spec.node_map().get(report.node_id) if spec else None
+        found = {"verdict": verdict, "node_id": node_id, "artifact": artifact,
+                 "on_refuted": str(output.get("on_refuted") or "degrade"),
+                 "report_label": writer.title if writer else report.node_id}
+    return found
+
+
 class _Masks:
     """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。"""
 
diff --git a/backend/app/engine/recompute.py b/backend/app/engine/recompute.py
index 0a6d3da0..f8590d4d 100644
--- a/backend/app/engine/recompute.py
+++ b/backend/app/engine/recompute.py
@@ -14,7 +14,8 @@
   - nodes.* / input.*：那个节点在口径卡执行之前最后一次的产出工件
   - vars.*：有 value_artifact（口径卡当时另存的完整值）才还原得出来；老运行照实说当时没有保存
   - 容器本身是一份查询结果（带 artifact 的 {columns, rows}，或者和某份快照的 rows 一模一样的列表）：排除行就去掉那几行
-- 遮罩列取来的输入不给值、也不拿来推演：改一个输入、排除一行，新值和原值一比就能算出遮住的那一格
+- 遮罩列取来的输入不给值、也不拿来推演：改一个输入、排除一行，新值和原值一比就能算出遮住的那一格。vars.res.rows[0][2]
+  这种按位置取自查询结果的标量没有定位，顺着取值链认出是哪一列（positional_cell，证据面板的指标步骤用同一个认法）
 - 落在读不出来（文件不在、与哈希不一致）的查询快照上的输入同样不给值、不推演：那一格遮没遮核对不了，宁可多遮
 
 有的指标不论改什么都推演不了（_Plan.blocked）：卡上没记全输入、用到了遮罩列、最外层的输入还原不了（容器的完整值
@@ -334,7 +335,7 @@ def _restore(item: dict[str, Any], remaining: set[str], nested: set[str], store:
             # 也不拿来推演。数据源现在设的遮罩照样认（_masked），认出来了就按遮罩说
             inp.unreadable = True
             inp.problem = inp.problem or (TAMPERED if trouble == "tampered" else UNREADABLE)
-    inp.masked = _masked(inp, locator, store)
+    inp.masked = _masked(inp, locator, store) or _positional_masked(inp, store)
     return inp
 
 
@@ -416,6 +417,110 @@ def _masked(inp: _Input, locator: dict[str, Any], store: _Store) -> bool:
     return any(isinstance(c, str) and c.lower() in hidden for c in columns)
 
 
+def _positional_masked(inp: _Input, store: _Store) -> bool:
+    """按位置取自查询结果的一格（vars.res.rows[0][2]）落在遮罩列上：卡上没记定位，顺着取值链找到那份结果、认出是哪一列。
+    认不出来的（结果里没带快照的 id、取值链走不通）不算：和以前一样照常推演。"""
+    if inp.snapshot is not None or inp.shape != "scalar" or inp.item.get("via") in ("tool_cell", "agent_field"):
+        return False
+    found = positional_cell(inp.path, inp.item.get("node_id"), lambda node_id: store.output(node_id)[0])
+    hidden = store.sources.masked.get(found["artifact"]) if found and found["artifact"] else None
+    return bool(hidden) and isinstance(found["column"], str) and found["column"].lower() in hidden
+
+
+#: 取值链没走通
+_GONE = object()
+
+
+def positional_cell(path: str, node_id: Any, output: Callable[[Any], Any]) -> dict[str, Any] | None:
+    """按位置取自一份查询结果的一格：vars.res.rows[0][2]、nodes.parse.rows[1]['amount'] 这种标量引用——查询结果经数据
+    整形解析成对象之后按行号、列号（或列名）取。返回 {artifact, source, row, column}：artifact、source 是那份结果里带着的
+    （数据源工具、合并查询交回的结果都带，没有的是 None），row 是从 0 数的行号，column 是列名。认不出来的返回 None。
+
+    遮罩要按列认（证据面板的指标步骤、推演都用这个）：cell() 和 Agent 字段有定位，这种引用卡上只记了路径和值。
+
+    output(节点 id) 交回那个节点当时的产出（读不出来给 None）。nodes.X 的根就是 X 的产出；vars.X 是写它的节点（输入记着
+    的 node_id）交给变量的那个值——数据整形、合并查询交的是产出本身，Agent、LLM、代码节点交的是产出里的 data，工具节点
+    交的是 JSON 文本，都试一遍，取值链走得通的那个就是。下标得是字面量：rows[vars.i] 要另取一个值才知道是哪一行，不认。
+    """
+    try:
+        tree = ast.parse(path, mode="eval").body
+    except SyntaxError:
+        return None
+    keys: list[Any] = []
+    while isinstance(tree, (ast.Attribute, ast.Subscript)):
+        if isinstance(tree, ast.Attribute):
+            keys.append(tree.attr)
+        else:
+            key = _literal_key(tree.slice)
+            if key is _GONE:
+                return None
+            keys.append(key)
+        tree = tree.value
+    keys.reverse()
+    if not isinstance(tree, ast.Name) or tree.id not in ("vars", "nodes") or len(keys) < 4 \
+            or not isinstance(keys[0], str):
+        return None
+    base = output(keys[0] if tree.id == "nodes" else node_id)
+    candidates = [base]
+    if isinstance(base, dict) and "data" in base:
+        candidates.append(base["data"])
+    if isinstance(base, str):
+        try:
+            candidates.append(json.loads(base))
+        except ValueError:
+            pass
+    for value in candidates:
+        if (found := _cell_in(value, keys[1:])) is not None:
+            return found
+    return None
+
+
+def _literal_key(node: ast.AST) -> Any:
+    """下标里的字面量：整数（含负数）或者文本。别的（变量、算式）交回 _GONE。"""
+    if isinstance(node, ast.Constant) and isinstance(node.value, (int, str)) and not isinstance(node.value, bool):
+        return node.value
+    if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub) and isinstance(node.operand, ast.Constant) \
+            and isinstance(node.operand.value, int) and not isinstance(node.operand.value, bool):
+        return -node.operand.value
+    return _GONE
+
+
+def _cell_in(value: Any, keys: list[Any]) -> dict[str, Any] | None:
+    """沿取值链往下走，走到一份查询结果（带 rows 列表的对象）、剩下的正好是 rows[行][列] 时交回那一格。"""
+    for i, key in enumerate(keys):
+        if isinstance(value, dict) and isinstance(value.get("rows"), list) and len(keys) - i == 3 and key == "rows":
+            return _cell_at(value, keys[i + 1], keys[i + 2])
+        if isinstance(value, dict) and isinstance(key, str):
+            value = value.get(key, _GONE)
+        elif isinstance(value, list) and isinstance(key, int) and -len(value) <= key < len(value):
+            value = value[key]
+        else:
+            return None
+        if value is _GONE:
+            return None
+    return None
+
+
+def _cell_at(result: dict[str, Any], row: Any, column: Any) -> dict[str, Any] | None:
+    rows = result["rows"]
+    if not isinstance(row, int) or not -len(rows) <= row < len(rows):
+        return None
+    row = row % len(rows)
+    record = rows[row]
+    columns = [str(c) for c in result.get("columns") or []]
+    if isinstance(column, int) and isinstance(record, (list, tuple)) and -len(columns) <= column < len(columns):
+        name: str | None = columns[column]
+    elif isinstance(column, str) and (column in columns or (isinstance(record, dict) and column in record)):
+        name = column
+    else:
+        name = None
+    if name is None:
+        return None
+    artifact, source = result.get("artifact"), result.get("source")
+    return {"artifact": artifact if isinstance(artifact, str) and artifact else None,
+            "source": source if isinstance(source, str) and source else None, "row": row, "column": name}
+
+
 def _static(path: str) -> bool:
     """取值路径的下标全是字面量（nodes.q.rows[0]），没有 nodes.q.rows[vars.i] 这种要另取一个值才知道是哪一项的。"""
     try:
@@ -840,6 +945,30 @@ def _fixed(inp: _Input) -> str | None:
     return None if value is None or is_number(value) else NOT_NUMBER
 
 
+def substituted_without(metric: dict[str, Any], paths: Iterable[str], token: str) -> str | None:
+    """卡上的代入式，只把 paths 这几个输入代成 token（证据面板给取自遮罩列的输入写「已遮罩」），别的照卡上代。
+
+    按卡上记的输入值重新代一遍：先核对一个都不换时代出来的和卡上记的一字不差，对不上（卡上记的不是原值、老卡的写法）
+    就整条不给，交回 None——宁可不显示代入式，也不能让遮着的值从对不上的那一处漏出来。卡上没代成字面量的（容器、太长的
+    文本）照旧留原路径。嵌在别的引用里的输入不单独换：外面那个引用代成了它的值，里面那一格本来就不出现。
+    """
+    expr, recorded = metric.get("expression"), metric.get("substituted")
+    if not isinstance(expr, str) or not isinstance(recorded, str):
+        return None
+    try:
+        tree, _, _ = parse_expression(expr)
+        kept = set(leaf_refs(parse_expression(recorded)[0]))
+    except ExpressionError:
+        return None
+    values = {str(i["path"]): i.get("value") for i in metric.get("inputs") or []
+              if isinstance(i, dict) and isinstance(i.get("path"), str) and i["path"] not in kept}
+    if substitute(tree, values) != recorded:
+        return None
+    hidden = set(paths)
+    swapped = ast.fix_missing_locations(_Bind({p: token for p in hidden}).visit(copy.deepcopy(tree)))
+    return substitute(swapped, {p: v for p, v in values.items() if p not in hidden})
+
+
 # --------------------------------------------------------------------------
 # 报告里引用的地方、下游口径卡
 # --------------------------------------------------------------------------
diff --git a/backend/tests/test_evidence_metric_masks.py b/backend/tests/test_evidence_metric_masks.py
new file mode 100644
index 00000000..586c9fac
--- /dev/null
+++ b/backend/tests/test_evidence_metric_masks.py
@@ -0,0 +1,233 @@
+"""证据面板指标步骤的遮罩（期 4 I2a）：取自遮罩列的输入，输入标签和代入式里都不显示它的值。
+
+以前只有查询步骤的结果窗口把遮罩列换成「已遮罩」，指标步骤的输入（「值 ← 路径」）和代入式照样写着原值——同一个
+号码，表格里遮着，上面的标签里亮着。现在：
+
+- cell() 取的格、Agent 交来的字段：按定位里的列认
+- vars.res.rows[0][2] 这种按位置取自查询结果的标量引用：顺着取值链找到那份查询结果（写变量的节点当时的产出），
+  按列号认出是哪一列
+- 遮的是「查询当时记下的」加「数据源现在设的」，和查询步骤同一套；数据源事后去掉了遮罩，当时遮着的照样遮
+- 取自遮罩列的输入值写「已遮罩」，带上是哪几列（masked），模型报的值、快照里的值一并不给；代入式里那一处写「已遮罩」，
+  遮罩列也进 redacted.columns（面板底下那句「遮罩不是安全边界」）
+- 没有遮罩的指标和以前一字不差；推演（api/whatif.py）同样认得出按位置取的遮罩列，不给值、不推演
+
+真跑一遍图（虚构的 SQLite 订单库，号码是编的），报告由 MockChatModel 照剧本写。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+from sqlalchemy import select
+
+from app.core import artifact_store
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run
+from app.engine.evidence import iter_segments
+from app.engine.runner import run_manager
+from app.main import app
+
+SQL = "SELECT id, region, phone, amount FROM orders ORDER BY id"
+#: 号码写成带横线的文本：全是数字的文本 cell() 会按数读（口径卡在算数），len() 就算不了
+ROWS = [(1, "华东", "139-0000-0001", 120.5), (2, "华南", "139-0000-0002", 80.0), (3, "华北", "139-0000-0003", 99.5)]
+PHONES = [r[2] for r in ROWS]
+#: 号码那一格按位置取：vars.res 是数据整形解析好的查询结果，第 2 列是 phone
+FIRST_PHONE = "vars.res.rows[0][2]"
+SECOND_PHONE = "vars.res.rows[1][2]"
+PHONE_CELL = "cell(nodes.fetch, 0, 'phone')"
+METRICS = [
+    {"id": "amt", "name": "首单金额", "unit": "元", "decimals": 1, "format": "thousands",
+     "expression": "cell(nodes.fetch, 0, 'amount')"},
+    {"id": "plen", "name": "号码位数", "unit": "位", "format": "thousands", "expression": f"len({PHONE_CELL})"},
+    {"id": "plen2", "name": "号码位数两倍", "unit": "位", "format": "thousands", "expression": f"len({FIRST_PHONE}) * 2"},
+    {"id": "mix", "name": "混合", "unit": "分", "decimals": 1, "format": "thousands",
+     "expression": f"round(cell(nodes.fetch, 1, 'amount') + len({SECOND_PHONE}), 1)"},
+]
+TEXT = "首单金额 [[m:amt]]，号码 [[m:plen]]，两倍 [[m:plen2]]，混合 [[m:mix]]。"
+
+
+@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
+
+
+async def set_source(path, options=None):
+    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)
+        row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
+        row.options = dict(options or {})
+        await session.commit()
+        return row.id
+
+
+@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, region TEXT, phone TEXT, amount REAL);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS)
+    db.commit()
+    db.close()
+    source_id = await set_source(path)
+    await engines.invalidate(source_id)
+    yield path
+    await engines.invalidate(source_id)
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def graph():
+    nodes = [
+        node("start", "input"),
+        node("fetch", "tool", tool="db_query__shop", args={"sql": SQL}),
+        node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"),
+        node("card", "metrics", caliber="遮罩口径", caliber_version="v1", metrics=METRICS),
+        node("write", "report", instructions="写一句", on_violation="flag"),
+        node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]),
+    ]
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+async def run_graph(monkeypatch) -> Run:
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=TEXT))
+    run_id = (await run_manager.start(graph=graph(), input_payload={})).id
+    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:
+            assert row.status == "succeeded", row.error
+            return row
+    raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
+
+
+async def segment(client, run_id: str, text: str) -> dict:
+    graph_body = (await client.get(f"/api/runs/{run_id}/evidence")).json()
+    [report] = graph_body["reports"]
+    doc = artifact_store.load(report["doc_artifact"])
+    seg = next(s for s in iter_segments(doc) if s["text"] == text)
+    res = await client.get(f"/api/runs/{run_id}/evidence/segments/{seg['id']}")
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+def inputs_of(body: dict) -> dict[str, dict]:
+    return {s["path"]: s for s in body["chain"] if s["step"] == "input"}
+
+
+def leaks(body: dict) -> list[str]:
+    raw = json.dumps(body, ensure_ascii=False)
+    return [p for p in PHONES if p in raw]
+
+
+# --------------------------------------------------------------------------
+
+
+async def test_a_masked_cell_is_hidden_from_the_input_label_and_the_substituted_expression(client, monkeypatch, shop):
+    await set_source(shop, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, "13位")
+    metric = body["chain"][0]
+    assert metric["step"] == "metric" and metric["expression"] == f"len({PHONE_CELL})"
+    assert metric["substituted"] == "len(已遮罩)"
+    [cell] = inputs_of(body).values()
+    assert cell["value"] == "已遮罩" and cell["masked"] == ["phone"] and cell["via"] == "tool_cell"
+    assert "snapshot_value" not in cell and "model_value" not in cell
+    # 那一格所在的查询步骤照旧遮着；面板底下的遮罩说明带上这一列
+    [query] = [s for s in body["chain"] if s["step"] == "query"]
+    assert query["masked"] == ["phone"] and body["redacted"]["columns"] == ["phone"]
+    assert leaks(body) == []
+
+
+async def test_a_scalar_taken_by_position_from_a_masked_column_is_hidden(client, monkeypatch, shop):
+    """vars.res.rows[0][2]：没有 cell() 的定位，顺着取值链认出取自查询结果的 phone 列。"""
+    await set_source(shop, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, "26位")
+    metric = body["chain"][0]
+    assert metric["substituted"] == "len(已遮罩) * 2"
+    [phone] = inputs_of(body).values()
+    assert phone["path"] == FIRST_PHONE and phone["value"] == "已遮罩" and phone["masked"] == ["phone"]
+    assert body["redacted"]["columns"] == ["phone"]
+    assert leaks(body) == []
+
+
+async def test_only_the_masked_input_is_hidden_when_inputs_are_mixed(client, monkeypatch, shop):
+    await set_source(shop, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, "93.0分")
+    assert body["chain"][0]["substituted"] == "round(80.0 + len(已遮罩), 1)"
+    inputs = inputs_of(body)
+    amount = inputs["cell(nodes.fetch, 1, 'amount')"]
+    assert amount["value"] == 80.0 and "masked" not in amount
+    assert inputs[SECOND_PHONE]["value"] == "已遮罩" and inputs[SECOND_PHONE]["masked"] == ["phone"]
+    assert leaks(body) == []
+    # 没用到遮罩列的指标和以前一样
+    plain = await segment(client, row.id, "120.5元")
+    assert plain["chain"][0]["substituted"] == "120.5"
+    [amt] = inputs_of(plain).values()
+    assert amt["value"] == 120.5 and "masked" not in amt
+
+
+async def test_masks_recorded_at_query_time_stay_after_the_source_drops_them(client, monkeypatch, shop):
+    await set_source(shop, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    await set_source(shop, {})
+    for text in ("13位", "26位", "93.0分"):
+        body = await segment(client, row.id, text)
+        assert "已遮罩" in body["chain"][0]["substituted"] and leaks(body) == [], text
+
+
+async def test_masks_added_after_the_run_apply_too(client, monkeypatch, shop):
+    row = await run_graph(monkeypatch)
+    before = await segment(client, row.id, "26位")
+    assert before["chain"][0]["substituted"] == "len('139-0000-0001') * 2"
+    assert inputs_of(before)[FIRST_PHONE]["value"] == "139-0000-0001" and "masked" not in inputs_of(before)[FIRST_PHONE]
+    await set_source(shop, {"mask_columns": "phone"})
+    after = await segment(client, row.id, "26位")
+    assert after["chain"][0]["substituted"] == "len(已遮罩) * 2" and leaks(after) == []
+
+
+async def test_without_masks_the_metric_step_is_unchanged(client, monkeypatch, shop):
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, "93.0分")
+    assert body["chain"][0]["substituted"] == "round(80.0 + len('139-0000-0002'), 1)"
+    assert all("masked" not in s for s in body["chain"] if s["step"] == "input")
+    assert body["redacted"] == {"columns": []}
+
+
+async def test_the_whatif_withholds_a_masked_scalar_taken_by_position(client, monkeypatch, shop):
+    """推演同一套：按位置取自遮罩列的输入不给值，整个指标推演不了，和证据面板说的一致。"""
+    await set_source(shop, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card", "metric": "plen2"})
+    assert res.status_code == 200, res.text
+    [plen2] = res.json()["metrics"]
+    assert plen2["status"] == "masked" and plen2["substituted"] is None
+    [phone] = plen2["inputs"]
+    assert phone["path"] == FIRST_PHONE and phone["masked"] is True and phone["old"] is None
+    assert not [p for p in PHONES if p in res.text]
+    # 改它的值同样不行：遮罩列不能拿来推演
+    res = await client.post(f"/api/runs/{row.id}/evidence/whatif",
+                            json={"card": "card", "metric": "plen2", "overrides": {"inputs": {FIRST_PHONE: 1}}})
+    assert res.status_code == 422 and "遮罩" in res.json()["detail"]
diff --git a/backend/tests/test_evidence_verify_status.py b/backend/tests/test_evidence_verify_status.py
new file mode 100644
index 00000000..e413faa8
--- /dev/null
+++ b/backend/tests/test_evidence_verify_status.py
@@ -0,0 +1,422 @@
+"""证据接口的核查状态（期 4 I2a）：口径卡指标那一步带双盲复算的结果，句子带反方审稿的判断。
+
+- 指标步骤 dual_check：{status, node_id, choice?, blind_rendered?, reason?}。status 是一致（agreed）、不一致（disagreed，
+  带审核人的选择，没选的不带）、无法复算（unchecked，带原因）、跳过（skipped：节点直通时 reason 是 disabled / no_inputs，
+  节点跑了、这个指标不在复算范围里时是 not_targeted）。读双盲复算节点封存的产出和它的台账条目，按台账里记的指标集工件
+  认它复算的是口径卡的哪一次结果；直通的没有台账条目，按画布上它复算的卡、和它跑完之前那张卡最后一次的结果认
+- 句子 unit.challenge：{verdict, node_id, artifact, on_refuted, report_label}。读反方审稿节点的产出和审稿记录，只认审的
+  就是这一份报告文档的；记录没存上时按画布上审的报告、和它跑完之前那份报告最后一次自查的文档认，判断取产出里列的那几句
+- 没有双盲复算、反方审稿的运行，答复里没有这两个键（老运行和以前一字不差）
+
+双盲复算、反方审稿各真跑一遍图（虚构的 SQLite 库，模型是 MockChatModel 剧本）；认「哪一次」的边界情形用手写的封存事件。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+import uuid
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage, ToolMessage
+from sqlalchemy import delete, select
+
+from app.api import evidence
+from app.core import artifact_store
+from app.core.artifact_store import put_json
+from app.data import introspect
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import Approval, DataSource, Run, Setting
+from app.engine import spend, verify_agent
+from app.engine.evidence import iter_segments, iter_units
+from app.engine.judge import SPEND_KEY as JUDGE_SPEND_KEY
+from app.engine.nodes import challenge as challenge_mod
+from app.engine.runner import run_manager
+from app.main import app
+from app.providers.mock_model import MockChatModel
+from tests.fixtures.dual_check_shop import DEFINITION, RIGHT, WITH_REFUNDS, Blind, add_source, build
+from tests.fixtures.sources import drop_source
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+
+    async def clear() -> None:
+        async with SessionLocal() as session:
+            await session.execute(delete(Setting).where(Setting.key.in_(
+                ["verify", "judge", spend.SPEND_KEY, JUDGE_SPEND_KEY])))
+            await session.commit()
+
+    await clear()
+    yield
+    await clear()
+    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, label=None, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
+
+
+def chain(*nodes):
+    return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) -> Run:
+    for _ in range(600):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None):
+            await run_manager.wait_idle(run_id)
+            return row
+    raise AssertionError(f"运行 {run_id} 停在 {row.status}：{row.error}")
+
+
+async def start(spec: dict, **kwargs) -> Run:
+    return await wait((await run_manager.start(graph=spec, input_payload={}, **kwargs)).id)
+
+
+async def report_of(client, run_id: str) -> dict:
+    body = (await client.get(f"/api/runs/{run_id}/evidence")).json()
+    [report] = body["reports"]
+    return artifact_store.load(report["doc_artifact"])
+
+
+async def open_segment(client, run_id: str, seg_id: str) -> dict:
+    res = await client.get(f"/api/runs/{run_id}/evidence/segments/{seg_id}")
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+async def segment(client, run_id: str, text: str) -> dict:
+    doc = await report_of(client, run_id)
+    return await open_segment(client, run_id, next(s for s in iter_segments(doc) if s["text"] == text)["id"])
+
+
+async def sentence(client, run_id: str, text: str | None = None, *, kind: str | None = None) -> dict:
+    """点开一整句（句末徽标）时证据面板请求的那个片段：这一句里第一个不是结构性的片段。按原文认句子，或者按句子的
+    种类认（标题）。"""
+    doc = await report_of(client, run_id)
+    md = doc["markdown"]
+    unit = next(u for _, u in iter_units(doc)
+                if (kind is not None and u.get("kind") == kind)
+                or (text is not None and md[u["span"][0]:u["span"][1]].strip() == text))
+    first = next(s for s in unit["segments"] if s["kind"] != "structural")
+    return await open_segment(client, run_id, first["id"])
+
+
+# --------------------------------------------------------------------------
+# 双盲复算：真跑
+# --------------------------------------------------------------------------
+
+GMV = {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands",
+       "expression": "cell(nodes.fetch, 0, 'gmv')"}
+#: 不在复算范围里的指标：没有来自查询的输入
+DAYS = {"id": "days", "name": "统计天数", "unit": "天", "format": "thousands", "expression": "7"}
+WRITER = "本周销售额 [[m:gmv]]，统计 [[m:days]]。"
+
+
+@pytest.fixture
+async def shop(tmp_path):
+    name, source_id = await add_source(build(tmp_path / "shop.db"))
+    yield name
+    await drop_source(source_id)
+
+
+def dual_graph(source: str, *, dual: bool = True, **config) -> dict:
+    """取数 → 口径卡 →（双盲复算）→ 报告 → 成果。"""
+    return chain(
+        node("start", "input"),
+        node("fetch", "tool", tool=f"db_query__{source}", args={"sql": RIGHT}),
+        node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=[GMV, DAYS]),
+        *([node("dual", "dual_check", "复核销售额", **{"definitions": {"gmv": DEFINITION}, **config})] if dual else []),
+        node("write", "report", "写周报", instructions="写一句周报", on_violation="flag"),
+        node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
+    )
+
+
+async def decide(client, run_id: str, choices: dict) -> Run:
+    async with SessionLocal() as session:
+        approval = (await session.execute(select(Approval).where(
+            Approval.run_id == run_id, Approval.status == "pending"))).scalar_one()
+    res = await client.post(f"/api/approvals/{approval.id}/decide",
+                            json={"approved": True, "value": {"choices": choices, "note": ""}})
+    assert res.status_code == 200, res.text
+    return await wait(run_id, ("succeeded", "failed"))
+
+
+async def test_an_agreeing_dual_check_shows_on_the_metric_step(client, monkeypatch, shop):
+    Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")}, writer=WRITER)
+    run = await start(dual_graph(shop))
+    assert run.status == "succeeded", run.error
+    step = (await segment(client, run.id, "330.0元"))["chain"][0]
+    assert step["step"] == "metric" and step["metric"] == "gmv"
+    assert step["dual_check"] == {"status": "agreed", "node_id": "dual", "blind_rendered": "330.0元"}
+    # 节点跑了、这个指标不在复算范围里（没有来自查询的输入）：照实说没复算，不是缺省
+    days = (await segment(client, run.id, "7天"))["chain"][0]
+    assert days["dual_check"] == {"status": "skipped", "node_id": "dual", "reason": "not_targeted"}
+
+
+@pytest.mark.parametrize("choice", ["b", None])
+async def test_a_disagreement_shows_what_the_reviewer_chose(client, monkeypatch, shop, choice):
+    Blind(monkeypatch, shop, {"销售额": (WITH_REFUNDS, "gmv")}, writer=WRITER)
+    run = await start(dual_graph(shop))
+    assert run.status == "interrupted", run.error
+    run = await decide(client, run.id, {"gmv": choice} if choice else {})
+    assert run.status == "succeeded", run.error
+    check = (await segment(client, run.id, "330.0元"))["chain"][0]["dual_check"]
+    assert check == {"status": "disagreed", "node_id": "dual", "blind_rendered": "380.0元",
+                     **({"choice": choice} if choice else {})}
+
+
+async def test_a_dual_check_turned_off_is_skipped(client, monkeypatch, shop):
+    Blind(monkeypatch, shop, {}, writer=WRITER)
+    run = await start(dual_graph(shop, enabled=False))
+    assert run.status == "succeeded", run.error
+    for text in ("330.0元", "7天"):
+        step = (await segment(client, run.id, text))["chain"][0]
+        assert step["dual_check"] == {"status": "skipped", "node_id": "dual", "reason": "disabled"}, text
+
+
+async def test_a_metric_the_blind_agent_could_not_compute_is_unchecked(client, monkeypatch, shop):
+    Blind(monkeypatch, shop, {}, writer=WRITER)
+    run = await start(dual_graph(shop))
+    assert run.status == "succeeded", run.error
+    check = (await segment(client, run.id, "330.0元"))["chain"][0]["dual_check"]
+    assert check["status"] == "unchecked" and check["node_id"] == "dual" and check["reason"]
+    assert "choice" not in check and "blind_rendered" not in check
+
+
+async def test_runs_without_verification_nodes_answer_as_before(client, monkeypatch, shop):
+    Blind(monkeypatch, shop, {}, writer=WRITER)
+    run = await start(dual_graph(shop, dual=False))
+    assert run.status == "succeeded", run.error
+    body = await segment(client, run.id, "330.0元")
+    assert "dual_check" not in body["chain"][0] and "challenge" not in body["unit"]
+
+
+# --------------------------------------------------------------------------
+# 反方审稿：真跑
+# --------------------------------------------------------------------------
+
+PRICED = "claude-sonnet-5"
+#: 各区域销售额（含退款）：华东 1500、华南 900、华北 400
+QUERY = "SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales GROUP BY region ORDER BY 销售额 DESC"
+NET = "SELECT region AS 区域, SUM(amount) AS 净销售额 FROM sales WHERE refunded = 0 GROUP BY region ORDER BY 净销售额 DESC"
+REFUNDS = "SELECT COUNT(*) AS 退款笔数 FROM sales WHERE refunded = 1"
+#: 一句有依据的结论、一句预埋的错误结论（华南其实低于华东），前面一个标题
+TEXT = "## 区域周报\n\n华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]华南销售额超过华东。[[see:Q1]]"
+EAST, SOUTH = "华东销售额 1,500，居各区域之首。", "华南销售额超过华东。"
+PLAYS = {
+    "华南销售额超过华东": ([NET], "refuted", "扣除退款后华南为 [[v:Q1.r1.净销售额]]，仍低于华东的 [[v:Q1.r0.净销售额]]。[[see:Q1]]"),
+    "居各区域之首": ([REFUNDS], "weakened", "华东含 [[v:Q1.r0.退款笔数]] 笔退款，扣除后仍居首，但销售额被高估。"),
+}
+
+
+@pytest.fixture
+async def sales(tmp_path):
+    path = tmp_path / f"{uuid.uuid4().hex[:8]}.db"
+    db = sqlite3.connect(path)
+    db.executescript(
+        "CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, amount INTEGER, refunded INTEGER);"
+        "INSERT INTO sales VALUES (1, '华东', 1200, 0), (2, '华东', 300, 1), (3, '华南', 900, 0), (4, '华北', 400, 0);")
+    db.commit()
+    db.close()
+    name = f"sales_{uuid.uuid4().hex[:8]}"
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, options={})
+        session.add(row)
+        await session.flush()
+        row.schema_cache = await introspect.introspect(row)
+        await session.commit()
+        source_id = row.id
+    yield name
+    await engines.invalidate(source_id)
+
+
+def reviewer(monkeypatch, source: str) -> None:
+    """写作者交 TEXT；反方 agent 按任务里的结论认出是哪一句，照 PLAYS 查询、交结论和理由。"""
+
+    def decide(model, messages):
+        if challenge_mod.CHALLENGE_SYSTEM not in str(messages[0].content):
+            return AIMessage(content=TEXT)
+        task = str(messages[1].content) if len(messages) > 1 else ""
+        queries, verdict, rationale = next((p for k, p in PLAYS.items() if k in task), ([], "stands", "没有找到反例。"))
+        if model.response_format:
+            return AIMessage(content=json.dumps({"verdict": verdict, "rationale": rationale}, ensure_ascii=False))
+        done = sum(1 for m in messages if isinstance(m, ToolMessage))
+        if done < len(queries):
+            call = {"name": f"db_query__{source}", "args": {"sql": queries[done]}, "id": f"q{done}-{uuid.uuid4().hex[:6]}"}
+            return AIMessage(content="查一下。", tool_calls=[call])
+        return AIMessage(content="查完了。")
+
+    monkeypatch.setattr(MockChatModel, "_decide", decide)
+
+    async def verify_model(session, spec):
+        return MockChatModel(model_name=PRICED, latency=0.0), PRICED
+
+    monkeypatch.setattr(verify_agent, "get_chat_model", verify_model)
+
+
+def challenge_graph(source: str, *, challenge: bool = True) -> dict:
+    return chain(
+        node("start", "input"),
+        node("fetch", "tool", tool=f"db_query__{source}", args={"sql": QUERY}),
+        node("write", "report", "写周报", instructions="写周报"),
+        *([node("chal", "challenge", "审稿", report="write")] if challenge else []),
+        node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
+             contract={"report_from": "write"}),
+    )
+
+
+async def challenge_output(run_id: str) -> dict:
+    from app.db.models import RunEvent
+
+    async with SessionLocal() as session:
+        [finished] = (await session.execute(select(RunEvent).where(
+            RunEvent.run_id == run_id, RunEvent.type == "node.finished", RunEvent.node_id == "chal"))).scalars()
+    return artifact_store.load(finished.data["artifact"])
+
+
+async def test_a_challenged_sentence_carries_the_verdict(client, monkeypatch, sales):
+    reviewer(monkeypatch, sales)
+    run = await start(challenge_graph(sales), run_class="formal")
+    assert run.status == "succeeded", run.error
+    record = (await challenge_output(run.id))["challenge_artifact"]
+    south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"]
+    assert south == {"verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "degrade",
+                     "report_label": "写周报"}
+    # 点开句子里的数字，片段接口答的这一句同样带着判断
+    east = (await segment(client, run.id, "1,500"))["unit"]["challenge"]
+    assert east["verdict"] == "weakened" and east["artifact"] == record
+    # 反方没审的（标题不是结论）：没有这个键
+    heading = await sentence(client, run.id, kind="heading")
+    assert heading["unit"]["kind"] == "heading" and "challenge" not in heading["unit"]
+
+
+async def test_a_lost_review_record_still_shows_the_listed_verdicts(client, monkeypatch, sales):
+    """审稿记录没存上（challenge_artifact 为空）：判断取产出里列的那几句，工件为空（理由和查询看不了）。"""
+    async def lost(self, record):
+        return None
+
+    monkeypatch.setattr(challenge_mod._Review, "_store", lost)
+    reviewer(monkeypatch, sales)
+    run = await start(challenge_graph(sales), run_class="formal")
+    assert run.status == "succeeded", run.error
+    assert (await challenge_output(run.id))["challenge_artifact"] is None
+    south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"]
+    assert south == {"verdict": "refuted", "node_id": "chal", "artifact": None, "on_refuted": "degrade",
+                     "report_label": "写周报"}
+
+
+async def test_reports_without_a_challenge_have_no_verdicts(client, monkeypatch, sales):
+    reviewer(monkeypatch, sales)
+    run = await start(challenge_graph(sales, challenge=False), run_class="formal")
+    assert run.status == "succeeded", run.error
+    assert "challenge" not in (await sentence(client, run.id, SOUTH))["unit"]
+
+
+# --------------------------------------------------------------------------
+# 认「哪一次」：手写的封存事件
+# --------------------------------------------------------------------------
+
+
+def sealed_of(graph: dict, events: list[tuple]) -> evidence._Sealed:
+    return evidence._Sealed(run_id="syn-run", graph=graph, events=events,
+                            seal={"sealed": True, "ok": True, "manifest_seq": events[-1][0], "legacy": False})
+
+
+def card_set(artifact: str) -> dict:
+    return {"evidence": [{"kind": "metric_set", "node_id": "card", "artifact": artifact}]}
+
+
+async def test_the_dual_check_belongs_to_the_card_result_it_checked():
+    """口径卡在循环里算了两次：双盲复算只复算了第一次的结果，第二次的指标没有复算状态；之后又跑了一次直通的，第二次的
+    结果记成跳过，第一次的不受影响。"""
+    first = await put_json({"kind": "metric_set", "metrics": [{"id": "gmv", "value": 1.5}], "n": 1}, kind="metric_set")
+    second = await put_json({"kind": "metric_set", "metrics": [{"id": "gmv", "value": 2.5}], "n": 2}, kind="metric_set")
+    checked = await put_json({"status": "agreed", "counts": {"agreed": 1, "disagreed": 0, "unchecked": 0},
+                              "metrics": [{"metric_id": "gmv", "name": "销售额", "unit": "元", "status": "agreed",
+                                           "original_rendered": "1.5元", "blind_rendered": "1.5元", "delta_pct": 0.0}],
+                              "model": "mock", "same_model": False}, kind="node_output")
+    skipped = await put_json({"status": "skipped", "reason": "disabled"}, kind="node_output")
+    graph = chain(node("card", "metrics"), node("dual", "dual_check", card="card"))
+    events = [
+        (1, "node.started", "card", {}),
+        (2, "node.finished", "card", card_set(first)),
+        (3, "node.started", "dual", {}),
+        (4, "node.finished", "dual", {"artifact": checked, "evidence": [{
+            "kind": "dual_check", "node_id": "dual", "exec": 1, "card": "card", "status": "agreed",
+            "counts": {"agreed": 1, "disagreed": 0, "unchecked": 0}, "metrics": [{"id": "gmv", "status": "agreed"}],
+            "metric_set": first}]}),
+        (5, "node.started", "card", {}),
+        (6, "node.finished", "card", card_set(second)),
+    ]
+    agreed = {"status": "agreed", "node_id": "dual", "blind_rendered": "1.5元"}
+    assert evidence._dual_check_of(sealed_of(graph, events), "card", first, "gmv") == agreed
+    assert evidence._dual_check_of(sealed_of(graph, events), "card", second, "gmv") is None
+    later = [*events, (7, "node.started", "dual", {}), (8, "node.finished", "dual", {"artifact": skipped})]
+    assert evidence._dual_check_of(sealed_of(graph, later), "card", second, "gmv") == {
+        "status": "skipped", "node_id": "dual", "reason": "disabled"}
+    assert evidence._dual_check_of(sealed_of(graph, later), "card", first, "gmv") == agreed
+    # 别的口径卡、卡上没有的指标：没有
+    assert evidence._dual_check_of(sealed_of(graph, events), "other", first, "gmv") is None
+    assert evidence._dual_check_of(sealed_of(graph, events), "card", first, "ghost") == {
+        "status": "skipped", "node_id": "dual", "reason": "not_targeted"}
+
+
+async def test_an_unreadable_dual_check_output_falls_back_to_its_ledger_entry():
+    """产出的文件不在了：状态和审核人的选择按台账条目说，盲算值说不了就不给。"""
+    card = await put_json({"kind": "metric_set", "metrics": [{"id": "gmv", "value": 3.5}], "n": 3}, kind="metric_set")
+    graph = chain(node("card", "metrics"), node("dual", "dual_check", card="card"))
+    events = [
+        (1, "node.finished", "card", card_set(card)),
+        (2, "node.finished", "dual", {"artifact": "f0" * 32, "evidence": [{
+            "kind": "dual_check", "node_id": "dual", "exec": 1, "card": "card", "status": "resolved",
+            "metrics": [{"id": "gmv", "status": "disagreed", "choice": "a"}], "metric_set": card}]}),
+    ]
+    assert evidence._dual_check_of(sealed_of(graph, events), "card", card, "gmv") == {
+        "status": "disagreed", "node_id": "dual", "choice": "a"}
+    assert evidence._dual_check_of(sealed_of(graph, events), "card", card, "other") == {
+        "status": "skipped", "node_id": "dual", "reason": "not_targeted"}
+
+
+def report(doc_artifact: str) -> evidence._Report:
+    return evidence._Report(node_id="write", doc_artifact=doc_artifact, ok=True, repairs=0, stats=None, doc=None,
+                            hash_ok=True, fields=[])
+
+
+@pytest.mark.parametrize("saved", [True, False])
+async def test_a_review_of_an_earlier_draft_does_not_mark_the_rewritten_report(saved):
+    """报告在反方审稿之后又重写了一次（循环、接着跑）：审的是前一稿，新的一稿上同一个句子编号说的可能是另一句话。
+    审稿记录存上了按记录里的文档认，没存上按它跑完之前那份报告最后一次自查的文档认。"""
+    draft = await put_json({"markdown": "第一稿", "n": 1}, kind="report_doc")
+    final = await put_json({"markdown": "第二稿", "n": 2}, kind="report_doc")
+    claims = [{"unit": "u1", "text": "华南销售额超过华东。", "verdict": "refuted"}]
+    record = await put_json({"report": "write", "doc_artifact": draft, "model": "mock",
+                             "claims": [{**c, "rationale": "", "rationale_doc": None, "queries": []} for c in claims]},
+                            kind="challenge") if saved else None
+    out = await put_json({"challenge_artifact": record, "counts": {"stands": 0, "weakened": 0, "refuted": 1,
+                                                                   "inconclusive": 0},
+                          "items": claims, "on_refuted": "flag"}, kind="node_output")
+    graph = chain(node("write", "report", "写周报"), node("chal", "challenge", "审稿", report="write"))
+    events = [
+        (1, "report.checked", "write", {"doc_artifact": draft, "ok": True}),
+        (2, "node.finished", "write", {}),
+        (3, "node.finished", "chal", {"artifact": out}),
+        (4, "report.checked", "write", {"doc_artifact": final, "ok": True}),
+    ]
+    sealed = sealed_of(graph, events)
+    assert evidence._challenge_of(sealed, report(draft), "u1") == {
+        "verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "flag", "report_label": "写周报"}
+    assert evidence._challenge_of(sealed, report(final), "u1") is None
+    assert evidence._challenge_of(sealed, report(draft), "u2") is None