运行对比加固（收尾）：区域等维度列设了遮罩时照样拿它核对两次引用的是不是同一批行，说明只写行号、不写值，挤出引用行段的和真的少了分得清，没挪的照常是数据变化；补单格反方向、归因维度按两边各自引用的行认、写入者改了配置时不严格拆分、并行支路查询四处守护用例

backend/app/engine/compare.py

@@ -1146,8 +1146,9 @@ class _Compare:
         归因说的南区就不是同一组数——归因按哪组维度说，就要按哪组维度核对。只按归因用的那组也不够：只引用了一行的，
         引用的行里每列都只有一个值，认不出维度；整份结果里分得清的维度（区域, 渠道）也要切在同一处。
 
-        遮罩列只用没遮的列认，说明里写的也只有没遮的列的值和行号，遮住的值不进说明；两组都分不出维度（没有不是数值的
-        列、维度列都设了遮罩）的，只能按行号认，返回 None。快照读不出来的说不清：只有单格会走到这里（行段在 _citation
+        先只用没遮的列认，说明里写的是没遮的列的值和行号；有列设了遮罩的，再连遮住的列一起核对一遍（区域设了遮罩时它
+        往往正是唯一认得出行的那一列，只用没遮的列等于对它不核对），这一遍的说明只写行号、不写是哪个值。两遍都分不出
+        维度（没有不是数值的列）的，只能按行号认，返回 None。快照读不出来的说不清：只有单格会走到这里（行段在 _citation
         里先核对整份结果，读不出来的已经说不清），那一行两次是不是同一个维度值核对不了，和行段一样不按行号当成同一处。
         快照里有的行和列对不上（格数不对、不是一行）的，认不出各行是哪个维度值，同样说不清。"""
         if qb.snap is None or qt.snap is None:
@@ -1167,12 +1168,29 @@ class _Compare:
         for dims in dict.fromkeys(tuple(d) for d in (whole, used) if d):
             if (found := self._cut(name, what, b, t, rb, rt, list(dims))) is not None:
                 return found
+        if hidden and any(str(c).lower() in hidden for snap in (qb.snap, qt.snap) for c in snap.get("columns") or []):
+            # 拿遮住的列核对过（哪怕最后没核对出不同）：整体说明照样提一句有列设了遮罩
+            self.masked_seen = True
+        if hidden:
+            # 遮罩列不当维度，是为了值不进说明；可它往往正是认得出行的那一列（区域设了遮罩），只拿没遮的列核对，等于对它
+            # 不核对，挤出引用的行就和真的没有了分不出来。连遮住的列一起再核对一遍，说明里只写行号、不写值
+            whole, _ = _dimensions(qb.snap, qt.snap, rb, rt, measure, frozenset())
+            used, _ = _dimensions(qb.snap, qt.snap, _attributed_rows(qb.snap, b) or [],
+                                  _attributed_rows(qt.snap, t) or [], measure, frozenset())
+            for dims in dict.fromkeys(tuple(d) for d in (whole, used) if d):
+                if (found := self._cut(name, what, b, t, rb, rt, list(dims), redact=True)) is not None:
+                    return found
         return None
 
     def _cut(self, name: str, what: str, b: Any, t: Any, rb: list[tuple[int, dict[str, Any]]],
-             rt: list[tuple[int, dict[str, Any]]], dims: list[str]) -> _Change | None:
+             rt: list[tuple[int, dict[str, Any]]], dims: list[str], *, redact: bool = False) -> _Change | None:
         """按 dims 这组维度核对两边引用的是不是同一批维度值（规则见 _left_out）：是返回 None，不是返回说明。rb、rt 是
-        两边整份结果的行。"""
+        两边整份结果的行。redact：dims 里有设了遮罩的列，说明里不写是哪个维度值，只写行号。"""
+
+        def named(key: tuple[Any, ...]) -> str:
+            return "其中一个维度值" if redact else _key_text(key)
+
+        masked_note = "（维度列设了遮罩，不写出是哪个值）" if redact else ""
 
         def split(records: list[tuple[int, dict[str, Any]]], inp: Any) -> tuple[dict[Any, int], dict[Any, list[int]]]:
             """{维度键: 引用的行里有几行}，{维度键: [引用的行外面的行号]}。一个键可能占几行（去掉日期这类每期都不同的
@@ -1209,9 +1227,9 @@ class _Compare:
                     # 那一行两次都是它（单格只有一行，nb、nt 都是 1）：还要是它同样的那一行，排在前面的行数一样
                     if ahead(ob) == ahead(ot):
                         continue
-                    return _Change("unknown", f"{name}两次都引用{cited}，两次都是{_key_text(key)}，可不是它同样的那一行"
+                    return _Change("unknown", f"{name}两次都引用{cited}，两次都是{named(key)}，可不是它同样的那一行"
                                               f"（它排在这一行前面的，基准运行有 {ahead(ob)} 行，这次有 {ahead(ot)} 行），"
-                                              f"取的不是{what}")
+                                              f"取的不是{what}{masked_note}")
                 # 不是同一个维度值：看只在一边被引用的它在另一边的快照里还有没有——有（只是不在这一行）是换了一个数，
                 # 没有是真的没有了、这次才有
                 if not (ot if nb > nt else ob):
@@ -1222,9 +1240,9 @@ class _Compare:
                     continue
                 if cb is not None and ct is not None and nb == nt:
                     # 都只引用了它的一部分、行数也一样，可排在引用的行前面的行数不同：引用的不是它同样的那几行
-                    return _Change("unknown", f"{name}两次都引用{cited}，两次都引用了{_key_text(key)}的 {nb} 行，可不是"
+                    return _Change("unknown", f"{name}两次都引用{cited}，两次都引用了{named(key)}的 {nb} 行，可不是"
                                               f"它同样的那几行（它排在引用的行前面的，基准运行有 {cb[0]} 行，这次有 "
-                                              f"{ct[0]} 行），取的不是{what}")
+                                              f"{ct[0]} 行），取的不是{what}{masked_note}")
             # 说引用的行外面有它的那一边；两边都有的，说引用得少的那一边
             if ot and (not ob or nt < nb):
                 where = (f"这次有 {len(ot)} 行在查询结果{_row_numbers(ot)}" if nt
@@ -1232,8 +1250,8 @@ class _Compare:
             else:
                 where = (f"在基准运行中有 {len(ob)} 行位于查询结果{_row_numbers(ob)}" if nb
                          else f"在基准运行中位于查询结果{_row_numbers(ob)}")
-            return _Change("unknown", f"{name}两次都引用{cited}，{_key_text(key)}{where}，不在引用的行里，"
-                                      f"取的不是{what}")
+            return _Change("unknown", f"{name}两次都引用{cited}，{named(key)}{where}，不在引用的行里，"
+                                      f"取的不是{what}{masked_note}")
         return None
 
     def _hidden(self, b: Any, t: Any) -> frozenset[str]:

backend/tests/test_compare_engine.py

@@ -1229,6 +1229,9 @@ def test_an_agent_citing_other_rows_is_not_a_data_change():
 
 
 #: 按区域排序的结果这次多了北区，排在南区前面，西区挪到第 4 行
+#: 区域列设了遮罩、西区挪出引用的第 1 至 3 行时的说明：只写行号，不写是哪个区域
+MASKED_MOVED = ("Agent「区域分析」两次都引用第 1 至 3 行，其中一个维度值这次仍在查询结果第 4 行，不在引用的行里，"
+                "取的不是同一组数（维度列设了遮罩，不写出是哪个值）")
 WITH_NORTH = [["东区", 1200.5, 100.0], ["北区", 90.0, 0.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0]]
 
 
@@ -1292,12 +1295,15 @@ def test_the_same_rows_are_not_the_same_citation_when_a_cited_region_moved_out_o
         (["东区"], 1200.0, 1300.0, 2.0), (["南区"], 800.0, 700.0, -2.0), (["西区"], 600.0, 650.0, 1.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"
+    total = by_id(result)["total"]
+    assert total["classification"] == "incomparable" and total["attribution"] is None
+    assert total["reasons"] == [MASKED_MOVED]
     text = json.dumps(result, ensure_ascii=False)
     assert all(region not in text for region in ("东区", "南区", "西区", "北区", "中区"))
 
@@ -1323,8 +1329,8 @@ def test_a_mask_on_one_side_only_still_hides_the_column():
             text = json.dumps(part, ensure_ascii=False)
             leaked = [r for r in ("东区", "南区", "西区", "北区", "中区") if r in text]
             assert not leaked, (label, where, leaked)
-        # 区域认不出，只能按行号认（同 test_the_same_rows_are_not_the_same_citation_when_a_cited_region_moved_out_of_them）
-        assert (total["classification"], total["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES]), label
+        # 照样拿遮住的区域列核对出西区挪出了引用的行，说明只写行号（同上一条用例的遮罩那一段）
+        assert (total["classification"], total["reasons"]) == ("incomparable", [MASKED_MOVED]), label
         # 整体说明照样提一句有列设了遮罩：归因少了区域这一列，看的人要知道是遮掉了，不是没有
         assert compare.MASKED_NOTE in result["notes"], label
 

backend/tests/test_compare_guards.py

@@ -0,0 +1,189 @@
+"""运行对比的守护用例：对抗复验找出的几处「拿掉规则、测试仍全绿」和一处遮罩下的放行。
+
+- 区域列设了遮罩时，部分行挤出引用的行段照样核对得出（说明只写行号），和「真的少了」分得清；什么都没挪的照常是数据变化
+- 单格「同一行」的规则两个方向都守住（基准运行里那个区域只有引用的那一行）
+- 归因按哪组维度说，就按哪组维度在两边各自引用的行上核对
+- 写入输入的节点改了配置时，严格拆分不拆
+- 记不出写入者时，并行支路上的查询也算进「它上游变了什么」
+
+夹具全部虚构，形状照 tests/fixtures/compare_runs.py。
+"""
+from __future__ import annotations
+
+import json
+
+from app.engine import compare
+from tests.fixtures.compare_runs import (
+    BASE_ROWS, COLUMNS, NEXT_ROWS, TOOL, TYPES, WEEK37, Fake, amounts_in, chain, metric, node, region_sql, scalar_in,
+)
+
+REGIONS = ("东区", "南区", "西区", "北区", "中区")
+
+#: 区域 × 日期：日期每期都不同、不当维度，同一个区域占几行
+DAY = {"sql": "SELECT region, day, SUM(amount) AS gmv FROM orders GROUP BY region, day ORDER BY region, day",
+       "columns": ["region", "day", "gmv"], "types": {"region": "text", "day": "date", "gmv": "number"}}
+#: 区域 × 渠道 × 日期
+CH = {"sql": "SELECT region, channel, day, SUM(amount) AS gmv FROM orders GROUP BY region, channel, day ORDER BY gmv DESC",
+      "columns": ["region", "channel", "day", "gmv"],
+      "types": {"region": "text", "channel": "text", "day": "date", "gmv": "number"}}
+
+
+def spanned(run_id, rows, span, *, columns, types, sql, masked=()):
+    """Agent 交一段行（span），口径卡对它求和。masked：这次运行每份快照都遮的列。"""
+    run = Fake(run_id)
+    run.tool("fetch", sql, columns, rows, types=types)
+    bot = run.agent("bot", sql, columns, rows, types=types, span=span)
+    first, last = span
+    at = columns.index("gmv")
+    run.card([metric("total", "sum(vars.kpi.amounts)", sum(r[at] for r in rows[first:last + 1]),
+                     [amounts_in(bot, rows, span=span)])])
+    if masked:
+        run.masked = {a: frozenset(masked) for a in run.queries}
+    return run
+
+
+def element(run_id, rows, span, *, index=0, sql=None, columns=COLUMNS, types=TYPES):
+    """口径卡取 Agent 数组字段的第 index 个元素（单格）。"""
+    run = Fake(run_id)
+    sql = sql or region_sql()
+    run.tool("fetch", sql, columns, rows, types=types)
+    bot = run.agent("bot", sql, columns, rows, types=types, span=span)
+    row, at = span[0] + index, columns.index("gmv")
+    item = {"path": f"vars.kpi.amounts[{index}]", "value": rows[row][at], "node_id": "bot", "via": "agent_field",
+            "field": f"amounts[{index}]", "ref": f"Q1.r{row}.gmv", "artifact": bot,
+            "locator": {"row": row, "column": "gmv"}, "eid": f"ev:cell:{bot[:8]}:{row}:gmv", "status": "verified"}
+    run.card([metric("first", f"vars.kpi.amounts[{index}]", rows[row][at], [item])])
+    return run
+
+
+def compared(base, target):
+    result = compare.compare(base.side(), target.side())
+    json.dumps(result, allow_nan=False)
+    return result
+
+
+def one(base, target, metric_id):
+    return {m["metric_id"]: m for m in compared(base, target)["metrics"]}[metric_id]
+
+
+# --------------------------------------------------------------------------- 遮罩下的核对
+
+#: 东、南、西各两天；引用第 1 至 4 行
+SIX = [["东区", "09-01", 100.0], ["东区", "09-02", 100.0], ["南区", "09-01", 50.0], ["南区", "09-02", 50.0],
+       ["西区", "09-01", 30.0], ["西区", "09-02", 30.0]]
+#: 东区这次多了一天，南区的第二天挤到第 5 行（还在快照里，只是不在引用的行里）
+PUSHED = [["东区", "09-07", 100.0], ["东区", "09-08", 100.0], ["东区", "09-09", 100.0], ["南区", "09-08", 50.0],
+          ["南区", "09-09", 50.0], ["西区", "09-08", 30.0], ["西区", "09-09", 30.0]]
+#: 对照：南区这次真的只有一天
+FEWER = [["东区", "09-07", 100.0], ["东区", "09-08", 100.0], ["东区", "09-09", 100.0], ["南区", "09-08", 50.0],
+         ["西区", "09-08", 30.0], ["西区", "09-09", 30.0]]
+
+
+def test_a_masked_dimension_still_tells_rows_pushed_out_from_rows_gone():
+    """区域列设了遮罩（两边都遮、只有一边遮）：它是唯一认得出行的那一列。以前遮了就不拿它认、只能按行号认，南区第二天
+    挤出引用的行段和南区真的少了一天的输出一字不差，都是「数据变化」。现在照样拿它核对，挤出去的说不清，说明只写行号；
+    真的少了的照常是数据变化。区域名在哪儿都不出现。"""
+    for label, mb, mt in (("两边都遮", ("region",), ("region",)), ("只有这次遮", (), ("region",)),
+                          ("只有基准遮", ("region",), ())):
+        moved = compared(spanned("r1", SIX, (0, 3), **DAY, masked=mb), spanned("r2", PUSHED, (0, 3), **DAY, masked=mt))
+        total = {m["metric_id"]: m for m in moved["metrics"]}["total"]
+        assert total["classification"] == "incomparable" and total["attribution"] is None, label
+        assert "第 5 行" in total["reasons"][0] and "维度列设了遮罩" in total["reasons"][0], label
+        assert compare.MASKED_NOTE in moved["notes"], label
+        gone = compared(spanned("r1", SIX, (0, 3), **DAY, masked=mb), spanned("r2", FEWER, (0, 3), **DAY, masked=mt))
+        total = {m["metric_id"]: m for m in gone["metrics"]}["total"]
+        assert total["classification"] == "data_only", (label, total["reasons"])
+        for result in (moved, gone):
+            text = json.dumps(result, ensure_ascii=False)
+            assert not [r for r in REGIONS if r in text], label
+
+
+def test_a_masked_dimension_with_nothing_moved_is_still_a_data_change():
+    """遮罩下照样核对，但没挪的不能因为看不见值就一律说不清：两次引用的是同样几个区域的同样几天、只是金额变了，照常是
+    数据变化（定时周报常见的取前几名、区域列设了遮罩，不能每周都说不清）。"""
+    changed = [[r, d, v * 1.1] for r, d, v in SIX]
+    total = one(spanned("r1", SIX, (0, 3), **DAY, masked=("region",)),
+                spanned("r2", changed, (0, 3), **DAY, masked=("region",)), "total")
+    assert total["classification"] == "data_only", total["reasons"]
+
+
+# --------------------------------------------------------------------------- 单格、归因维度
+
+def test_an_element_whose_region_had_one_row_at_base_is_not_the_same_row():
+    """单格（Agent 两次都交第 2 至 3 行，口径卡取第一个元素即第 2 行）：基准运行那是南区唯一的一天（50）；这次东区没有
+    数据，南区两天排在第 1、2 行，第 2 行是南区的第二天（45），同一天的南区其实是 60。两次都是南区，可不是同样的那一行，
+    说不清——按行号认会写成 50 → 45，正负号都反了。基准运行那边南区只有引用的那一行时，「排在前面的行数」也要比。"""
+    before = [["东区", "2026-09-01", 100.0], ["南区", "2026-09-01", 50.0], ["西区", "2026-09-01", 30.0]]
+    after = [["南区", "2026-09-08", 60.0], ["南区", "2026-09-09", 45.0], ["西区", "2026-09-08", 30.0]]
+    m = one(element("r1", before, (1, 2), **DAY), element("r2", after, (1, 2), **DAY), "first")
+    assert m["classification"] == "incomparable", m["reasons"]
+    m = one(element("r1", after, (1, 2), **DAY), element("r2", before, (1, 2), **DAY), "first")
+    assert m["classification"] == "incomparable", m["reasons"]
+
+
+def test_attribution_dimensions_are_checked_on_each_side_own_cited_rows():
+    """渠道在引用的行里两边没有相同的值（不当维度），归因按区域说。南区这次有一行（线下）在引用的行外面：按两边各自
+    引用的行认维度，查得出南区不是同一组数；只拿一边的引用行认维度（两边都用基准的、或都用这次的），会放过它，归因
+    把南区写成 80 → 90，和南区真的只有一行的对照一字不差。两个方向都核对。"""
+    base = [["东区", "线上", "09-01", 100.0], ["东区", "线下", "09-01", 50.0], ["南区", "线上", "09-01", 80.0],
+            ["西区", "线上", "09-01", 30.0]]
+    this = [["东区", "团购", "09-08", 100.0], ["东区", "团购", "09-09", 60.0], ["南区", "团购", "09-08", 90.0],
+            ["南区", "线下", "09-08", 40.0], ["西区", "线上", "09-08", 30.0]]
+    for b, t in ((base, this), (this, base)):
+        m = one(spanned("r1", b, (0, 2), **CH), spanned("r2", t, (0, 2), **CH), "total")
+        assert m["classification"] == "incomparable", m["reasons"]
+
+
+def test_an_element_on_a_misshapen_snapshot_is_not_taken_as_the_same():
+    """单格，一边快照里有一行和列对不上（少了一格）：认不出各行是哪个维度值，说不清，不按行号当成同一个数。"""
+    broken = [["北区", 90.0, 0.0], *NEXT_ROWS[:2], ["西区", 600.0]]
+    for b, t in ((BASE_ROWS, broken), (broken, BASE_ROWS)):
+        m = one(element("r1", b, (0, 2)), element("r2", t, (0, 2)), "first")
+        assert m["classification"] == "incomparable" and "同一个维度值" in m["reasons"][0], m["reasons"]
+
+
+# --------------------------------------------------------------------------- 严格拆分与写入者
+
+def test_no_strict_split_when_the_writer_of_an_input_was_reconfigured():
+    """口径 ×2 改成 ×3（v1 → v2），输入 vars.adj 由代码节点「调整」写入，这次「调整」也改了代码：旧口径用到的输入本身
+    就因为配置变了而不同，拆成「口径效应 12、数据效应 4」会把代码改动带来的变化算成数据效应。不拆，说明原因。"""
+
+    def run(run_id, rows, value, factor, version, code):
+        graph = chain(node("start", "input", "开始"), node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": region_sql()}),
+                      node("calc", "code", "调整", code=code, assign_to="adj"),
+                      node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version=version),
+                      node("out", "output", "成果"))
+        r = Fake(run_id, graph=graph)
+        r.tool("fetch", region_sql(), rows=rows)
+        r.code("calc", value, "adj")
+        r.card([metric("calc", f"vars.adj * {factor}", value * factor,
+                       [scalar_in("vars.adj", value, "calc", "code")])], version=version)
+        return r
+
+    m = one(run("r1", BASE_ROWS, 10, 2, "v1", "x = 1"), run("r2", NEXT_ROWS, 12, 3, "v2", "x = 2"), "calc")
+    assert m["decomposition"] is None and "无法严格拆分" in m["decomposition_note"], m
+
+
+def test_an_unknown_writer_counts_queries_on_parallel_branches():
+    """记不出 vars.adj 的写入者：并行支路（开始 → 旁路取数 → 成果）上的查询在口径卡之前可能执行过，也算进它上游。只有
+    旁路取数的结果变了是数据变化；旁路取数只是代入的日期不同，是参数变化。"""
+
+    def run(run_id, value, aside_rows, aside_sql):
+        graph = chain(node("start", "input", "开始"),
+                      node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": region_sql()}),
+                      node("calc", "code", "调整", code="x", assign_to="adj"),
+                      node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1"),
+                      node("out", "output", "成果"))
+        graph["nodes"].append(node("aside", "tool", "旁路取数", tool=TOOL, args={"sql": region_sql()}))
+        graph["edges"] += [{"source": "start", "target": "aside"}, {"source": "aside", "target": "out"}]
+        r = Fake(run_id, graph=graph)
+        r.tool("fetch", region_sql(), rows=BASE_ROWS)
+        r.tool("aside", aside_sql, rows=aside_rows)
+        r.code("calc", value, "adj")
+        r.card([metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, None, "state")])])
+        return r
+
+    m = one(run("r1", 10, BASE_ROWS, region_sql()), run("r2", 12, NEXT_ROWS, region_sql()), "calc")
+    assert m["classification"] == "data_only", m["reasons"]
+    m = one(run("r1", 10, BASE_ROWS, region_sql()), run("r2", 12, NEXT_ROWS, region_sql(WEEK37)), "calc")
+    assert m["classification"] == "params_only", m["reasons"]