双盲复算复核整改（第 3 轮）：驳回过的有效记录条件不再借着同一次裁决里的合并写回，合起来写时已有的段只写一次，补改名后分组的遮罩列用例

- 同一次裁决里几个指标落到有效记录条件上、要合成一句时，驳回检查原来只拿合起来的整句和驳回的说法比：「status <> 'refunded'」和「is_test = 0」AND 在一起就不再等于驳回的那句，被写成人工确认，和审批卡上「不写」的预告正相反。现在合并前按每个指标自己的说法过一遍驳回检查（_screen）：带回驳回说法的指标拿出来不写，原因和它的预告是同一句话，其余的照写，备注只列真写进去的指标；合起来以后又包含了驳回的整句的也不写。有效记录条件的驳回检查改按段比，!= 和 <>、列名大小写、先后不同都认得出。
- 合起来的条件再和目录里已有的条件合写时，原来拿并好的整句去接，和已有条件部分重叠的整句原样接上，写出「status <> 'refunded' AND status <> 'refunded' AND is_test = 0」。现在接的是每个指标各自的条件，_and_join 只接还没出现过的段；已有条件是人工确认过的，原因里只说另外多出的那几段。最外层是 OR 的条件只在接了别的条件时加括号。
- Fact.also（其余几个指标）改成 members（并进来的每一个，各带自己的说法）：并好的整句不再顶掉第一个指标自己的说法。
- 补用例：合并时驳回的条件拿出来、其余照写（facts 级和图级，真写和预告一字不差）；驳回的整句由几个指标拼回来时不写；驳回时写法不同也认得出；合写不重复、已确认时只说多出的段；_and_join 的段级去重和括号；遮罩列改了名再分组（region AS area）时不收窄。

backend/app/engine/dual_check_facts.py

@@ -23,17 +23,23 @@ is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不
 
 **怎么写。** 照对话提案的规矩核对（plan_patch：列在不在表结构里、取值格式），记为人工填写、已确认，备注写明来自哪次
 运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖，照实交回原因；已经是同样的说法就什么都
-不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边（几个指标并成一项时是每一个）
+不写（再写一遍也不升版本）；这一项被人工驳回过、要写的又带回了驳回的说法（有效记录条件按段比：驳回的那句每一段都在
-也按它筛了的，和新条件合起来写（已确认的不改，照实说）；有一个没按它筛的，两者的关系拿不准，不动。推断的去重规则
+要写的条件里），两次人工判断对不上，不写。有效记录条件是一整句条件：目录里已有的条件被采纳的一边（几个指标并成一项
-说的未必是同一件事，也不替换。整个写入由节点放在 ask() 之后的一个 @task 里：审批恢复重放、接着运行时不会再写一遍。
+时是每一个）也按它筛了的，和新条件合起来写，已有的段只写一次（已确认的不改，照实说多出了哪几段）；有一个没按它筛
+的，两者的关系拿不准，不动。推断的去重规则说的未必是同一件事，也不替换。整个写入由节点放在 ask() 之后的一个 @task
+里：审批恢复重放、接着运行时不会再写一遍。
 
 **同一次裁决里几个指标落到同一项上。** 说法一样的写一次，备注里列出每个指标。有效记录条件是能用 AND 叠加的排除条件
 （这里只收排除无效记录的），几个指标各自采纳的条件都有人的选择撑着，合成一句写——「排除退款」和「排除测试单」并不
-矛盾。业务日期一张表只有一个，去重规则说的未必是同一件事，说法不同就一个都不写。
+矛盾。业务日期一张表只有一个，去重规则说的未必是同一件事，说法不同就一个都不写。合起来之前先按每个指标自己的说法
+过一遍驳回检查：自己采纳的说法曾被人工驳回的指标拿出来不写、照实说原因，其余的照写；合起来以后又包含了驳回的整句的，
+也不写。
 
 **预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么（preview）：按发起审批时的目录，选原流程、选盲算各会
-写什么、不写的为什么。和真写是同一套判断（derive 认事实、_plan 对目录），所以只在两处可能和最后写的不一样：审批停下
+写什么、不写的为什么。和真写是同一套判断（derive 认事实、_plan 对目录、_revives 认驳回），所以只在两处可能和最后写
-期间目录被人改了，或者同一次裁决里几个指标落到同一项上（上一段）。
+的不一样：审批停下期间目录被人改了；同一次裁决里几个指标落到同一项上（上一段）——写进去的是合起来的说法，预告说会
+写的也可能因为说法互相矛盾、和已有条件的关系拿不准、合起来又成了驳回过的说法而不写。反过来不会：预告说不写的，
+不会因为和别的指标并成一项被写进去。
 
 catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。
 """
@@ -41,7 +47,7 @@ from __future__ import annotations
 
 import logging
 import re
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, replace
 from typing import Any
 
@@ -86,8 +92,9 @@ _SCOPE_DIGITS = 2
 @dataclass(frozen=True)
 class Fact:
     """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明；sql、dialect 是被采纳
-    那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。also 是同一次裁决里落到同一项上、
+    那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。同一次裁决里几个指标落到同一项上
-    并进这一项的其余几个指标（各自的指标、选择、写法），value 是合起来之后的说法。"""
+    时并成一项：members 是并进来的每个指标各自的那一项（各带自己的说法，审批卡上各自的预告就是按它算的），value 是
+    合起来之后的说法；单个指标的 members 为空。"""
 
     source: str
     table: str
@@ -97,17 +104,17 @@ class Fact:
     choice: str
     sql: str = ""
     dialect: str | None = None
-    also: tuple[Fact, ...] = ()
+    members: tuple[Fact, ...] = ()
 
 
 def _members(fact: Fact) -> tuple[Fact, ...]:
-    """这一项背后的每个指标的选择：并过的连同并进来的。"""
+    """这一项背后的每个指标的选择：并成一项的是并进来的每一个（各带自己的说法），单个指标的是它自己。"""
-    return (fact, *fact.also)
+    return fact.members or (fact,)
 
 
-def _names(fact: Fact) -> str:
+def _names(members: Sequence[Fact]) -> str:
-    """说明里的指标叫法：「销售额」，几个指标并成一项时「销售额」、「订单金额」。"""
+    """说明里的指标叫法：「销售额」，几个指标时「销售额」、「订单金额」。"""
-    names = list(dict.fromkeys(m.metric for m in _members(fact)))
+    names = list(dict.fromkeys(m.metric for m in members))
     return "、".join(f"「{n}」" for n in names[:4]) + (f"等 {len(names)} 个" if len(names) > 4 else "")
 
 
@@ -422,39 +429,57 @@ 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:
+def _keys(condition: Any, read: str | None) -> set[str] | None:
-    """几句条件用 AND 接成一句：已经有的（base 里的、前面接过的，按 _normalized 比）不再接；最外层是 OR 的加括号，
+    """一句条件按 AND 拆开后每一段的比较用写法；不是文字、解析不了返回 None。"""
-    不然后面接的条件只并到 OR 的后一半上。每句保留原来的写法。有一句解析不了返回 None。"""
+    tree = _parse(condition, read) if isinstance(condition, str) and condition.strip() else None
+    return {_normalized(part, read) for part in _and_parts(tree)} if tree is not None else None
+
+
+def _and_join(base: str | None, values: list[str], dialect: str | None, *, keep_base: bool = True) -> str | None:
+    """几句条件用 AND 接成一句：已经有的段（base 里的、前面接过的，按 _normalized 比）不再接。一句里每一段都是新的，
+    原样接上（保留原来的写法）；只有几段是新的，只接那几段（按方言重新写）——整句接上会把已有的段再写一遍。
+    keep_base 为假时 base 只用来认哪些段已经有了、不写进结果（说「另外多出哪些条件」用），什么都没多出时为空串。
+    不止一段时，最外层是 OR 的加括号，不然后面接的条件只并到 OR 的后一半上。有一句解析不了返回 None。"""
     from sqlglot import exp
 
     read = _read(dialect)
-    texts: list[str] = []
+    pieces: list[tuple[str, bool]] = []  # (写法, 最外层是不是 OR)
     seen: set[str] = set()
-    for text in [*([base] if base else []), *values]:
+    for i, text in enumerate([*([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)}
+        parts = _and_parts(tree)
-        if keys <= seen:
+        fresh = []
+        for part in parts:
+            key = _normalized(part, read)
+            if key not in seen:
+                seen.add(key)
+                fresh.append(part)
+        if not fresh or (base and i == 0 and not keep_base):
             continue
-        seen |= keys
+        if len(fresh) == len(parts):
-        texts.append(f"({text})" if isinstance(tree, exp.Or) else str(text))
+            pieces.append((str(text).strip(), isinstance(tree, exp.Or)))
-    return " AND ".join(texts)
+        else:
+            pieces += [(part.sql(dialect=read), isinstance(part, exp.Or)) for part in fresh]
+    if len(pieces) == 1:
+        return pieces[0][0]
+    return " AND ".join(f"({text})" if is_or else text for text, is_or in pieces)
 
 
 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 = _parse(existing, read)
+    have = _keys(existing, read)
     combined = _and_join(existing, [str(m.value) for m in _members(fact)], fact.dialect) \
-        if have_tree is not None else None
+        if have is not None else None
-    if combined is None:
+    if have is None or combined is None:
         return None, f"「{fact.table}」已有的{label}「{existing}」无法解析，与这次的条件合不起来，未改动"
-    have = {_normalized(p, read) for p in _and_parts(have_tree)}
     for member in _members(fact):
         if not have <= (_conditions_of(member.sql, read) or set()):
             return None, (f"「{fact.table}」已有的{label}「{existing}」，指标「{member.metric}」被采纳的"
@@ -560,28 +585,82 @@ def _differing(facts: list[Fact]) -> str:
 
 
 def _merged(members: list[Fact]) -> tuple[Fact | None, str | None]:
-    """同一次裁决里落到同一项上的几个说法合成一项：(合起来的那一项, None) 或 (None, 为什么都不写)。
+    """同一次裁决里落到同一项上的几个说法合成一项：(合起来的那一项, None) 或 (None, 为什么都不写)。只有一个时就是它。
 
     说法都一样的写一次。有效记录条件用 AND 合成一句（derive 只收排除无效记录的条件，几句叠起来不会互相打架）。
-    业务日期一张表只有一个、去重规则说的未必是同一件事，说法不同就都不写。"""
+    业务日期一张表只有一个、去重规则说的未必是同一件事，说法不同就都不写。每个指标各自的那一项原样留在 members 里：
-    first, rest = members[0], tuple(members[1:])
+    驳回检查（_screen）、和目录里已有条件的合并（_combined_filter）都要按各自的说法算。"""
-    if not rest:
+    first = members[0]
+    if len(members) == 1:
         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))
+        return (replace(first, value=value, members=tuple(members)), None) if value is not None \
-    if all(_same_fact(first.item, m.value, first.value) for m in rest):
+            else (None, _differing(members))
-        return replace(first, also=rest), None
+    if all(_same_fact(first.item, m.value, first.value) for m in members[1:]):
+        return replace(first, members=tuple(members)), None
     return None, _differing(members)
 
 
+def _rejected(notes: Mapping[str, Any], item: str) -> tuple[bool, Any]:
+    """目录里这一项是不是被人工驳回的：(是不是, 驳回的说法)。"""
+    stored = notes.get(item)
+    if isinstance(stored, Mapping) and stored.get("status") == "rejected":
+        return True, stored.get("value")
+    return False, None
+
+
+def _revives(fact: Fact, rejected: Any) -> bool:
+    """要写的说法是不是又带回了曾被人工驳回的说法。业务日期、去重规则：同一个说法。有效记录条件：驳回的那句的每一段
+    都在要写的条件里（按 _normalized 比，!= 和 <>、列名大小写、先后不同都认得出；有一边解析不了按原文比）——驳回的
+    那句只是要写的一部分也算，写进去就等于替人把驳回过的条件又确认了一遍。"""
+    if fact.item != "valid_filter":
+        return _same_fact(fact.item, rejected, fact.value)
+    read = _read(fact.dialect)
+    mine, theirs = _keys(fact.value, read), _keys(rejected, read)
+    if mine is None or theirs is None:
+        return _same_fact(fact.item, rejected, fact.value)
+    return bool(theirs) and theirs <= mine
+
+
+def _rejected_text(fact: Fact, rejected: Any, members: Sequence[Fact]) -> str:
+    """驳回检查拦下时的说法：驳回的是哪一句（照目录里记的原文），哪几个指标的选择又带回了它。审批卡上的预告和真写
+    拦下时用的是同一句话。"""
+    label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
+    return (f"「{fact.table}」的{label}「{_value_text(fact.item, rejected)}」曾被人工驳回，与这次指标{_names(members)}"
+            "的选择相反，未写入。如需更改，请到数据目录中修改")
+
+
+def _screen(notes: Mapping[str, Any], fact: Fact) -> tuple[Fact | None, list[str]]:
+    """几个指标并成的一项，先拿掉自己采纳的说法曾被人工驳回的指标：(剩下的重新并成的一项, 拿掉的为什么不写)，一个
+    都不剩时前者为 None。单个指标的、这一项没被驳回过的原样返回，由 _plan 照常判断。
+
+    审批卡上每个指标的预告是按它自己的说法单独算的（preview → _plan），在那里被驳回检查拦下、卡上写着「不写」的，
+    并进来以后也得拦下。只拿合起来的整句去比，「status <> 'refunded'」和别的条件 AND 在一起就不再等于驳回的那句，
+    会被写成人工确认，和卡上说的正相反。拿掉的指标照实说原因（和它的预告同一句话），其余的照写，也和各自的预告一致。"""
+    rejected, said = _rejected(notes, fact.item)
+    if not rejected or not fact.members:
+        return fact, []
+    out: list[Fact] = []
+    kept: list[Fact] = []
+    for member in fact.members:
+        (out if _revives(member, said) else kept).append(member)
+    if not out:
+        return fact, []
+    reasons = [_rejected_text(fact, said, out)]
+    if not kept:
+        return None, reasons
+    narrowed, why = _merged(kept)
+    return narrowed, reasons if why is None else [*reasons, why]
+
+
 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：说法一样的、有效记录条件合起来写，说法
-    不同的整组不写），逐项写。"""
+    不同的整组不写），逐项写（_write：按读到的目录先拿掉各自带回驳回说法的指标，其余的照写）。"""
     skipped: list[dict[str, Any]] = []
     facts: list[Fact] = []
     for candidate in candidates:
@@ -611,39 +690,39 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no
         if fact is None:
             skipped.append({"reason": why})
             continue
-        done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor)
+        done, reasons = 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})
-        else:
+        skipped += [{"reason": r} for r in reasons]
-            skipped.append({"reason": why})
     return {"written": written, "skipped": skipped}
 
 
-async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, str | None]:
+async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, list[str]]:
+    """写一项：(写了没有, 没写的原因)。几个指标并成的一项可能写了、同时有几个指标被拿出来没写，原因照样交回。"""
     try:
         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}」已不存在，指标{_names(fact)}的选择未写入数据目录"
+                return False, [f"数据源「{fact.source}」已不存在，指标{_names(_members(fact))}的选择未写入数据目录"]
-            return await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title),
+            return await _write(session, row, fact, run_id=run_id, node_title=node_title, actor=actor)
-                                actor=actor)
     except Exception:  # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记
         logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", fact.source)
-        return False, f"指标{_names(fact)}的选择写入数据目录时出错，未写入"
+        return False, [f"指标{_names(_members(fact))}的选择写入数据目录时出错，未写入"]
 
 
 def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]:
     """按目录里这张表现在的说法，算出这一项要写成什么：(要写的取值, None) 或 (None, 为什么不写)。不读库、不写库：
-    真写（_write）和审批卡上的预告（preview）用的是这同一套判断。"""
+    真写（_write）和审批卡上的预告（preview）用的是这同一套判断。几个指标并成的一项，先由 _screen 拿掉各自带回驳回
+    说法的指标再交到这里。"""
     label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
-    names = _names(fact)
+    names = _names(_members(fact))
     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"),
+    rejected, said = _rejected(notes, fact.item)
-                                                                                 fact.value):
+    if rejected and _revives(fact, said):
-        # 同一个说法以前被人明确驳回过：两次人工判断对不上，不替人拿主意
+        # 同一个说法以前被人明确驳回过：两次人工判断对不上，不替人拿主意。并成的一项各自的说法都不是驳回的那句、
-        return None, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回，"
+        # 合起来却又包含了它，也在这里拦下
-                      f"与这次指标{names}的选择相反，未写入。如需更改，请到数据目录中修改")
+        return None, _rejected_text(fact, said, _members(fact))
-    live = stored if stored is not None and stored.get("status") != "rejected" else None
+    live = stored if stored is not None and not rejected else None
     value = fact.value
     if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \
             and live["value"].get("column") == value.get("column"):
@@ -655,9 +734,13 @@ def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None
         if combined is None:
             return None, why
         if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]):
-            adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.also else "这次几个指标被采纳的写法"
+            # 只说已确认的说法里没有的那几段：已有的段算成「多出」，审核人会以为连它也被改了
+            extra = _and_join(live["value"].strip(), [str(m.value) for m in _members(fact)], fact.dialect,
+                              keep_base=False)
+            adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.members \
+                else "这次几个指标被采纳的写法"
             return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，{adopted}"
-                          f"另外多出条件「{fact.value}」，未改动。如需补上，请到数据目录中修改")
+                          f"另外多出条件「{extra or 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):
@@ -702,34 +785,41 @@ def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None
     return out
 
 
-async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]:
+async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node_title: str,
-    """写一项。(写了没有, 没写的原因)。写什么、写不写由 _plan 按读到的目录定；版本对不上（别人刚改过）重读再来。"""
+                 actor: str | None) -> tuple[bool, list[str]]:
+    """写一项：(写了没有, 没写的原因)。几个指标并成的一项先按读到的目录拿掉各自带回驳回说法的指标（_screen），写什么、
+    写不写再由 _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
+    reasons: list[str] = []
     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)
+        kept, reasons = _screen(notes, fact)
+        if kept is None:
+            return False, reasons
+        value, why = _plan(notes, kept, tables)
         if why is not None:
-            return False, why
+            return False, [*reasons, why]
+        note = note_of(kept, run_id=run_id, node_title=node_title)
         at = catalog.now_iso()
         try:
-            merged = catalog.apply_patch(notes, [{"path": fact.item, "value": value, "reason": note}],
+            merged = catalog.apply_patch(notes, [{"path": kept.item, "value": value, "reason": note}],
-                                         table=fact.table, tables=tables, at=at)
+                                         table=kept.table, tables=tables, at=at)
         except catalog.CatalogInvalid as e:
-            return False, f"「{fact.table}」的{label}未写入数据目录：{e}"
+            return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
         # 不论原来是谁提的：这一项现在出自人的选择，记为人工填写、已确认，备注写明出处
-        merged[fact.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at)
+        merged[kept.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)
+            await catalog.write_entry(session, row.id, kept.table, merged, if_version=version, actor=actor)
         except catalog.CatalogConflict:
             await session.rollback()
             continue
         except catalog.CatalogInvalid as e:
-            return False, f"「{fact.table}」的{label}未写入数据目录：{e}"
+            return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
-        return True, None
+        return True, reasons
-    return False, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"
+    return False, [*reasons, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"]
 
 
 __all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"]

backend/tests/test_dual_check_facts.py

@@ -7,9 +7,10 @@
 - 指标自己的筛选范围（只看华东、只看线上）不是有效记录条件：取值出现在指标名称、口径说明、所在分组、运行输入里的，
   条件用到分组的列的，不是在排除退款、作废、测试这类无效记录的，都不写
 - 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖；
-  目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动
+  目录里已有的有效记录条件被采纳的一边也用着的，合起来写（已有的段只写一次），没用着的不动；驳回过的说法不写回
+  （!= 和 <>、列名大小写、先后不同也认得出）
 - 同一次裁决里几个指标落到同一项上：说法一样的写一次；有效记录条件（都在排除无效记录）用 AND 合成一句写；业务日期
-  说法不同的都不写
+  说法不同的都不写；自己的说法驳回过的指标拿出来不写、其余照写（和各自的预告一致），合起来又成了驳回的整句也不写
 - 审批卡上的预告（preview）：选 a、选 b 各会写什么、不写的为什么，按目录里已有的说法算，和真写同一套判断
 
 差异条目一律由 sql_semantics.explain_sql_diff 真算出来，和节点交给审核人看的是同一份。表名、数字都是虚构的。
@@ -422,6 +423,110 @@ async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_
     assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "channel <> 'test'"
 
 
+async def test_a_rejected_condition_is_not_written_back_through_a_combined_decision(shop):
+    """目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」「订单数」采纳了排除退款、「订单金额」采纳了
+    排除测试单：审批卡上预告前两个不写（曾被人工驳回）、后一个写「is_test = 0」，真写也照这样。驳回过的条件不能借着
+    合并又写成人工确认，拿出来的指标照实说原因。"""
+    name, source_id = shop
+    rejected = {"valid_filter": catalog.make_item("status <> 'refunded'", "llm", "rejected")}
+    await seed(source_id, rejected)
+    refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name)
+    tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)
+    counted = candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name)
+    notes = {"orders": (await entry_of(source_id)).notes}
+    assert preview(refunds, SCHEMA, notes)["b"]["write"] is False
+    assert preview(tests, SCHEMA, notes)["b"]["value"] == "is_test = 0"
+    out = await record_choices([refunds, tests, counted], run_id="run-1", node_title="复核", actor=None)
+    assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
+    [skipped] = out["skipped"]
+    assert "「status <> 'refunded'」曾被人工驳回" in skipped["reason"] and "「销售额」、「订单数」" in skipped["reason"]
+    assert "订单金额" not in skipped["reason"]
+    item = (await entry_of(source_id)).notes["valid_filter"]
+    assert item["value"] == "is_test = 0" and (item["source"], item["status"]) == ("human", "confirmed")
+    assert "「订单金额」" in item["note"] and "销售额" not in item["note"] and "订单数" not in item["note"]
+
+    # 并成一项的每个指标采纳的都是驳回过的说法：一个都不写，一条原因说清是哪几个指标
+    await seed_again(source_id, rejected)
+    out = await record_choices([refunds, counted], run_id="run-2", node_title="复核", actor=None)
+    assert out["written"] == [] and len(out["skipped"]) == 1, out
+    assert "曾被人工驳回" in out["skipped"][0]["reason"] and "「销售额」、「订单数」" in out["skipped"][0]["reason"]
+    assert (await entry_of(source_id)).notes["valid_filter"]["status"] == "rejected"
+
+
+async def test_a_rejected_statement_is_not_rebuilt_from_several_metrics(shop):
+    """驳回的是「is_test = 0 AND status != 'refunded'」整句：单看每个指标都不是这句（预告各写各的），合起来却正是这句
+    （!= 和 <>、先后不同也是同一个说法）。两次人工判断对不上，都不写；再并进一个别的条件，驳回的整句仍在里面，照样不写。"""
+    name, source_id = shop
+    rejected = catalog.make_item("is_test = 0 AND status != 'refunded'", "llm", "rejected")
+    await seed(source_id, {"valid_filter": rejected})
+    refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name)
+    tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)
+    channel = candidate(COUNT_WEEK, extra("channel <> 'test'", COUNT_WEEK), "b", metric="订单数", source=name)
+    notes = {"orders": (await entry_of(source_id)).notes}
+    assert all(preview(c, SCHEMA, notes)["b"]["write"] is True for c in (refunds, tests, channel))
+    for picked, names in (([refunds, tests], "「销售额」、「订单金额」"),
+                          ([refunds, tests, channel], "「销售额」、「订单金额」、「订单数」")):
+        out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None)
+        assert out["written"] == [], out
+        [skipped] = out["skipped"]
+        assert "曾被人工驳回" in skipped["reason"] and names in skipped["reason"]
+    entry = await entry_of(source_id)
+    assert entry.version == 1 and entry.notes["valid_filter"]["status"] == "rejected"
+
+
+@pytest.mark.parametrize("rejected", ["status != 'refunded'", "STATUS <> 'refunded'"])
+async def test_a_rejected_condition_is_recognised_however_it_was_written(shop, rejected):
+    """驳回时写的是 != 、列名大写：和这次要写的「status <> 'refunded'」是同一个说法，预告和真写都不写。"""
+    name, source_id = shop
+    await seed(source_id, {"valid_filter": catalog.make_item(rejected, "llm", "rejected")})
+    picked = candidate(WITH_REFUNDS, RIGHT, "b", source=name)
+    said = preview(picked, SCHEMA, {"orders": (await entry_of(source_id)).notes})
+    assert said["b"]["write"] is False and f"「{rejected}」曾被人工驳回" in said["b"]["reason"]
+    out = await record_choices([picked], run_id="run-1", node_title="复核", actor=None)
+    assert out == {"written": [], "skipped": [{"reason": said["b"]["reason"]}]}
+    assert (await entry_of(source_id)).version == 1
+
+
+async def test_combined_filters_do_not_repeat_a_condition_already_in_the_catalog(shop):
+    """目录里推断的「status <> 'refunded'」：「销售额」采纳了带着它的盲算，「订单金额」两边都按它筛、盲算另外排除了
+    测试单。合起来写时已有的条件只出现一次（正是「订单金额」的预告）；已有的条件是人工确认过的，原因里只说另外多出
+    的那一段。"""
+    name, source_id = shop
+    await seed(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "llm")})
+    base = extra("status <> 'refunded'")
+    picked = [candidate(SUM_WEEK, base, "b", metric="销售额", source=name),
+              candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
+    said = preview(picked[1], SCHEMA, {"orders": (await entry_of(source_id)).notes})
+    out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None)
+    assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
+    item = (await entry_of(source_id)).notes["valid_filter"]
+    assert item["value"] == said["b"]["value"] == "status <> 'refunded' AND is_test = 0"
+
+    await seed_again(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "human", "confirmed")})
+    out = await record_choices(picked, run_id="run-2", node_title="复核", actor=None)
+    assert out["written"] == []
+    reason = out["skipped"][0]["reason"]
+    assert "已有人工确认过的说法「status <> 'refunded'」" in reason and "另外多出条件「is_test = 0」" in reason
+    assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "status <> 'refunded'"
+
+
+def test_conditions_are_joined_without_repeating_a_part():
+    """几句条件接成一句：已经有的段不再接（整句接上会把已有的段再写一遍）；只说多出的时不带原有的；最外层是 OR 的
+    只在接了别的条件时加括号。"""
+    from app.engine.dual_check_facts import _and_join
+
+    assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite") == \
+        "status <> 'refunded' AND is_test = 0"
+    assert _and_join("status <> 'refunded'", ["STATUS != 'refunded'", "is_test = 0", "is_test = 0"], "sqlite") == \
+        "status <> 'refunded' AND is_test = 0"
+    assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite",
+                     keep_base=False) == "is_test = 0"
+    assert _and_join("status = 'void' OR is_test = 1", ["(status = 'void' OR is_test = 1)"], "sqlite") == \
+        "status = 'void' OR is_test = 1"
+    assert _and_join("status = 'void' OR is_test = 1", ["is_test = 0"], "sqlite") == \
+        "(status = 'void' OR is_test = 1) AND is_test = 0"
+
+
 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),

backend/tests/test_dual_check_node.py

@@ -7,9 +7,10 @@
 - 不一致：挂起，审批 payload 和对照工件按约定，payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往
   数据目录里写什么的预告；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject 为 fail 时失败
 - 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比，HAVING 一并去掉）；分组的列设了
-  遮罩时不收窄，遮罩列的取值不进审批卡、对照工件、产出和事件
+  遮罩时（改了名再分组的也算）不收窄，遮罩列的取值不进审批卡、对照工件、产出和事件
 - 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的
-  筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写
+  筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写；人工驳回过的条件不会借着和别的
+  指标并成一项写回去，真写的和审批卡上的预告一致
 - 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询
 - 触顶、分组结果的维度（只给取值、遮罩列不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的
   盲算结果不采用
@@ -588,6 +589,37 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c
     assert entry.version == 1 and entry.notes["business_date"]["value"] == {"column": "order_date"}
 
 
+async def test_a_rejected_filter_is_not_written_back_along_with_another_metric(monkeypatch, shop, client, tmp_path):
+    """目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」采纳了排除退款的盲算、「订单金额」采纳了排除
+    测试单的盲算：审批卡上预告前者不写、后者写「is_test = 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()
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, source_id, "orders", {
+            "valid_filter": catalog.make_item("status <> 'refunded'", "human", "rejected")}, if_version=0, actor="数据组")
+    original = f"SELECT SUM(amount) AS gmv, SUM(amount) AS total FROM orders WHERE paid_date {WEEK}"
+    no_test = f"SELECT SUM(amount) AS total FROM orders WHERE is_test = 0 AND paid_date {WEEK}"
+    total = {**GMV, "id": "total", "name": "订单金额", "expression": "cell(nodes.fetch, 0, 'total')"}
+    Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单金额": (no_test, "total")})
+    run = await start(graph(name, original, metrics=[GMV, total], dual={"definitions": {}}))
+    assert run.status == "interrupted", run.error
+    said = {s["metric_id"]: s["catalog"]["b"] for s in (await pending(run.id)).payload["summary"]}
+    assert said["gmv"]["write"] is False and "曾被人工驳回" in said["gmv"]["reason"]
+    assert said["total"]["write"] is True and said["total"]["value"] == "is_test = 0"
+    done = await decide(client, run.id, {"gmv": "b", "total": "b"})
+    assert done.status == "succeeded", done.error
+    [wrote] = logs(await events(run.id), "dual_check_catalog")
+    assert wrote["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}]
+    assert [s["reason"] for s in wrote["skipped"]] == [said["gmv"]["reason"]]
+    item = (await notes_of(source_id)).notes["valid_filter"]
+    assert item["value"] == said["total"]["value"] and (item["source"], item["status"]) == ("human", "confirmed")
+    assert "「订单金额」" in item["note"] and "「销售额」" not in item["note"]
+
+
 async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatch, shop, client):
     """「华东销售额」原流程按指标名只看华东，盲算漏了地区：审核人选原流程，但 region = '华东' 是这个指标自己的筛选
     范围，写成表的有效记录条件，以后所有查订单表的工作流（包括下一次盲算）都会只看华东。"""
@@ -845,6 +877,26 @@ async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monk
     assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"]
 
 
+async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, masked_shop, client):
+    """原流程把遮罩列改了名再分组（region AS area）：结果列名 area 不在遮罩里，分组值却还是 region 的取值。收窄时看的
+    是分组键用到的列，一样不拿这一行的分组值补等值条件，「华东」不进审批卡、对照工件、产出和事件。"""
+    name, _ = masked_shop
+    renamed = (f"SELECT region AS area, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' "
+               f"AND paid_date {WEEK} GROUP BY region ORDER BY region")
+    first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
+    Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")})
+    run = await start(graph(name, renamed, metrics=[first], dual={"definitions": {}}))
+    assert run.status == "interrupted", run.error
+    payload = (await pending(run.id)).payload
+    assert payload["summary"][0]["original_rendered"].startswith("300")
+    card = artifact_store.load(payload["card_artifact"])
+    done = await decide(client, run.id, {"first": "a"})
+    assert done.status == "succeeded", done.error
+    _no_masked_value(payload, card, done, await events(run.id))
+    # 照整条比：原流程分组、盲算没分组都列出来
+    assert len(payload["summary"][0]["diff"]) > 1
+
+
 async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client):
     """盲算自己按遮罩列分组、交了华东那一行：同样不拿这一行的分组值改写盲算的 SQL。"""
     name, _ = masked_shop