M2 证据面板和推演认得出改了名字的遮罩列，指标本身的原值也不露：查询步骤、指标的输入、推演的勾行与遮罩、合并查询按 SQL 认取自遮罩列的结果列（c.phone AS tel 这类别名、函数、子查询、WITH），和可验证报告、核查 agent 同一套认法；用到遮罩列的指标，推演答复里的 old、rendered_old 和证据面板指标步骤的 value 写「已遮罩」

验收里证据面板对 c.phone AS tel 的快照给出 masked=[]、三个号码原样交出；推演把 phone_raw 判成 masked、输入置空，指标的 old 却还是那串号码。现在：
- 认遮罩列只有一个函数 api/evidence.py 的 _snapshot_masks（查询当时记下的加数据源现在设的，再经 engine/masking.py 的 _derived_hidden 按 SQL 认改了名字的；SQL 解析不了又写着遮罩列的整份遮），查询步骤、_InputMasks、推演的 _mask_of 都调它；_Masks 顺带记数据源类型，按它的方言解析
- 合并查询按逐格来历遮：输入快照里改了名字的遮罩列（输入是合并结果的再往下追）也遮到合并结果里
- 指标的原值按 recompute.hides_value 一个判断：有输入取自遮罩列（含只嵌在下标里的），或者最外层的输入遮没遮核对不了；推演结果标 withheld，报告里引用它的地方 rendered_old 同样写「已遮罩」
- 面板底下的遮罩说明改成「这些列取自数据源中设置了遮罩的列」：遮的可能是别名列

backend/app/api/evidence.py

@@ -127,7 +127,8 @@ NONE_NOTE = "本次运行没有报告文档，也没有出具契约，没有可
 NO_DOC_NOTE = "本次运行的出具契约采用引用模式核对，但报告没有产出可核对的文档"
 
 MASKED = "已遮罩"
-MASK_NOTE = "这些列在数据源中设置了遮罩，面板上不显示原值。遮罩只减少暴露，不是安全边界"
+#: 面板底下的遮罩说明。遮的列可能是遮罩列本身，也可能是 SQL 里给它起了别名、套了函数得来的结果列（c.phone AS tel）
+MASK_NOTE = "这些列取自数据源中设置了遮罩的列，面板上不显示原值。遮罩只减少暴露，不是安全边界"
 #: 查询条目的数据源现在找不到了（改名或删掉了）：只能按快照里记下的、查询当时的遮罩来遮
 SOURCE_GONE = "数据源「{source}」已不存在（可能已改名或删除），按查询时记录的遮罩处理"
 #: 查询步骤给被引用的行前后各带几行
@@ -878,6 +879,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
     wanted: dict[str, list[tuple[Any, Any]]] = {}
     hide = _InputMasks(sealed, masks, str(entry.get("node_id") or ""), artifact)
     withheld: set[str] = set()
+    unverified: set[str] = set()
     for item in (metric or {}).get("inputs") or []:
         one = {"step": "input", **item}
         artifact_of = item.get("artifact")
@@ -889,10 +891,16 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
         if (hidden := await hide.check(item)) is not None:
             one = _withheld(one, *hidden)
             withheld.add(str(item.get("path")))
+            if hidden[1] is not None:
+                unverified.add(str(item.get("path")))
         inputs.append(one)
     if withheld:
         # 代入式里那几处同样写「已遮罩」：不然标签里遮着、代入式里亮着，同一个值换个地方照样看得到
         step["substituted"] = recompute.substituted_without(metric or {}, withheld, MASKED)
+    if step["value"] is not None and recompute.hides_value(metric or {}, withheld - unverified, unverified):
+        # 指标的值本身也不给，和推演的原值同一个判断：指标可能就是遮罩列的那一格（cell(nodes.q, 0, 'phone')），输入里
+        # 遮着、这里亮着等于没遮。报告上的字（rendered）是读者已经看到的，照旧
+        step["value"] = MASKED
     queries = [s for a, cells in wanted.items() for s in await _query_steps(a, cells, doc, sealed, masks)]
     return [step, *inputs, *queries]
 
@@ -916,8 +924,8 @@ SNAPSHOT_UNVERIFIED = "查询快照无法读取，核对不了这一格是否取
 
 
 class _InputMasks:
-    """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询按它
-    合并的那几个数据源算。
+    """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_snapshot_masks）：查询当时记下的加数据源现在设的，按 SQL
+    认出改了名字的（c.phone AS tel 的 tel）也算，合并查询按它合并的那几个数据源和逐格来历算。
 
     取自哪份查询结果的哪几列按 recompute.input_origin 认，推演、可验证报告是同一个认法：cell() 的格、Agent 交来的字段按
     定位里的列认；vars.res.rows[0][2] 这种按位置取的一格卡上没记定位，顺着取值链认，从写变量的节点在口径卡算这一次之前
@@ -975,13 +983,8 @@ class _InputMasks:
     async def _hidden(self, artifact: str | None, source: str | None, snap: dict[str, Any] | None) -> set[str]:
         key = (artifact, source)
         if key not in self._memo:
-            if snap is not None and snap.get("source") == MERGE_SOURCE:
-                hidden, _ = await _merge_masks(snap, self._masks)
-            else:
-                info = self._sealed.queries.get(artifact) if artifact else None
-                current, _ = await self._masks.of((info or {}).get("source") or source or (snap or {}).get("source"))
-                hidden = set(current) | (_recorded_masks(snap) if snap is not None else set())
-            self._memo[key] = hidden
+            info = self._sealed.queries.get(artifact) if artifact else None
+            self._memo[key], _ = await _snapshot_masks(snap, (info or {}).get("source") or source, self._masks)
         return self._memo[key]
 
 
@@ -1283,10 +1286,12 @@ def _challenge_counts(raw: Any, claims: list[dict[str, Any]]) -> dict[str, int]:
 
 
 class _Masks:
-    """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。"""
+    """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。顺带记下数据源的类型（kind），
+    按 SQL 认改了名字的遮罩列时照它的方言解析（_snapshot_masks）；数据源不在了的按通用读法。"""
 
     def __init__(self) -> None:
         self._memo: dict[str, tuple[set[str], bool]] = {}
+        self._kinds: dict[str, str | None] = {}
 
     async def of(self, source: Any) -> tuple[set[str], bool]:
         name = str(source or "")
@@ -1296,8 +1301,56 @@ class _Masks:
             async with SessionLocal() as session:
                 row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
             self._memo[name] = ({c.lower() for c in masked_columns(row.options if row else None)}, row is not None)
+            self._kinds[name] = row.kind if row is not None else None
         return self._memo[name]
 
+    async def kind(self, source: Any) -> str | None:
+        name = str(source or "")
+        if not name:
+            return None
+        await self.of(name)
+        return self._kinds.get(name)
+
+
+def _derived(snap: dict[str, Any], hidden: set[str], kind: str | None) -> set[str]:
+    """一份查询快照里值取自遮罩列（hidden，小写）的结果列（小写，含 hidden 本身）：除了列名对上的，SQL 里给遮罩列起了
+    别名、套了函数、经子查询或 WITH 转一道得来的也算。认法是 engine/masking.py 的 _derived_hidden，核查 agent 在工具边界
+    遮、可验证报告判快照附不附带用的同一套；SQL 解析不了、里面又写着遮罩列（或者 * 这类不写列名也会带出遮罩列的写法）
+    的，整份快照的列都算。合并查询的快照它不看 SQL，原样交回（合并的遮罩由 _merge_masks 按逐格来历算）。
+
+    分析本身出了意外：整份的列都算（宁可多遮）。纯计算，一条 SQL 不到 1 毫秒，在调用方的线程里算：推演在工作线程里
+    等事件循环查库（api/whatif.py 的 _Hidden），这里再开线程会和它争同一个线程池。"""
+    if not hidden:
+        return set()
+    try:
+        # masking 要用这里的说法（MASKED），也是用到时才导入：两边都不在模块顶上互相导入
+        from app.engine.masking import _derived_hidden
+
+        return set(hidden) | _derived_hidden(snap, set(hidden), kind)[0]
+    except Exception:  # noqa: BLE001 - 认不准就整份遮掉
+        logger.warning("证据面板：按 SQL 认改了名字的遮罩列时出错，整份快照的列都按遮罩处理", exc_info=True)
+        return set(hidden) | {str(c).lower() for c in snap.get("columns") or []}
+
+
+async def _snapshot_masks(snap: dict[str, Any] | None, source: Any, masks: _Masks, *,
+                          depth: int = 0) -> tuple[set[str], list[str]]:
+    """一份查询快照要遮的列（小写）和已经不在的数据源：证据面板的查询步骤、指标的输入、推演、报告自查（engine/nodes/
+    report.py）都按这一个函数认。
+
+    - 遮的是「查询当时记下的」和「数据源现在设的」两者之和：事后加的遮罩照样生效，数据源改名、删掉了也不会把当时遮着
+      的列亮出来；数据源不在了的，它的名字进第二项
+    - 列名对上的之外，按 SQL 认出改了名字的遮罩列（_derived）
+    - 合并查询不是数据源：按它合并的那几个数据源和逐格来历算（_merge_masks）
+    - snap 为 None（快照读不出来）：只能按 source 现在设的遮罩算，没有 SQL 可认"""
+    if snap is not None and snap.get("source") == MERGE_SOURCE:
+        return await _merge_masks(snap, masks, depth=depth)
+    name = source or (snap or {}).get("source")
+    hidden, found = await masks.of(name)
+    hidden = set(hidden) | (_recorded_masks(snap) if snap is not None else set())
+    if snap is not None and hidden:
+        hidden = _derived(snap, hidden, await masks.kind(name))
+    return hidden, ([] if found or not name else [str(name)])
+
 
 async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str, Any], sealed: _Sealed,
                       masks: _Masks, *, hint: bool = False) -> dict[str, Any]:
@@ -1343,22 +1396,15 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
     rows = snap["rows"]
     step["source"] = step["source"] or snap.get("source")
     merged = snap.get("source") == MERGE_SOURCE
+    # 遮的列：查询当时记下的加数据源现在设的，按 SQL 认出改了名字的（c.phone AS tel）也遮。合并查询不是数据源，按它
+    # 合并的那几个数据源和逐格来历算，不能拿「合并查询」去找数据源，找不到再说它「已不存在」（_snapshot_masks）
+    hidden, gone = await _snapshot_masks(snap, step["source"], masks)
+    if gone:
+        step["mask_note"] = SOURCE_GONE.format(source="、".join(gone))
     if merged:
-        # 合并查询不是数据源：遮罩按它合并的那几个数据源算（_merge_masks），不能拿「合并查询」去找数据源，
-        # 找不到再说它「已不存在」
-        hidden, gone = await _merge_masks(snap, masks)
-        if gone:
-            step["mask_note"] = SOURCE_GONE.format(source="、".join(gone))
         step["merge"] = _merge_view(snap, cells, catalog, sealed)
         if hint and _leads_to_upload(snap, cells):
             step["provenance"] = True
-    else:
-        # 遮的是「查询当时记下的」和「数据源现在设的」两者之和：事后加的遮罩照样生效，数据源改名、
-        # 删掉了也不会把当时遮着的列亮出来
-        hidden, found = await masks.of(step["source"])
-        hidden = hidden | _recorded_masks(snap)
-        if not found:
-            step["mask_note"] = SOURCE_GONE.format(source=step["source"])
     masked = {i for i, c in enumerate(columns) if c.lower() in hidden}
     index = sorted({i for r in rows_hit for i in range(r - WINDOW, r + WINDOW + 1) if 0 <= i < len(rows)})
     step.update(sql=snap.get("sql"), columns=columns, total_rows=len(rows), truncated=bool(snap.get("truncated")),
@@ -1497,12 +1543,13 @@ def _input_checks(artifact: str) -> list[dict[str, Any]]:
     return _query_checks(snap) if isinstance(snap, dict) else []
 
 
-async def _merge_masks(snap: dict[str, Any], masks: _Masks) -> tuple[set[str], list[str]]:
+async def _merge_masks(snap: dict[str, Any], masks: _Masks, *, depth: int = 0) -> tuple[set[str], list[str]]:
     """合并结果要遮的列（小写）和已经不在的数据源：
 
     - 合并当时记下的（快照的 mask_columns：各输入当时的遮罩列，加上来自遮罩列、换了名字的结果列）；
     - 它合并的每个数据源现在设的（按列名）；
-    - 结果列有逐格来历、来自某个数据源现在遮着的列的，按来历遮（换了名字也遮住）。宁可多遮。"""
+    - 结果列有逐格来历的，按来历遮：来自的那一列在输入里要遮（_merge_input_masks：输入快照按查询步骤同一套认，数据源
+      现在设的、SQL 里改了名字的都算，输入本身是合并结果的再往下追），换了名字也遮住。宁可多遮。"""
     hidden = _recorded_masks(snap)
     gone: list[str] = []
     for name in snap.get("sources") or []:
@@ -1513,17 +1560,33 @@ async def _merge_masks(snap: dict[str, Any], masks: _Masks) -> tuple[set[str], l
     lineage = snap.get("lineage") if isinstance(snap.get("lineage"), dict) else {}
     origins = lineage.get("sources") or []
     by_alias = {str(i.get("alias")): i for i in snap.get("inputs") or [] if isinstance(i, dict)}
+    inputs: dict[str, set[str]] = {}
     for name, target in zip(snap.get("columns") or [], lineage.get("columns") or []):
         if not isinstance(target, dict) or not isinstance(target.get("source"), int) \
                 or not 0 <= target["source"] < len(origins):
             continue
         origin = origins[target["source"]] if isinstance(origins[target["source"]], dict) else {}
-        source = (by_alias.get(str(origin.get("input"))) or {}).get("source")
-        if source and source != MERGE_SOURCE and str(target.get("column") or "").lower() in (await masks.of(source))[0]:
+        alias = str(origin.get("input"))
+        if alias not in inputs:
+            inputs[alias] = await _merge_input_masks(by_alias.get(alias) or {}, masks, depth)
+        if str(target.get("column") or "").lower() in inputs[alias]:
             hidden.add(str(name).lower())
     return hidden, gone
 
 
+async def _merge_input_masks(item: dict[str, Any], masks: _Masks, depth: int) -> set[str]:
+    """合并的一个输入（快照 inputs 里的一项）的结果里要遮的列（小写）。读得出输入快照的按 _snapshot_masks 认（SQL 里给
+    遮罩列起了别名的也算）；读不出来、追得太深的，只能按它的数据源现在设的列名算。"""
+    source = item.get("source")
+    artifact = str(item.get("artifact") or "")
+    content, ok = _fetch(artifact) if artifact and depth < MERGE_DEPTH else (None, None)
+    if ok is True and isinstance(content, dict):
+        return (await _snapshot_masks(content, source, masks, depth=depth + 1))[0]
+    if source and source != MERGE_SOURCE:
+        return set((await masks.of(source))[0])
+    return set()
+
+
 def _leads_to_upload(snap: dict[str, Any], cells: list[tuple[Any, Any]], depth: int = 0) -> bool:
     """被引用的格经合并追下去，最后落在上传表格的查询快照上（有 data_version）：界面才去请求推断来源。"""
     if depth >= MERGE_DEPTH:

backend/app/api/whatif.py

@@ -10,7 +10,8 @@ GET /api/runs/{run_id}/evidence/whatif/rows：推演时勾行用，按页取封
 只认这次运行封存范围内的工件，和证据接口同一套白名单（api/evidence.py 的 _sealed / _sealed_loader）：口径卡的指标集
 从封存的 node.finished 台账找，排除行只认封存范围内交回过的查询快照，nodes.* 的完整值从封存的节点产出工件取，
 vars.* 的完整值（value_artifact）只认封存的指标集里写着的。还在跑、停在审批的运行也能推演，结果里 sealed 为假：
-数据可能还会变。遮罩列（查询当时记下的加数据源现在设的）不给值、也不能拿来推演，见 engine/recompute.py。
+数据可能还会变。遮罩列（查询当时记下的加数据源现在设的，SQL 里改了名字的也算）不给值、也不能拿来推演，用到它的指标
+原值也不给，见 engine/recompute.py。
 
 重算本身是纯函数（engine/recompute.py）。这里负责：认请求、找卡、准备工件和遮罩、找报告里引用的地方、标出可能受影响
 的下游口径卡。
@@ -31,17 +32,15 @@ from app.api.evidence import (
     REPORT_NOT_FOUND,
     TAMPERED_QUERY,
     _Masks,
-    _merge_masks,
-    _recorded_masks,
     _reports,
     _Sealed,
     _sealed,
     _sealed_cards,
     _sealed_loader,
+    _snapshot_masks,
 )
 from app.core.artifact_store import load
 from app.engine import recompute
-from app.engine.merge_query import MERGE_SOURCE
 
 router = APIRouter(prefix="/api/runs", tags=["whatif"])
 
@@ -115,7 +114,8 @@ async def evidence_whatif_rows(run_id: str, artifact: str | None = None, offset:
     """一页查询快照的行：{artifact, columns, rows, row_offset, row_index, total_rows, masked, hash_ok, sealed, note?}。
 
     形状和证据面板的查询步骤一致（界面用同一个函数读）。只认这次运行封存范围内交回过的快照；遮罩列（查询当时记下的
-    加数据源现在设的）写「已遮罩」；读不出来、与哈希不一致的不给行，note 照实说。只读：不追加事件、不落工件。
+    加数据源现在设的，SQL 里改了名字的也算）写「已遮罩」；读不出来、与哈希不一致的不给行，note 照实说。只读：不追加
+    事件、不落工件。
     """
     if not artifact or not artifact.strip():
         raise CodedHTTPException(422, "请求格式有误：artifact 应为查询快照的工件 ID", BAD_REQUEST)
@@ -312,9 +312,9 @@ def _loader(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]]) -> An
 class _Hidden(Mapping[str, frozenset[str]]):
     """封存范围内的查询快照 → 要遮的列（小写），推演用到哪份才算哪份。
 
-    遮的列和证据面板查询步骤同一套（api/evidence.py 的 _query_step）：查询当时记下的加数据源现在设的，合并查询按它
-    合并的那几个数据源算。快照读不出来、被改过时照样按台账记的数据源算：遮罩不能因为快照坏了就失效（那一格的值另由
-    重算不给）。
+    遮的列和证据面板查询步骤同一套（api/evidence.py 的 _snapshot_masks）：查询当时记下的加数据源现在设的，按 SQL 认出
+    改了名字的也算，合并查询按它合并的那几个数据源和逐格来历算。快照读不出来、被改过时照样按台账记的数据源算：遮罩
+    不能因为快照坏了就失效（那一格的值另由重算不给）。
 
     在重算的工作线程里取用：快照经推演用的同一个 loader 读（重算本来也要读这一份，读过的不再读盘）；查数据源现在
     设的遮罩要访问库，交回事件循环去查（_Masks 是异步的），线程在这里等它。以前每次请求在事件循环里把这次运行的
@@ -353,9 +353,8 @@ class _Hidden(Mapping[str, frozenset[str]]):
 
 
 async def _mask_of(snap: dict[str, Any] | None, info: dict[str, Any], masks: _Masks) -> set[str]:
-    """一份查询快照要遮的列（小写）。snap 为 None（读不出来）时只能按台账记的数据源现在设的遮罩算。"""
-    if snap is not None and snap.get("source") == MERGE_SOURCE:
-        hidden, _ = await _merge_masks(snap, masks)
-        return hidden
-    hidden, _ = await masks.of(info.get("source") or (snap or {}).get("source"))
-    return hidden | (_recorded_masks(snap) if snap is not None else set())
+    """一份查询快照要遮的列（小写），和证据面板的查询步骤同一个函数（api/evidence.py 的 _snapshot_masks）：查询当时记下的
+    加数据源现在设的，SQL 里给遮罩列起了别名、套了函数的结果列也算。snap 为 None（读不出来）时只能按台账记的数据源现在
+    设的遮罩算。"""
+    hidden, _ = await _snapshot_masks(snap, info.get("source"), masks)
+    return hidden

backend/app/engine/recompute.py

@@ -15,7 +15,9 @@
   - vars.*：有 value_artifact（口径卡当时另存的完整值）才还原得出来；老运行照实说当时没有保存
   - 容器本身是一份查询结果（带 artifact 的 {columns, rows}，或者和某份快照的 rows 一模一样的列表）：排除行就去掉那几行
 - 遮罩列取来的输入不给值、也不拿来推演：改一个输入、排除一行，新值和原值一比就能算出遮住的那一格。vars.res.rows[0][2]
-  这种按位置取自查询结果的标量没有定位，顺着取值链认出是哪一列（positional_cell，证据面板的指标步骤用同一个认法）
+  这种按位置取自查询结果的标量没有定位，顺着取值链认出是哪一列（positional_cell，证据面板的指标步骤用同一个认法）。
+  用到了遮罩列的指标，它本身的原值同样写「已遮罩」（hides_value，证据面板指标步骤的值同一个判断）：指标的值可能就是
+  那一格，输入里拿掉了、指标上照给，等于没遮
 - 落在读不出来（文件不在、与哈希不一致）的查询快照上的输入同样不给值、不推演：那一格遮没遮核对不了，宁可多遮
 
 有的指标不论改什么都推演不了（_Plan.blocked）：卡上没记全输入、用到了遮罩列、最外层的输入还原不了（容器的完整值
@@ -209,6 +211,13 @@ class _Plan:
                 return "unsupported", f"无法推演：{path}（{problem}）"
         return None
 
+    @property
+    def hides(self) -> bool:
+        """指标的原值也不给（hides_value 同一个判断）：用到了遮罩列，或者最外层的输入遮没遮核对不了。成立时一定推演
+        不了（blocked 不为 None），新值本来就是空的。"""
+        return _hides(self.tops, [p for p, i in self.inputs.items() if i.masked],
+                      [p for p, i in self.inputs.items() if i.unreadable])
+
     @property
     def stuck(self) -> str | None:
         """blocked 的原因换成不说「推演」的说法：recompute 交给双盲复算的预检、盲值，那里说「不能推演」驴唇不对马嘴。
@@ -872,7 +881,7 @@ def _result(plan: _Plan, status: str, new: Any, reason: str | None, values: dict
         else substitute(plan.tree, values) if status == "ok" and plan.tree is not None else None
     if any(inp.masked or inp.unreadable for inp in plan.inputs.values()):
         substituted = None      # 代入式里代进去的就是遮罩列（或者遮没遮核对不了）的那一格
-    return {
+    out = {
         "id": metric.get("id"), "name": metric.get("name") or metric.get("id"), "unit": metric.get("unit") or "",
         "decimals": metric.get("decimals"), "format": metric.get("format") or DEFAULT_FORMAT,
         "old": _finite(old), "new": _finite(new), "delta": delta,
@@ -882,6 +891,13 @@ def _result(plan: _Plan, status: str, new: Any, reason: str | None, values: dict
         "status": status, "reason": reason, "substituted": substituted,
         "inputs": [_input_view(inp, values, notes, blocked=blocked) for inp in plan.inputs.values()],
     }
+    if plan.hides:
+        # 指标的值就是（或者算自）遮罩列的那一格：原值同样写「已遮罩」。以前输入里拿掉了、指标的 old 照给，同一个号码
+        # 换个地方照样看得到；原因在 reason 里（推演不了的原因）。本来就没有值的照旧是空的，缺输入还得看得出来
+        out["withheld"] = True
+        if old is not None:
+            out["old"] = out["rendered_old"] = _masked_word()
+    return out
 
 
 def _render(metric: dict[str, Any], value: Any, conv: str | None = None) -> str:
@@ -1026,6 +1042,35 @@ def _fixed(inp: _Input) -> str | None:
     return None if value is None or is_number(value) else NOT_NUMBER
 
 
+def hides_value(metric: dict[str, Any], masked: Iterable[str], unverified: Iterable[str] = ()) -> bool:
+    """指标本身的值要不要一并不给（推演的原值、证据面板指标步骤的值）：masked、unverified 是取自遮罩列、遮没遮核对不了
+    的输入的路径。
+
+    - 有输入取自遮罩列：不给，哪怕它只嵌在下标里（推演把整个指标按遮罩处理，_Plan.blocked）。指标的值可能就是那一格
+      （cell(nodes.q, 0, 'phone')），也可能是算出来的数，宁可多遮：改一个别的输入、拿指标的值一减，就能算出遮住的那一格
+    - 最外层的输入遮没遮核对不了：不给，它的值就在指标值里。只嵌在下标里的不算：它只决定取哪一格，取到的值卡上另记着
+    认不出最外层引用（表达式有误）时，有一个遮没遮核对不了就不给。"""
+    masked, unverified = set(masked), set(unverified)
+    if masked or not unverified:
+        return bool(masked)
+    try:
+        tops, _ = _refs(parse_expression(str(metric.get("expression") or ""))[0])
+    except ExpressionError:
+        return True
+    return _hides(tops, masked, unverified)
+
+
+def _hides(tops: Iterable[str], masked: Iterable[str], unverified: Iterable[str]) -> bool:
+    return any(True for _ in masked) or bool(set(unverified) & set(tops))
+
+
+def _masked_word() -> str:
+    """「已遮罩」：证据面板、推演同一个说法。api.evidence 导入了这个模块，用到时再导入。"""
+    from app.api.evidence import MASKED
+
+    return MASKED
+
+
 def substituted_without(metric: dict[str, Any], paths: Iterable[str], token: str) -> str | None:
     """卡上的代入式，只把 paths 这几个输入代成 token（证据面板给取自遮罩列的输入写「已遮罩」），别的照卡上代。
 
@@ -1062,6 +1107,7 @@ def report_segments(reports: Iterable[tuple[str, dict[str, Any]]], *, card_node:
     只认指向这张卡这一份指标集的引用（目录条目的 node_id、artifact）：循环里同一张卡算过几次，报告引用的是哪一次
     就是哪一次。新值按片段自己的换算（|万、|pct……）渲染，和报告里的字同一套规则。指标推演不了（unsupported、masked）
     时 rendered_new、changed 为 None：不知道会变成什么。缺输入、当时就显示「—」的引用也列上：推演里补上了值，它会变。
+    原值不给的指标（withheld），rendered_old 同样写「已遮罩」。
     """
     by_id = {str(r.get("id")): r for r in results}
     out: list[dict[str, Any]] = []
@@ -1078,8 +1124,10 @@ def report_segments(reports: Iterable[tuple[str, dict[str, Any]]], *, card_node:
                 if result is None:
                     continue
                 new = None if result.get("rendered_new") is None else _render(result, result.get("new"), cite.get("conv"))
+                # 原值不给的指标（_Plan.hides）：报告里引用它的地方同样写「已遮罩」，推演的答复里不出现它的值
+                old = _masked_word() if result.get("withheld") and result.get("old") is not None else seg.get("text")
                 out.append({"report": report_node, "unit": unit.get("id"), "sid": seg.get("id"), "metric": result["id"],
-                            "rendered_old": seg.get("text"), "rendered_new": new,
+                            "rendered_old": old, "rendered_new": new,
                             "changed": None if new is None else new != seg.get("text")})
     return out
 

backend/tests/test_evidence_masked_aliases.py

@@ -0,0 +1,309 @@
+"""证据面板认得出改了名字的遮罩列：查询步骤、指标的输入、推演的勾行和遮罩，按 SQL 认。
+
+验收发现：正式运行的报告没写号码，可验证报告也把 top 查询（c.phone AS tel）的快照只给了哈希，点开正文里的消费额，
+片段接口的查询步骤却是 columns=[customer, tel, spend]、masked=[]，三个号码原样交出来。证据面板以前只遮列名恰好等于
+遮罩列的那几列；可验证报告、核查 agent 早已按 SQL 认出别名、函数、子查询、WITH 转一道得来的结果列。现在证据面板
+按同一套认法（engine/masking.py 的 _derived_hidden）：
+
+- 查询步骤：c.phone AS tel 的 tel 列遮掉，面板底下的遮罩说明带上它
+- 指标的输入：cell() 取 tel 那一格、按位置取的 vars.res.rows[0][1] 都不给值；指标本身的值也不给
+- 推演：勾行接口遮 tel，用到 tel 的指标推演不了、原值不给
+- SQL 解析不了、里面又写着遮罩列的，整份快照的列都遮（保守规则）；没写着的只遮遮罩列本身
+- 合并查询：输入快照里改了名字的遮罩列，按逐格来历遮到合并结果里
+- 没设遮罩时一切照旧
+
+真跑一遍图（虚构的 SQLite 商城库，号码是编的），报告由 MockChatModel 照剧本写。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+from sqlalchemy import select
+
+from app.core import artifact_store
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run
+from app.engine.evidence import iter_segments
+from app.engine.runner import run_manager
+from app.main import app
+
+CUSTOMERS = [(1, "客户甲", "199-0000-0101"), (2, "客户乙", "199-0000-0102"), (3, "客户丙", "199-0000-0103"),
+             (4, "客户丁", "199-0000-0104")]
+ORDERS = [(1, 1, 120.5), (2, 2, 300.0), (3, 2, 80.0), (4, 3, 210.25), (5, 4, 50.0), (6, 1, 99.5)]
+PHONES = [c[2] for c in CUSTOMERS]
+#: 验收原样的写法：号码起了别名 tel，和消费额排在同一份结果里
+SQL_TOP = ("SELECT c.name AS customer, c.phone AS tel, ROUND(SUM(o.amount), 2) AS spend "
+           "FROM orders o JOIN customers c ON c.id = o.customer_id GROUP BY c.id, c.name, c.phone ORDER BY spend DESC")
+TEL_CELL = "cell(nodes.top, 0, 'tel')"
+#: 号码那一格按位置取：vars.res 是数据整形解析好的查询结果，第 1 列是 tel
+TEL_POS = "vars.res.rows[0][1]"
+METRICS = [
+    {"id": "top_spend", "name": "最高消费", "unit": "元", "decimals": 2, "format": "thousands",
+     "expression": "cell(nodes.top, 0, 'spend')"},
+    {"id": "tel_len", "name": "号码位数", "unit": "位", "format": "thousands", "expression": f"len({TEL_CELL})"},
+    {"id": "tel_pos2", "name": "号码位数两倍", "unit": "位", "format": "thousands", "expression": f"len({TEL_POS}) * 2"},
+]
+TEXT = "消费最高的客户花了 [[v:Q1.r0.spend]]，口径卡记 [[m:top_spend]]；号码 [[m:tel_len]]，两倍 [[m:tel_pos2]]。"
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def set_source(path, options=None):
+    async with SessionLocal() as session:
+        row = (await session.execute(select(DataSource).where(DataSource.name == "mall"))).scalar_one_or_none()
+        if row is None:
+            row = DataSource(name="mall", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            session.add(row)
+        row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
+        row.options = dict(options or {})
+        await session.commit()
+        return row.id
+
+
+@pytest.fixture
+async def mall(tmp_path):
+    path = tmp_path / "mall.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);"
+                     "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL);")
+    db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?)", ORDERS)
+    db.commit()
+    db.close()
+    source_id = await set_source(path)
+    await engines.invalidate(source_id)
+    yield path
+    await engines.invalidate(source_id)
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def chain(*nodes) -> dict:
+    return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+def graph() -> dict:
+    return chain(
+        node("start", "input"),
+        node("top", "tool", tool="db_query__mall", args={"sql": SQL_TOP}),
+        node("parse", "transform", mode="json", template="{{ nodes.top }}", assign_to="res"),
+        node("card", "metrics", caliber="商城口径", caliber_version="v1", metrics=METRICS),
+        node("write", "report", instructions="写一句", on_violation="flag"),
+        node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]),
+    )
+
+
+async def run_graph(monkeypatch, spec: dict | None = None, text: str = TEXT) -> Run:
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
+    run_id = (await run_manager.start(graph=spec or graph(), input_payload={})).id
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            assert row.status == "succeeded", row.error
+            return row
+    raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
+
+
+async def doc_of(client, run_id: str) -> dict:
+    [report] = (await client.get(f"/api/runs/{run_id}/evidence")).json()["reports"]
+    return artifact_store.load(report["doc_artifact"])
+
+
+async def segment(client, run_id: str, *, text: str | None = None, ref: str | None = None) -> dict:
+    """按正文里的字或者引用（v:Q1.r0.spend）找片段，取片段接口的答复。"""
+    doc = await doc_of(client, run_id)
+    seg = next(s for s in iter_segments(doc) if (text is not None and s["text"] == text)
+               or (ref is not None and s.get("ref") == ref))
+    res = await client.get(f"/api/runs/{run_id}/evidence/segments/{seg['id']}")
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+def leaks(body) -> list[str]:
+    raw = json.dumps(body, ensure_ascii=False)
+    return [p for p in PHONES if p in raw]
+
+
+def steps(body: dict, kind: str) -> list[dict]:
+    return [s for s in body["chain"] if s["step"] == kind]
+
+
+# --------------------------------------------------------------------------
+# 查询步骤
+# --------------------------------------------------------------------------
+
+
+async def test_a_renamed_masked_column_is_masked_in_the_query_step(client, monkeypatch, mall):
+    """验收原样：点开正文里直接引用的消费额，查询步骤里 c.phone AS tel 的 tel 列遮掉，号码一个都不交出来。"""
+    await set_source(mall, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, ref="v:Q1.r0.spend")
+    [query] = steps(body, "query")
+    assert query["columns"] == ["customer", "tel", "spend"] and query["masked"] == ["tel"], query["masked"]
+    assert [r[1] for r in query["rows"]] == ["已遮罩"] * len(query["rows"]) and query["rows"][0][2] == 380.0
+    assert body["redacted"]["columns"] == ["tel"] and body["redacted"]["note"]
+    assert leaks(body) == []
+
+
+async def test_the_metric_chain_masks_the_renamed_column_too(client, monkeypatch, mall):
+    """口径卡指标 → 输入 → 查询：查询步骤同一套认法，tel 列同样遮掉（验收的 s13）。"""
+    await set_source(mall, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, text="380.00元")
+    [query] = steps(body, "query")
+    assert query["masked"] == ["tel"] and leaks(body) == []
+    [amount] = steps(body, "input")
+    assert amount["value"] == 380.0 and "masked" not in amount
+
+
+async def test_an_input_taken_from_a_renamed_masked_column_is_withheld(client, monkeypatch, mall):
+    """cell() 取 tel 那一格：输入不给值、写明取自哪一列，代入式里那一处写「已遮罩」，指标本身的值也不给。"""
+    await set_source(mall, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, text="13位")
+    metric = body["chain"][0]
+    assert metric["step"] == "metric" and metric["substituted"] == "len(已遮罩)"
+    assert metric["value"] == "已遮罩" and metric["rendered"] == "13位"
+    [cell] = steps(body, "input")
+    assert cell["value"] == "已遮罩" and cell["masked"] == ["tel"] and cell["path"] == TEL_CELL
+    assert body["redacted"]["columns"] == ["tel"] and leaks(body) == []
+
+
+async def test_a_scalar_taken_by_position_from_a_renamed_masked_column_is_withheld(client, monkeypatch, mall):
+    """vars.res.rows[0][1]：没有定位，顺着取值链认出取自查询结果的 tel 列，tel 又是 phone 改的名字。"""
+    await set_source(mall, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, text="26位")
+    assert body["chain"][0]["substituted"] == "len(已遮罩) * 2" and body["chain"][0]["value"] == "已遮罩"
+    [pos] = steps(body, "input")
+    assert pos["path"] == TEL_POS and pos["value"] == "已遮罩" and pos["masked"] == ["tel"]
+    assert leaks(body) == []
+
+
+async def test_without_masks_nothing_changes(client, monkeypatch, mall):
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, ref="v:Q1.r0.spend")
+    [query] = steps(body, "query")
+    assert query["masked"] == [] and query["rows"][0][1] == "199-0000-0102" and body["redacted"] == {"columns": []}
+    plen = await segment(client, row.id, text="13位")
+    assert plen["chain"][0]["value"] == 13 and plen["chain"][0]["substituted"] == "len('199-0000-0102')"
+
+
+async def test_a_mask_set_after_the_run_also_reaches_renamed_columns(client, monkeypatch, mall):
+    """事后才给数据源加的遮罩：按数据源现在设的列认 SQL，改了名字的照样遮。"""
+    row = await run_graph(monkeypatch)
+    await set_source(mall, {"mask_columns": "phone"})
+    body = await segment(client, row.id, ref="v:Q1.r0.spend")
+    assert steps(body, "query")[0]["masked"] == ["tel"] and leaks(body) == []
+
+
+# --------------------------------------------------------------------------
+# 推演：勾行接口、用到 tel 的指标
+# --------------------------------------------------------------------------
+
+
+async def test_the_whatif_masks_the_renamed_column(client, monkeypatch, mall):
+    await set_source(mall, {"mask_columns": ["phone"]})
+    row = await run_graph(monkeypatch)
+    body = await segment(client, row.id, ref="v:Q1.r0.spend")
+    snapshot = steps(body, "query")[0]["artifact"]
+    page = (await client.get(f"/api/runs/{row.id}/evidence/whatif/rows", params={"artifact": snapshot})).json()
+    assert page["masked"] == ["tel"] and all(r[1] == "已遮罩" for r in page["rows"]) and leaks(page) == []
+    res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card"})
+    assert res.status_code == 200, res.text
+    metrics = {m["id"]: m for m in res.json()["metrics"]}
+    assert metrics["tel_len"]["status"] == "masked" and metrics["tel_pos2"]["status"] == "masked"
+    assert metrics["top_spend"]["status"] == "unchanged" and metrics["top_spend"]["old"] == 380.0
+    assert leaks(res.json()) == []
+    res = await client.post(f"/api/runs/{row.id}/evidence/whatif",
+                            json={"card": "card", "overrides": {"inputs": {TEL_CELL: 1}}})
+    assert res.status_code == 422 and "遮罩" in res.json()["detail"]
+
+
+# --------------------------------------------------------------------------
+# 合并查询：输入里改了名字的遮罩列，按逐格来历遮到合并结果里
+# --------------------------------------------------------------------------
+
+
+async def test_a_merged_result_masks_a_column_renamed_in_its_input(client, monkeypatch, mall):
+    await set_source(mall, {"mask_columns": ["phone"]})
+    spec = chain(
+        node("start", "input"),
+        node("q_cust", "tool", tool="db_query__mall", args={"sql": "SELECT id, phone AS tel FROM customers ORDER BY id"}),
+        node("q_ord", "tool", tool="db_query__mall",
+             args={"sql": "SELECT customer_id AS id, SUM(amount) AS amount FROM orders GROUP BY customer_id ORDER BY id"}),
+        node("merged", "merge", inputs={"a": "q_cust", "b": "q_ord"},
+             sql="SELECT a.id, a.tel, b.amount FROM a JOIN b ON a.id = b.id ORDER BY a.id"),
+        node("write", "report", instructions="写一句", on_violation="flag"),
+        node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]),
+    )
+    row = await run_graph(monkeypatch, spec, text="一号客户共消费 [[v:Q3.r0.amount]]。")
+    body = await segment(client, row.id, ref="v:Q3.r0.amount")
+    merged = next(s for s in steps(body, "query") if s.get("merge"))
+    assert "tel" in merged["masked"] and all(r[merged["columns"].index("tel")] == "已遮罩" for r in merged["rows"])
+    assert leaks(body) == []
+
+
+# --------------------------------------------------------------------------
+# 手搭的快照：SQL 的几种写法，解析不了时的保守规则
+# --------------------------------------------------------------------------
+
+#: 库里没有这个数据源：遮罩只有查询当时记在快照里的那一份（phone）
+GHOST = "ghost_mall"
+
+
+async def query_step(sql: str, columns: list[str], rows: list[list]) -> dict:
+    """手搭一份封存范围内的查询快照（查询当时记下了遮罩列 phone），直接走查询步骤（evidence._query_step）。"""
+    from app.api import evidence
+
+    snap = await artifact_store.put_json({"columns": columns, "rows": rows, "source": GHOST, "sql": sql,
+                                          "mask_columns": ["phone"]}, kind="query")
+    sealed = evidence._Sealed(run_id="syn-alias", graph={}, events=[],
+                              seal={"sealed": True, "ok": True, "manifest_seq": 1, "legacy": False})
+    sealed.queries = {snap: {"node_id": "q", "tool": f"db_query__{GHOST}", "source": GHOST}}
+    return await evidence._query_step(snap, [(0, columns[0])], {"catalog": {}}, sealed, evidence._Masks())
+
+
+@pytest.mark.parametrize("sql, columns, masked", [
+    # 函数套一层再经 WITH 转一道
+    ("WITH t AS (SELECT name, UPPER(phone) AS p FROM customers) SELECT name AS customer, p AS contact FROM t",
+     ["customer", "contact"], ["contact"]),
+    # 子查询里起的别名，外层再选出来
+    ("SELECT x.who, x.num FROM (SELECT name AS who, phone AS num FROM customers) x", ["who", "num"], ["num"]),
+    # 截一段再拼上星号：值仍取自遮罩列
+    ("SELECT name, substr(phone, 1, 3) || '****' AS short_tel FROM customers", ["name", "short_tel"], ["short_tel"]),
+    # 只数个数：带不出值，不遮
+    ("SELECT COUNT(DISTINCT phone) AS n FROM customers", ["n"], []),
+    # 解析不了、又写着遮罩列：整份快照的列都遮
+    ("SELECT name, phone AS p FROM customers WHERE ((", ["name", "p"], ["name", "p"]),
+    # 解析不了、没写着遮罩列：只遮遮罩列本身（结果里没有）
+    ("SELECT name, id AS p FROM customers WHERE ((", ["name", "p"], []),
+])
+async def test_the_query_step_reads_the_sql_like_the_bundle_and_the_verify_agent(sql, columns, masked):
+    rows = [["客户甲" if i == 0 else "199-0000-0101" for i in range(len(columns))]]
+    step = await query_step(sql, columns, rows)
+    assert step["masked"] == masked, step["masked"]
+    assert [v == "已遮罩" for v in step["rows"][0]] == [c in masked for c in columns]

backend/tests/test_whatif_masked_values.py

@@ -0,0 +1,268 @@
+"""推演和证据面板不露指标本身的原值：指标的输入取自遮罩格（或者指标的值本身就是遮罩格）时，原值写「已遮罩」。
+
+验收发现：phone_raw = cell(nodes.q_cust, 0, 'phone') 推演时判成 masked，输入里置了空，指标的 old、rendered_old 却还是那串
+号码；按位置取的 phone_pos 也一样。同一份答复里输入拿掉了、指标上照给，等于没遮。现在：
+
+- 推演（engine/recompute.py）：用到了遮罩列的指标（哪怕只嵌在下标里），old、rendered_old 写「已遮罩」，标 withheld，
+  原因照旧写在 reason 里；new、rendered_new 本来就是空的（推演不了）；报告里引用它的地方 rendered_old 同样写「已遮罩」
+- 最外层的输入遮没遮核对不了（快照读不出来）的也一样；只嵌在下标里、遮没遮核对不了的那一格不碍事，原值照给
+- 证据面板的指标步骤：value 按同一个判断写「已遮罩」（recompute.hides_value），报告上的字（rendered）照旧
+- 没用到遮罩列的指标一字不差
+
+前半是纯函数（手写的虚构卡和快照），后半真跑一遍图（虚构的 SQLite 客户库，号码是编的）。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+from sqlalchemy import select
+
+from app.core import artifact_store
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run
+from app.engine import recompute
+from app.engine.evidence import iter_segments
+from app.engine.recompute import Changes, Sources, whatif
+from app.engine.runner import run_manager
+from app.main import app
+
+# --------------------------------------------------------------------------
+# 纯函数：手写的卡和快照
+# --------------------------------------------------------------------------
+
+SNAP, OTHER = "a1" * 32, "b2" * 32
+STORE = {
+    SNAP: {"columns": ["id", "phone", "amount"], "source": "crm", "sql": "SELECT id, phone, amount FROM members",
+           "rows": [[1, "19900000201", 120.5], [2, "19900000202", 80.0]]},
+    OTHER: {"columns": ["n"], "source": "crm", "sql": "SELECT 2 AS n", "rows": [[2]]},
+}
+PHONE = "cell(nodes.crm, 0, 'phone')"
+AMOUNT = "cell(nodes.crm, 0, 'amount')"
+
+
+def cell(row, column, value, artifact=SNAP):
+    return {"path": f"cell(nodes.crm, {row}, '{column}')", "value": value, "node_id": "crm", "via": "tool_cell",
+            "locator": {"row": row, "column": column}, "artifact": artifact, "status": "verified"}
+
+
+def scalar(path, value):
+    return {"path": path, "value": value, "node_id": "fetch", "via": "transform", "status": "ok"}
+
+
+def card(mid, expression, substituted, value, inputs, unit=""):
+    return {"kind": "metric_set", "caliber": "会员口径", "caliber_version": "v1", "metrics": [
+        {"id": mid, "name": mid, "unit": unit, "value": value, "format": "plain", "expression": expression,
+         "substituted": substituted, "inputs": inputs, "status": "ok"}]}
+
+
+def sources(**kw):
+    base = {"load": STORE.get, "queries": (SNAP, OTHER), "masked": {SNAP: frozenset({"phone"})}}
+    return Sources(**{**base, **kw})
+
+
+def text_of(result) -> str:
+    return json.dumps(result, ensure_ascii=False)
+
+
+def test_the_value_of_a_metric_that_is_a_masked_cell_is_withheld():
+    """验收的 phone_raw：指标就是遮罩列的那一格。old、rendered_old 都写「已遮罩」，号码不出现在答复的任何地方。"""
+    result = whatif(card("phone_raw", PHONE, "19900000201", 19900000201, [cell(0, "phone", 19900000201)]),
+                    Changes(), sources())
+    [m] = result["metrics"]
+    assert (m["status"], m["old"], m["rendered_old"], m["withheld"]) == ("masked", "已遮罩", "已遮罩", True)
+    assert [m[k] for k in ("new", "rendered_new", "delta", "rendered_delta", "changed", "substituted")] == [None] * 6
+    assert "遮罩" in m["reason"] and m["inputs"][0]["masked"] is True and m["inputs"][0]["old"] is None
+    assert "19900000201" not in text_of(result)
+
+
+def test_a_metric_computed_from_a_masked_cell_withholds_its_value_too():
+    """号码位数这类算出来的数同样不给：改一个别的输入、拿指标的值一减，就能算出遮住的那一格（宁可多遮）。"""
+    [m] = whatif(card("mix", f"len({PHONE}) + {AMOUNT}", "len('19900000201') + 120.5", 131.5,
+                      [cell(0, "phone", "19900000201"), cell(0, "amount", 120.5)]), Changes(), sources())["metrics"]
+    assert (m["status"], m["old"], m["rendered_old"]) == ("masked", "已遮罩", "已遮罩")
+    inputs = {i["path"]: i for i in m["inputs"]}
+    assert inputs[AMOUNT]["old"] == 120.5 and inputs[PHONE]["old"] is None
+
+
+def test_a_missing_value_stays_missing():
+    """本来就没有值的（缺输入）照旧是空的、显示「—」：缺输入还得看得出来。"""
+    [m] = whatif(card("gone", PHONE, None, None, [cell(0, "phone", None)]), Changes(), sources())["metrics"]
+    assert m["status"] == "masked" and m["old"] is None and m["rendered_old"] == "—" and m["withheld"] is True
+
+
+def test_a_top_level_cell_on_an_unreadable_snapshot_withholds_the_value():
+    """那一格所在的快照读不出来：遮没遮核对不了，输入不给值，指标的原值同样不给（和代入式一样）。"""
+    [m] = whatif(card("east", AMOUNT, "120.5", 120.5, [cell(0, "amount", 120.5)]), Changes(),
+                 sources(load=lambda a: None if a == SNAP else STORE.get(a)))["metrics"]
+    assert m["status"] == "unsupported" and m["old"] == "已遮罩" and m["withheld"] is True
+    assert "120.5" not in text_of(m)
+
+
+def test_a_cell_only_choosing_a_row_on_an_unreadable_snapshot_does_not_hide_the_value():
+    """只嵌在下标里、遮没遮核对不了的那一格：它只决定取哪一项，取到的值卡上另记着，原值照给、照样推演。"""
+    pick = f"vars.kpi.list[{AMOUNT.replace('amount', 'id')}] + vars.kpi.k"
+    top = f"vars.kpi.list[{AMOUNT.replace('amount', 'id')}]"
+    metric = card("pick", pick, "20 + 2", 22, [scalar(top, 20), cell(0, "id", 1), scalar("vars.kpi.k", 2)])
+    [m] = whatif(metric, Changes(), sources(load=lambda a: None if a == SNAP else STORE.get(a)))["metrics"]
+    assert m["status"] == "unchanged" and m["old"] == 22 and "withheld" not in m
+    [m] = whatif(metric, Changes(inputs={"vars.kpi.k": 5}), sources(load=lambda a: None if a == SNAP else STORE.get(a)))[
+        "metrics"]
+    assert (m["status"], m["old"], m["new"]) == ("ok", 22, 25)
+
+
+def test_an_unmasked_metric_is_unchanged():
+    result = whatif(card("east", AMOUNT, "120.5", 120.5, [cell(0, "amount", 120.5)]), Changes(), sources())
+    [m] = result["metrics"]
+    assert (m["status"], m["old"], m["rendered_old"]) == ("unchanged", 120.5, "120.5") and "withheld" not in m
+
+
+def test_report_segments_of_a_withheld_metric_do_not_carry_the_value():
+    """报告里引用了原值不给的指标：推演答复里那一处的 rendered_old 同样写「已遮罩」。"""
+    metric = card("phone_raw", PHONE, "19900000201", 19900000201, [cell(0, "phone", 19900000201)])
+    result = whatif(metric, Changes(), sources())
+    entry = {"kind": "metric", "node_id": "card", "artifact": "c" * 64, "locator": {"metric": "phone_raw"}}
+    doc = {"catalog": {"m:phone_raw": entry}, "blocks": [{"units": [{"id": "u0", "segments": [
+        {"id": "s0", "kind": "number", "text": "19900000201", "state": "deterministic",
+         "cite": {"alias": "m:phone_raw", "kind": "metric", "status": "resolved"}}]}]}]}
+    [seg] = recompute.report_segments([("write", doc)], card_node="card", card_artifact="c" * 64,
+                                      results=result["metrics"])
+    assert (seg["rendered_old"], seg["rendered_new"], seg["changed"]) == ("已遮罩", None, None)
+
+
+@pytest.mark.parametrize("masked, unverified, expected", [
+    ({PHONE}, set(), True),                  # 取自遮罩列
+    ({"vars.kpi.i"}, set(), True),           # 遮罩列的那一格只嵌在下标里：推演整个按遮罩处理，原值也不给
+    (set(), {AMOUNT}, True),                 # 最外层的输入遮没遮核对不了
+    (set(), {"vars.kpi.i"}, False),          # 只嵌在下标里、遮没遮核对不了：不碍事
+    (set(), set(), False),
+])
+def test_hides_value_is_one_rule_for_the_whatif_and_the_evidence_panel(masked, unverified, expected):
+    metric = {"expression": f"{AMOUNT} + vars.kpi.list[vars.kpi.i]"}
+    assert recompute.hides_value(metric, masked, unverified) is expected
+    # 表达式读不懂：有一个遮没遮核对不了的就不给
+    assert recompute.hides_value({"expression": "1 +"}, set(), {AMOUNT}) is True
+
+
+# --------------------------------------------------------------------------
+# 真跑：验收原样的口径卡（单格原值、按位置取、按位置取的位数、编号）
+# --------------------------------------------------------------------------
+
+CUSTOMERS = [(1, "客户甲", "19900000301"), (2, "客户乙", "19900000302"), (3, "客户丙", "19900000303")]
+PHONES = [c[2] for c in CUSTOMERS]
+CARD_METRICS = [
+    {"id": "phone_raw", "name": "首位客户手机号（单格原值）", "format": "plain",
+     "expression": "cell(nodes.q_cust, 0, 'phone')"},
+    {"id": "phone_pos", "name": "首位客户手机号（按位置）", "format": "plain", "expression": "vars.cust.rows[0][2]"},
+    {"id": "phone_pos_len", "name": "首位客户手机号位数", "unit": "位", "format": "plain",
+     "expression": "len(str(vars.cust.rows[0][2]))"},
+    {"id": "id0", "name": "首位客户编号", "format": "plain", "expression": "cell(nodes.q_cust, 0, 'id')"},
+]
+TEXT = "首位客户编号 [[m:id0]]，号码位数 [[m:phone_pos_len]]。"
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client(engine_up):
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+async def members(tmp_path):
+    path = tmp_path / "members.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);")
+    db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = (await session.execute(select(DataSource).where(DataSource.name == "members"))).scalar_one_or_none()
+        if row is None:
+            row = DataSource(name="members", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            session.add(row)
+        row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
+        row.options = {"mask_columns": ["phone"]}
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield path
+    await engines.invalidate(source_id)
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def graph() -> dict:
+    nodes = [
+        node("start", "input"),
+        node("q_cust", "tool", tool="db_query__members",
+             args={"sql": "SELECT id, name, phone FROM customers ORDER BY id LIMIT 5"}),
+        node("parse_cust", "transform", mode="json", template="{{ nodes.q_cust }}", assign_to="cust"),
+        node("card_c", "metrics", caliber="遮罩取法对照", caliber_version="v1", metrics=CARD_METRICS),
+        node("write", "report", instructions="写一句", on_violation="flag"),
+        node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]),
+    ]
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+async def run_graph(monkeypatch) -> Run:
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=TEXT))
+    run_id = (await run_manager.start(graph=graph(), input_payload={})).id
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            assert row.status == "succeeded", row.error
+            return row
+    raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
+
+
+def leaks(raw: str) -> list[str]:
+    return [p for p in PHONES if p in raw]
+
+
+async def test_the_whatif_answer_carries_no_masked_value(client, monkeypatch, members):
+    """验收原样：推演答复里号码一个都没有（以前在 phone_raw、phone_pos 的 old、rendered_old 里）。"""
+    row = await run_graph(monkeypatch)
+    res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card_c"})
+    assert res.status_code == 200, res.text
+    assert leaks(res.text) == [], leaks(res.text)
+    metrics = {m["id"]: m for m in res.json()["metrics"]}
+    for mid in ("phone_raw", "phone_pos", "phone_pos_len"):
+        m = metrics[mid]
+        assert (m["status"], m["old"], m["rendered_old"], m["withheld"]) == ("masked", "已遮罩", "已遮罩", True), m
+        assert m["new"] is None and m["rendered_new"] is None and "遮罩" in m["reason"]
+    assert (metrics["id0"]["status"], metrics["id0"]["old"]) == ("unchanged", 1) and "withheld" not in metrics["id0"]
+    # 报告里引用了号码位数：那一处的原值也写「已遮罩」
+    [seg] = [s for s in res.json()["segments"] if s["metric"] == "phone_pos_len"]
+    assert seg["rendered_old"] == "已遮罩" and seg["rendered_new"] is None
+
+
+async def test_the_evidence_panel_withholds_the_metric_value_by_the_same_rule(client, monkeypatch, members):
+    """证据面板的指标步骤：用到遮罩列的指标 value 写「已遮罩」，报告上的字照旧；没用到的照给。"""
+    row = await run_graph(monkeypatch)
+    [report] = (await client.get(f"/api/runs/{row.id}/evidence")).json()["reports"]
+    doc = artifact_store.load(report["doc_artifact"])
+    by_text = {s["text"]: s["id"] for s in iter_segments(doc) if s.get("ref")}
+    body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{by_text['11位']}")).json()
+    metric = body["chain"][0]
+    assert (metric["step"], metric["value"], metric["rendered"]) == ("metric", "已遮罩", "11位")
+    assert leaks(json.dumps(body, ensure_ascii=False)) == []
+    plain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{by_text['1']}")).json()
+    assert plain["chain"][0]["value"] == 1