证据接口复核整改（后端）：反方审稿带整份报告的真实计数、跑停运行的审稿照样认得，双盲复算的跳过原因按指标说，遮罩核对不了的宁可多遮

反方审稿：unit.challenge 带上节点产出里整份报告的计数（counts）和没能反查的句数（unreviewed，产出里有才带）。证据面板借审批卡的
展示画这一句，以前凭这一句拼出「被推翻 1 句、其余 0 句」，读着像整份报告的结果。节点设为「报错并停止运行」、句子被推翻时节点报错、
没有 node.finished，以前证据接口认不出这次审稿；现在按审完那一刻的 challenge_done 日志认（只认审稿记录存上了的，处置写 fail），
运行因此失败了也看得到被推翻的是哪几句。

双盲复算：节点以 no_inputs 直通时，那句说的是整张卡。以前原样套到卡上每个指标，配了「复算的指标」子集时，明明取自查询、只是没被
选上的指标也写成「口径卡上没有来自查询的输入」。现在按这个指标自己说：不在子集里、值不是数的是 not_targeted，自己没有来自查询的
输入的才是 no_inputs，关闭了的照旧 disabled；节点跑了、没复算的指标同样这样分。

遮罩：认哪个输入取自哪份查询结果的哪几列抽成 recompute.input_origin，证据面板、推演（之后还有可验证报告）同一个认法。核对不了的
宁可多遮，和推演、可验证报告的原则一致：cell()、Agent 字段落在的快照读不出来；按位置取的一格（以 .rows[行][列] 结尾）取值链上的
产出读不出来；那份结果没存快照、只带着数据源名。证据面板不给值，mask_unverified 写原因、不进遮罩列；推演按读不出来处理，不给值、
不推演。既没带快照 id 也没带数据源名的结果（数据整形改过形状）认不出出处，照常显示，写进交付说明。

recompute.recompute 给双盲复算的预检和盲值用，整个指标重算不了时原因不再说「推演」（_Plan.stuck，判断同 blocked，只换措辞）。
按位置的遮罩接进 recompute 之后，双盲复算的预检对「cell() 输入 + 按位置取自遮罩列的一格」的指标判无法复算（以前照常盲算，遮着的
号码还会作为字面量进盲算模型的计算式），运行对比里这种输入两边都不给值：各补一个用例钉住。

一个指标里同一件工件只读一次：按位置取了好几格时，写变量的节点那份可能几 MB 的产出以前每格重读一遍。

补测：遮罩列上 Agent 字段、cell() 对不上时 model_value / snapshot_value 不在答复里；代入式重新代不出来时整条不给。
test_recompute 的夹具补上数据整形节点 parse 的产出：以前缺着，按位置取的那一格现在会因此核对不了。
去掉实现后新用例各自变红（计数、跑停的审稿、子集的跳过原因、两处宁可多遮、推演、预检措辞、按位置遮罩、只读一次）。

backend/app/api/evidence.py

@@ -865,7 +865,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
         "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):
+    if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id, metric):
         step["dual_check"] = checked
     catalog = doc.get("catalog") or {}
     inputs: list[dict[str, Any]] = []
@@ -881,8 +881,8 @@ 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)
+        if (hidden := await hide.check(item)) is not None:
+            one = _withheld(one, *hidden)
             withheld.add(str(item.get("path")))
         inputs.append(one)
     if withheld:
@@ -892,23 +892,38 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
     return [step, *inputs, *queries]
 
 
-def _withheld(step: dict[str, Any], columns: list[str]) -> dict[str, Any]:
-    """取自遮罩列的输入：值写「已遮罩」（本来就没有值的照旧是空的，缺输入还得看得出来），模型报的值、快照里的值一并
-    拿掉，masked 写明是哪几列。"""
+def _withheld(step: dict[str, Any], columns: list[str], unverified: str | None) -> dict[str, Any]:
+    """不给值的输入：值写「已遮罩」（本来就没有值的照旧是空的，缺输入还得看得出来），模型报的值、快照里的值一并拿掉。
+    取自遮罩列的，masked 写明是哪几列；遮没遮核对不了的（_InputMasks），mask_unverified 写原因、不进 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
+    if unverified is None:
+        out["masked"] = columns
+    else:
+        out["mask_unverified"] = unverified
     return out
 
 
-class _InputMasks:
-    """一个指标的输入里哪些取自遮罩列。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询
-    按它合并的那几个数据源算；快照读不出来时只能按数据源现在设的算。
+#: tool_cell、Agent 字段落在的查询快照读不出来：只知道数据源现在设的遮罩，查询当时记下的在快照里，遮没遮核对不了
+SNAPSHOT_UNVERIFIED = "查询快照无法读取，核对不了这一格是否取自设置了遮罩的列"
 
-    cell() 的格、Agent 交来的字段按定位里的列认。vars.res.rows[0][2] 这种按位置取自查询结果的标量卡上没记定位：顺着
-    取值链找到那份结果（recompute.positional_cell，推演用同一个认法），结果里带着快照的 id 和数据源。取值链从写变量的
-    节点在口径卡算这一次之前的产出起走（_card_outputs）。认不出来的不遮——卡上本来就只记着路径和值。
+
+class _InputMasks:
+    """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询按它
+    合并的那几个数据源算。
+
+    取自哪份查询结果的哪几列按 recompute.input_origin 认，推演、可验证报告是同一个认法：cell() 的格、Agent 交来的字段按
+    定位里的列认；vars.res.rows[0][2] 这种按位置取的一格卡上没记定位，顺着取值链认，从写变量的节点在口径卡算这一次之前
+    的产出起走（_card_outputs）。
+
+    遮没遮核对不了的宁可多遮，和推演一致：快照读不出来（数据源现在没遮这一列，可查询当时记下的遮罩在快照里）、按位置取的
+    一格取值链上的产出读不出来、那份结果没存快照只带着数据源名。认得出、可结果既没带快照的 id 也没带数据源名的（数据整形
+    改过形状）不遮：认不出出处，卡上本来就只记着路径和值。
+
+    一个指标里同一件工件只读一次（_fetch 记在 _read）：按位置取了十几格的指标，不必把写变量的节点那几 MB 的产出重读
+    十几遍。查询步骤（_query_step）另读一遍快照：它的接口好几条链都在用，读两遍比读十几遍是小事。
     """
 
     def __init__(self, sealed: _Sealed, masks: _Masks, card_node: str, artifact: str) -> None:
@@ -916,34 +931,45 @@ class _InputMasks:
         self._card, self._artifact = card_node, artifact
         self._outputs: dict[str, str] | None = None
         self._memo: dict[tuple[str | None, str | None], set[str]] = {}
+        self._read: dict[str, tuple[Any, bool | None]] = {}
 
-    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"]]
+    def _fetch(self, artifact: str) -> tuple[Any, bool | None]:
+        if artifact not in self._read:
+            self._read[artifact] = _fetch(artifact)
+        return self._read[artifact]
+
+    async def check(self, item: dict[str, Any]) -> tuple[list[str], str | None] | None:
+        """None：给值。(遮罩列, None)：取自这几列遮罩列（列名照原样，按出现的先后）。(用到的列, 原因)：遮没遮核对不了，
+        原因给人看（认不出是哪一列时列为空）。"""
+        origin = recompute.input_origin(item, self._output)
+        if origin is None:
+            return None
+        if origin.unverified is not None:
+            return list(origin.columns), origin.unverified
+        used = list(dict.fromkeys(origin.columns))
         if not used:
-            return []
-        hidden = await self._hidden(artifact, source)
-        return list(dict.fromkeys(c for c in used if c.lower() in hidden))
+            return None
+        snap, ok = self._fetch(origin.artifact) if origin.artifact else (None, None)
+        readable = ok is True and isinstance(snap, dict)
+        hidden = await self._hidden(origin.artifact, origin.source, snap if readable else None)
+        if masked := [c for c in used if c.lower() in hidden]:
+            return masked, None
+        if origin.artifact and not readable:
+            return used, SNAPSHOT_UNVERIFIED
+        if not origin.artifact and origin.source:
+            return used, recompute.POSITION_NO_SNAPSHOT
+        return None
 
     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)
+        content, ok = self._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]:
+    async def _hidden(self, artifact: str | None, source: str | None, snap: dict[str, Any] | 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:
@@ -1013,8 +1039,10 @@ def _spec_of(sealed: _Sealed) -> GraphSpec | None:
         return None
 
 
-def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: Any) -> dict[str, Any] | None:
+def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: Any,
+                   metric: dict[str, Any] | None = None) -> dict[str, Any] | None:
     """这个指标的双盲复算状态：{status, node_id, choice?, blind_rendered?, reason?}。没有双盲复算复核过这份指标集的是 None。
+    metric 是卡上的这个指标（认它为什么没被复算用），读不到卡时为 None。
 
     认的是复核了口径卡的哪一次结果：台账条目（kind dual_check）记着它复算的指标集工件，和指标步骤的工件对上才算——口径卡
     在循环里算了几次时，报告引用的那一次才是。直通了的（设置里关了、口径卡上没有来自查询的输入）没有台账条目：按画布上
@@ -1022,8 +1050,9 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An
 
     - agreed / disagreed / unchecked：节点产出里这个指标的结果。不一致的带审核人的选择（a / b / neither，没选的不带）；
       一致、不一致的带盲算值（渲染好的）；无法复算的带原因
-    - skipped：节点直通，reason 同节点产出（disabled / no_inputs）；或者节点跑了、这个指标不在复算范围里（没选它、值不是
-      数、没有来自查询的输入），reason 为 not_targeted
+    - skipped：节点没复算这个指标，reason 说为什么（_not_checked）：关闭了（disabled，节点直通）；不在节点配置的
+      「复算的指标」里、或者值不是数（not_targeted）；这个指标没有一个输入来自查询快照（no_inputs）。节点直通时产出里的
+      no_inputs 说的是整张卡，不原样套到每个指标上：卡上确有来自查询的指标、只是没被选上的，写它的说法是 not_targeted
     产出读不出来（文件不在、与哈希不一致）时按台账条目说状态和选择，不带盲算值和原因。
     """
     if not card_node or not artifact or metric_id is None:
@@ -1032,6 +1061,7 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An
     if not checkers:
         return None
     spec: GraphSpec | None = None
+    spec_read = False
     #: 口径卡节点 → 到这一刻为止它最后一次的指标集
     latest: dict[str, str] = {}
     found: dict[str, Any] | None = None
@@ -1047,25 +1077,54 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An
         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 not spec_read:
+            spec, spec_read = _spec_of(sealed), True
+        node = spec.node_map().get(node_id) if spec else 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))
+        found = _dual_status(node_id, output, entry, str(metric_id), metric, _picked(node))
     return found
 
 
-def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | None, metric_id: str) -> dict[str, Any]:
+def _picked(node: Any) -> list[str] | None:
+    """画布上这个双盲复算节点配的「复算的指标」，认法同 dual_check._options：没配、配了空列表都是 None（复算全部）。"""
+    wanted = node.config.get("metrics") if node is not None else None
+    picked = [str(m) for m in wanted if isinstance(m, str) and m.strip()] if isinstance(wanted, list) else []
+    return picked or None
+
+
+def _query_input(item: Any) -> bool:
+    """来自查询快照的输入（同 dual_check._from_query）：双盲复算只复算有这种输入的指标。"""
+    return isinstance(item, dict) and item.get("via") in ("tool_cell", "agent_field") \
+        and isinstance(item.get("artifact"), str) and bool(item["artifact"])
+
+
+def _not_checked(node_reason: str, metric: dict[str, Any] | None, picked: list[str] | None, metric_id: str) -> str:
+    """节点没复算这个指标的原因。关闭了（disabled）照节点的说；别的按这个指标自己说：不在配置的「复算的指标」里是
+    not_targeted；卡上读得到它、它没有一个输入来自查询快照是 no_inputs，有（那就是值不是数）是 not_targeted。读不到卡时
+    照节点直通的原因说，节点跑了的说 not_targeted。"""
+    if node_reason == "disabled":
+        return node_reason
+    if picked is not None and metric_id not in picked:
+        return "not_targeted"
+    if metric is not None:
+        return "not_targeted" if any(_query_input(i) for i in metric.get("inputs") or []) else "no_inputs"
+    return node_reason or "not_targeted"
+
+
+def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | None, metric_id: str,
+                 metric: dict[str, Any] | None, picked: list[str] | None) -> dict[str, Any]:
     """一次双盲复算里这个指标的状态（_dual_check_of 的答复）。"""
     if output.get("status") == "skipped":
-        return {"status": "skipped", "node_id": node_id, "reason": str(output.get("reason") or "")}
+        reason = _not_checked(str(output.get("reason") or ""), metric, picked, metric_id)
+        return {"status": "skipped", "node_id": node_id, "reason": reason}
     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:
@@ -1075,7 +1134,7 @@ def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | N
         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"}
+        return {"status": "skipped", "node_id": node_id, "reason": _not_checked("", metric, picked, metric_id)}
     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"):
@@ -1088,14 +1147,26 @@ def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | N
     return out
 
 
+#: 反方审稿的四种判断（engine/nodes/challenge.py 的 VERDICTS）：审稿概况的计数照这个顺序
+_CHALLENGE_VERDICTS = ("stands", "weakened", "refuted", "inconclusive")
+
+
 def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, Any] | None:
-    """这一句的反方审稿判断：{verdict, node_id, artifact, on_refuted, report_label}。没审到这一句的是 None。
+    """这一句的反方审稿判断：{verdict, node_id, artifact, on_refuted, report_label, counts, unreviewed?}。没审到这一句的
+    是 None。
 
     只认审的就是正文这一份报告文档的：审稿记录（challenge_artifact）里记着它审的报告节点和文档工件，和这份对上才算——
     报告在审稿之后又重写了（循环、接着跑），同一个句子编号在新的一稿里说的可能是另一句话。记录没存上（产出里的
     challenge_artifact 为空）或读不出来时，按画布上它审的报告、和它跑完之前那份报告最后一次自查记下的文档认，判断取
     产出里列的那几句（最多 10 句，被推翻、被削弱的排在前面），artifact 照产出写（没存上的为空）。几个反方审稿审了同一
-    份文档时取最后跑完、审到了这一句的那个。on_refuted 是实际生效的处置（degrade / flag），report_label 是报告节点的名字。
+    份文档时取最后跑完、审到了这一句的那个。on_refuted 是实际生效的处置，report_label 是报告节点的名字。
+
+    counts 是这次审稿四种判断各几句——整份报告的，取节点产出里的那一份；unreviewed 是没能反查的句数，产出里有才带。
+    证据面板借审批卡上的展示画这一句，那一块的概况得是这份报告真实的数，不能凭这一句拼出一个「被推翻 0 句」。
+
+    节点设为「报错并停止运行」、又有句子被推翻时，节点报错、没有 node.finished：审完那一刻的 challenge_done 日志记着
+    审稿记录和计数，按它认（只认审稿记录存上了的），on_refuted 为 fail。运行因此失败了，被推翻的是哪几句、反方凭什么，
+    证据面板里照样看得到。
     """
     reviewers = set(sealed.nodes_of("challenge"))
     if not reviewers or not unit_id:
@@ -1103,6 +1174,8 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A
     spec: GraphSpec | None = None
     #: 报告节点 → 到这一刻为止它最后一次自查记下的文档
     drafts: dict[str, str] = {}
+    #: 反方审稿节点 → 这一次执行里审完时的日志：节点报错时它就是审稿结果
+    done: dict[str, dict[str, Any]] = {}
     found: dict[str, Any] | None = None
     for _, etype, node_id, data in sealed.events:
         if not node_id or not isinstance(data, dict):
@@ -1110,17 +1183,34 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A
         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:
+        if node_id not in reviewers:
+            continue
+        if etype == "node.started":
+            done.pop(node_id, None)
+            continue
+        if etype == "log" and data.get("code") == "challenge_done":
+            done[node_id] = data
+            continue
+        if etype == "node.finished":
+            done.pop(node_id, None)
+            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    # 直通了（关闭、没有可审的句子）或者产出读不出来
+        elif etype == "node.failed" and node_id in done:
+            log = done.pop(node_id)
+            if not isinstance(log.get("artifact"), str) or not log["artifact"]:
+                continue
+            output = {"challenge_artifact": log["artifact"], "counts": log.get("counts"), "on_refuted": "fail"}
+        else:
             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")
+        elif etype == "node.failed":
+            continue    # 节点报错时没有产出可退：认不出审的是哪一稿
         else:
             spec = spec or _spec_of(sealed)
             node = spec.node_map().get(node_id) if spec else None
@@ -1128,17 +1218,33 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A
             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)
+        claims = [c for c in claims or [] if isinstance(c, dict)]
+        verdict = next((c.get("verdict") for c in claims if 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
+        unreviewed = output.get("unreviewed")
         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}
+                 "report_label": writer.title if writer else report.node_id,
+                 "counts": _challenge_counts(output.get("counts"), claims),
+                 **({"unreviewed": unreviewed} if isinstance(unreviewed, int) and not isinstance(unreviewed, bool)
+                    and unreviewed >= 0 else {})}
     return found
 
 
+def _challenge_counts(raw: Any, claims: list[dict[str, Any]]) -> dict[str, int]:
+    """四种判断各几句：取节点产出（或审完时的日志）里的计数；一个数都认不出来时按认得的句子数（审稿记录里的是全部，
+    记录没存上时只有产出里列的那几句）。"""
+    given = raw if isinstance(raw, dict) else {}
+    known = {v: given[v] for v in _CHALLENGE_VERDICTS
+             if isinstance(given.get(v), int) and not isinstance(given[v], bool) and given[v] >= 0}
+    if known:
+        return {v: known.get(v, 0) for v in _CHALLENGE_VERDICTS}
+    return {v: sum(1 for c in claims if c.get("verdict") == v) for v in _CHALLENGE_VERDICTS}
+
+
 class _Masks:
     """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。"""
 

backend/app/engine/recompute.py

@@ -61,6 +61,13 @@ NOT_RESTORABLE = "这类输入当时没有保存完整值"
 MASKED_INPUT = "这一列在数据源中设置了遮罩，不能用于推演"
 MASKED_TABLE = "这份查询结果里有设置了遮罩的列，不能用于推演"
 MASKED_METRIC = "这个指标用到了设置了遮罩的列，不能推演"
+#: 按位置取自查询结果的一格（vars.res.rows[0][2]）遮没遮核对不了：取值链上节点的产出读不出来，或者那份结果没存快照、
+#: 只带着数据源名（查询当时记下的遮罩在快照里）。和读不出来的快照一样宁可多遮（_positional）
+POSITION_UNREADABLE = "查询结果无法读取，核对不了这一格是否取自设置了遮罩的列"
+POSITION_NO_SNAPSHOT = "查询结果没有保存快照，核对不了这一格是否取自设置了遮罩的列"
+#: recompute（双盲复算的预检、盲值）里整个指标重算不了的说法：同 _Plan.blocked，只是不说「推演」——那里不是推演
+NO_LINEAGE_PLAIN = "这个指标生成于系统升级前，口径卡没有记录输入"
+MASKED_PLAIN = "这个指标用到了设置了遮罩的列"
 SELECTOR = "这个输入决定取哪一行、哪一项，修改它需要重新取数，不能推演"
 EXCLUDED = "排除后无法计算"
 UNREADABLE = "查询快照无法读取"
@@ -170,6 +177,8 @@ class _Plan:
     tops: list[str] = field(default_factory=list)
     inputs: dict[str, _Input] = field(default_factory=dict)
     problem: str | None = None
+    #: problem 不说「推演」的说法（recompute 给双盲复算用）；和 problem 一样时不填
+    plain: str | None = None
     #: 嵌在别的引用里的输入 → 包着它的最外层引用（_refs）
     within: dict[str, set[str]] = field(default_factory=dict)
 
@@ -200,6 +209,20 @@ class _Plan:
                 return "unsupported", f"无法推演：{path}（{problem}）"
         return None
 
+    @property
+    def stuck(self) -> str | None:
+        """blocked 的原因换成不说「推演」的说法：recompute 交给双盲复算的预检、盲值，那里说「不能推演」驴唇不对马嘴。
+        判断和先后与 blocked 一字不差，只换措辞；推演得了的是 None。"""
+        blocked = self.blocked
+        if blocked is None:
+            return None
+        if self.problem:
+            return self.plain or self.problem
+        if blocked[0] == "masked":
+            return MASKED_PLAIN
+        path = next(p for p in self.tops if self.inputs[p].problem)
+        return f"输入 {path} 无法还原（{self.inputs[path].problem}）"
+
 
 # --------------------------------------------------------------------------
 # 入口
@@ -232,8 +255,14 @@ def whatif(card: dict[str, Any], changes: Changes, sources: Sources, *, metric:
 
 
 def recompute(metric: dict[str, Any], changes: Changes, sources: Sources) -> dict[str, Any]:
-    """重算一个指标，不管改动碰没碰到它，也不校验改动。核对还原是否忠实时用：什么都不改，重算出来应当和卡上记的一样。"""
-    return _outcome(_plan(metric, _Store(sources)), changes, sources, force=True)[0]
+    """重算一个指标，不管改动碰没碰到它，也不校验改动。核对还原是否忠实时用：什么都不改，重算出来应当和卡上记的一样。
+
+    调用方是双盲复算（预检、按盲算的输入代回原式），不是推演：整个指标重算不了时原因不说「推演」（_Plan.stuck）。"""
+    plan = _plan(metric, _Store(sources))
+    result = _outcome(plan, changes, sources, force=True)[0]
+    if (plain := plan.stuck) is not None:
+        result["reason"] = plain
+    return result
 
 
 # --------------------------------------------------------------------------
@@ -245,7 +274,7 @@ def _plan(metric: dict[str, Any], store: _Store) -> _Plan:
     expr = str(metric.get("expression") or "").strip()
     items = metric.get("inputs")
     if not expr or not isinstance(items, list):
-        return _Plan(metric, problem=NO_LINEAGE)
+        return _Plan(metric, problem=NO_LINEAGE, plain=NO_LINEAGE_PLAIN)
     try:
         tree, _, _ = parse_expression(expr)
     except ExpressionError as e:
@@ -258,7 +287,8 @@ def _plan(metric: dict[str, Any], store: _Store) -> _Plan:
             inputs[item["path"]] = _restore(item, remaining, set(within), store)
     plan = _Plan(metric, tree, tops, inputs, within=within)
     if lost := [p for p in tops if p not in inputs]:
-        plan.problem = f"口径卡缺少输入 {'、'.join(lost[:3])} 的记录，不能推演"
+        plan.plain = f"口径卡缺少输入 {'、'.join(lost[:3])} 的记录"
+        plan.problem = f"{plan.plain}，不能推演"
     return plan
 
 
@@ -335,7 +365,9 @@ 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) or _positional_masked(inp, store)
+    inp.masked = _masked(inp, locator, store)
+    if not inp.masked:
+        _positional(inp, store)
     return inp
 
 
@@ -417,14 +449,56 @@ 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、取值链走不通）不算：和以前一样照常推演。"""
+def _positional(inp: _Input, store: _Store) -> None:
+    """按位置取自查询结果的一格（vars.res.rows[0][2]）：卡上没记定位，顺着取值链找到那份结果、认出是哪一列（input_origin）。
+
+    落在遮罩列上的按遮罩（masked）。遮没遮核对不了的——取值链上节点的产出读不出来，或者那份结果没存快照、只带着数据源名
+    ——和读不出来的快照同一个处理：值不给、也不拿来推演（unreadable，原因记在 problem），宁可多遮。认得出、可那份结果
+    既没带快照的 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
+        return
+    origin = input_origin(inp.item, lambda node_id: store.output(node_id)[0])
+    if origin is None:
+        return
+    if origin.unverified is None and origin.artifact:
+        hidden = store.sources.masked.get(origin.artifact)
+        inp.masked = bool(hidden) and any(c.lower() in hidden for c in origin.columns)
+        return
+    if origin.unverified is not None or origin.source:
+        inp.unreadable = True
+        inp.problem = inp.problem or origin.unverified or POSITION_NO_SNAPSHOT
+
+
+@dataclass(frozen=True)
+class Origin:
+    """一个输入取自哪份查询结果的哪几列（input_origin）。
+
+    - artifact、source：那份查询结果的快照 id 和数据源名，结果里没带的是 None
+    - columns：用到的列（列名照原样）。按位置取的一格、取值链上的产出读不出来时认不出是哪一列，为空
+    - unverified：遮没遮核对不了的原因（取值链上节点的产出读不出来）；核对得了的是 None
+    """
+
+    artifact: str | None
+    source: str | None
+    columns: tuple[str, ...]
+    unverified: str | None = None
+
+
+def input_origin(item: dict[str, Any], output: Callable[[Any], Any]) -> Origin | None:
+    """卡上的一个输入取自哪份查询结果的哪几列，遮罩按列认时用（推演、证据面板的指标步骤、可验证报告是同一个认法）。
+
+    cell() 的格、Agent 交来的字段按定位里的列认；vars.res.rows[0][2] 这种按位置取的一格顺着取值链认（positional_cell，
+    output 同它）。不是取自查询结果的格（常量、运行输入、Agent 没按出处交的字段、加工过的数）是 None。"""
+    if item.get("via") in ("tool_cell", "agent_field") and isinstance(item.get("artifact"), str) and item["artifact"]:
+        locator = item.get("locator") if isinstance(item.get("locator"), dict) else {}
+        columns = [locator.get("column")] if locator.get("column") else list((locator.get("columns") or {}).values())
+        return Origin(item["artifact"], None, tuple(c for c in columns if isinstance(c, str) and c))
+    found = positional_cell(str(item.get("path") or ""), item.get("node_id"), output)
+    if found is None:
+        return None
+    if found.get("unreadable"):
+        return Origin(None, None, (), POSITION_UNREADABLE)
+    return Origin(found["artifact"], found["source"], (found["column"],))
 
 
 #: 取值链没走通
@@ -441,6 +515,10 @@ def positional_cell(path: str, node_id: Any, output: Callable[[Any], Any]) -> di
     output(节点 id) 交回那个节点当时的产出（读不出来给 None）。nodes.X 的根就是 X 的产出；vars.X 是写它的节点（输入记着
     的 node_id）交给变量的那个值——数据整形、合并查询交的是产出本身，Agent、LLM、代码节点交的是产出里的 data，工具节点
     交的是 JSON 文本，都试一遍，取值链走得通的那个就是。下标得是字面量：rows[vars.i] 要另取一个值才知道是哪一行，不认。
+
+    路径就是按位置取一格的样子（以 .rows[行号][列] 结尾），可那个节点的产出读不出来（文件不在、与哈希不一致、不知道是
+    谁写的）时返回 {"unreadable": True}：这一格取自哪一列核对不了，调用方宁可多遮。卡上记着这个输入的值，算的时候
+    取值链是走得通的，现在读不出来只能是产出丢了或被改了。
     """
     try:
         tree = ast.parse(path, mode="eval").body
@@ -461,6 +539,9 @@ def positional_cell(path: str, node_id: Any, output: Callable[[Any], Any]) -> di
             or not isinstance(keys[0], str):
         return None
     base = output(keys[0] if tree.id == "nodes" else node_id)
+    if base is None:
+        positional = keys[-3] == "rows" and isinstance(keys[-2], int) and isinstance(keys[-1], (int, str))
+        return {"unreadable": True} if positional else None
     candidates = [base]
     if isinstance(base, dict) and "data" in base:
         candidates.append(base["data"])

backend/tests/test_evidence_metric_masks.py

@@ -231,3 +231,186 @@ async def test_the_whatif_withholds_a_masked_scalar_taken_by_position(client, mo
     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"]
+
+
+# --------------------------------------------------------------------------
+# 手搭的封存范围：模型报的值、快照里的值，以及遮没遮核对不了的几种
+# --------------------------------------------------------------------------
+
+#: 一份带遮罩的快照：phone 在查询当时就设了遮罩（记在快照里）。数据源名是编的，库里没有这个数据源（现在的遮罩为空）
+GHOST = "ghost_mask_src"
+PHONE_SNAP = {"columns": ["id", "phone"], "rows": [[1, "139-0000-0009"]], "source": GHOST,
+              "sql": "SELECT id, phone FROM members", "mask_columns": ["phone"]}
+#: 这几条用例里出现过的号码，答复里一个都不能有
+SECRETS = ["139-0000-0009", "139-0000-0008", "139-0000-0007", "139-0000-0006", "139-0000-0005", "139-0000-0004"]
+CELL = "cell(nodes.fetch, 0, 'phone')"
+
+
+def card_metric(mid, expression, substituted, inputs, value=13):
+    return {"id": mid, "name": mid, "unit": "位", "value": value, "rendered": f"{value}位", "format": "thousands",
+            "expression": expression, "substituted": substituted, "recompute_ok": True, "inputs": inputs,
+            "status": "ok"}
+
+
+async def hand_chain(metrics, *, queries=(), outputs=None, metric_id=None):
+    """手搭一次封存了的运行（取数 → 数据整形 → 口径卡），直接走指标步骤（evidence._metric_chain）。queries 是封存范围内
+    交回过的查询快照，outputs 是口径卡算之前各节点的产出工件。"""
+    from app.api import evidence
+    from app.core.artifact_store import put_json
+    from app.engine.evidence import make_eid
+
+    card = await put_json({"kind": "metric_set", "caliber": "遮罩口径", "caliber_version": "v1", "metrics": metrics},
+                          kind="metric_set")
+    events = [(i + 1, "node.finished", nid, {"artifact": art}) for i, (nid, art) in enumerate((outputs or {}).items())]
+    events += [(len(events) + 1, "node.started", "card", {}),
+               (len(events) + 2, "node.finished", "card", {"evidence": [
+                   {"kind": "metric_set", "node_id": "card", "artifact": card}]})]
+    sealed = evidence._Sealed(run_id="syn-mask", graph=graph(), events=events,
+                              seal={"sealed": True, "ok": True, "manifest_seq": len(events), "legacy": False})
+    sealed.queries = {a: {"node_id": "fetch", "tool": f"db_query__{GHOST}", "source": GHOST} for a in queries}
+    sealed.ledger = {("card", card)}
+    mid = metric_id or metrics[0]["id"]
+    eid = make_eid("metric", card, {"metric": mid})
+    cite = {"alias": f"m:{mid}", "kind": "metric", "locator": {"metric": mid}, "status": "resolved", "eid": eid}
+    entry = {"alias": f"m:{mid}", "kind": "metric", "artifact": card, "node_id": "card", "locator": {"metric": mid},
+             "eid": eid}
+    seg = {"id": "s1", "kind": "number", "text": "13位", "state": "deterministic", "cite": cite}
+    return await evidence._metric_chain(seg, cite, entry, {"catalog": {f"m:{mid}": entry}}, sealed, evidence._Masks())
+
+
+def said(chain) -> list[str]:
+    raw = json.dumps(chain, ensure_ascii=False)
+    return [p for p in SECRETS if p in raw]
+
+
+async def test_what_the_model_reported_and_what_the_snapshot_held_are_withheld_too():
+    """对不上的输入另记着两边的值：Agent 字段的 model_value（模型报的）、cell() 的 snapshot_value（快照里的）。取自遮罩列
+    时一并拿掉——界面上不显示它们，接口答复里同样不能带着。"""
+    from app.core.artifact_store import put_json
+
+    snap = await put_json(PHONE_SNAP, kind="query")
+    cell = card_metric("plen", f"len({CELL})", "len('139-0000-0008')", [
+        {"path": CELL, "value": "139-0000-0008", "node_id": "fetch", "via": "tool_cell", "artifact": snap,
+         "locator": {"row": 0, "column": "phone"}, "status": "mismatch", "snapshot_value": "139-0000-0009"}])
+    agent = card_metric("alen", "len(vars.kpi.phone)", "len('139-0000-0009')", [
+        {"path": "vars.kpi.phone", "value": "139-0000-0009", "node_id": "bot", "via": "agent_field", "field": "phone",
+         "artifact": snap, "locator": {"row": 0, "column": "phone"}, "status": "mismatch",
+         "model_value": "139-0000-0007"}])
+    for metric in (cell, agent):
+        chain = await hand_chain([metric], queries=[snap])
+        [step] = [s for s in chain if s["step"] == "input"]
+        assert step["value"] == "已遮罩" and step["masked"] == ["phone"], metric["id"]
+        assert "model_value" not in step and "snapshot_value" not in step, metric["id"]
+        assert chain[0]["substituted"] == "len(已遮罩)" and said(chain) == [], (metric["id"], said(chain))
+
+
+async def test_a_substituted_expression_that_does_not_reproduce_is_not_shown():
+    """卡上记的代入式和按卡上的输入重新代出来的对不上（老卡、卡上记的不是原值）：整条代入式不给，宁可不显示，也不能让
+    遮着的值从对不上的那一处漏出来。"""
+    from app.core.artifact_store import put_json
+    from app.engine import recompute
+
+    snap = await put_json(PHONE_SNAP, kind="query")
+    odd = card_metric("odd", f"len({CELL})", "len('139-0000-0006')", [
+        {"path": CELL, "value": "139-0000-0005", "node_id": "fetch", "via": "tool_cell", "artifact": snap,
+         "locator": {"row": 0, "column": "phone"}, "status": "verified"}])
+    assert recompute.substituted_without(odd, {CELL}, "已遮罩") is None
+    chain = await hand_chain([odd], queries=[snap])
+    assert chain[0]["substituted"] is None and said(chain) == [], said(chain)
+    # 对得上的照常换掉那一处
+    assert recompute.substituted_without({**odd, "substituted": "len('139-0000-0005')"}, {CELL}, "已遮罩") == "len(已遮罩)"
+
+
+async def test_a_cell_on_a_snapshot_that_cannot_be_read_is_withheld():
+    """cell() 落在的快照读不出来（文件不在了）：数据源现在没遮这一列，可查询当时记下的遮罩在快照里，遮没遮核对不了——
+    宁可多遮，和推演一致。说法另写，不说「这一列设置了遮罩」（不进 masked）。"""
+    gone = "dd" * 32
+    metric = card_metric("plen", f"len({CELL})", "len('139-0000-0004')", [
+        {"path": CELL, "value": "139-0000-0004", "node_id": "fetch", "via": "tool_cell", "artifact": gone,
+         "locator": {"row": 0, "column": "phone"}, "status": "verified"}])
+    chain = await hand_chain([metric], queries=[gone])
+    [step] = [s for s in chain if s["step"] == "input"]
+    assert step["value"] == "已遮罩" and "masked" not in step
+    assert step["mask_unverified"] == "查询快照无法读取，核对不了这一格是否取自设置了遮罩的列"
+    assert chain[0]["substituted"] == "len(已遮罩)" and said(chain) == []
+
+
+PARSED = {"columns": ["id", "phone"], "rows": [[1, "139-0000-0004"]]}
+POSITIONAL = "vars.res.rows[0][1]"
+
+
+@pytest.mark.parametrize("parsed, why", [
+    # 写变量的节点的产出读不出来：取值链走不通，那一格取自哪一列核对不了
+    (None, "查询结果无法读取，核对不了这一格是否取自设置了遮罩的列"),
+    # 结果没存快照、只带着数据源名：查询当时记下的遮罩在快照里，核对不了
+    ({**PARSED, "source": GHOST}, "查询结果没有保存快照，核对不了这一格是否取自设置了遮罩的列"),
+])
+async def test_a_positional_cell_that_cannot_be_traced_is_withheld(parsed, why):
+    from app.core.artifact_store import put_json
+
+    out = await put_json(parsed, kind="node_output") if parsed is not None else "ee" * 32
+    metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [
+        {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]",
+         "status": "ok"}], value=26)
+    chain = await hand_chain([metric], outputs={"parse": out})
+    [step] = [s for s in chain if s["step"] == "input"]
+    assert step["value"] == "已遮罩" and step["mask_unverified"] == why and "masked" not in step
+    assert chain[0]["substituted"] == "len(已遮罩) * 2" and said(chain) == []
+
+
+async def test_a_reshaped_result_without_a_snapshot_or_a_source_is_shown_as_is():
+    """数据整形改过形状的结果（既没带快照的 id，也没带数据源名）认不出出处：照常显示。已知的缺口，写在交付说明里。"""
+    from app.core.artifact_store import put_json
+
+    out = await put_json(PARSED, kind="node_output")
+    metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [
+        {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]",
+         "status": "ok"}], value=26)
+    chain = await hand_chain([metric], outputs={"parse": out})
+    [step] = [s for s in chain if s["step"] == "input"]
+    assert step["value"] == "139-0000-0004" and "mask_unverified" not in step and "masked" not in step
+
+
+@pytest.mark.parametrize("parsed, withheld", [
+    (None, True),
+    ({**PARSED, "source": GHOST}, True),
+    (PARSED, False),
+])
+def test_the_whatif_withholds_the_same_untraceable_cells(parsed, withheld):
+    """推演同一个认法：取值链走不通、结果没存快照的那一格不给值、不推演；认不出出处的照常推演。"""
+    from app.engine import recompute
+
+    store = {"out": parsed}
+    metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [
+        {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]",
+         "status": "ok"}], value=26)
+    sources = recompute.Sources(load=store.get, outputs={"parse": "out"} if parsed is not None else {})
+    [result] = recompute.whatif({"metrics": [metric]}, recompute.Changes(), sources)["metrics"]
+    [inp] = result["inputs"]
+    if withheld:
+        assert inp["old"] is None and inp.get("withheld") is True and result["status"] == "unsupported"
+        assert result["substituted"] is None and "139-0000-0004" not in json.dumps(result, ensure_ascii=False)
+    else:
+        assert inp["old"] == "139-0000-0004" and result["status"] == "unchanged"
+
+
+async def test_the_writer_output_is_read_once_for_all_cells_taken_by_position(monkeypatch):
+    """一个指标按位置取了好几格：写变量的节点的产出（可能是几 MB 的查询结果）一次请求里只读一次，快照也只读一次。"""
+    from app.api import evidence
+    from app.core.artifact_store import put_json
+
+    snap = await put_json({**PHONE_SNAP, "rows": [[1, "139-0000-0009"], [2, "139-0000-0008"]]}, kind="query")
+    out = await put_json({"columns": ["id", "phone"], "rows": [[1, "139-0000-0009"], [2, "139-0000-0008"]],
+                          "artifact": snap, "source": GHOST}, kind="node_output")
+    paths = ["vars.res.rows[0][1]", "vars.res.rows[1][1]", "vars.res.rows[0][0]"]
+    metric = card_metric("many", " + ".join(f"len(str({p}))" for p in paths), "len('139-0000-0009') + 0", [
+        {"path": p, "value": v, "node_id": "parse", "via": "transform", "field": p[9:], "status": "ok"}
+        for p, v in zip(paths, ["139-0000-0009", "139-0000-0008", 1])], value=27)
+    reads: list[str] = []
+    real = evidence.load
+    monkeypatch.setattr(evidence, "load", lambda artifact: (reads.append(artifact), real(artifact))[1])
+    chain = await hand_chain([metric], outputs={"parse": out})
+    inputs = {s["path"]: s for s in chain if s["step"] == "input"}
+    assert inputs[paths[0]]["masked"] == ["phone"] and inputs[paths[1]]["masked"] == ["phone"]
+    assert inputs[paths[2]]["value"] == 1 and "masked" not in inputs[paths[2]]
+    assert reads.count(out) == 1 and reads.count(snap) == 1, reads

backend/tests/test_evidence_verify_status.py

@@ -1,11 +1,15 @@
 """证据接口的核查状态（期 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}。读反方审稿节点的产出和审稿记录，只认审的
-  就是这一份报告文档的；记录没存上时按画布上审的报告、和它跑完之前那份报告最后一次自查的文档认，判断取产出里列的那几句
+  带审核人的选择，没选的不带）、无法复算（unchecked，带原因）、跳过（skipped）。跳过的原因按这个指标自己说：关闭了
+  （disabled）；不在节点配置的「复算的指标」里、或值不是数（not_targeted）；这个指标没有来自查询的输入（no_inputs）——
+  节点直通时产出里的 no_inputs 说的是整张卡，卡上确有来自查询的指标、只是没被选上的不能也说成没有。读双盲复算节点封存的
+  产出和它的台账条目，按台账里记的指标集工件认它复算的是口径卡的哪一次结果；直通的没有台账条目，按画布上它复算的卡、
+  和它跑完之前那张卡最后一次的结果认
+- 句子 unit.challenge：{verdict, node_id, artifact, on_refuted, report_label, counts, unreviewed?}。读反方审稿节点的产出
+  和审稿记录，只认审的就是这一份报告文档的；记录没存上时按画布上审的报告、和它跑完之前那份报告最后一次自查的文档认，
+  判断取产出里列的那几句。counts 是整份报告的审稿计数（节点产出里的那一份），证据面板借审批卡的展示画这一句时，概况那
+  一行不能凭这一句拼。节点设为「报错并停止运行」、句子被推翻、运行因此失败时，按审完那一刻的日志认
 - 没有双盲复算、反方审稿的运行，答复里没有这两个键（老运行和以前一字不差）
 
 双盲复算、反方审稿各真跑一遍图（虚构的 SQLite 库，模型是 MockChatModel 剧本）；认「哪一次」的边界情形用手写的封存事件。
@@ -161,9 +165,21 @@ async def test_an_agreeing_dual_check_shows_on_the_metric_step(client, monkeypat
     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"}
+    assert days["dual_check"] == {"status": "skipped", "node_id": "dual", "reason": "no_inputs"}
+
+
+async def test_a_metric_left_out_of_the_subset_is_not_said_to_lack_query_inputs(client, monkeypatch, shop):
+    """「复算的指标」只选了统计天数：它没有来自查询的输入，节点以 no_inputs 直通。销售额明明取自查询，只是没被选上——
+    它的说法是不在复算范围内，不能套用节点那句「没有来自查询的输入」。"""
+    Blind(monkeypatch, shop, {}, writer=WRITER)
+    run = await start(dual_graph(shop, metrics=["days"]))
+    assert run.status == "succeeded", run.error
+    gmv = (await segment(client, run.id, "330.0元"))["chain"][0]["dual_check"]
+    assert gmv == {"status": "skipped", "node_id": "dual", "reason": "not_targeted"}
+    days = (await segment(client, run.id, "7天"))["chain"][0]["dual_check"]
+    assert days == {"status": "skipped", "node_id": "dual", "reason": "no_inputs"}
 
 
 @pytest.mark.parametrize("choice", ["b", None])
@@ -291,13 +307,16 @@ async def test_a_challenged_sentence_carries_the_verdict(client, monkeypatch, sa
     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"]
+    output = await challenge_output(run.id)
+    record = output["challenge_artifact"]
     south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"]
+    # counts 是整份报告的审稿计数（节点产出里的那一份）：一句被推翻、一句被削弱，不是只数这一句
     assert south == {"verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "degrade",
-                     "report_label": "写周报"}
+                     "report_label": "写周报", "counts": {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0}}
+    assert south["counts"] == output["counts"]
     # 点开句子里的数字，片段接口答的这一句同样带着判断
     east = (await segment(client, run.id, "1,500"))["unit"]["challenge"]
-    assert east["verdict"] == "weakened" and east["artifact"] == record
+    assert east["verdict"] == "weakened" and east["artifact"] == record and east["counts"] == output["counts"]
     # 反方没审的（标题不是结论）：没有这个键
     heading = await sentence(client, run.id, kind="heading")
     assert heading["unit"]["kind"] == "heading" and "challenge" not in heading["unit"]
@@ -315,7 +334,22 @@ async def test_a_lost_review_record_still_shows_the_listed_verdicts(client, monk
     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": "写周报"}
+                     "report_label": "写周报", "counts": {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0}}
+
+
+async def test_a_review_that_stopped_the_run_still_marks_its_sentences(client, monkeypatch, sales):
+    """「结论被推翻时」设为报错并停止运行：节点报错、没有 node.finished，运行失败。被推翻的是哪几句、反方凭什么，
+    证据面板里照样要看得到——按审完那一刻的日志（challenge_done）认审稿记录，处置写 fail。"""
+    reviewer(monkeypatch, sales)
+    graph = challenge_graph(sales)
+    next(n for n in graph["nodes"] if n["id"] == "chal")["data"]["config"]["on_refuted"] = "fail"
+    run = await start(graph, run_class="formal")
+    assert run.status == "failed" and "推翻" in (run.error or ""), run.error
+    south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"]
+    assert south["verdict"] == "refuted" and south["on_refuted"] == "fail" and south["node_id"] == "chal"
+    assert south["artifact"] and south["counts"] == {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0}
+    east = (await segment(client, run.id, "1,500"))["unit"]["challenge"]
+    assert east["verdict"] == "weakened" and east["artifact"] == south["artifact"]
 
 
 async def test_reports_without_a_challenge_have_no_verdicts(client, monkeypatch, sales):
@@ -417,6 +451,54 @@ async def test_a_review_of_an_earlier_draft_does_not_mark_the_rewritten_report(s
     ]
     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": "写周报"}
+        "verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "flag", "report_label": "写周报",
+        "counts": {"stands": 0, "weakened": 0, "refuted": 1, "inconclusive": 0}}
     assert evidence._challenge_of(sealed, report(final), "u1") is None
     assert evidence._challenge_of(sealed, report(draft), "u2") is None
+
+
+async def test_the_counts_are_the_whole_reviews_and_unreviewed_rides_along():
+    """counts 照产出（整份报告审了 12 句），没能反查的句数产出里有就带上；产出里的计数认不出来时按审稿记录数。"""
+    doc = await put_json({"markdown": "周报", "n": 3}, kind="report_doc")
+    claims = [{"unit": "u1", "text": "华南销售额超过华东。", "verdict": "weakened"},
+              {"unit": "u2", "text": "华北持平。", "verdict": "stands"}]
+    record = await put_json({"report": "write", "doc_artifact": doc, "model": "mock",
+                             "claims": [{**c, "rationale": "", "rationale_doc": None, "queries": []} for c in claims]},
+                            kind="challenge")
+    graph = chain(node("write", "report", "写周报"), node("chal", "challenge", "审稿", report="write"))
+
+    async def answer(output: dict) -> dict:
+        out = await put_json({"challenge_artifact": record, "items": claims, "on_refuted": "degrade", **output},
+                             kind="node_output")
+        events = [(1, "report.checked", "write", {"doc_artifact": doc, "ok": True}),
+                  (2, "node.finished", "chal", {"artifact": out})]
+        return evidence._challenge_of(sealed_of(graph, events), report(doc), "u1")
+
+    got = await answer({"counts": {"stands": 7, "weakened": 3, "refuted": 2, "inconclusive": 0}, "unreviewed": 1})
+    assert got["counts"] == {"stands": 7, "weakened": 3, "refuted": 2, "inconclusive": 0} and got["unreviewed"] == 1
+    # 计数格式不对（一个数都认不出来）：按审稿记录里的句子数，不写四个 0
+    got = await answer({"counts": {"stands": "7"}, "unreviewed": True})
+    assert got["counts"] == {"stands": 1, "weakened": 1, "refuted": 0, "inconclusive": 0} and "unreviewed" not in got
+
+
+async def test_a_review_that_failed_the_node_needs_its_record():
+    """节点报错：认审完那一刻的日志。重跑（node.started）之后没再审完的不算；日志里没有审稿记录（保存失败）认不出审的
+    是哪一稿，也不算。"""
+    doc = await put_json({"markdown": "周报", "n": 4}, kind="report_doc")
+    claims = [{"unit": "u1", "text": "华南销售额超过华东。", "verdict": "refuted", "rationale": "", "rationale_doc": None,
+               "queries": []}]
+    record = await put_json({"report": "write", "doc_artifact": doc, "model": "mock", "claims": claims},
+                            kind="challenge")
+    graph = chain(node("write", "report", "写周报"), node("chal", "challenge", "审稿", report="write",
+                                                       on_refuted="fail"))
+    counts = {"stands": 0, "weakened": 0, "refuted": 1, "inconclusive": 0}
+    head = [(1, "report.checked", "write", {"doc_artifact": doc, "ok": True}), (2, "node.started", "chal", {})]
+    done = (3, "log", "chal", {"code": "challenge_done", "counts": counts, "artifact": record})
+    failed = (4, "node.failed", "chal", {"error": "反方审稿推翻了 1 句结论"})
+    got = evidence._challenge_of(sealed_of(graph, [*head, done, failed]), report(doc), "u1")
+    assert got == {"verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "fail",
+                   "report_label": "写周报", "counts": counts}
+    retried = [*head, done, (4, "node.started", "chal", {}), (5, "node.failed", "chal", {"error": "模型不可用"})]
+    assert evidence._challenge_of(sealed_of(graph, retried), report(doc), "u1") is None
+    lost = [*head, (3, "log", "chal", {"code": "challenge_done", "counts": counts, "artifact": None}), failed]
+    assert evidence._challenge_of(sealed_of(graph, lost), report(doc), "u1") is None

backend/tests/test_positional_masks_elsewhere.py

@@ -0,0 +1,113 @@
+"""按位置取自遮罩列的一格（vars.res.rows[1][2]、nodes.fetch.rows[0][0]）在推演之外的两处：期 4 I2a 把它接进了
+recompute 的还原（_positional），用 recompute 的地方都跟着变，这里把变化钉住。
+
+- 双盲复算的预检（dual_check._plan_metric 什么都不改先重算一遍）：以前这种指标照常盲算，遮着的那一格还会作为字面量
+  写进给盲算模型的计算式；现在预检判它用到了遮罩列，按无法复算记，原因不说「推演」（这里不是推演）
+- 运行对比（engine/compare.py）：这种输入两边的值都不给，标 masked，和 cell() 取的遮罩格一样
+
+库和数字都是编的：号码写成带横线的文本，区域名是「东区」这类。
+"""
+from __future__ import annotations
+
+import json
+
+from app.engine import compare
+from app.engine.runner import run_manager
+from tests.fixtures.compare_runs import BASE_ROWS, NEXT_ROWS, metric, scalar_in, week
+from tests.test_evidence_metric_masks import (  # noqa: F401 - 夹具按名字注入（engine_up 自动生效）
+    METRICS,
+    PHONES,
+    SQL,
+    client,
+    engine_up,
+    node,
+    set_source,
+    shop,
+)
+
+
+def dual_graph():
+    """取数 → 数据整形（vars.res）→ 口径卡 → 双盲复算（只复算 mix）→ 成果。"""
+    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("dual", "dual_check", metrics=["mix"]),
+        node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.card.text }}"}]),
+    ]
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+async def dual_output(run_id: str) -> dict:
+    from sqlalchemy import select
+
+    from app.core import artifact_store
+    from app.db.base import SessionLocal
+    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 == "dual"))).scalars()
+    return artifact_store.load(finished.data["artifact"])
+
+
+async def test_the_dual_check_does_not_blind_compute_a_metric_reading_a_masked_cell_by_position(shop):
+    """mix = round(cell(nodes.fetch, 1, 'amount') + len(vars.res.rows[1][2]), 1)：cell() 那一格不在遮罩列上，按位置取的
+    号码在。预检判它用到了遮罩列，按无法复算记缺口，原因不说「推演」；没遮罩时预检照常通过、指标进入盲算。"""
+    import asyncio
+
+    from app.db.base import SessionLocal
+    from app.db.models import Run
+
+    async def run(options):
+        await set_source(shop, options)
+        run_id = (await run_manager.start(graph=dual_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:
+                await run_manager.wait_idle(run_id)
+                assert row.status == "succeeded", row.error
+                return await dual_output(run_id)
+        raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
+
+    masked = await run({"mask_columns": ["phone"]})
+    [mix] = masked["metrics"]
+    assert mix["metric_id"] == "mix" and mix["status"] == "unchecked"
+    assert mix["reason"] == "无法按原表达式重算这个指标（这个指标用到了设置了遮罩的列）"
+    assert "推演" not in json.dumps(masked, ensure_ascii=False)
+    assert not [p for p in PHONES if p in json.dumps(masked, ensure_ascii=False)]
+    assert any("遮罩" in gap for gap in masked["issuance_gaps"])
+
+    # 没遮罩：预检通过，原因不再是遮罩（盲算模型没配，按盲算没交回数记，但已经走到了盲算这一步）
+    plain = await run({})
+    [mix] = plain["metrics"]
+    assert "遮罩" not in (mix.get("reason") or "") and "无法按原表达式重算" not in (mix.get("reason") or "")
+
+
+def region_len(fetch, bot, rows, bot_rows):
+    """东区名称的字数：按位置取自取数结果的第 0 行第 0 列（region）。"""
+    return [metric("rlen", "len(nodes.fetch.rows[0][0])", len(rows[0][0]),
+                   [scalar_in("nodes.fetch.rows[0][0]", rows[0][0], "fetch", "tool")], name="首个区域名字数",
+                   unit="字", decimals=0)]
+
+
+def test_the_compare_page_withholds_a_masked_cell_read_by_position():
+    """运行对比：按位置取自遮罩列（region）的一格两边都不给值、标 masked，和 cell() 取的遮罩格一样；没遮罩时照给。"""
+    def masked(run):
+        run.masked = {a: frozenset({"region"}) for a in run.queries}
+        return run
+
+    result = compare.compare(masked(week("r1", metrics=region_len)).side(),
+                             masked(week("r2", NEXT_ROWS, metrics=region_len)).side())
+    [rlen] = [m for m in result["metrics"] if m["metric_id"] == "rlen"]
+    [inp] = rlen["inputs"]
+    assert inp["masked"] is True and inp["base"] is None and inp["target"] is None
+    assert "东区" not in json.dumps(result, ensure_ascii=False)
+
+    shown = compare.compare(week("r1", metrics=region_len).side(), week("r2", NEXT_ROWS, metrics=region_len).side())
+    [rlen] = [m for m in shown["metrics"] if m["metric_id"] == "rlen"]
+    [inp] = rlen["inputs"]
+    assert not inp.get("masked") and inp["base"] == BASE_ROWS[0][0] and inp["target"] == NEXT_ROWS[0][0]

backend/tests/test_recompute.py

@@ -29,6 +29,9 @@ BOT_OUT = "d4" * 32
 FETCH_OUT = "e5" * 32
 ITEMS_VALUE = "f6" * 32
 RES_ROWS = "a7" * 32
+#: 数据整形节点 parse 解析好的 S1（数据源工具交回的结果原样解析，带着快照的 id）：vars.res 就是它。按位置取的
+#: vars.res.rows[0][1] 顺着它认得出取自 S1 的 amount 列；没有这份产出时那一格遮没遮核对不了，推演不给值
+PARSE_OUT = "c9" * 32
 STORE = {
     **SNAPSHOTS,
     BOT_OUT: {"text": "四个区域合计 340", "data": {"amounts": [120.5, 80.0, 99.5, 40.0]},
@@ -37,6 +40,7 @@ STORE = {
     FETCH_OUT: {"items": ["甲", "乙", "丙"], "k": 2},
     ITEMS_VALUE: {"path": "vars.kpi.items", "value": ["甲", "乙", "丙"]},
     RES_ROWS: {"path": "vars.res.rows", "value": SNAPSHOTS[S1]["rows"]},
+    PARSE_OUT: {**SNAPSHOTS[S1], "artifact": S1},
 }
 
 
@@ -47,7 +51,7 @@ def load(artifact):
 
 
 def sources(**kw):
-    base = {"load": load, "queries": (S1, S2, BAD), "outputs": {"bot": BOT_OUT, "fetch": FETCH_OUT},
+    base = {"load": load, "queries": (S1, S2, BAD), "outputs": {"bot": BOT_OUT, "fetch": FETCH_OUT, "parse": PARSE_OUT},
             "masked": {S2: frozenset({"phone"})}}
     return Sources(**{**base, **kw})