双盲复算写目录：人工驳回过的同一说法不再写回；节点说明补上盲算 agent 看得到什么

- 数据目录里这一项以前被人明确驳回、驳回的正是这次要写的说法：两次人工判断对不上，不替人拿主意，不写并照实说原因；驳回的是
  别的说法时照写。业务日期只按哪一列比，规则、时区是补充。
- 节点模块说明和盲算的系统提示补上：结果分了组的另给那一行的维度，统计期看本次运行的输入。

backend/app/engine/dual_check_facts.py

@@ -229,6 +229,13 @@ def _value_text(item: str, value: Any) -> str:
     return str(value)
 
 
+def _same_fact(item: str, a: Any, b: Any) -> bool:
+    """目录里的一项和要写的是不是同一个说法：业务日期只看哪一列（规则、时区是补充），文字压掉多余的空白比。"""
+    if item == "business_date":
+        return isinstance(a, Mapping) and isinstance(b, Mapping) and a.get("column") == b.get("column")
+    return isinstance(a, str) and isinstance(b, str) and " ".join(a.split()) == " ".join(b.split())
+
+
 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}]}。永远不抛：数据源没了、
@@ -268,8 +275,13 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor:
         entry = await catalog.read_entry(session, row.id, fact.table)
         notes = entry.notes if entry else {}
         version = entry.version if entry else 0
-        live = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None
-        live = live if live is not None and live.get("status") != "rejected" else None
+        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)}」曾被人工驳回，"
+                           f"与这次指标「{fact.metric}」的选择相反，未写入。如需更改，请到数据目录中修改")
+        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) \
                 and live["value"].get("column") == value.get("column"):

backend/app/engine/nodes/dual_check.py

@@ -2,9 +2,10 @@
 
 **按输入复算。** 指标由口径卡的原式算出，原式本身确定、可复算（metrics 的 recompute_ok）；会错的是喂给它的那几个数：
 查哪张表、按哪天归属、排没排除退款……所以盲算的单位是「一个来自查询快照的输入」。盲算 agent（engine/verify_agent.py）
-只看得到指标名、口径说明（节点配置的 definitions）、这个输入在计算式里是哪一项、数据源的数据目录，看不到原 SQL 和
-原结果；它自己写只读 SQL，最后交一个格子引用，由代码从它自己的查询快照里取值。盲值是拿原式代入盲算的输入算出来的
-（engine/recompute.py，取整同口径卡），和原值按相对误差比，原值为 0 时按绝对误差。两个指标用到同一格的只盲算一次。
+只看得到指标名、口径说明（节点配置的 definitions）、这个输入在计算式里是哪一项（结果分了组的另给那一行的维度）、本次
+运行的输入参数和数据源的数据目录，看不到原 SQL 和原结果；它自己写只读 SQL，最后交一个格子引用，由代码从它自己的查询
+快照里取值。盲值是拿原式代入盲算的输入算出来的（engine/recompute.py，取整同口径卡），和原值按相对误差比，原值为 0 时
+按绝对误差。两个指标用到同一格的只盲算一次。
 
 **一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff），
 完整对照落成一件工件（审批 payload 经 runner._safe 会截断，长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用
@@ -94,8 +95,8 @@ BLIND_SYSTEM = (
     "计算式里的一个数，交给人和原流程比对。\n"
     "- 业务含义只按下面给的指标名称、口径说明和数据目录理解。数据目录里人工确认过的说明（业务日期按哪一列、有效记录"
     "条件、去重规则）要遵守，标「推断，未确认」的谨慎使用\n"
-    "- 时间范围、要排除哪些行、要不要去重，都按口径说明来；口径说明没写到的，按数据目录和最常见的业务理解，并在理由里"
-    "写明你的取舍\n"
+    "- 时间范围（统计期看本次运行的输入）、要排除哪些行、要不要去重，都按口径说明来；口径说明没写到的，按数据目录和"
+    "最常见的业务理解，并在理由里写明你的取舍\n"
     "- 尽量用一条 SQL 直接算出要交的那个数（聚合放在 SQL 里做），最后交出它所在的格子"
 )
 BLIND_TASK_PROMPT = (

backend/tests/test_dual_check_facts.py

@@ -188,6 +188,21 @@ async def test_the_same_column_inferred_is_confirmed_and_keeps_its_rule(shop):
     assert (item["source"], item["status"]) == ("human", "confirmed")
 
 
+async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_does_not_block(shop):
+    """同一个说法以前被人驳回过：两次人工判断对不上，不写；驳回的是别的说法：照写。"""
+    name, source_id = shop
+    await seed(source_id, {"business_date": catalog.make_item({"column": "paid_date"}, "llm", "rejected")})
+    out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1",
+                               node_title="复核", actor=None)
+    assert out["written"] == [] and "曾被人工驳回" in out["skipped"][0]["reason"]
+    assert (await entry_of(source_id)).version == 1
+
+    out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", source=name)], run_id="run-2",
+                               node_title="复核", actor=None)
+    assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}]
+    assert (await entry_of(source_id)).notes["business_date"]["value"] == {"column": "order_date"}
+
+
 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"),