双盲复算复核整改（第 2 轮）：遮罩的分组列不收窄，收窄时去掉 HAVING，审批卡带上处置和写目录的预告，同一次裁决里的有效记录条件合起来写

节点（nodes/dual_check.py）
- 差异说明收窄到这一格时，分组键的结果列或它用到的列设了遮罩，就不收窄、照整条比。原流程、盲算各按自己那份快照
  的遮罩（查询当时记下的加数据源现在设的）判断。以前按这一行的分组值补等值条件，「region = '华东'」这样的遮罩列
  取值会跟着差异说明进审批 payload、human.requested 事件、对照工件和 dual_check_catalog 日志
- 按行收窄时连 HAVING 一起去掉：这一行既然在结果里，它所在的组就满足 HAVING。留着会多出一条「分组后的筛选」，
  还会占掉「只差一个方面」的名额，真正的差异（比如时间列）就写不进目录
- 审批 payload 在约定的几项之外带上实际生效的 on_reject。摘要里每个指标另带 catalog：选 a、选 b 时，按发起审批
  时的目录各会往数据目录里写什么（{write: true, source, table, item, value}），不写的给出原因（{write: false,
  reason}）。审核人要看得出不采用原流程会不会让运行失败、某个选择会不会被记成人工确认的目录项
  （on_reject 以前只能从图快照里读，子工作流里的节点读不到）

写数据目录（dual_check_facts.py）
- 写入的判断拆成不读库的 _plan：真写（_write）和审批卡上的预告（preview）用同一套判断，不会说一套写一套
- 同一次裁决里落到同一项上的几个说法合成一项（_merged）：
  - 说法一样的只写一次，备注列出每个指标
  - 有效记录条件用 AND 合成一句。derive 只收排除无效记录的条件，「排除退款」和「排除测试单」并不矛盾，以前两条
    都被丢掉，原因还写成「互相矛盾」
  - 业务日期、去重规则说法不同的仍然都不写，原因改为「给出了不同的说法」
- 合并后再和目录里已有的推断条件合起来写时，要求每个被采纳的一边都按已有条件筛过；有一个没按它筛，就不动并说明
  是哪个指标

测试
- 遮罩分组列（原流程侧、盲算侧）的取值不出现在 payload、对照工件、节点产出和节点事件里
- HAVING 收窄后只剩时间列一处差异，并写进目录
- payload 的 on_reject、摘要里的预告（一致、已确认不覆盖、指标自己的范围几种情形）
- 有效记录条件合并（含和已有推断条件合写、有一个被采纳的一边没用已有条件时不动），同一说法只写一次
- 补上复核指出的三处接线缺口，各有一条图级用例：
  - 带取舍的列名（gmv_by_order_date、paid_orders）不交给盲算
  - 取值在运行输入里的排除条件不写目录
  - 分组列上的 is_test = 0 不写目录
- 13 处变异（遮罩判断两侧、HAVING、上面三处接线、合并、预告、on_reject 等）去掉后，相关测试都会变红

backend/app/engine/dual_check_facts.py

@@ -23,10 +23,17 @@ is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不
 
 **怎么写。** 照对话提案的规矩核对（plan_patch：列在不在表结构里、取值格式），记为人工填写、已确认，备注写明来自哪次
 运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖，照实交回原因；已经是同样的说法就什么都
-不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边也按它筛了的，和新条件合起来写
+不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边（几个指标并成一项时是每一个）
-（已确认的不改，照实说）；被采纳的一边没按它筛的，两者的关系拿不准，不动。推断的去重规则说的未必是同一件事，也不
+也按它筛了的，和新条件合起来写（已确认的不改，照实说）；有一个没按它筛的，两者的关系拿不准，不动。推断的去重规则
-替换。同一次裁决里几个指标对同一项给出不同说法的，一个都不写。整个写入由节点放在 ask() 之后的一个 @task 里：审批
+说的未必是同一件事，也不替换。整个写入由节点放在 ask() 之后的一个 @task 里：审批恢复重放、接着运行时不会再写一遍。
-恢复重放、接着运行时不会再写一遍。
+
+**同一次裁决里几个指标落到同一项上。** 说法一样的写一次，备注里列出每个指标。有效记录条件是能用 AND 叠加的排除条件
+（这里只收排除无效记录的），几个指标各自采纳的条件都有人的选择撑着，合成一句写——「排除退款」和「排除测试单」并不
+矛盾。业务日期一张表只有一个，去重规则说的未必是同一件事，说法不同就一个都不写。
+
+**预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么（preview）：按发起审批时的目录，选原流程、选盲算各会
+写什么、不写的为什么。和真写是同一套判断（derive 认事实、_plan 对目录），所以只在两处可能和最后写的不一样：审批停下
+期间目录被人改了，或者同一次裁决里几个指标落到同一项上（上一段）。
 
 catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。
 """
@@ -35,7 +42,7 @@ from __future__ import annotations
 import logging
 import re
 from collections.abc import Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any
 
 from sqlalchemy import select
@@ -79,7 +86,8 @@ _SCOPE_DIGITS = 2
 @dataclass(frozen=True)
 class Fact:
     """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明；sql、dialect 是被采纳
-    那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。"""
+    那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。also 是同一次裁决里落到同一项上、
+    并进这一项的其余几个指标（各自的指标、选择、写法），value 是合起来之后的说法。"""
 
     source: str
     table: str
@@ -89,6 +97,18 @@ class Fact:
     choice: str
     sql: str = ""
     dialect: str | None = None
+    also: tuple[Fact, ...] = ()
+
+
+def _members(fact: Fact) -> tuple[Fact, ...]:
+    """这一项背后的每个指标的选择：并过的连同并进来的。"""
+    return (fact, *fact.also)
+
+
+def _names(fact: Fact) -> str:
+    """说明里的指标叫法：「销售额」，几个指标并成一项时「销售额」、「订单金额」。"""
+    names = list(dict.fromkeys(m.metric for m in _members(fact)))
+    return "、".join(f"「{n}」" for n in names[:4]) + (f"等 {len(names)} 个" if len(names) > 4 else "")
 
 
 def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
@@ -402,28 +422,45 @@ def _conditions_of(sql: str, read: str | None) -> set[str] | None:
     return {_normalized(part, read) for where in tree.find_all(exp.Where) for part in _and_parts(where.this)}
 
 
+def _and_join(base: str | None, values: list[str], dialect: str | None) -> str | None:
+    """几句条件用 AND 接成一句：已经有的（base 里的、前面接过的，按 _normalized 比）不再接；最外层是 OR 的加括号，
+    不然后面接的条件只并到 OR 的后一半上。每句保留原来的写法。有一句解析不了返回 None。"""
+    from sqlglot import exp
+
+    read = _read(dialect)
+    texts: list[str] = []
+    seen: set[str] = set()
+    for text in [*([base] if base else []), *values]:
+        tree = _parse(str(text), read)
+        if tree is None:
+            return None
+        keys = {_normalized(p, read) for p in _and_parts(tree)}
+        if keys <= seen:
+            continue
+        seen |= keys
+        texts.append(f"({text})" if isinstance(tree, exp.Or) else str(text))
+    return " AND ".join(texts)
+
+
 def _combined_filter(existing: str, fact: Fact) -> tuple[str | None, str | None]:
     """目录里已有的有效记录条件和这次的新条件合起来：(合起来的整句条件, None) 或 (None, 为什么不动)。
 
     有效记录条件是一整句（copilot 提案也要求「原有条件也要带上」）。只写新条件会把两边都按着筛的原有条件静默替换掉；
-    原有条件被采纳的一边没按它筛，审核人的选择说明不了它对不对，也不能替人删掉。"""
+    原有条件有一个被采纳的一边没按它筛，审核人的选择说明不了它对不对（合起来写就等于替人确认了它），也不能替人删掉。"""
     label = catalog.TABLE_FIELD_LABEL["valid_filter"]
     read = _read(fact.dialect)
-    have_tree, new_tree = _parse(existing, read), _parse(str(fact.value), read)
+    have_tree = _parse(existing, read)
-    if have_tree is None or new_tree is None:
+    combined = _and_join(existing, [str(m.value) for m in _members(fact)], fact.dialect) \
+        if have_tree is not None else None
+    if combined is None:
         return None, f"「{fact.table}」已有的{label}「{existing}」无法解析，与这次的条件合不起来，未改动"
-    have = [_normalized(p, read) for p in _and_parts(have_tree)]
+    have = {_normalized(p, read) for p in _and_parts(have_tree)}
-    used = _conditions_of(fact.sql, read) or set()
+    for member in _members(fact):
-    if not set(have) <= used:
+        if not have <= (_conditions_of(member.sql, read) or set()):
-        return None, (f"「{fact.table}」已有的{label}「{existing}」，被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}"
+            return None, (f"「{fact.table}」已有的{label}「{existing}」，指标「{member.metric}」被采纳的"
-                      "没有按它筛选，两者的关系无法确定，未改动。如需更改，请到数据目录中修改")
+                          f"{SIDE_LABEL.get(member.choice, member.choice)}没有按它筛选，两者的关系无法确定，未改动。"
-    if {_normalized(p, read) for p in _and_parts(new_tree)} <= set(have):
+                          "如需更改，请到数据目录中修改")
-        return existing, None
+    return combined, None
-    from sqlglot import exp
-
-    # 原有条件最外层是 OR 的，接 AND 之前要加括号，不然新条件只并到 OR 的后一半上
-    head = f"({existing})" if isinstance(have_tree, exp.Or) else existing
-    return f"{head} AND {fact.value}", None
 
 
 def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any],
@@ -491,9 +528,13 @@ def _aggregate_pair(a: Any, b: Any, dialect: Any) -> tuple[str, tuple[bool, bool
 
 
 def note_of(fact: Fact, *, run_id: str, node_title: str) -> str:
-    """目录项上的备注：来自哪次运行、哪个节点、哪个指标、采纳了哪一边。"""
+    """目录项上的备注：来自哪次运行、哪个节点、哪些指标、各采纳了哪一边。"""
-    return (f"双盲复算（运行 {run_id}，节点「{node_title}」）：审核人在指标「{fact.metric}」上"
+    parts = []
-            f"采用了{SIDE_LABEL.get(fact.choice, fact.choice)}的写法")
+    for side, label in SIDE_LABEL.items():
+        names = list(dict.fromkeys(m.metric for m in _members(fact) if m.choice == side))
+        if names:
+            parts.append(f"在指标{'、'.join(f'「{n}」' for n in names)}上采用了{label}的写法")
+    return f"双盲复算（运行 {run_id}，节点「{node_title}」）：审核人" + "，".join(parts)
 
 
 def _value_text(item: str, value: Any) -> str:
@@ -509,21 +550,38 @@ def _same_fact(item: str, a: Any, b: Any) -> bool:
     return isinstance(a, str) and isinstance(b, str) and " ".join(a.split()) == " ".join(b.split())
 
 
-def _contradiction(facts: list[Fact]) -> str:
+def _differing(facts: list[Fact]) -> str:
     """同一次裁决里对同一项给出的不同说法，说清是谁、各说了什么。"""
     first = facts[0]
     label = catalog.TABLE_FIELD_LABEL.get(first.item, first.item)
     said = "；".join(f"指标「{f.metric}」采纳的是「{_value_text(f.item, f.value)}」" for f in facts[:4])
-    return (f"同一次裁决里对「{first.table}」的{label}给出的选择互相矛盾（{said}），无法确定哪一个是数据事实，"
+    return (f"同一次裁决里对「{first.table}」的{label}给出了不同的说法（{said}），无法确定哪一个是数据事实，"
             "都未写入数据目录")
 
 
+def _merged(members: list[Fact]) -> tuple[Fact | None, str | None]:
+    """同一次裁决里落到同一项上的几个说法合成一项：(合起来的那一项, None) 或 (None, 为什么都不写)。
+
+    说法都一样的写一次。有效记录条件用 AND 合成一句（derive 只收排除无效记录的条件，几句叠起来不会互相打架）。
+    业务日期一张表只有一个、去重规则说的未必是同一件事，说法不同就都不写。"""
+    first, rest = members[0], tuple(members[1:])
+    if not rest:
+        return first, None
+    if first.item == "valid_filter":
+        value = _and_join(None, [str(m.value) for m in members], first.dialect)
+        return (replace(first, value=value, also=rest), None) if value is not None else (None, _differing(members))
+    if all(_same_fact(first.item, m.value, first.value) for m in rest):
+        return replace(first, also=rest), None
+    return None, _differing(members)
+
+
 async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str,
                          actor: str | None) -> dict[str, list[dict[str, Any]]]:
     """把审核人的选择写进数据目录：{written: [{source, table, item}], skipped: [{reason}]}。永远不抛：数据源没了、
     目录写不进去，都落在 skipped 里照实说。
 
-    先把每个候选认成目录项，再按（数据源、表、项）归拢：同一项上说法互相矛盾的整组不写，剩下的逐个写。"""
+    先把每个候选认成目录项，再按（数据源、表、项）归拢成一项（_merged：说法一样的、有效记录条件合起来写，说法
+    不同的整组不写），逐项写。"""
     skipped: list[dict[str, Any]] = []
     facts: list[Fact] = []
     for candidate in candidates:
@@ -549,10 +607,10 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no
         groups.setdefault((fact.source, fact.table, fact.item), []).append(fact)
     written: list[dict[str, Any]] = []
     for members in groups.values():
-        if any(not _same_fact(f.item, f.value, members[0].value) for f in members[1:]):
+        fact, why = _merged(members)
-            skipped.append({"reason": _contradiction(members)})
+        if fact is None:
+            skipped.append({"reason": why})
             continue
-        for fact in members:
         done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor)
         if done:
             written.append({"source": fact.source, "table": fact.table, "item": fact.item})
@@ -566,29 +624,25 @@ async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None
         async with SessionLocal() as session:
             row = (await session.execute(select(DataSource).where(DataSource.name == fact.source))).scalar_one_or_none()
             if row is None:
-                return False, f"数据源「{fact.source}」已不存在，指标「{fact.metric}」的选择未写入数据目录"
+                return False, f"数据源「{fact.source}」已不存在，指标{_names(fact)}的选择未写入数据目录"
             return await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title),
                                 actor=actor)
     except Exception:  # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记
         logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", fact.source)
-        return False, f"指标「{fact.metric}」的选择写入数据目录时出错，未写入"
+        return False, f"指标{_names(fact)}的选择写入数据目录时出错，未写入"
 
 
-async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]:
+def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]:
-    """写一项。(写了没有, 没写的原因)。"""
+    """按目录里这张表现在的说法，算出这一项要写成什么：(要写的取值, None) 或 (None, 为什么不写)。不读库、不写库：
+    真写（_write）和审批卡上的预告（preview）用的是这同一套判断。"""
     label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
-    side = SIDE_LABEL.get(fact.choice, fact.choice)
+    names = _names(fact)
-    tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None
-    for _ in range(_RETRIES):
-        entry = await catalog.read_entry(session, row.id, fact.table)
-        notes = entry.notes if entry else {}
-        version = entry.version if entry else 0
     stored = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None
     if stored is not None and stored.get("status") == "rejected" and _same_fact(fact.item, stored.get("value"),
                                                                                  fact.value):
         # 同一个说法以前被人明确驳回过：两次人工判断对不上，不替人拿主意
-            return False, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回，"
+        return None, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回，"
-                           f"与这次指标「{fact.metric}」的选择相反，未写入。如需更改，请到数据目录中修改")
+                      f"与这次指标{names}的选择相反，未写入。如需更改，请到数据目录中修改")
     live = stored if stored is not None and stored.get("status") != "rejected" else None
     value = fact.value
     if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \
@@ -599,36 +653,74 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor:
             and live["value"].strip() and not _same_fact(fact.item, live["value"], value):
         combined, why = _combined_filter(live["value"].strip(), fact)
         if combined is None:
-                return False, why
+            return None, why
         if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]):
-                return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，被采纳的"
+            adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.also else "这次几个指标被采纳的写法"
-                               f"{side}另外多一个条件「{fact.value}」，未改动。如需补上，请到数据目录中修改")
+            return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，{adopted}"
+                          f"另外多出条件「{fact.value}」，未改动。如需补上，请到数据目录中修改")
         value = combined
     elif fact.item == "dedup" and isinstance(live, dict) and live.get("status") != "confirmed" \
             and not _same_fact(fact.item, live.get("value"), value):
         # 推断的去重规则说的未必是同一件事（按哪一列去重、哪种汇总要去重），替换掉就把它丢了
-            return False, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」，"
+        return None, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」，"
-                           f"与这次指标「{fact.metric}」采纳的「{value}」不同，无法确定两者的关系，未覆盖。"
+                      f"与这次指标{names}采纳的「{value}」不同，无法确定两者的关系，未覆盖。"
                       "如需更改，请到数据目录中修改")
-        changes = [{"path": fact.item, "value": value, "reason": note}]
+    plan = catalog.plan_patch(notes, [{"path": fact.item, "value": value}], table=fact.table, tables=tables)
-        plan = catalog.plan_patch(notes, changes, table=fact.table, tables=tables)
     if plan.problems:
-            return False, f"「{fact.table}」的{label}未写入数据目录：{plan.problems[0]}"
+        return None, f"「{fact.table}」的{label}未写入数据目录：{plan.problems[0]}"
     change = plan.changes[0]
     if change.state == "same":
-            return False, (f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」，"
+        return None, f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」，无需重复写入"
-                           "无需重复写入")
     if change.before_status == "confirmed":
-            return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」，"
+        return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」，"
-                           f"与这次指标「{fact.metric}」采纳的「{_value_text(fact.item, value)}」不同，未覆盖。"
+                      f"与这次指标{names}采纳的「{_value_text(fact.item, value)}」不同，未覆盖。"
                       "如需更改，请到数据目录中修改")
+    return change.after, None
+
+
+def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
+            notes: Mapping[str, Mapping[str, Any]] | None = None) -> dict[str, dict[str, Any]]:
+    """审批卡上的预告：审核人在这个指标上选原流程（a）、选盲算（b），按目前的数据目录各会往目录里写什么。
+
+    candidate 同 derive（choice 不用给）；notes 是这个数据源的目录（{表名: notes}）。返回 {a: …, b: …}，每边是
+    {write: true, source, table, item, value}（value 是真要写进去的说法：和目录里已有的有效记录条件合起来的，是合起来
+    之后的整句）或 {write: false, reason}。和恢复后真写是同一套判断，只在模块说明里说的两处可能不一样。永远不抛。"""
+    notes = notes or {}
+    tables = schema_cache.get("tables") if isinstance(schema_cache, Mapping) else None
+    out: dict[str, dict[str, Any]] = {}
+    for side in SIDE_LABEL:
+        try:
+            fact, why = derive({**candidate, "choice": side}, schema_cache, notes)
+            value = None
+            if fact is not None:
+                value, why = _plan(notes.get(fact.table) or {}, fact, tables)
+        except Exception:  # noqa: BLE001 - 预告算不出来不能拦住审批：照实说，以写入后的结果为准
+            logger.exception("双盲复算预告数据目录的写入失败（指标 %s）", candidate.get("metric"))
+            fact, why = None, f"指标「{candidate.get('metric')}」的选择核对数据目录时出错，无法预告是否写入"
+        out[side] = {"write": True, "source": fact.source, "table": fact.table, "item": fact.item, "value": value} \
+            if fact is not None and why is None else {"write": False, "reason": why}
+    return out
+
+
+async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]:
+    """写一项。(写了没有, 没写的原因)。写什么、写不写由 _plan 按读到的目录定；版本对不上（别人刚改过）重读再来。"""
+    label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
+    tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None
+    for _ in range(_RETRIES):
+        entry = await catalog.read_entry(session, row.id, fact.table)
+        notes = entry.notes if entry else {}
+        version = entry.version if entry else 0
+        value, why = _plan(notes, fact, tables)
+        if why is not None:
+            return False, why
         at = catalog.now_iso()
         try:
-            merged = catalog.apply_patch(notes, changes, table=fact.table, tables=tables, at=at)
+            merged = catalog.apply_patch(notes, [{"path": fact.item, "value": value, "reason": note}],
+                                         table=fact.table, tables=tables, at=at)
         except catalog.CatalogInvalid as e:
             return False, f"「{fact.table}」的{label}未写入数据目录：{e}"
         # 不论原来是谁提的：这一项现在出自人的选择，记为人工填写、已确认，备注写明出处
-        merged[fact.item] = catalog.make_item(change.after, "human", "confirmed", note=note, at=at)
+        merged[fact.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at)
         try:
             await catalog.write_entry(session, row.id, fact.table, merged, if_version=version, actor=actor)
         except catalog.CatalogConflict:
@@ -640,4 +732,4 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor:
     return False, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"
 
 
-__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "record_choices"]
+__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"]

backend/app/engine/nodes/dual_check.py

@@ -14,11 +14,14 @@
 里看不出日期、口径的取舍时才给，好让它分得清哪一项是哪一项。
 
 **一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）：
-两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件），一条查询算好几列、分组后取一行时
+两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件，HAVING 一并去掉），一条查询算好几列、
-才不会混进「选出的列」「分组」这类和这一格无关的差别。完整对照落成一件工件（审批 payload 经 runner._safe 会截断，
+分组后取一行时才不会混进「选出的列」「分组」这类和这一格无关的差别。分组的列设了遮罩时不收窄：这一行的分组值写进
+SQL，就跟着差异说明进了审批卡、对照工件和目录日志。完整对照落成一件工件（审批 payload 经 runner._safe 会截断，
 长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有
 缺口；b、neither、没选的写进产出的 issuance_gaps，成果节点的出具契约据此降档（on_reject 为 fail 时节点直接失败）。
 两边只差一个方面、又对得上数据目录事实的，选择写进数据目录（engine/dual_check_facts.py，指标自己的筛选范围不写）。
+选了会怎样要在卡上看得出：payload 在约定的几项之外带上实际生效的 on_reject，摘要里每个指标带 catalog——选 a、选 b
+按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）。
 盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。
 
 **产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial；
@@ -54,6 +57,7 @@ from app.db.models import DataSource
 from app.engine import recompute
 from app.engine.approval import read_decision
 from app.engine.context import NodeContext, NodeError
+from app.engine.dual_check_facts import preview, record_choices
 from app.engine.evidence import build_catalog, is_number, next_exec
 from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \
     parse_expression
@@ -907,10 +911,14 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
     """不一致的指标逐个说清差在哪，完整对照落成工件，组好审批的 payload。返回 {payload, facts, card_artifact}。
 
     两条 SQL 先收窄到这一格再比（_narrowed）：原流程一条查询算好几列、分组后取一行时，整条拿去比会混进「选出的列」
-    「分组」这类和这一格无关的差别，审批卡上只列前几条，审核人看到的多半是噪声，只差一个方面的也就写不进目录。
+    「分组」这类和这一格无关的差别，审批卡上只列前几条，审核人看到的多半是噪声，只差一个方面的也就写不进目录。两边
+    各按自己那份快照要遮的列收窄（原流程的、盲算的快照各记着查询当时的遮罩，再加数据源现在设的）。
     facts 是每个不一致指标写数据目录要用的东西（dual_check_facts.derive 的候选，少一个 choice）：只有一个输入对不上时是
     那一处完整的差异条目（带两边比较的写法和涉及的表、列），几个输入都对不上时只说「不止一处」；scope 是这个指标自己的
-    范围（口径说明、所在分组、运行输入、分组的列），有效记录条件据此排除指标自己的筛选范围。
+    范围（口径说明、所在分组、运行输入、分组的列），有效记录条件据此排除指标自己的筛选范围。scope 只在节点里用、不进
+    payload 和工件（所在分组的取值里可能有遮罩列的）。
+    payload 在约定的几项之外带 on_reject（实际生效的处置），摘要里每个指标带 catalog（选 a、选 b 各会往数据目录里写
+    什么，dual_check_facts.preview）：审核人在卡上要看得出没采用原流程会不会让运行失败、选了会不会记成人工确认的目录项。
     """
     rows = [r for r in checked["rows"] if r["status"] == DIFFERENT]
     results = checked["results"]
@@ -938,9 +946,11 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
             if inp["key"] not in row["differing"]:
                 continue
             blind_snap, _ = sources.snapshot(str(blind["artifact"])) if blind.get("artifact") else (None, None)
-            a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind) or inp["sql"]
+            blind_hidden = await sources.hidden(str(blind["artifact"])) if blind.get("artifact") else frozenset()
-            b_sql = _narrowed(blind.get("sql"), blind.get("column"), _record(blind_snap, blind.get("row")), kind) \
+            a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind,
-                or blind.get("sql")
+                              await sources.hidden(inp["artifact"])) or inp["sql"]
+            b_sql = _narrowed(blind.get("sql"), blind.get("column"), _record(blind_snap, blind.get("row")), kind,
+                              blind_hidden) or blind.get("sql")
             diff = await _explain(a_sql, b_sql, infos[inp["source"]])
             per_input.append((inp, blind, diff))
             compared.append((a_sql, str(b_sql or "")))
@@ -960,22 +970,24 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
                      "items": [{"aspect": i.get("aspect"), "text": i.get("text")} for i in merged["items"]],
                      "fallback": merged["fallback"], "text_diff": merged["text_diff"]},
         })
-        summary.append({
-            "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"],
-            "original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"],
-            "delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]],
-            "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]),
-        })
         first = per_input[0][0] if per_input else (row["inputs"][0] if row["inputs"] else {})
         single = compared[0] if len(per_input) == 1 else None
+        info = infos.get(str(first.get("source"))) or {}
         facts[row["metric_id"]] = {
             "metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"),
-            "dialect": (infos.get(str(first.get("source"))) or {}).get("kind"),
+            "dialect": info.get("kind"),
             "sql": {"a": single[0], "b": single[1]} if single else {},
             "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None,
                      "items": merged["items"]},
             "scope": {"texts": [t for t in dict.fromkeys(scope_texts) if t], "columns": sorted(scope_columns)},
         }
+        summary.append({
+            "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"],
+            "original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"],
+            "delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]],
+            "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]),
+            "catalog": preview(facts[row["metric_id"]], info.get("schema"), info.get("notes")),
+        })
     content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))}
     try:
         artifact = await artifact_store.put_json(content, kind="dual_check", run_id=ctx.run.run_id,
@@ -985,7 +997,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
     payload = {
         "kind": "dual_check", "node_id": ctx.node.id, "mode": "input",
         "title": f"双盲复算：{len(rows)} 个指标两边结果不一致，请选择采用哪一边的结果",
-        "card_artifact": artifact, "summary": summary,
+        "card_artifact": artifact, "summary": summary, "on_reject": options.on_reject,
     }
     names = "、".join(f"「{r['name']}」" for r in rows[:5]) + (f"等 {len(rows)} 个" if len(rows) > 5 else "")
     ctx.emit(EventType.LOG, level="warn", code="dual_check_disagreed",
@@ -995,17 +1007,18 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
 
 
 async def _source_info(sources: _Sources, name: str) -> dict[str, Any]:
-    """比较两条 SQL 要用的：方言、表结构、列的中文名（数据目录里的）。数据源没了就都不给，照样比原文。"""
+    """比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名和整份数据目录。数据源没了就都不给，照样比原文。"""
     from app.data import catalog
     from app.data.sql_semantics import catalog_labels
 
     row = await sources.source(name)
     if row is None:
-        return {"kind": None, "schema": None, "names": {}}
+        return {"kind": None, "schema": None, "names": {}, "notes": {}}
     async with SessionLocal() as session:
         entries = await catalog.read_catalog(session, row.id)
+    notes = {table: entry.notes for table, entry in entries.items()}
     return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None,
-            "names": catalog_labels({table: entry.notes for table, entry in entries.items()})}
+            "names": catalog_labels(notes), "notes": notes}
 
 
 async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]:
@@ -1088,12 +1101,16 @@ def _sql_literal(value: Any) -> Any:
     return None
 
 
-def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str | None) -> str | None:
+def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str | None,
+              hidden: frozenset[str] = frozenset()) -> str | None:
     """把一条查询收窄到这一格：最外层 SELECT 只留算出这一列的那一项；分组的去掉分组，按这一行的分组值补上等值条件
-    （华东那一行 → WHERE … AND region = '华东'），排序、取行数跟着去掉。
+    （华东那一行 → WHERE … AND region = '华东'），HAVING、排序、取行数跟着去掉——这一行既然在结果里，它所在的组就
+    满足 HAVING，留着只会多出一条和这一格无关的差异。
 
     差异说明拿收窄后的两条比，才只说这一格是怎么算出来的。认不准的（SELECT *、DISTINCT、窗口函数、集合运算、分组键
-    对不上结果列、这一行没有分组值）返回 None，调用方照整条比。"""
+    对不上结果列、这一行没有分组值）返回 None，调用方照整条比。hidden 是这份快照要遮的列（小写）：分组键的结果列、
+    或者它用到的列在里面时也返回 None——分组值写进 SQL 就跟着差异说明进了审批卡、对照工件和目录日志，遮罩列的值
+    不能出现在那些地方。"""
     import sqlglot
     from sqlglot import exp
 
@@ -1128,6 +1145,8 @@ def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str
             if name is None or name not in values:
                 return None
             expr = projections[index].this if isinstance(projections[index], exp.Alias) else projections[index]
+            if name in hidden or any(col.name.lower() in hidden for col in expr.find_all(exp.Column)):
+                return None
             if values[name] is None:
                 conditions.append(exp.Is(this=expr.copy(), expression=exp.Null()))
                 continue
@@ -1135,7 +1154,7 @@ def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str
             if literal is None:
                 return None
             conditions.append(exp.EQ(this=expr.copy(), expression=literal))
-        for arg in ("group", "order", "limit", "offset"):
+        for arg in ("group", "having", "order", "limit", "offset"):
             tree.set(arg, None)
         tree = tree.where(*conditions, copy=False)
     elif tree.args.get("limit") is None and tree.args.get("offset") is None:
@@ -1233,7 +1252,6 @@ def _resolved_text(rows: list[dict[str, Any]], choices: dict[str, str]) -> str:
 
 async def _catalog(ctx: NodeContext, candidates: list[dict[str, Any]]) -> dict[str, Any]:
     from app.data.catalog import TABLE_FIELD_LABEL
-    from app.engine.dual_check_facts import record_choices
 
     result = await record_choices(candidates, run_id=ctx.run.run_id, node_title=ctx.node.title, actor=ctx.actor())
     written, skipped = result["written"], result["skipped"]

backend/tests/test_dual_check_facts.py

@@ -7,7 +7,10 @@
 - 指标自己的筛选范围（只看华东、只看线上）不是有效记录条件：取值出现在指标名称、口径说明、所在分组、运行输入里的，
   条件用到分组的列的，不是在排除退款、作废、测试这类无效记录的，都不写
 - 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖；
-  目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动；同一次裁决里互相矛盾的选择都不写
+  目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动
+- 同一次裁决里几个指标落到同一项上：说法一样的写一次；有效记录条件（都在排除无效记录）用 AND 合成一句写；业务日期
+  说法不同的都不写
+- 审批卡上的预告（preview）：选 a、选 b 各会写什么、不写的为什么，按目录里已有的说法算，和真写同一套判断
 
 差异条目一律由 sql_semantics.explain_sql_diff 真算出来，和节点交给审核人看的是同一份。表名、数字都是虚构的。
 """
@@ -18,7 +21,7 @@ import pytest
 from app.data import catalog
 from app.data.sql_semantics import explain_sql_diff
 from app.db.base import SessionLocal
-from app.engine.dual_check_facts import derive, record_choices
+from app.engine.dual_check_facts import derive, preview, record_choices
 from tests.fixtures.dual_check_shop import (
     BOTH_WRONG,
     BUYERS,
@@ -237,6 +240,13 @@ async def seed(source_id: str, notes: dict) -> None:
         await catalog.write_entry(session, source_id, "orders", notes, if_version=0, actor="数据组")
 
 
+async def seed_again(source_id: str, notes: dict) -> None:
+    """把目录这张表整份改回 notes（在现有版本上写）。"""
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, source_id, "orders")
+        await catalog.write_entry(session, source_id, "orders", notes, if_version=entry.version, actor="数据组")
+
+
 async def test_a_choice_is_written_as_a_confirmed_human_item_with_its_origin(shop):
     name, source_id = shop
     out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-42",
@@ -359,7 +369,8 @@ async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_do
 
 
 async def test_contradictory_choices_in_one_decision_write_nothing(shop):
-    """同一次裁决里，一个指标采纳了按下单日、另一个采纳了按支付日：写哪个都是替人拿主意，都不写。"""
+    """同一次裁决里，一个指标采纳了按下单日、另一个采纳了按支付日：一张表只有一个业务日期，写哪个都是替人拿主意，
+    都不写。"""
     name, source_id = shop
     out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", metric="销售额", source=name),
                                 candidate(BY_ORDER_DATE, RIGHT, "b", metric="复核销售额", source=name)],
@@ -367,10 +378,94 @@ async def test_contradictory_choices_in_one_decision_write_nothing(shop):
     assert out["written"] == []
     [skipped] = out["skipped"]
     assert "「销售额」" in skipped["reason"] and "「复核销售额」" in skipped["reason"]
-    assert "order_date" in skipped["reason"] and "paid_date" in skipped["reason"] and "互相矛盾" in skipped["reason"]
+    assert "order_date" in skipped["reason"] and "paid_date" in skipped["reason"]
+    assert "给出了不同的说法" in skipped["reason"] and "无法确定哪一个是数据事实" in skipped["reason"]
     assert await entry_of(source_id) is None
 
 
+async def test_complementary_valid_filters_in_one_decision_are_combined(shop):
+    """同一次裁决里，一个指标采纳了排除退款、另一个采纳了排除测试单：两条都在排除无效记录，各有人的选择撑着，
+    不互相矛盾，用 AND 合成一句写。"""
+    name, source_id = shop
+    out = await record_choices([
+        candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name),
+        candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name),
+        # 第三个指标采纳的是和第一个一样的条件：合并时不重复
+        candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name),
+    ], run_id="run-1", node_title="复核", actor=None)
+    assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
+    entry = await entry_of(source_id)
+    item = entry.notes["valid_filter"]
+    assert item["value"] == "status <> 'refunded' AND is_test = 0" and entry.version == 1
+    assert (item["source"], item["status"]) == ("human", "confirmed")
+    assert all(f"「{m}」" in item["note"] for m in ("销售额", "订单金额", "订单数")) and "run-1" in item["note"]
+
+
+async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_uses(shop):
+    """目录里推断的「channel <> 'test'」：几个被采纳的一边都按它筛了，合起来写；有一个没按它筛的，关系拿不准，不动。"""
+    name, source_id = shop
+    await seed(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")})
+    base = extra("channel <> 'test'")
+    both = [candidate(base, extra("status <> 'refunded'", base), "b", metric="销售额", source=name),
+            candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
+    out = await record_choices(both, run_id="run-1", node_title="复核", actor=None)
+    assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
+    assert (await entry_of(source_id)).notes["valid_filter"]["value"] == \
+        "channel <> 'test' AND status <> 'refunded' AND is_test = 0"
+
+    await seed_again(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")})
+    one_without = [both[0], candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)]
+    out = await record_choices(one_without, run_id="run-2", node_title="复核", actor=None)
+    assert out["written"] == []
+    reason = out["skipped"][0]["reason"]
+    assert "「channel <> 'test'」" in reason and "指标「订单金额」被采纳的盲算没有按它筛选" in reason and "未改动" in reason
+    assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "channel <> 'test'"
+
+
+async def test_the_same_choice_from_two_metrics_is_written_once(shop):
+    name, source_id = shop
+    out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", metric="销售额", source=name),
+                                candidate(BY_ORDER_DATE, RIGHT, "b", metric="订单金额", source=name)],
+                               run_id="run-1", node_title="复核", actor=None)
+    assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []}
+    entry = await entry_of(source_id)
+    assert entry.version == 1 and "「销售额」" in entry.notes["business_date"]["note"]
+    assert "「订单金额」" in entry.notes["business_date"]["note"]
+
+
+# --------------------------------------------------------------------------
+# 审批卡上的预告：选哪一边各会写什么（和真写同一套判断）
+# --------------------------------------------------------------------------
+
+
+def test_the_preview_says_what_each_choice_would_write():
+    picked = candidate(BY_ORDER_DATE, RIGHT, "a")
+    assert preview(picked, SCHEMA, {}) == {
+        "a": {"write": True, "source": "shop", "table": "orders", "item": "business_date",
+              "value": {"column": "order_date"}},
+        "b": {"write": True, "source": "shop", "table": "orders", "item": "business_date",
+              "value": {"column": "paid_date"}},
+    }
+    # 只有被采纳的一边有条件时才写：选另一边照实说为什么不写
+    said = preview(candidate(WITH_REFUNDS, RIGHT, "a"), SCHEMA, {})
+    assert said["a"]["write"] is False and "被采纳的原流程没有这个条件" in said["a"]["reason"]
+    assert said["b"] == {"write": True, "source": "shop", "table": "orders", "item": "valid_filter",
+                         "value": "status <> 'refunded'"}
+
+
+def test_the_preview_reads_the_catalog_as_it_is():
+    """目录里已有的说法照写入时的规矩算：确认过的不覆盖、同样的说法不重写、推断的有效记录条件合起来写。"""
+    confirmed = {"orders": {"business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}}
+    said = preview(candidate(BY_ORDER_DATE, RIGHT, "a"), SCHEMA, confirmed)
+    assert said["a"]["write"] is False and "无需重复写入" in said["a"]["reason"]
+    assert said["b"]["write"] is False and "已有人工确认过的说法「order_date」" in said["b"]["reason"]
+
+    inferred = {"orders": {"valid_filter": catalog.make_item("is_test = 0", "llm")}}
+    base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}"
+    said = preview(candidate(base, extra("status <> 'refunded'", base), "a"), SCHEMA, inferred)
+    assert said["b"]["write"] is True and said["b"]["value"] == "is_test = 0 AND status <> 'refunded'"
+
+
 async def test_a_missing_source_or_an_unrecordable_choice_is_reported(shop):
     name, _ = shop
     out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source="no_such_source"),

backend/tests/test_dual_check_node.py

@@ -4,13 +4,15 @@
 - 一致：不挂起，产出 agreed，日志 dual_check_agreed；盲算的查询发 tool.end（role: verify），台账只有一条 dual_check，
   用量计入运行；盲算 agent 的任务里没有原 SQL，也没有原流程结果的列名
 - 一个都没核成是 unchecked、只核成一部分是 partial，日志 dual_check_unchecked 说清哪几个、为什么
-- 不一致：挂起，审批 payload 和对照工件按约定；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject
+- 不一致：挂起，审批 payload 和对照工件按约定，payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往
-  为 fail 时失败
+  数据目录里写什么的预告；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject 为 fail 时失败
-- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比）
+- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比，HAVING 一并去掉）；分组的列设了
+  遮罩时不收窄，遮罩列的取值不进审批卡、对照工件、产出和事件
 - 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的
-  筛选范围不写，两条 SQL 规范化后一致的不写
+  筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写
 - 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询
-- 触顶、分组结果的维度（只给取值、遮罩列不给）、两个指标用到同一格只盲算一次、遮罩列上的盲算结果不采用
+- 触顶、分组结果的维度（只给取值、遮罩列不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的
+  盲算结果不采用
 """
 from __future__ import annotations
 
@@ -234,6 +236,23 @@ async def test_the_blind_agent_is_not_told_which_date_column_the_original_groupe
     assert "order_date" not in task and "gmv" not in task
 
 
+async def test_column_names_carrying_a_choice_are_not_given_as_hints(monkeypatch, shop):
+    """几项输入都来自查询、又没写口径说明时，列名才交给盲算区分计算式里的几项；列名里带着原流程的取舍的（按下单日、
+    已支付）不给：「gmv_by_order_date」等于告诉它按下单日归属。看不出取舍的（total）照给。"""
+    name, _ = shop
+    wide = (f"SELECT SUM(amount) AS gmv_by_order_date, COUNT(*) AS paid_orders, SUM(amount) AS total FROM orders "
+            f"WHERE status <> 'refunded' AND order_date {WEEK}")
+    aov = {"id": "aov", "name": "客单价", "unit": "元", "decimals": 2, "format": "plain",
+           "expression": "(cell(nodes.fetch, 0, 'gmv_by_order_date') + cell(nodes.fetch, 0, 'total')) "
+                         "/ cell(nodes.fetch, 0, 'paid_orders')"}
+    blind = Blind(monkeypatch, name, {})    # 盲算都交不出数：这里只看任务原文
+    await start(graph(name, wide, metrics=[aov], dual={"definitions": {}}))
+    assert len(blind.tasks) == 3
+    said = "\n".join(blind.tasks)
+    assert "gmv_by_order_date" not in said and "paid_orders" not in said
+    assert "「total」" in said
+
+
 async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop):
     name, _ = shop
     share = {"id": "gmv_k", "name": "销售额（千元）", "unit": "千元", "decimals": 2, "format": "thousands",
@@ -352,16 +371,25 @@ async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch,
     approval = await pending(run.id)
     assert approval.mode == "input" and approval.node_id == "dual"
     payload = approval.payload
-    assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary"}
+    # 约定的六个键，另加 on_reject（实际生效的处置）：审核人在卡上要看得出没采用原流程会不会让运行失败
+    assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary", "on_reject"}
     assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input"
+    assert payload["on_reject"] == "degrade"
     [item] = payload["summary"]
     assert set(item) == {"metric_id", "name", "unit", "original_rendered", "blind_rendered", "delta_pct", "diff",
-                         "fallback"}
+                         "fallback", "catalog"}
     assert (item["metric_id"], item["name"], item["unit"]) == ("gmv", "销售额", "元")
     assert item["original_rendered"].startswith("310") and item["blind_rendered"].startswith("330")
     assert item["delta_pct"] == round((330 - 310) / 310 * 100, 2)
     assert item["fallback"] is False and len(item["diff"]) == 1
     assert "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] and "原流程" in item["diff"][0]
+    # 选哪一边会往数据目录里写什么，按发起审批时的目录预告：只差时间列一处，选哪边就记哪一列
+    assert item["catalog"] == {
+        "a": {"write": True, "source": name, "table": "orders", "item": "business_date",
+              "value": {"column": "order_date"}},
+        "b": {"write": True, "source": name, "table": "orders", "item": "business_date",
+              "value": {"column": "paid_date"}},
+    }
 
     card = artifact_store.load(payload["card_artifact"])
     assert set(card) == {"metrics", "model", "same_model"} and card["same_model"] is False
@@ -470,6 +498,8 @@ async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client):
     name, source_id = shop
     Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
     run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"}))
+    # 审批卡上看得出：这一步没采用原流程，运行就会停下
+    assert (await pending(run.id)).payload["on_reject"] == "fail"
     done = await decide(client, run.id, {"gmv": "b"})
     assert done.status == "failed" and "审核人没有采用原流程的结果" in (done.error or "") and "销售额" in done.error
     evs = await events(run.id)
@@ -546,6 +576,10 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c
             actor="数据组")
     Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
     run = await start(graph(name, BY_ORDER_DATE))
+    # 预告按目录里已有的说法算：选原流程是同一个说法、不用再写，选盲算不会覆盖已确认的
+    preview = (await pending(run.id)).payload["summary"][0]["catalog"]
+    assert preview["a"]["write"] is False and "无需重复写入" in preview["a"]["reason"]
+    assert preview["b"]["write"] is False and "已有人工确认过的说法「order_date」" in preview["b"]["reason"]
     done = await decide(client, run.id, {"gmv": "b"})
     [wrote] = logs(await events(done.id), "dual_check_catalog")
     assert wrote["written"] == []
@@ -564,6 +598,9 @@ async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatc
     run = await start(graph(name, original, metrics=[east], dual={"definitions": {"east": "华东地区已支付订单的金额"}}))
     [item] = (await pending(run.id)).payload["summary"]
     assert len(item["diff"]) == 1 and "region = '华东'" in item["diff"][0]
+    # 预告和真写的是同一套判断：两边选哪个都不写，原因照实
+    assert item["catalog"]["a"]["write"] is False and "可能是指标自己的筛选范围" in item["catalog"]["a"]["reason"]
+    assert item["catalog"]["b"]["write"] is False and "被采纳的盲算没有这个条件" in item["catalog"]["b"]["reason"]
     done = await decide(client, run.id, {"east": "a"})
     assert done.status == "succeeded", done.error
     [wrote] = logs(await events(run.id), "dual_check_catalog")
@@ -631,6 +668,61 @@ async def test_a_grouped_query_is_compared_on_the_row_it_was_taken_from(monkeypa
     assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
 
 
+async def test_a_having_clause_is_dropped_when_narrowing_to_one_row(monkeypatch, shop, client):
+    """原流程分组后另有 HAVING：取的这一行既然在结果里，它所在的组就满足 HAVING，对这一格是噪声。收窄时连 HAVING
+    一起去掉，差异只剩真正的那一处（时间列），写得进目录。"""
+    name, source_id = shop
+    by_region = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} "
+                 "GROUP BY region HAVING SUM(amount) > 10 ORDER BY region")
+    east_paid = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ")
+    east = {**GMV, "id": "east", "name": "华东销售额"}
+    Blind(monkeypatch, name, {"华东销售额": (east_paid, "gmv")})
+    run = await start(graph(name, by_region, metrics=[east], dual={"definitions": {}}))
+    [item] = (await pending(run.id)).payload["summary"]
+    assert item["original_rendered"].startswith("200") and item["blind_rendered"].startswith("300")
+    assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0]
+    await decide(client, run.id, {"east": "b"})
+    assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
+
+
+async def test_a_condition_on_a_grouped_column_is_not_written(monkeypatch, shop, client, tmp_path):
+    """原流程按 is_test 分组、口径卡取 is_test = 0 那一行：收窄后原流程多出「is_test = 0」，看着像排除测试单的有效
+    记录条件，可它用的是原流程分组的列——是这个指标按组取的范围，不写目录。取值 0 太短、不在任何文字里，只有分组的列
+    拦得下它。"""
+    name, source_id = shop
+    db = sqlite3.connect(tmp_path / "shop.db")
+    db.execute("INSERT INTO orders VALUES (6, '华东', 105, 40.0, 'paid', '2026-09-03', '2026-09-03', 'online', 1)")
+    db.commit()
+    db.close()
+    by_flag = (f"SELECT is_test, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} "
+               "GROUP BY is_test ORDER BY is_test")
+    Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
+    run = await start(graph(name, by_flag, dual={"definitions": {}}))
+    [item] = (await pending(run.id)).payload["summary"]
+    assert item["original_rendered"].startswith("330") and item["blind_rendered"].startswith("370")
+    assert len(item["diff"]) == 1 and "is_test = 0" in item["diff"][0]
+    done = await decide(client, run.id, {"gmv": "a"})
+    [wrote] = logs(await events(done.id), "dual_check_catalog")
+    assert wrote["written"] == [] and "分组的列" in wrote["skipped"][0]["reason"]
+    assert await notes_of(source_id) is None
+
+
+async def test_a_condition_matching_a_run_input_is_not_written(monkeypatch, shop, client):
+    """排除退款的条件看着像有效记录条件，可取值正是本次运行的输入（这一期要排除的状态）：是这个指标这一期的筛选范围，
+    不是整张表的事实，不写目录。没有这个输入时同样的选择会写（test_a_refund_filter_becomes_the_valid_row_filter）。"""
+    name, source_id = shop
+    spec = graph(name, RIGHT, dual={"definitions": {}})
+    spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "exclude_status"}]
+    Blind(monkeypatch, name, {"销售额": (WITH_REFUNDS, "gmv")})
+    run = await start(spec, {"exclude_status": "refunded"})
+    [item] = (await pending(run.id)).payload["summary"]
+    assert len(item["diff"]) == 1 and "refunded" in item["diff"][0]
+    done = await decide(client, run.id, {"gmv": "a"})
+    [wrote] = logs(await events(done.id), "dual_check_catalog")
+    assert wrote["written"] == [] and "本次运行的输入" in wrote["skipped"][0]["reason"]
+    assert await notes_of(source_id) is None
+
+
 async def test_two_differing_inputs_do_not_write_the_catalog(monkeypatch, shop, client):
     """一个指标的两个输入都对不上：审核人选的是整个指标哪边对，拆不出是哪一处，不写目录。"""
     name, source_id = shop
@@ -733,6 +825,50 @@ async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch
     assert "华东" not in task and "region" not in task
 
 
+async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monkeypatch, masked_shop, client):
+    """原流程按遮罩列（region）分组、口径卡取华东那一行：差异说明收窄到这一行时，不能拿这一行的分组值补等值条件——
+    那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄，照整条比。"""
+    name, _ = masked_shop
+    first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
+    Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")})
+    run = await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}}))
+    assert run.status == "interrupted", run.error
+    payload = (await pending(run.id)).payload
+    card = artifact_store.load(payload["card_artifact"])
+    done = await decide(client, run.id, {"first": "a"})
+    assert done.status == "succeeded", done.error
+    evs = await events(run.id)
+    _no_masked_value(payload, card, done, evs)
+    # 照整条比：原流程分组、盲算没分组都列出来，不止差一个方面，选了也不写目录
+    assert len(payload["summary"][0]["diff"]) > 1
+    [wrote] = logs(evs, "dual_check_catalog")
+    assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"]
+
+
+async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client):
+    """盲算自己按遮罩列分组、交了华东那一行：同样不拿这一行的分组值改写盲算的 SQL。"""
+    name, _ = masked_shop
+    Blind(monkeypatch, name, {"销售额": (BY_REGION, "gmv")})
+    run = await start(graph(name, BY_ORDER_DATE))
+    assert run.status == "interrupted", run.error
+    payload = (await pending(run.id)).payload
+    assert payload["summary"][0]["blind_rendered"].startswith("300")
+    card = artifact_store.load(payload["card_artifact"])
+    done = await decide(client, run.id, {"gmv": "b"})
+    assert done.status == "succeeded", done.error
+    _no_masked_value(payload, card, done, await events(run.id))
+
+
+def _no_masked_value(payload: dict, card: dict, done: Run, evs: list[RunEvent], value: str = "华东") -> None:
+    """遮罩列上的取值不出现在审批 payload、对照工件、节点产出和节点发的事件里。查询本身的 tool.end、模型的往来按
+    设计带着原始结果（证据面板显示时再遮），不在此列。"""
+    seen = {"payload": payload, "card": card, "output": dual_output(done, evs),
+            "events": [e.data for e in evs if (e.node_id == "dual" or e.type.startswith("human.") or e.type == "log")
+                       and not e.type.startswith(("tool.", "llm."))]}
+    for where, content in seen.items():
+        assert value not in json.dumps(content, ensure_ascii=False), where
+
+
 # --------------------------------------------------------------------------
 # 设置：全局开关、节点覆盖、发起时的快照
 # --------------------------------------------------------------------------