运行对比复核修正（第 3 轮）：行号相同的引用还要引用同一批维度值，记不出写入者时只看口径卡之前的节点，默认设置的登记按节点类型逐类核对

- Agent 两次引用的行号相同、都不是整份结果时，一边引用了的维度值（区域……）在另一边的快照里却没被引用（按区域排序多了北区、按金额取前 3 名名次变了），判为说不清，说明写出是哪个维度值、在第几行；不再把还在快照里的区域算成「数据没了」、给出远超 100% 的占比。另一边快照里真的没有了的照常是数据变化。数组字段里的单个元素同样核对。维度按两边整份结果认，遮罩列不拿来认，分不出维度的仍按行号认 - 两边都记不出写入者时，范围从整张图改为口径卡之前可能执行过的节点（它的上游加并行支路）：口径卡自己、只在它下游的写周报写不了它读到的值，不再被点名成来源；并行支路上的节点不说成「上游」 - 「整份结果」统一按引用了哪几行判断，容器一支不再单列 - 默认设置登记表的核对测试改为按节点类型：从各执行器出发顺着用到的函数（含就地 import、回调、类）找 cfg 读法，在字面量元组上循环读的键名认得出，认不出的变量键名直接失败；核对节点模块里每个读配置的函数都走到了，登记了不读的也算错。据此补登记成果节点出具契约读的报告节点三项（evidence_from、metrics_from、entities） - 补测试守住上一轮的几条规则：Agent 交的不同而代码节点的数没变是没变，只有这次运行的快照读不出来时说这次，默认设置同时改了结论句裁判时 Agent 只写模型，调不调模型由默认设置决定，报告节点在代码节点上游，改了几个节点时点名画布上靠前的
2026-11-15 05:41:57 +00:00 · 2026-10-05 12:28:38 +08:00
parent 3254744f6d
commit 36569e1fa4
2 changed files with 417 additions and 73 deletions
@@ -23,10 +23,12 @@
    结果时，再按别名一路追到喂它的每一份上游快照。只差字面量时还要看图里的 SQL：图里的 SQL 改过（改了写死的筛选值、
    系数）是模板变化，图里一字未改的才是运行时代入的参数
  - Agent 按出处交来的字段还要引用同一列、同一段行（两边都引用整份结果的不算换了；行段相同、一边是整份结果一边
-    不是的算换了）：取哪一列、哪几行是 Agent 每次运行时定的，换了就不是同一个数
+    不是的算换了；行号相同、都不是整份结果的，一边引用了的维度值在另一边的快照里、却没被引用的也算换了——取前几名
+    时名次变了、按区域排序时多了一个区域）：取哪一列、哪几行是 Agent 每次运行时定的，换了就不是同一个数
  - 不落在快照上的：模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）没有出处，值变了说不清是数据还是它
-    本身的差异；代码、数据整形这类节点看它自己查过的库，不查库的看它的配置和画布上的全部上游。上游的查询结果两次
-    一模一样、值却变了的同样说不清；上游有模型、人交出值的节点、两次交出的不同的也说不清
+    本身的差异；代码、数据整形这类节点看它自己查过的库，不查库的看它的配置和画布上的全部上游；记不出写入者的看
+    口径卡之前可能执行过的节点（口径卡自己、只在它下游的不算）。上游的查询结果两次一模一样、值却变了的同样说不清；
+    上游有模型、人交出值的节点、两次交出的不同的也说不清
  - 图级的变化落到节点上：连线有增删的，终点节点算改了；默认设置改了的，运行时会读这几项（_CFG_KEYS）、又没有自行
    设置的节点算改了
 - 严格拆分（只对口径变了的）：旧表达式的每个叶子路径都能在这次的口径卡输入里取得值时，
@@ -97,10 +99,12 @@ _MODEL_KEYS = frozenset({"provider", "model", "temperature", "max_tokens", "thin
 #: 各类节点运行时按 NodeContext.cfg 读的配置项：节点上没写这一项就回退到图级默认设置（GraphSpec.defaults），默认
 #: 设置改了这一项，没自行设置它的这类节点交出的值可能跟着变；不读它的不受影响（结论句裁判只有报告节点读）。读了也
 #: 不影响交出的值的见 _QUIET_DEFAULTS。合并查询、双盲复算、反方审稿不经 cfg 读配置。执行器新读了一项要在这里登记，
-#: tests/test_compare_engine.py 按源码核对
+#: tests/test_compare_engine.py 按各类节点的执行器（连同它用到的函数）的源码逐类核对
 _CFG_KEYS: dict[str, frozenset[str]] = {
    "input": frozenset({"fields"}),
-    "output": frozenset({"fields", "contract"}),
+    # 出具契约按报告节点同一套参数重建证据目录：后三项读的是报告节点上的，报告节点没写同样回退到默认设置。成果节点
+    # 自己从不写这三项，默认设置改了就算它跟着变（报告节点自己写了的也算，宁可多算）
+    "output": frozenset({"fields", "contract", "evidence_from", "metrics_from", "entities"}),
    "transform": frozenset({"mode", "expression", "template", "assign_to"}),
    "llm": _MODEL_KEYS | {"system", "skills", "use_history", "prompt", "output_schema", "emit_message", "assign_to"},
    "agent": _MODEL_KEYS | {"system", "skills", "use_history", "prompt", "output_schema", "cite_fields", "tools",
@@ -691,6 +695,20 @@ def _worst(changes: Iterable[_Change]) -> _Change:
    return max(found, key=lambda c: _WEIGHT[c.kind]) if found else _Change("same")


+def _reach(node_id: str, links: Mapping[str, set[str]], memo: dict[str, set[str]]) -> set[str]:
+    """顺着连线（links：节点 → 上一层或下一层的节点）能走到的全部节点，不含它自己（成环的除外）。"""
+    if node_id not in memo:
+        seen: set[str] = set()
+        todo = [node_id]
+        while todo:
+            for nxt in links.get(todo.pop(), ()):
+                if nxt not in seen:
+                    seen.add(nxt)
+                    todo.append(nxt)
+        memo[node_id] = seen
+    return memo[node_id]
+
+
 def _from_input(path: str, item: dict[str, Any]) -> bool:
    head = _REF_HEAD.match(path)
    return item.get("via") == "input" or (head is not None and head.group(1) == "input")
@@ -709,6 +727,7 @@ class _Compare:
        self._norm: dict[tuple[str, str | None], NormalizedSql] = {}
        self._shapes: dict[tuple[int, int], _Shape] = {}
        self._ancestors: dict[str, set[str]] = {}
+        self._descendants: dict[str, set[str]] = {}
        self.base_nodes, self.target_nodes = _nodes(base.graph), _nodes(target.graph)
        self.base_queries, self.target_queries = _queries(base), _queries(target)
        self.pairs = _pair(self.base_queries, self.target_queries, self._norm_of)
@@ -719,8 +738,10 @@ class _Compare:
        self.order = {n: i for i, n in enumerate(dict.fromkeys([*self.target_nodes, *self.base_nodes]))}
        self.base_outputs, self.target_outputs = _outputs(base), _outputs(target)
        self.parents: dict[str, set[str]] = {}
+        self.children: dict[str, set[str]] = {}
        for source, target_id, _ in _edges(base.graph) | _edges(target.graph):
            self.parents.setdefault(target_id, set()).add(source)
+            self.children.setdefault(source, set()).add(target_id)

    # ---------------------------------------------------------------- 结果

@@ -921,24 +942,26 @@ class _Compare:

    def _upstream(self, node_id: str) -> set[str]:
        """画布上 node_id 的全部上游（两次运行的图合起来，宁可多算）。"""
-        if node_id not in self._ancestors:
-            seen: set[str] = set()
-            todo = [node_id]
-            while todo:
-                for parent in self.parents.get(todo.pop(), ()):
-                    if parent not in seen:
-                        seen.add(parent)
-                        todo.append(parent)
-            self._ancestors[node_id] = seen
-        return self._ancestors[node_id]
+        return _reach(node_id, self.parents, self._ancestors)
+
+    def _downstream(self, node_id: str) -> set[str]:
+        """画布上 node_id 的全部下游（两次运行的图合起来）。"""
+        return _reach(node_id, self.children, self._descendants)
+
+    def _before(self, card: str) -> set[str]:
+        """口径卡执行之前可能执行过的节点：画布上它的上游，加上和它不在一条线上的（并行的支路可能先写了变量）。它自己、
+        只在它下游的节点在它之后才执行，写不了它读到的值（循环回边连回来的算上游）。"""
+        after = self._downstream(card) - self._upstream(card)
+        return {n for n in self.titles if n != card and n not in after}

    def _type_of(self, node_id: str) -> str | None:
        node = self.target_nodes.get(node_id) or self.base_nodes.get(node_id)
        return str(node.get("type")) if node is not None else None

-    def _moved(self, node_id: str) -> str:
-        """一个影响取值的节点变化（self.changed）写成一句话。"""
-        name = f"上游节点「{self.title(node_id)}」"
+    def _moved(self, node_id: str, *, upstream: bool = True) -> str:
+        """一个影响取值的节点变化（self.changed）写成一句话。upstream 是它在不在画布上的上游（并行支路上的节点不在，
+        不说成「上游」）。"""
+        name = f"{'上游节点' if upstream else '节点'}「{self.title(node_id)}」"
        change = self.changed.get(node_id, "")
        if change == "defaults":
            keys = self.inherited.get(node_id, [])
@@ -959,9 +982,9 @@ class _Compare:
        before, after = self.base_outputs.get(node_id, ()), self.target_outputs.get(node_id, ())
        return before != after or None in before or None in after

-    def provenance(self, path: str, b: Any, t: Any, *, moved: bool = True) -> _Change:
-        """一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入）。moved 是这个指标的值变没变：输入是
-        容器、完整值又还原不出来，核对不了它变没变时，按它推断。"""
+    def provenance(self, path: str, b: Any, t: Any, *, card: str, moved: bool = True) -> _Change:
+        """一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入，card 是读它的口径卡节点）。moved 是这个
+        指标的值变没变：输入是容器、完整值又还原不出来，核对不了它变没变时，按它推断。"""
        if b is None or t is None:
            return _Change("unknown", f"输入「{path}」只在一边的口径卡里有记录")
        if b.snapshot or t.snapshot:
@@ -992,21 +1015,30 @@ class _Compare:
            return own
        if writer and writer in self.changed:
            return _Change("template", self._moved(writer))
-        # 写它的节点不查库（代码、数据整形……）：看画布上它的全部上游；写它的节点不知道就看整张图
-        scope = self._upstream(writer) if writer else set(self.titles)
+        # 写它的节点不查库（代码、数据整形……）：看画布上它的全部上游。写它的节点不知道时，看口径卡之前可能执行过的
+        # 节点（_before）：口径卡自己、只在它下游的节点（写周报……）写不了它读到的值，点名它们就把下游的变化说成了
+        # 它的来源。above 是点名时算「上游」的节点
+        if writer:
+            scope = above = self._upstream(writer)
+        else:
+            scope, above = self._before(card), self._upstream(card)
        moved_nodes = self._ordered(n for n in scope if n in self.changed)
        worst = _worst(self._query_change(b2, t2) for b2, t2 in self.pairs if (t2 or b2).node_id in scope)
        where = f"来自「{self.title(writer)}」" if writer else "不知道由哪个节点写入"
        if moved_nodes:
-            return _Change("unknown", f"输入「{path}」{where}，{self._moved(moved_nodes[0])}，无法判断变化来源")
+            first = moved_nodes[0]
+            return _Change("unknown", f"输入「{path}」{where}，{self._moved(first, upstream=first in above)}，"
+                                      "无法判断变化来源")
        if worst.kind in ("sql", "template", "unknown"):
            return _Change("unknown", f"输入「{path}」{where}，对不上具体的查询；它的上游有变化，无法判断变化来源")
        if not same and (spoken := self._ordered(n for n in scope if self._regenerated(n))):
            # 上游有模型、人交出的值，两次不同：它交的数变了，可能是数据变了，也可能是模型本身的差异（它可能用的
            # 正是 Agent 没按出处交的数）。上游的查询结果变没变都一样说不清
-            kind = TYPE_LABEL.get(self._type_of(spoken[0]) or "", "节点")
-            return _Change("unknown", f"输入「{path}」{where}，它的上游节点「{self.title(spoken[0])}」（{kind}）两次"
-                                      "交出的结果不同，无法判断变化是否来自数据")
+            first = spoken[0]
+            who = ("它的上游节点" if writer else "上游节点") if first in above else "节点"
+            kind = TYPE_LABEL.get(self._type_of(first) or "", "节点")
+            return _Change("unknown", f"输入「{path}」{where}，{who}「{self.title(first)}」（{kind}）两次交出的结果不同，"
+                                      "无法判断变化是否来自数据")
        if worst.kind == "same" and not same and not self._data_moved(scope):
            # 上游查到的数据两次一模一样，它交的值却变了：不是数据变的。上游只有运行输入变了的，算参数
            if (self.base.inputs or {}) != (self.target.inputs or {}) \
@@ -1039,7 +1071,8 @@ class _Compare:
    def _citation(self, agent: Any, b: Any, t: Any, qb: _Query, qt: _Query) -> _Change | None:
        """Agent 按出处交来的字段两次引用的是不是同一处：同一列（对象数组是同一组列）、同一行或同一段行。两边引用的都是
        整份结果的算同一处（这次多了几行也是）；行段相同、一边是整份结果一边不是的不算——没引用的那几行还在这次的快照
-        里，按维度配对会把它们算成「数据没了」。不是同一处、核对不了是不是同一处的，返回说明。"""
+        里，按维度配对会把它们算成「数据没了」。行号相同、都不是整份结果的，还要引用同一批维度值（_left_out）。不是
+        同一处、核对不了是不是同一处的，返回说明。"""
        name = f"Agent「{self.title(agent)}」" if agent else "Agent"
        lb, lt = _locator(b), _locator(t)
        cb, ct = lb.get("column") or lb.get("columns"), lt.get("column") or lt.get("columns")
@@ -1050,11 +1083,13 @@ class _Compare:
        if b.row is not None or t.row is not None:
            # 单格（数组字段里的一个元素、按出处交的一格）：取哪一行由口径卡的下标和 Agent 交的行段一起定，同一行才是
            # 同一处
-            return None if (b.row, b.rows) == (t.row, t.rows) else changed
+            return self._left_out(name, what, b, t, qb, qt, cb) if (b.row, b.rows) == (t.row, t.rows) else changed
        wb, wt = _whole(b, qb), _whole(t, qt)
        same_span = b.rows == t.rows
-        if (wb and wt) or (same_span and wb is False and wt is False):
+        if wb and wt:
            return None
+        if same_span and wb is False and wt is False:
+            return self._left_out(name, what, b, t, qb, qt, cb)
        if (wb is None or wt is None) and (same_span or (wb is not False and wt is not False)):
            # 快照读不出来，核对不了是不是整份结果：行段相同的可能一边整份一边不是，行段不同的也可能两边都是整份
            side = "基准运行" if wb is None else "这次运行"
@@ -1067,6 +1102,45 @@ class _Compare:
        return _Change("unknown", f"{name}在基准运行中只引用了{_rows_text(b)}（结果共 {len(qb.snap['rows'])} 行），"
                                  "这次引用了整份结果，取的不是同一组数")

+    def _left_out(self, name: str, what: str, b: Any, t: Any, qb: _Query, qt: _Query, column: Any) -> _Change | None:
+        """行号相同、都不是整份结果的引用，引用的是不是同一批维度值（区域……）。一边引用了的，另一边没引用、却还在那边的
+        快照里，取的就不是同一组数：按区域排序的结果多了北区，西区挪出了引用的行段；按金额取前 3 名，北区挤了进来——
+        按维度配对会把还在的西区算成「数据没了」。另一边的快照里也没有的，是数据真的没了，照常比。维度按两边整份结果
+        认（和归因同一个规则，_dimensions）；分不出维度（没有不是数值的列、维度列设了遮罩）、快照读不出来、行的形状
+        不对的，只能按行号认，返回 None。"""
+        if qb.snap is None or qt.snap is None:
+            return None
+        rb, rt = _records(qb.snap, None), _records(qt.snap, None)
+        if rb is None or rt is None:
+            return None
+        dims, _ = _dimensions(qb.snap, qt.snap, rb, rt, column if isinstance(column, str) else "", self._hidden(b, t))
+        if not dims:
+            return None
+
+        def split(records: list[tuple[int, dict[str, Any]]], inp: Any) -> tuple[dict[Any, int], dict[Any, int]]:
+            """{维度键: 第一次出现的行号}：引用的行里的，引用的行外面的。"""
+            first, last = _cited(inp, len(records))
+            inside: dict[Any, int] = {}
+            outside: dict[Any, int] = {}
+            for i, row in records:
+                (inside if first <= i <= last else outside).setdefault(tuple(_hashable(row.get(d)) for d in dims), i)
+            return inside, outside
+
+        (in_b, out_b), (in_t, out_t) = split(rb, b), split(rt, t)
+        for key in in_b:
+            if key not in in_t and key in out_t:
+                return _Change("unknown", f"{name}两次都引用{_rows_text(t)}，{_key_text(key)}这次仍在查询结果第 "
+                                          f"{out_t[key] + 1} 行，不在引用的行里，取的不是{what}")
+        for key in in_t:
+            if key not in in_b and key in out_b:
+                return _Change("unknown", f"{name}两次都引用{_rows_text(b)}，{_key_text(key)}在基准运行中位于查询结果第 "
+                                          f"{out_b[key] + 1} 行，不在引用的行里，取的不是{what}")
+        return None
+
+    def _hidden(self, b: Any, t: Any) -> frozenset[str]:
+        """两边输入落在的快照上要遮的列（小写）。"""
+        return (self.base.masked.get(b.snapshot) or frozenset()) | (self.target.masked.get(t.snapshot) or frozenset())
+
    def _generated(self, path: str, writer: str, same: bool) -> _Change:
        """模型、人交出的值（Agent 没按出处交、模型调用、人工审批……）：没有出处，值变了说不清是数据变了还是它本身的
        差异——同样的数据、同样的配置，模型两次交出的数也可能不同。它自己查的库结构变了、它的配置变了照常标出。"""
@@ -1241,7 +1315,7 @@ class _Compare:
        moved = not recompute._same(bm.get("value"), tm.get("value"))
        changes: dict[str, _Change] = {}
        if lineage and bp is not None and tp is not None:
-            changes = {p: self.provenance(p, bp.inputs.get(p), tp.inputs.get(p), moved=moved)
+            changes = {p: self.provenance(p, bp.inputs.get(p), tp.inputs.get(p), card=card_id, moved=moved)
                       for p in dict.fromkeys([*tp.inputs, *bp.inputs])}
        if caliber:
            # 口径优先；喂它的取数也变了、说不清的，接在口径的说明后面照样写出来，不因为口径变了就不提。只在一边才有
@@ -1276,7 +1350,7 @@ class _Compare:
            if metric.get("status") == "missing_input":
                out["reasons"].append(f"{side}这个指标缺少输入，没有值")
        if out["classification"] in _ATTRIBUTED and bp is not None and tp is not None:
-            out["attribution"] = self._attribution(bp, tp)
+            out["attribution"] = self._attribution(bp, tp, card_id)
        if any(inp.masked for plan in (bp, tp) if plan is not None for inp in plan.inputs.values()):
            self.masked_seen = True
        return out
@@ -1325,7 +1399,7 @@ class _Compare:
        if any(inp.masked for inp in used) or any(inp.masked for plan in plans for inp in plan.inputs.values()):
            return None, SPLIT_MASKED
        for path in leaves:
-            change = self.provenance(path, bp.inputs.get(path), t.inputs[path])
+            change = self.provenance(path, bp.inputs.get(path), t.inputs[path], card=t.node_id)
            if change.kind in ("sql", "template"):
                return None, f"旧口径用到的输入取数也有变化，无法严格拆分：{change.reason}"
            if change.kind == "unknown":
@@ -1375,7 +1449,7 @@ class _Compare:

    # ---------------------------------------------------------------- 维度归因

-    def _attribution(self, bp: Any, tp: Any) -> dict[str, Any] | None:
+    def _attribution(self, bp: Any, tp: Any, card: str) -> dict[str, Any] | None:
        """挑一个落在查询快照上、两边都变了的输入做维度归因：先找给得出占比的，没有就取第一个列得出行的，再没有
        就取第一个说明（截断、度量列遮罩……）。一个都挑不出来返回 None。"""
        found: list[dict[str, Any]] = []
@@ -1383,7 +1457,7 @@ class _Compare:
            b = bp.inputs.get(path)
            if b is None or not (b.snapshot and t.snapshot) or recompute._same(b.value, t.value):
                continue
-            if self.provenance(path, b, t).kind not in ("same", "params"):
+            if self.provenance(path, b, t, card=card).kind not in ("same", "params"):
                continue
            result = self._attribute(b, t)
            if result is None:
@@ -1409,12 +1483,12 @@ class _Compare:
            if snap.get("truncated"):
                # 截断的结果只是前几行：各键的和不是完整的和，按它归因会把没取回的行的变化算到别的键头上
                return {**empty, "note": TRUNCATED.format(side=side, rows=len(snap["rows"]))}
-        hidden = (self.base.masked.get(b.snapshot) or frozenset()) | (self.target.masked.get(t.snapshot) or frozenset())
+        hidden = self._hidden(b, t)
        if any(str(c).lower() in hidden for c in [*(sb.get("columns") or []), *(st.get("columns") or [])]):
            self.masked_seen = True
        if measure.lower() in hidden:
            return {**empty, "note": MASKED_MEASURE.format(column=measure)}
-        rb, rt = _records(sb, b), _records(st, t)
+        rb, rt = _records(sb, b.rows), _records(st, t.rows)
        if rb is None or rt is None:
            return None
        values = (_input_total(b), _input_total(t))
@@ -1546,14 +1620,28 @@ def _same_input(b: Any, t: Any) -> bool | None:
    return recompute._same(b.item.get("value"), t.item.get("value"))


+def _cited(inp: Any, count: int) -> tuple[int, int]:
+    """输入引用了快照的哪几行（第一行, 最后一行），count 是快照共几行：单格是那一行，数组字段是它那一段行，容器本身
+    是一份查询结果的是全部的行。"""
+    if inp.row is not None:
+        return inp.row, inp.row
+    if inp.rows is not None:
+        return inp.rows
+    return 0, count - 1
+
+
 def _whole(inp: Any, q: _Query) -> bool | None:
-    """这个输入引用的是不是整份查询结果：容器本身就是整份结果，或者数组字段的行段从第一行到最后一行。快照读不出来、
-    不知道结果共几行的返回 None。"""
-    if inp.table:
-        return True
+    """这个输入引用的是不是整份查询结果：引用的行从第一行到最后一行（容器本身就是整份结果）。快照读不出来、不知道
+    结果共几行的返回 None。"""
    if q.snap is None:
        return None
-    return inp.rows is not None and inp.rows == (0, len(q.snap["rows"]) - 1)
+    count = len(q.snap["rows"])
+    return _cited(inp, count) == (0, count - 1)
+
+
+def _key_text(key: tuple[Any, ...]) -> str:
+    """维度值写进说明：几列的用「 / 」连起来，和对比页上归因行的写法一样。"""
+    return "「" + " / ".join(_shown(v) for v in key) + "」"


 def _rows_text(inp: Any) -> str:
@@ -1578,11 +1666,12 @@ def _merge_inputs(q: _Query) -> dict[str, str] | None:
            if isinstance(i, dict) and i.get("alias") and i.get("artifact")}


-def _records(snap: dict[str, Any], inp: Any) -> list[tuple[int, dict[str, Any]]] | None:
-    """归因用到的行：(行号, {列: 值})。数组字段只用它那一段行，单格用整份结果。"""
+def _records(snap: dict[str, Any], span: tuple[int, int] | None) -> list[tuple[int, dict[str, Any]]] | None:
+    """快照里的行：(行号, {列: 值})。span 给了只取那一段行（归因时数组字段只用它那一段），None 是整份结果（单格按
+    整份结果归因）。行的形状不对返回 None。"""
    columns = [str(c) for c in snap.get("columns") or []]
    rows = snap.get("rows") or []
-    first, last = (inp.rows if inp.rows is not None else (0, len(rows) - 1))
+    first, last = span if span is not None else (0, len(rows) - 1)
    out = []
    for i in range(first, min(last, len(rows) - 1) + 1):
        row = rows[i]
@@ -10,20 +10,27 @@ SQL，Agent 节点 bot 自己再查一次、把各区域的金额作为数组字
 - 遮罩列的值绝不出现在结果里；用到遮罩列的指标不拆分
 - 模板对比：新增、删除的节点，配置有变化的节点和配置项名
 - 输入的出处：合并查询追到它的上游查询；Agent 按出处交的字段还要引用同一列、同一段行（行段相同、一边是整份结果
-  一边不是的不算）；没按出处交的数、上游数据没变数却变了的、上游的 Agent 两次交出的结果不同的，说不清变化从哪来
-  （incomparable）；图里写死的字面量改了、连线变了、读到的默认设置变了都算模板变化
+  一边不是的不算；都不是整份结果的，一边引用了的区域另一边还在快照里却没引用的也不算）；没按出处交的数、上游数据
+  没变数却变了的、上游的 Agent 两次交出的结果不同的，说不清变化从哪来（incomparable），记不出写入者的只看口径卡
+  之前可能执行过的节点；图里写死的字面量改了、连线变了、读到的默认设置变了都算模板变化
 """
 from __future__ import annotations

+import ast
 import copy
+import importlib
 import inspect
 import json
-import re
+import pkgutil
+import textwrap
 from decimal import Decimal
+from types import SimpleNamespace

+import app.engine.nodes as nodes_package
 from app.engine import compare
 from app.engine.compiler import RUNNERS
 from app.engine.merge_query import MERGE_SOURCE
+from app.engine.schema import NodeType
 from tests.fixtures.compare_runs import (
    BASE_ROWS,
    COLUMNS,
@@ -610,6 +617,9 @@ def test_a_code_node_after_an_agent_that_answered_differently_is_incomparable():
    assert calc["classification"] == "incomparable" and calc["attribution"] is None
    assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「区域分析」（Agent）两次交出的结果不同，"
                               "无法判断变化是否来自数据"]
+    # Agent 交的不同，代码节点交的数两次一样：照常没变，没有可说不清的变化
+    calc = by_id(run_compare(make("r1", BASE_ROWS, 0.46, value=1.0), make("r2", NEXT_ROWS, 0.52, value=1.0)))["calc"]
+    assert (calc["classification"], calc["reasons"]) == ("unchanged", [])

    # Agent 的产出两次都没落成工件（写工件失败）：一样不一样核对不了，同样不报成数据变化
    def unstored(run):
@@ -622,20 +632,89 @@ def test_a_code_node_after_an_agent_that_answered_differently_is_incomparable():


 def test_a_reconfigured_node_upstream_of_a_code_node_is_named():
-    """代码节点上游的数据整形节点改了表达式（SQL 一字没改）：代码节点交的数变了说不清从哪来，说明里点出改的是哪个节点。"""
-    def make(run_id, expression, rows, value):
-        shape = node("shape", "transform", "整理", mode="expression", expression=expression, assign_to="kept")
-        run = Fake(run_id, graph=weekly_graph(after_fetch=(shape, calc_node())))
+    """代码节点上游的数据整形节点改了表达式（SQL 一字没改）：代码节点交的数变了说不清从哪来，说明里点出改的是哪个节点；
+    改了几个的，点名画布上最靠前的那个（节点 ID 的先后和画布上的先后不同）。"""
+    def make(run_id, expressions, rows, value):
+        shapes = [node(nid, "transform", label, mode="expression", expression=expression, assign_to=f"kept_{nid}")
+                  for (nid, label), expression in zip((("shape_m", "整理"), ("shape_z", "再整理"), ("shape_a", "汇总")),
+                                                      expressions)]
+        run = Fake(run_id, graph=weekly_graph(after_fetch=(*shapes, calc_node())))
        run.tool("fetch", region_sql(), rows=rows)
        run.agent("bot", region_sql(), rows=rows)
        run.code("calc", value, "adj")
        run.card(code_metrics(value)(None, None, rows, rows))
        return run

-    calc = by_id(run_compare(make("r1", "nodes.fetch.rows", BASE_ROWS, 10),
-                             make("r2", "nodes.fetch.rows[:2]", NEXT_ROWS, 12)))["calc"]
+    same = ["nodes.fetch.rows"] * 3
+    calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10),
+                             make("r2", ["nodes.fetch.rows[:2]", *same[1:]], NEXT_ROWS, 12)))["calc"]
    assert calc["classification"] == "incomparable"
    assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"]
+    calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10),
+                             make("r2", ["nodes.fetch.rows[:2]"] * 3, NEXT_ROWS, 12)))["calc"]
+    assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"]
+
+
+def test_a_report_upstream_of_a_code_node_that_wrote_differently_is_incomparable():
+    """代码节点上游有报告撰写节点（文字是模型写的），两次写得不同：代码节点交的数变了，可能来自数据，也可能来自模型
+    本身的差异；两次写得一字不差的（产出工件一样）照常是数据变化。"""
+    def make(run_id, rows, text, value):
+        run = Fake(run_id, graph=weekly_graph(after_fetch=(node("draft", "report", "初稿"), calc_node())))
+        run.tool("fetch", region_sql(), rows=rows)
+        run.add("node.started", "draft", {"node_type": "report"})
+        run.add("node.finished", "draft", {"artifact": run.put({"text": text})})
+        run.agent("bot", region_sql(), rows=rows)
+        run.code("calc", value, "adj")
+        run.card(code_metrics(value)(None, None, rows, rows))
+        return run
+
+    calc = by_id(run_compare(make("r1", BASE_ROWS, "上周初稿", 10), make("r2", NEXT_ROWS, "本周初稿", 12)))["calc"]
+    assert calc["classification"] == "incomparable"
+    assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「初稿」（报告撰写）两次交出的结果不同，"
+                               "无法判断变化是否来自数据"]
+    calc = by_id(run_compare(make("r1", BASE_ROWS, "初稿", 10), make("r2", NEXT_ROWS, "初稿", 12)))["calc"]
+    assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM])
+
+
+def test_an_unknown_writer_is_looked_for_only_among_nodes_before_the_card():
+    """两边都记不出 vars.adj 是哪个节点写的：只看口径卡之前可能执行过的节点。口径卡自己改了配置、它下游的写周报两次写得
+    不同，都写不了它读到的值，不能点名成它的上游；它上游的节点改了配置照常说不清，并行支路上的节点也可能先写了变量，
+    同样说不清，只是不说成「上游」。"""
+    def make(run_id, rows, value, *, other="1", report="上周", reshape="nodes.fetch.rows", aside="nodes.fetch.rows"):
+        graph = chain(node("start", "input", "开始"),
+                      node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": region_sql()}),
+                      node("shape", "transform", "整理", mode="expression", expression=reshape, assign_to="kept"),
+                      node("calc", "code", "调整", code="x", assign_to="adj"),
+                      node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1",
+                           metrics=[{"id": "other", "expression": other}]),
+                      node("write", "report", "写周报"), node("out", "output", "成果"))
+        # 并行支路：开始 → 旁路整理 → 成果，和口径卡不在一条线上
+        graph["nodes"].append(node("aside", "transform", "旁路整理", mode="expression", expression=aside,
+                                   assign_to="extra"))
+        graph["edges"] += [{"source": "start", "target": "aside"}, {"source": "aside", "target": "out"}]
+        run = Fake(run_id, graph=graph)
+        run.tool("fetch", region_sql(), rows=rows)
+        run.code("calc", value, "adj")
+        run.card([metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, None, "state")])])
+        run.add("node.started", "write", {"node_type": "report"})
+        run.add("node.finished", "write", {"artifact": run.put({"text": report})})
+        return run
+
+    # 口径卡自己改了另一个指标的定义，写周报两次写得不同：都在它之后执行，变化照常来自上游数据
+    calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, other="2", report="本周")))["calc"]
+    assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM_SCOPED])
+    calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, report="本周")))["calc"]
+    assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM])
+    # 口径卡上游的整理节点改了表达式
+    calc = by_id(run_compare(make("r1", BASE_ROWS, 10),
+                             make("r2", NEXT_ROWS, 12, reshape="nodes.fetch.rows[:2]")))["calc"]
+    assert calc["classification"] == "incomparable"
+    assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，上游节点「整理」的配置有变化，无法判断变化来源"]
+    # 并行支路上的旁路整理改了表达式
+    calc = by_id(run_compare(make("r1", BASE_ROWS, 10),
+                             make("r2", NEXT_ROWS, 12, aside="nodes.fetch.rows[:2]")))["calc"]
+    assert calc["classification"] == "incomparable"
+    assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，节点「旁路整理」的配置有变化，无法判断变化来源"]


 def test_a_run_input_upstream_of_a_code_node_is_a_parameter_change():
@@ -798,6 +877,11 @@ def test_a_changed_default_model_is_a_template_change_for_agents_that_follow_it(
    many["defaults"] = {"provider": "openai", "model": "model-b", "temperature": 0.5, "max_tokens": 2000}
    share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=many)))["share"]
    assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型接入、模型、温度等 4 项）有变化"]
+    # 同时改了 Agent 不读的结论句裁判：只写它跟着变的模型
+    judged = copy.deepcopy(target_graph)
+    judged["defaults"]["judge"] = {"max_cost_usd": 1.0}
+    share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=judged)))["share"]
+    assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型）有变化"]
    # Agent 自己指定了模型：默认设置变了它不跟着变
    for graph in (base_graph, target_graph):
        graph["nodes"][2]["data"]["config"]["model"] = "model-x"
@@ -823,17 +907,19 @@ def test_defaults_that_agents_do_not_read_leave_them_alone():
 def test_a_model_default_reaches_every_node_that_calls_the_model():
    """默认模型换了，跟着换的不只 Agent：让模型分类的条件分支、让模型修复的结构校验（默认开着）、用模型重排的知识检索
    也是。它们在代码节点上游时，代码节点交的数变了不能说来自上游数据；不调模型的照常。"""
-    def calc(extra):
-        def make(run_id, model, rows, value):
+    def calc(extra, before=None, after=None):
+        def make(run_id, defaults, rows, value):
            graph = weekly_graph(after_fetch=(extra, calc_node()))
-            graph["defaults"] = {"provider": "anthropic", "model": model}
+            graph["defaults"] = defaults
            run = Fake(run_id, graph=graph)
            run.tool("fetch", region_sql(), rows=rows)
            run.agent("bot", region_sql(), rows=rows)
            run.code("calc", value, "adj")
            run.card(code_metrics(value)(None, None, rows, rows))
            return run
-        return by_id(run_compare(make("r1", "model-a", BASE_ROWS, 10), make("r2", "model-b", NEXT_ROWS, 12)))["calc"]
+        before = before or {"provider": "anthropic", "model": "model-a"}
+        after = after or {"provider": "anthropic", "model": "model-b"}
+        return by_id(run_compare(make("r1", before, BASE_ROWS, 10), make("r2", after, NEXT_ROWS, 12)))["calc"]

    cases = [{"key": "peak", "condition": "true"}]
    for extra in (node("route", "branch", "分流", mode="llm", instruction="判断是否旺季", cases=cases),
@@ -849,29 +935,121 @@ def test_a_model_default_reaches_every_node_that_calls_the_model():
        found = calc(extra)
        assert found["classification"] == "data_only", (extra["type"], found["reasons"])

+    # 调不调模型也可能由默认设置定：条件分支没写判断方式、默认设置写了让模型判断的，跟着默认模型变；结构校验的默认
+    # 设置关了让模型修复的，不跟着变
+    asks = node("route", "branch", "分流", instruction="判断是否旺季", cases=cases)
+    found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "llm", "model": "model-b"})
+    assert found["classification"] == "incomparable", found["reasons"]
+    assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（模型）有变化，"
+                                "无法判断变化来源"]
+    found = calc(node("route", "validate", "校验", schema={"type": "object"}),
+                 {"repair_with_llm": False, "model": "model-a"}, {"repair_with_llm": False, "model": "model-b"})
+    assert found["classification"] == "data_only", found["reasons"]
+    # 默认的判断方式这次改成了按表达式：基准运行那次让模型判断过，跟着变的是判断方式和模型两项
+    found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "expression", "model": "model-b"})
+    assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（判断方式、模型）"
+                                "有变化，无法判断变化来源"]
+
+
+#: 只顺着节点模块里的函数往下找：别处的函数不经 NodeContext.cfg 读节点配置
+_NODE_MODULES = "app.engine.nodes."
+
+
+def _defined_here(obj):
+    """obj 是定义在节点模块里的函数或类（装饰过的取原函数），否则 None。"""
+    obj = inspect.unwrap(obj) if inspect.isfunction(obj) else getattr(obj, "func", obj)
+    if (inspect.isfunction(obj) or inspect.isclass(obj)) and obj.__module__.startswith(_NODE_MODULES):
+        return obj
+    return None
+
+
+def _settings_read(runner):
+    """一类节点的执行器运行时按 NodeContext.cfg 读的配置项（节点上没写就回退到默认设置）：执行器自己的函数体，加上它
+    用到的、定义在节点模块里的函数和类，一层层往下找（函数里就地 import 的也算）。返回 (配置项, 认不出键名的读法,
+    走到的函数)。键名是变量的，只认在字面量元组上循环出来的（Agent 的预算三项）；别的变量键名认不出是哪几项，
+    原样交回，测试据此失败。"""
+    read, unknown, seen = set(), [], set()
+    todo = [runner]
+    while todo:
+        fn = _defined_here(todo.pop())
+        if fn is None or (fn.__module__, fn.__qualname__) in seen:
+            continue
+        seen.add((fn.__module__, fn.__qualname__))
+        names = vars(importlib.import_module(fn.__module__))        # 它所在模块里的名字（类没有 __globals__）
+        parts = list(ast.walk(ast.parse(textwrap.dedent(inspect.getsource(fn)))))
+        imported = {alias.asname or alias.name: (part.module, alias.name) for part in parts
+                    if isinstance(part, ast.ImportFrom) and part.module for alias in part.names}
+        looped = {}         # 循环变量 → 它在字面量元组上取到的键名：{k: ctx.cfg(k) for k in ("max_steps", …)}
+        for part in parts:
+            for gen in getattr(part, "generators", None) or ([part] if isinstance(part, ast.For) else []):
+                if isinstance(gen.target, ast.Name) and isinstance(gen.iter, (ast.Tuple, ast.List)) and all(
+                        isinstance(e, ast.Constant) and isinstance(e.value, str) for e in gen.iter.elts):
+                    looped[gen.target.id] = {e.value for e in gen.iter.elts}
+        for part in parts:
+            called = part.func if isinstance(part, ast.Call) else None
+            if isinstance(called, ast.Attribute) and called.attr == "cfg":
+                key = part.args[0] if part.args else None
+                if isinstance(key, ast.Constant) and isinstance(key.value, str):
+                    read.add(key.value)
+                elif isinstance(key, ast.Name) and key.id in looped:
+                    read |= looped[key.id]
+                else:
+                    unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}")
+            elif isinstance(called, ast.Attribute) and called.attr == "approval_mode":
+                read.add("approval")        # NodeContext.approval_mode 同样先读节点、再回退到默认设置
+            elif isinstance(called, ast.Name) and called.id == "_setting":
+                # 报告模块的 _setting(spec, 节点, 键) 和 NodeContext.cfg 同一个规则
+                key = part.args[2] if len(part.args) > 2 else None
+                if isinstance(key, ast.Constant) and isinstance(key.value, str):
+                    read.add(key.value)
+                else:
+                    unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}")
+            elif isinstance(part, ast.Name) and isinstance(part.ctx, ast.Load):
+                # 调用、当回调传出去的函数都算用到了
+                target = names.get(part.id)
+                if target is None and part.id in imported:
+                    module, name = imported[part.id]
+                    target = getattr(importlib.import_module(module), name, None)
+                todo.append(target)
+            elif isinstance(part, ast.Attribute) and isinstance(part.value, ast.Name):
+                module = names.get(part.value.id)
+                if inspect.ismodule(module):
+                    todo.append(getattr(module, part.attr, None))
+    return read, unknown, seen
+

 def test_every_setting_a_node_reads_is_known_to_the_comparison():
    """节点运行时按 NodeContext.cfg 读配置，节点上没写就回退到图级默认设置。对比按「哪类节点读哪几项」把默认设置的
-    变化落到节点上（compare._CFG_KEYS）：执行器新读了一项、这里没登记，默认设置改了这一项就会被漏掉。"""
-    by_module: dict = {}
+    变化落到节点上（compare._CFG_KEYS）：执行器新读了一项、这里没登记，默认设置改了这一项就会被漏掉；登记了不读的，
+    默认设置改了它就会被误标。按节点类型比，不按模块比：一个模块管几类节点（Agent 和模型调用、调用工具和代码……），
+    其中一类读了、另一类没读的，不能拿另一类的登记充数。"""
+    reached = set()
    for kind, runner in RUNNERS.items():
-        by_module.setdefault(inspect.getmodule(runner), set()).add(str(kind))
-    for module, kinds in by_module.items():
-        source = inspect.getsource(module)
-        read = set(re.findall(r'ctx\.cfg\(\s*"(\w+)"', source))
-        read |= set(re.findall(r'_setting\(spec, node, "(\w+)"', source))
-        if "ctx.approval_mode()" in source:
-            read.add("approval")        # NodeContext.approval_mode 同样先读节点、再回退到默认设置
-        known = set().union(
-            *(compare._CFG_KEYS.get(k, frozenset()) | compare._CFG_KEYS_IF.get(k, frozenset()) for k in kinds))
-        assert read <= known, (module.__name__, sorted(read - known))
+        read, unknown, seen = _settings_read(runner)
+        assert not unknown, (str(kind), unknown)
+        known = compare._CFG_KEYS.get(str(kind), frozenset()) | compare._CFG_KEYS_IF.get(str(kind), frozenset())
+        assert read == known, (str(kind), "没登记", sorted(read - known), "登记了不读", sorted(known - read))
+        reached |= seen
+    # 键名不是字面量的读法也认得出来
+    assert {"max_steps", "budget_tokens", "budget_usd"} <= _settings_read(RUNNERS[NodeType.AGENT])[0]
+    # 节点模块里每个按 cfg 读配置的函数都走到了：走不到的说明上面漏了某种用法，查出来的配置项就不全
+    holders = set()
+    for info in pkgutil.iter_modules(nodes_package.__path__):
+        module = importlib.import_module(_NODE_MODULES + info.name)
+        for top in ast.parse(inspect.getsource(module)).body:
+            if isinstance(top, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and any(
+                    isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
+                    and n.func.attr in ("cfg", "approval_mode") for n in ast.walk(top)):
+                holders.add((module.__name__, top.name))
+    assert holders and holders <= reached, sorted(holders - reached)


-def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None):
+def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None, sql=None):
    """Agent 按出处交回数组字段：引用快照的哪一列、哪一段行由调用方定。"""
    run = Fake(run_id)
-    run.tool("fetch", region_sql(), rows=rows)
-    bot = run.agent("bot", region_sql(), rows=rows, column=column, span=span)
+    sql = sql or region_sql()
+    run.tool("fetch", sql, rows=rows)
+    bot = run.agent("bot", sql, rows=rows, column=column, span=span)
    first, last = span or (0, len(rows) - 1)
    values = [r[COLUMNS.index(column)] for r in rows[first:last + 1]]
    run.card([metric("total", "sum(vars.kpi.amounts)", sum(values),
@@ -917,6 +1095,65 @@ def test_the_same_rows_are_not_the_same_citation_when_only_one_side_is_the_whole
    assert total["classification"] == "data_only" and total["attribution"]["rows"]


+def test_the_same_rows_are_not_the_same_citation_when_a_cited_region_moved_out_of_them():
+    """两边都只引用第 1 至 3 行、都不是整份结果：基准运行引用了西区，这次北区排进了前 3 行，西区挪到第 4 行，还在这次
+    的快照里，只是没被引用。取的不是同一组数，不能把西区算成「数据没了」（占比会远超 100%）；反过来（这次引用的区域
+    在基准运行的快照里、没被引用）也一样。西区这次真的没有了的，照常是数据变化。"""
+    with_middle = [["东区", 1200.5, 100.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]]
+    total = by_id(run_compare(cited("r1", with_middle, span=(0, 2)), cited("r2", WITH_NORTH, span=(0, 2))))["total"]
+    assert total["classification"] == "incomparable" and total["attribution"] is None
+    assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」这次仍在查询结果第 4 行，不在引用的"
+                                "行里，取的不是同一组数"]
+    total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), cited("r2", with_middle, span=(0, 2))))["total"]
+    assert total["classification"] == "incomparable" and total["attribution"] is None
+    assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」在基准运行中位于查询结果第 4 行，不在"
+                                "引用的行里，取的不是同一组数"]
+
+    # 按金额从大到小取前 3 名：北区这次挤进了前 3，西区落到第 4 行
+    top = region_sql().replace("ORDER BY region", "ORDER BY gmv DESC")
+    before = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]]
+    after = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["西区", 600.0, 0.0]]
+    total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top),
+                              cited("r2", after, span=(0, 2), sql=top)))["total"]
+    assert total["classification"] == "incomparable" and total["attribution"] is None
+    assert "「西区」这次仍在查询结果第 4 行" in total["reasons"][0]
+    # 对照：西区这次真的没有了（不在这次的快照里），照常是数据变化，按区域归因，占比之和为 1
+    gone = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["中区", 30.0, 0.0]]
+    total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top),
+                              cited("r2", gone, span=(0, 2), sql=top)))["total"]
+    assert (total["classification"], total["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES])
+    rows = total["attribution"]["rows"]
+    assert [(r["key"], r["base"], r["target"], r["share"]) for r in rows] == [
+        (["北区"], None, 700.0, 7.0), (["西区"], 600.0, None, -6.0), (["东区"], 1200.0, 1200.0, 0.0),
+        (["南区"], 800.0, 800.0, 0.0)]
+    assert sum(exact(r["share"]) for r in rows) == 1
+
+    # 区域列设了遮罩：不拿它认是哪一区，只能按行号认，遮住的区域名不出现在结果里
+    def masked(run):
+        run.masked = {a: frozenset({"region"}) for a in run.queries}
+        return run
+    result = run_compare(masked(cited("r1", with_middle, span=(0, 2))), masked(cited("r2", WITH_NORTH, span=(0, 2))))
+    assert by_id(result)["total"]["classification"] == "data_only"
+    text = json.dumps(result, ensure_ascii=False)
+    assert all(region not in text for region in ("东区", "南区", "西区", "北区", "中区"))
+
+
+def test_an_input_cites_the_whole_result_only_when_it_covers_every_row():
+    """怎么认整份结果（compare._whole）：容器本身就是一份查询结果的（没有行号、行段）、数组字段的行段从第一行到最后
+    一行的是；单格、只有一段行的不是；快照读不出来的不知道。Agent 的字段在真实运行里总带着行号或行段，容器这一支
+    只是兜底，造不出真实的运行来走它，在这里直接核对。"""
+    query = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1, snap={"rows": [["东区"], ["南区"], ["西区"]]})
+    lost = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1)
+
+    def cites(row=None, rows=None):
+        return SimpleNamespace(row=row, rows=rows)
+    assert compare._whole(cites(), query) is True
+    assert compare._whole(cites(rows=(0, 2)), query) is True
+    assert compare._whole(cites(rows=(0, 1)), query) is False
+    assert compare._whole(cites(row=0), query) is False
+    assert compare._whole(cites(rows=(0, 2)), lost) is None
+
+
 def test_cited_rows_on_an_unreadable_snapshot_are_not_taken_as_the_same():
    """基准运行 Agent 那次查询的快照文件不在了：两边都引用第 1 至 3 行，可基准那次是不是整份结果核对不了，不报成
    数据变化。"""
@@ -929,6 +1166,18 @@ def test_cited_rows_on_an_unreadable_snapshot_are_not_taken_as_the_same():
    total = by_id(run_compare(base, cited("r2", WITH_NORTH)))["total"]
    assert total["classification"] == "incomparable" and "无法核对" in total["reasons"][0]

+    # 反过来只有这次运行的快照读不出来：说的是这次运行
+    def unreadable(run):
+        del run.store[run.queries[1]]
+        return run
+    total = by_id(run_compare(cited("r1"), unreadable(cited("r2", WITH_NORTH, span=(0, 2)))))["total"]
+    assert total["classification"] == "incomparable"
+    assert total["reasons"] == ["这次运行的查询快照无法读取，无法核对 Agent「区域分析」引用的是不是整份结果"]
+    # 基准运行只引用了一部分、这次的行段又不同：这次是不是整份结果都不是同一处，照实说行段换了
+    total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), unreadable(cited("r2", span=(1, 2)))))["total"]
+    assert total["classification"] == "incomparable"
+    assert total["reasons"] == ["Agent「区域分析」引用的行由第 1 至 3 行改为第 2 至 3 行，取的不是同一组数"]
+

 def test_an_element_of_a_cited_array_follows_its_row():
    """口径卡取 Agent 数组字段里的第一个元素：Agent 交的行段换了起点，第一个元素落到了别的行，不是同一个数；起点没变
@@ -949,6 +1198,12 @@ def test_an_element_of_a_cited_array_follows_its_row():
    assert first["reasons"] == ["Agent「区域分析」引用的行由第 1 行改为第 2 行，取的不是同一个数"]
    first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), make("r2", NEXT_ROWS, (0, 2))))["first"]
    assert (first["classification"], first["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES])
+    # 起点没变，可这次北区排在了第一行：基准运行取的是东区，东区这次还在第 2 行，取的不是同一个数
+    first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)),
+                              make("r2", [["北区", 90.0, 0.0], *NEXT_ROWS], (0, 3))))["first"]
+    assert first["classification"] == "incomparable"
+    assert first["reasons"] == ["Agent「区域分析」两次都引用第 1 行，「东区」这次仍在查询结果第 2 行，不在引用的行里，"
+                                "取的不是同一个数"]


 def test_no_split_when_an_old_leaf_now_cites_another_column():