K3：核查 agent 看到的查询结果先遮、存下的快照也遮

双盲复算、反方审稿、审批质询共用的 engine/verify_agent.py 以前把查询结果原样交给模型（db_query 自己落快照，ToolMessage 里是原样的结果），遮罩全靠审批质询事后按值认，期 4 起六轮复核每轮都找到新的漏法；运行详情「工件」页签走的通用接口 GET /api/artifacts/{id} 还把核查的查询快照原样给出。根本的修法是不让模型看到遮罩值。 - tools/datasource：查询工具加遮罩开关 mask_results（缺省关，Agent 节点、调用工具节点照旧）。打开时数据源设了遮罩的，先按 engine/masking.py 的 mask_query_result 遮好再交给模型、再落快照：列名直接对上的、按 SQL 认出取自遮罩列的（别名、函数、子查询、WITH、SELECT *，复用 _derived_hidden；SQL 解析不了又写着遮罩列的，整份结果的列都遮）整列换成「已遮罩」、类型记成文本；同一份结果里别的格子整格是遮罩值的换掉，文字格里夹着的那一截换掉（换了写法的号码也认）；列名不换。结果里一列都没取自遮罩列的（数值的汇总、只数个数、只在条件里用到）原样。快照里 mask_columns 另记下查询时遮了的、改了名字的结果列（证据面板、质询工件据此画成遮罩列），masked_at_query 记下查询时就整列遮掉的列。这一次遮掉的值并进 ToolContext.extra[MASKED_VALUES_KEY]，只在内存里。 - verify_agent：建查询工具时和 force_readonly 一样打开遮罩开关；格子引用指到查询时就遮了的列，照实说是遮罩列、不采信；遮掉的值经 VerifyOutcome.masked_values 交回调用方（不进 as_dict、不往外交）；系统提示补一句「已遮罩」是什么。 - 审批质询：按值遮降为第二道防线。agent 自己的快照查询时就遮好了，收不到遮罩值，改由 masked_values 并进来（_plan 的 seen），回答、查询清单照旧按值遮；「已遮罩」本身不收成遮罩值。事后取质询工件时内存里的值已经不在，SQL 检查结果的片段、说明照事件里记下的那条 SQL 换（_recorded_swaps：逐词对出回答时换掉的字面量和 LIKE 里换掉的那一截）。 - 双盲复算取盲算值、反方审稿的引用、证据面板、可验证报告（带遮罩的快照只给哈希，照旧）、推演、运行对比（不收核查的查询）都按快照里已经是「已遮罩」照常工作，文件不用改。测试：核查 agent 选出遮罩列（直接、别名、函数、子查询、WITH、SELECT *）时交给假模型的 ToolMessage、落库的快照、通用工件接口、tool.end 预览、收尾交给模型的查询结果里都是「已遮罩」；没碰遮罩列的聚合查询原样、照常取值；文字格夹着的号码换掉那一截；引用遮罩格子不采信；SQL 认不出时整份遮；开关缺省关、没设遮罩的源快照一字不差；列名不改；审批质询、反方审稿、双盲复算各补一条端到端；质询工件的检查片段照记下的 SQL 换。逐处去掉实现（开关、并进内存的值、照记下的 SQL 换、跳过「已遮罩」、遮罩格子的说法、记下改了名字的列、文字格按子串换、不改列名）对应的用例都会变红。
2026-11-15 01:11:59 +00:00 · 2026-10-06 17:36:55 +08:00
parent 1136b2d09d
commit 450758185d
10 changed files with 576 additions and 47 deletions
@@ -12,7 +12,11 @@
   上传表格不算，见 _usable_sources）；上限按设置（每次的查询次数、时长，全局每日金额）。回答先给结论，数字只能引用 agent
   自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 compose_doc 组装，裸数字被标出来（提问人自己写的数除外）。
   封存之后的追问查的是数据源现在的数据，任务里给报告生成的时间，回答末尾注明；触顶、查询失败、引用解析不了也在末尾注明
-3. 遮罩（_plan）：哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列
+3. 遮罩（_plan）：agent 的查询结果在核查 agent 的工具边界就遮好了（engine/masking.py 的 mask_query_result）：取自
+   遮罩列的格子交给模型、落进快照之前就是「已遮罩」，模型看不到遮罩值。这里是第二道防线：查询时遮掉的值由 verify_agent
+   交回（masked_values，只在内存里），连同报告自己快照里的遮罩值，照下面的办法按值遮；事后取质询工件时内存里的值已经
+   不在，SQL 检查结果照事件里记下的 SQL 换（_recorded_swaps）。
+   哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列
   （起别名、套函数、经子查询转一道、和 * 写在同一支里，_derived_hidden；认不出落在哪一列的，整份快照遮掉；按遮罩列
   分档、当分组键或套在汇总里数个数的不算，_band_only）；快照按列遮，再按值遮（别的列里写着遮罩值的格子，_masked_view；质询
   工件里，文字格里夹着的遮罩值另按子串换）。回答正文、查询清单里的遮罩值换成「已遮罩」（_Secrets、_scrub_sql）：
@@ -73,8 +77,10 @@
 - 质询工件里 SQL 检查结果的片段是检查器按语法树重新写出来的，按整条 SQL 里整个换掉了的字面量照值换：同一个值在
  片段里别处出现也换（多遮，算式里换掉的 0、1 这类也是）；截断处半截的值去掉（停在字符串、带引号的名字、注释里的，
  从开头的引号、注释起整个去掉）。说明里码值表列出的码值和别的字一样按值换，恰好是遮罩值的码值也换（多遮）
- 运行详情的「工件」页签、通用的工件接口按工件原样给：质询的查询快照记在这次运行名下（同报告自己的查询快照），从
-  那里打开不经这里的遮罩
+- 运行详情的「工件」页签、通用的工件接口按工件原样给：质询的查询快照落库时就遮好了（工具边界先遮），从那里打开也
+  看不到取自遮罩列的值；但工具边界只认这份结果里的遮罩值，别的查询、报告快照里遮罩列的值写在这份结果没设遮罩的列里时
+  （没设遮罩的联系方式列、备注），从那里打开照原样（质询工件接口按回答时的办法换）。报告自己的查询快照（写作时的查询）
+  仍原样存着，从那里打开不经这里的遮罩
 - 遮罩值最多收 SECRETS_MAX 种写法，再多的按值认不出（按列遮照旧）
 - 换了写法的号码只认数字之间夹着一到三个空格、连字符、点、括号（含全角），带 +86 / 0086 / 86 前缀、全角数字：写成
  汉字（一三九……）、中间夹着别的字、拆成几处写的认不出；和别的数字只隔着一个连字符或点连着的（13900000001-2、
@@ -84,6 +90,7 @@ from __future__ import annotations

 import asyncio
 import contextlib
+import difflib
 import itertools
 import logging
 import operator
@@ -91,7 +98,7 @@ import re
 import time
 from collections import OrderedDict
 from collections.abc import Callable, Iterable, Iterator
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from datetime import datetime, timezone
 from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal
 from typing import Any
@@ -1618,13 +1625,15 @@ async def _source_kinds(names: Iterable[str]) -> dict[str, str | None]:

 async def _plan(own: dict[str, dict[str, Any]], others: list[dict[str, Any]], given: list[str | None],
                shown: list[tuple[str, str, str]] | None = None, live: dict[str, list[str]] | None = None, *,
-                embedded: frozenset[str] = frozenset()) -> _Plan:
+                embedded: frozenset[str] = frozenset(), seen: _Secrets | None = None) -> _Plan:
    """own：agent 自己的查询快照（工件 id → 内容）；others：报告自己的查询快照（被问的那句可能直接写着遮罩列的值，
    agent 会照着复述、写进查询条件）；given：问题、选中的原文、报告里列出的指标和运行输入的值（都已经摆在读者眼前，
    里面的值哪里都不换）；shown：agent 执行的查询，(SQL, 数据源名, 快照工件 id)（照原样留在 SQL 里的阈值、LIMIT，
    回答正文里不换，见 _Secrets.show；取质询工件不用：快照里的格子照旧按值遮）；live：用得上的数据源现在设的遮罩列
    （快照没取到的查询按它认 SQL 里的比较，见 _compared）；embedded：这几份快照的文字格里夹着的遮罩值也按子串换
-    （只给要取的那份质询工件，见 _masked_view）。
+    （只给要取的那份质询工件，见 _masked_view）；seen：agent 查询时在工具边界遮掉的值（verify_agent 交回的
+    masked_values，只在内存里）。agent 自己的快照查询时就遮好了，遮罩列里收不到值；模型照着别处看到的号码写进回答、
+    查询条件时，按值遮要靠它认出来。

    查数据源的遮罩要读库，在这里读；解析 SQL、收上万个值、逐格比对是纯计算，放进线程：放在事件循环里会挡住同一进程
    里别的运行（同 tools/datasource 的 _sql_checks）。"""
@@ -1636,13 +1645,13 @@ async def _plan(own: dict[str, dict[str, Any]], others: list[dict[str, Any]], gi
    based = [(snapshot, (await _hidden_columns(snapshot, masks))[0]) for snapshot in others]
    kinds = await _source_kinds([*(str(s.get("source") or "") for s in [*own.values(), *others]),
                                 *(name for _, name, _ in shown)])
-    return await asyncio.to_thread(_settle_plan, own, named, based, kinds, given, shown, live or {}, embedded)
+    return await asyncio.to_thread(_settle_plan, own, named, based, kinds, given, shown, live or {}, embedded, seen)


 def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str], list[str]]],
                 based: list[tuple[dict[str, Any], set[str]]], kinds: dict[str, str | None],
                 given: list[str | None], shown: list[tuple[str, str, str]], live: dict[str, list[str]],
-                 embedded: frozenset[str]) -> _Plan:
+                 embedded: frozenset[str], seen: _Secrets | None = None) -> _Plan:
    secrets = _Secrets()
    hidden: dict[str, set[str]] = {}
    tainted: dict[str, set[str]] = {}
@@ -1654,6 +1663,7 @@ def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str]
        kind = kinds.get(str(snapshot.get("source") or ""))
        secrets.add(snapshot, _derived_hidden(snapshot, base, kind)[0], labels=_snapshot_labels(snapshot, kind),
                    base=base)
+    secrets.merge(seen)
    secrets.protect(given)
    secrets.show(_shown_literals([(sql, kinds.get(name), _compared(tainted, artifact, live, name))
                                  for sql, name, artifact in shown], secrets))
@@ -1755,7 +1765,8 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict
        output_schema=INQUIRY_SCHEMA, max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"),
        daily_max_usd=settings.get("daily_max_usd"), run_id=op.run_id, data_versions=op.versions)

-    # 遮罩：agent 自己的查询快照、报告自己的查询快照（被问的那句可能直接写着遮罩列的值，agent 会照着复述）
+    # 遮罩：agent 自己的查询快照（查询时就遮好了，遮掉的值在 outcome.masked_values 里）、报告自己的查询快照（被问的
+    # 那句可能直接写着遮罩列的值，agent 会照着复述）
    own: dict[str, dict[str, Any]] = {}
    for query in outcome.queries:
        snapshot = _quiet_load(artifact_store.load, query.get("artifact"))
@@ -1765,7 +1776,7 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict
              and _query_shaped(snapshot := _quiet_load(op.loader, entry.get("artifact")))]
    plan = await _plan(own, others, [ask.question, ask.quote, *_ref_values(refs)],
                       [(str(q.get("sql") or ""), str(q.get("source") or ""), str(q.get("artifact") or ""))
-                        for q in outcome.queries], live)
+                        for q in outcome.queries], live, seen=outcome.masked_values)

    text, doc_id, failed = "", None, None
    source = (outcome.rationale or {}).get("source") if outcome.result is not None else None
@@ -2059,22 +2070,57 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]:
    elif isinstance(shown.get("sql"), str):
        shown = {**shown, "sql": _scrub_sql(shown["sql"], plan.secrets, compared, kind)}
    if "checks" in shown:
-        shown = {**shown, "checks": _masked_checks(shown["checks"], content.get("sql"), plan.secrets, compared, kind)}
+        whole, pieces = _recorded_swaps(content.get("sql"), recorded, kind) if recorded is not None else (set(), set())
+        shown = {**shown, "checks": _masked_checks(shown["checks"], content.get("sql"), plan.secrets, compared, kind,
+                                                   swapped=whole, pieces=pieces)}
    out: dict[str, Any] = {"id": artifact_id, "content": shown, "masked": plan.masked[artifact_id]}
    if gone := plan.gone.get(artifact_id):
        out["mask_note"] = SOURCE_GONE.format(source="、".join(gone))
    return out


-def _masked_checks(checks: Any, sql: Any, secrets: _Secrets, compared: set[str], kind: str | None
+def _recorded_swaps(sql: Any, recorded: str, kind: str | None) -> tuple[set[str], set[str]]:
+    """回答时这条 SQL 里换掉的值：快照里原样的那条和事件里记下的那条（_scrub_sql 换过的）逐词对，原样那条里的字面量
+    到了记下的那条里整个换成了「已遮罩」的（第一项，字面量的字），或者其中一截换成了「已遮罩」的（第二项，那一截，
+    LIKE '%13900000001%' 这类）。_scrub_sql 只换字面量、名字和注释里的值，一个词换成一个词，两条分出来的词一一对得上；
+    对不上（分词出错这类）、没有记下的那条时，什么都不认。"""
+    if not isinstance(sql, str) or not sql.strip() or not recorded or sql == recorded:
+        return set(), set()
+    before, after = _sql_tokens(sql, kind), _sql_tokens(recorded, kind)
+    if before is None or after is None or len(before[0]) != len(after[0]):
+        return set(), set()
+    word = _masked_word()
+    whole: set[str] = set()
+    pieces: set[str] = set()
+    for old, kind_of, new in zip(before[0], before[1], after[0]):
+        if not _is_literal(kind_of) or old.text == new.text or word not in new.text:
+            continue
+        if new.text == word:
+            whole.add(old.text)
+            continue
+        for op, i1, i2, j1, j2 in difflib.SequenceMatcher(None, old.text, new.text, autojunk=False).get_opcodes():
+            if op == "replace" and new.text[j1:j2] == word and old.text[i1:i2].strip():
+                pieces.add(old.text[i1:i2])
+    return whole, pieces
+
+
+def _masked_checks(checks: Any, sql: Any, secrets: _Secrets, compared: set[str], kind: str | None, *,
+                   swapped: set[str] | frozenset[str] = frozenset(), pieces: set[str] | frozenset[str] = frozenset()
                   ) -> list[dict[str, Any]]:
    """快照里记下的 SQL 检查结果（数据源查询工具对照数据目录查出的问题）交给质询工件的样子：字段照证据面板
    （_query_checks），给模型的那句改法不交出去——它照抄了 SQL 片段。片段（sql_excerpt）、说明（message）摘自、照引
    同一条 SQL（快照里原样的那条），照它在查询清单里的换法换掉遮罩值（_literal_swaps）：SQL 换了、片段和说明没换，
-    等于把换掉的值从旁边露出来；片段单拿出来认不出在找哪一个值，不能自己另认一遍。"""
+    等于把换掉的值从旁边露出来；片段单拿出来认不出在找哪一个值，不能自己另认一遍。
+    swapped / pieces：回答时这条 SQL 里整个换掉的字面量、换掉的一截（_recorded_swaps，照事件里记下的那条认）。回答时
+    agent 自己查询里遮掉的值还在内存里，事后取工件时已经不在了（快照查询时就遮好了，收不到值），照记下的 SQL 补上。"""
    from app.api.evidence import _query_checks

    whole = _literal_swaps(sql, secrets, compared, kind)
+    if swapped:
+        whole = (whole or set()) | set(swapped)
+    if pieces:
+        secrets = replace(secrets, strong=secrets.strong | set(pieces), weak=secrets.weak | set(pieces), _index=None,
+                          _compiled={})
    out: list[dict[str, Any]] = []
    for check in _query_checks({"checks": checks}):
        if isinstance(check.get("sql_excerpt"), str):
@@ -2,18 +2,24 @@

 数据源在 options.mask_columns 里设了遮罩的列（手机号、姓名这类），证据面板按列名遮（api/evidence.py）。核查 agent
 （engine/verify_agent.py：双盲复算、反方审稿、审批质询共用）自己写 SQL，起别名、套函数、经子查询转一道，结果列名就
-对不上遮罩列；按值认又有门槛和写法的问题。认法集中在这里，审批质询（engine/inquiry.py）、双盲复算（nodes/dual_check.py，
-经 inquiry 的同名导出）共用：
+对不上遮罩列；按值认又有门槛和写法的问题。认法集中在这里，核查 agent 的查询工具（tools/datasource 的遮罩开关）、审批
+质询（engine/inquiry.py）、双盲复算（nodes/dual_check.py，经 inquiry 的同名导出）共用：
+
+- 工具边界先遮（mask_query_result）：核查 agent 的查询结果交给模型、落进快照之前，取自遮罩列的格子换成「已遮罩」，
+  同一份结果里别的文字格夹着的遮罩值也换掉。模型看不到遮罩值，就不会把它写进回答、理由、SQL——事后按值认每一轮
+  复核都找到新的漏法，根治靠这一步；审批质询回答里的按值遮降为第二道防线

 - 哪些列（_derived_hidden）：从遮罩列本身起，按 SQL 一层层认出值取自遮罩列的结果列（别名、函数、子查询、WITH、UNION 的
  另一支、和 * 写在同一支里没起别名的表达式）；COUNT 只数个数、窗口的分区键只决定次序、按遮罩列分档又当分组键或套在
  汇总里数个数的不算（_value_columns、_band_only）。认不出落在哪一列的、SQL 解析不了又写着遮罩列的，整份快照的列都算
 - 只装着分档标签的列（_label_columns）：标签是 agent 写在 SQL 里的字，不当遮罩值
 - 哪些值（_Secrets）：遮罩列里出现过的值，按值认有门槛（SECRET_MIN、SECRET_DIGITS），换的时候先按词、按开头几个字筛出
-  正文里真有的那几个值，不现编上万个分支的正则
+  正文里真有的那几个值，不现编上万个分支的正则；正文、文字格里换了写法的长号码（199-1474-2785、+86 …、全角数字）按
+  纯数字写法在集合里查（loose_spans）
 - 快照遮好的样子（_masked_view）

-这里的函数原来都在 engine/inquiry.py 里，名字不变，inquiry 照旧导出（别处按 inquiry 的名字引用）。
+除 mask_query_result、换了写法的号码之外，这里的函数原来都在 engine/inquiry.py 里，名字不变，inquiry 照旧导出（别处按
+inquiry 的名字引用）。
 """
 from __future__ import annotations

@@ -810,19 +816,17 @@ def _loose_whole(text: str, run: re.Match[str], groups: list[tuple[int, int, str


 def _loose_span(text: str, run: re.Match[str], groups: list[tuple[int, int, str]], i: int, j: int) -> tuple[int, int]:
-    """第 i 到第 j 截这个号码在 text 里换掉的那一段：从第一截到最后一截，从这一串开头起的连同前面的 + 和括号。括号配成
-    对：号码里有落单的左括号，紧跟着的右括号一起换掉；配不上的落单左括号留着不换（「（电话 199-1474-2785）」两个括号
-    都留着）；落单的右括号，紧挨在前面的左括号一起换掉（(199)1474-2785 从第二截起认出来时）。"""
+    """第 i 到第 j 截这个号码在 text 里换掉的那一段：从第一截到最后一截，紧挨在前面的 + 一起换。括号只换号码里的：号码
+    里有落单的右括号（(199)1474-2785、(+86)199…），把前面配对的左括号一起换掉；落单的左括号，把后面紧跟着的右括号一起
+    换掉。整个号码外面套着的一对括号是标点（「来自（13900000001）」），留着不换。"""
    start, end = groups[i][0], groups[j][1]
-    if i == 0:
-        start = run.start()
+    if i == 0 and start > run.start() and text[start - 1] in "+＋":
+        start -= 1
    depth = sum((c in _OPENS) - (c in _CLOSES) for c in text[start:end])
-    while depth > 0 and end < len(text) and text[end] in _CLOSES:
-        end, depth = end + 1, depth - 1
-    while depth > 0 and text[start] in _OPENS:
-        start, depth = start + 1, depth - 1
    while depth < 0 and start > 0 and text[start - 1] in _OPENS:
        start, depth = start - 1, depth + 1
+    while depth > 0 and end < len(text) and text[end] in _CLOSES:
+        end, depth = end + 1, depth - 1
    return start, end


@@ -844,8 +848,9 @@ def _pattern(value: str) -> str:

@dataclass
 class _Secrets:
-    """遮罩列里出现过的值。模型看得到原始的查询结果（工具交回的是原样），回答里照着复述、或者把值写进查询条件，
-    都会把它带进回答正文和查询清单；按值找出来换成「已遮罩」。这只减少暴露，不是安全边界（同证据面板的遮罩）。
+    """遮罩列里出现过的值。模型照着报告正文、提问、别处的文字格里看到的值复述、或者把值写进查询条件，都会把它带进
+    回答正文和查询清单；按值找出来换成「已遮罩」。核查 agent 自己的查询结果在工具边界就遮好了（mask_query_result），
+    这里是第二道防线。这只减少暴露，不是安全边界（同证据面板的遮罩）。

    strong 是按值认得出的写法（够长的数、两个字以上的文字），正文、SQL 里的字面量、名字和注释、别的列里的格子、列名都按
    它换；weak 是遮罩列里的全部写法（含短的数），只用在 SQL 里和遮罩列做等值比较的字面量上；shown 是 SQL 里照原样
@@ -870,6 +875,8 @@ class _Secrets:
        written = {name: texts for name, texts in (labels or {}).items() if name not in own}
        columns = [str(c) for c in snapshot.get("columns") or []]
        index = [i for i, c in enumerate(columns) if c.lower() in hidden]
+        # 核查 agent 的查询快照在工具边界就遮好了（mask_query_result）：遮罩列的格子是「已遮罩」，不是遮罩值
+        word = _masked_word()
        for row in snapshot["rows"]:
            if len(self.weak) >= SECRETS_MAX:
                break
@@ -880,7 +887,7 @@ class _Secrets:
            else:
                continue
            for name, value in cells:
-                if value is None or isinstance(value, bool):
+                if value is None or isinstance(value, bool) or value == word:
                    continue
                forms = _forms(value)
                if forms & written.get(name, set()):
@@ -891,6 +898,18 @@ class _Secrets:
                        self.strong.add(form)
        self._index = None

+    def merge(self, other: _Secrets | None) -> None:
+        """并进另一份收好的遮罩值（核查 agent 查询时在工具边界遮掉的，见 mask_query_result）。合起来最多 SECRETS_MAX 种。"""
+        if other is None or not other.weak:
+            return
+        for form in other.weak:
+            if len(self.weak) >= SECRETS_MAX:
+                break
+            self.weak.add(form)
+            if form in other.strong:
+                self.strong.add(form)
+        self._index, self._compiled = None, {}
+
    def protect(self, texts: Iterable[str | None]) -> None:
        """已经摆在读者眼前的值不换：提问人自己写进问题、选中原文里的（「13900000001 这位客户」「排除 300 以下的」），
        报告里列出的指标、运行输入的值——换了只会让回答、查询对不上问题和报告。正文、格子、SQL 里都不换。"""
@@ -1039,8 +1058,8 @@ class _Secrets:
        return self._index


-def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, *, embedded: bool = False
-                 ) -> tuple[dict[str, Any], list[str]]:
+def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, *, embedded: bool = False,
+                 rename: bool = True) -> tuple[dict[str, Any], list[str]]:
    """一份查询快照遮好的样子和遮掉的列：要遮的列（hidden，小写）每一格都换成「已遮罩」、类型记成文本（渲染时不当数）；
    别的列里写着按值认得出的遮罩值的格子也换（另一张表里没设遮罩的同一个手机号、改了名字没认出来的遮罩列）。
    列名里写着遮罩值的（agent 把遮罩值写进了别名，「13900000001 的单」），列名也换（同 SQL 里的名字，_scrub_sql）。
@@ -1048,7 +1067,8 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets,

    embedded 只给质询工件用：别的列里一格文字夹着遮罩值（备注里写着手机号，换了写法的也算），把那一截换成「已遮罩」，
    这一格别的字照原样，门槛同整格（_Secrets.scrub）；这一列不记进遮掉的列——界面把记进去的列整列画成「已遮罩」，这一列在数据源
-    里也没设遮罩。回答一路不这样换：渲染出来露了就整份不交（_leaked），不然回答被问的那一格看着像查到了别的东西。"""
+    里也没设遮罩。回答一路不这样换：渲染出来露了就整份不交（_leaked），不然回答被问的那一格看着像查到了别的东西。
+    rename 为假时列名不换（工具边界先遮时：列名是模型自己写的，换了它引用的列名就对不上）。"""
    columns = [str(c) for c in snapshot.get("columns") or []]
    index = {i for i, c in enumerate(columns) if c.lower() in hidden}
    by_value = bool(secrets.strong)
@@ -1058,7 +1078,7 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets,
    touched = set(index)
    edited = False
    position = {c: i for i, c in enumerate(columns)}
-    names = [secrets.scrub(c) for c in columns] if by_value else columns
+    names = [secrets.scrub(c) for c in columns] if by_value and rename else columns

    def cell(i: int | None, value: Any) -> Any:
        nonlocal edited
@@ -1097,3 +1117,53 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets,
    types.update({names[i]: "text" for i in index})
    view = {**snapshot, "rows": rows, "column_types": types, **({"columns": names} if renamed else {})}
    return view, [names[i] for i in sorted(touched)]
+
+
+# --------------------------------------------------------------------------
+# 工具边界：查询结果交给核查 agent 之前先遮
+# --------------------------------------------------------------------------
+
+
+@dataclass
+class QueryMask:
+    """一份查询结果在工具边界遮好的样子（mask_query_result）。
+
+    payload：遮好的结果，交给模型的、落进快照的都是它。columns：查询时就整列遮掉的结果列（结果里的列名，按先后）。
+    secrets：这份结果里遮掉的值，只在内存里交给调用方（审批质询按值遮的第二道防线：模型照着别处看到的号码写进回答、
+    查询条件时认得出），不进快照、不交给模型。"""
+
+    payload: dict[str, Any]
+    columns: list[str]
+    secrets: _Secrets
+
+
+def mask_query_result(payload: dict[str, Any], masked: Iterable[str], kind: str | None) -> QueryMask:
+    """一份查询结果（tools/datasource 的查询工具交回的样子：columns / rows / column_types / sql）在交给核查 agent、落进
+    快照之前先遮。masked 是数据源设了遮罩的列，kind 是数据源的类型（SQL 按它的方言解析）。
+
+    - 整列遮：列名直接对上遮罩列的，加上按 SQL 认出值取自遮罩列的（_derived_hidden：别名、函数、子查询、WITH、
+      SELECT *；COUNT、只在条件里用到、分档当分组键的不算）；SQL 解析不了、又写着遮罩列的，整份结果的列都遮。这几列的
+      格子换成「已遮罩」，类型记成文本
+    - 别的列：整格就是这份结果里某个遮罩值的换掉，文字格里夹着的那一截换掉（换了写法的号码也认，_Secrets.loose_spans）；
+      门槛同按值遮（SECRET_DIGITS、SECRET_MIN），只装着分档标签的列里的标签不算遮罩值
+    - 列名不换（模型自己写的，换了它引用的列名就对不上）；结果里一列都没取自遮罩列的原样交回：数值的汇总、只数个数的
+      查询不受影响
+
+    分析本身出了意外，整份结果的列都遮：这一步挡在模型前面，分析的缺陷不能把原值放过去（宁可多遮）。"""
+    base = {str(c).strip().lower() for c in masked if str(c).strip()}
+    columns = [str(c) for c in payload.get("columns") or []]
+    if not base or not columns or not _query_shaped(payload):
+        return QueryMask(payload, [], _Secrets())
+    try:
+        hidden, _ = _derived_hidden(payload, base, kind)
+        if not any(c.lower() in hidden for c in columns):
+            return QueryMask(payload, [], _Secrets())
+        secrets = _Secrets()
+        secrets.add(payload, hidden, labels=_snapshot_labels(payload, kind), base=base)
+        secrets.prepare()
+        view, _ = _masked_view(payload, hidden, secrets, embedded=True, rename=False)
+    except Exception:  # noqa: BLE001 - 认不准就整份遮掉，原值不交给模型
+        logger.warning("核查查询的遮罩分析出错，整份结果按遮罩处理", exc_info=True)
+        hidden, secrets = {c.lower() for c in columns}, _Secrets()
+        view, _ = _masked_view(payload, hidden, secrets, rename=False)
+    return QueryMask(view, [c for c in columns if c.lower() in hidden], secrets)
@@ -6,6 +6,13 @@
 **只读。** 工具只建给定数据源的 db_schema / db_query，建的时候 force_readonly：可写源上的写语句也被守卫拒绝
 （tools/datasource.py）。这一路没有审批关卡，不能指望人在半路拦下一条 DELETE。

+**先遮再交。** 建查询工具时还打开遮罩开关（tools/datasource 的 mask_results）：数据源设了遮罩时，取自遮罩列的格子
+（按 SQL 认，起别名、套函数、经子查询或 WITH 转一道、SELECT * 都算）在交给模型、落进查询快照之前就换成「已遮罩」，
+同一份结果里别的文字格夹着的遮罩值也换掉（engine/masking.py 的 mask_query_result）。模型看不到遮罩值，就不会把它
+写进回答、理由、SQL；以前是事后按值认，每一轮复核都找到新的漏法。快照里的 masked_at_query 记下查询时就遮了的列，
+格子引用指到这几列的不采信。遮掉的值只在内存里交回调用方（VerifyOutcome.masked_values，审批质询按值遮的第二道防线
+用），不进快照、不交给模型、不进 as_dict。
+
 **数从快照取。** 每次成功的查询是一份查询快照（工具自己落的 put_json），按这个 agent 自己的顺序编 Q1、Q2…。
 最后交结果时，调用方结构里的数值字段一律改成格子引用 {alias, row, column}，由代码按引用从快照里取值：模型自己
 写的数字不采信，取不到就是 None，不兜底成 0。数值字段的各种写法都认（type 列表、anyOf 里的数、$ref 引用进来的、
@@ -45,7 +52,7 @@ import json
 import logging
 import time
 from collections.abc import Callable, Iterable
-from dataclasses import asdict, dataclass, field
+from dataclasses import asdict, dataclass, field, replace
 from enum import Enum
 from typing import Any

@@ -59,13 +66,14 @@ from app.engine import spend
 from app.engine.evidence import build_catalog, cell_eid, compose_doc, query_entry_fields, schema_ledger_entry
 from app.engine.expressions import CellError, column_kind, locate_cell, numeric_text
 from app.engine.judge import FALLBACK_MAX_TOKENS, MAX_TOKENS, _json_in, _tokens, _usd
+from app.engine.masking import _Secrets
 from app.engine.nodes.llm import _DIGEST_BUDGET, _digest, _split_structured, _typed, explain_model_error
 from app.engine.state import message_text
 from app.engine.toolcalls import ToolTimeout, limit_fields, limit_of, run_bounded
 from app.providers import catalog as pricing
 from app.providers.budget import BudgetFallback, budget_refused, cut_off, cut_words, error_usage, truncated_error
 from app.providers.factory import ModelSpec, ProviderNotConfigured, bind_tools_safely, get_chat_model
-from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX, build_datasource_tools
+from app.tools.datasource import MASKED_VALUES_KEY, QUERY_PREFIX, SCHEMA_PREFIX, build_datasource_tools
 from app.tools.registry import ToolArgsError, ToolContext, Versions, args_model_of, prepare_args

 logger = logging.getLogger(__name__)
@@ -103,6 +111,8 @@ VERIFY_SYSTEM = (
    "- 先用 db_schema 工具看表结构，确认表名、字段名，再用 db_query 工具写只读 SQL。不能改数据，写语句会被拒绝\n"
    "- 每次成功的查询前面都标着编号【证据 Qn】，最后交结果时用它说明每个数取自哪次查询的哪一格\n"
    "- {limit}先想清楚要查什么，不要重复同样的查询\n"
+    "- 数据源设了遮罩的列（手机号、姓名这类），查询结果里显示为「已遮罩」：不要设法绕过，交结果、写理由时也不要引用"
+    "这些格子\n"
    "- 不要编造数字，查不到就照实说"
 )
 VERIFY_FINAL_PROMPT = (
@@ -149,6 +159,8 @@ class VerifyOutcome:
    output_tokens, total_tokens, cost_usd, calls, unpriced_calls}。limit：触顶原因（LIMITS 之一），没触顶为 None。
    answered：模型是不是自己停下来的（没被上限掐断，也没因为输出到了上限被截断）。被截断的一律没有结果，error 是
    TRUNCATED_ERROR。
+    masked_values：查询时在工具边界遮掉的值（engine/masking 的 _Secrets）。只在内存里交给调用方按值遮（审批质询的回答），
+    不进 as_dict、不往外交。
    """

    result: dict[str, Any] | None
@@ -162,9 +174,12 @@ class VerifyOutcome:
    answered: bool = False
    notes: list[str] = field(default_factory=list)
    duration_ms: int = 0
+    masked_values: Any = field(default=None, repr=False, compare=False)

    def as_dict(self) -> dict[str, Any]:
-        return asdict(self)
+        out = asdict(replace(self, masked_values=None))
+        out.pop("masked_values", None)
+        return out


 # --------------------------------------------------------------------------
@@ -407,6 +422,12 @@ class _Cells:
            self.cells.append({"path": path, "status": "unresolved", "alias": alias, "artifact": query["artifact"],
                               "reason": f"{alias} {e}"})
            return None
+        if str(name).lower() in _masked_at_query(content):
+            # 查询时就遮了的列：格子里是「已遮罩」，不是数据源里的值。照实说是遮罩列，不说成「不是数值」
+            self.cells.append({"path": path, "status": "unresolved", "ref": f"{alias}.r{row}.{name}", "alias": alias,
+                               "row": row, "column": name, "artifact": query["artifact"],
+                               "reason": f"{alias} 的「{name}」取自设置了遮罩的列，查询时已遮罩，不采信"})
+            return None
        truth = _typed(raw, {"type": sorted(types)}, column_kind(content, name))
        cited = {"ref": f"{alias}.r{row}.{name}", "alias": alias, "row": row, "column": name,
                 "artifact": query["artifact"], "eid": cell_eid(query["artifact"], row, name), "value": truth}
@@ -502,6 +523,12 @@ class _Cells:
        return None


+def _masked_at_query(content: Any) -> set[str]:
+    """查询快照里查询时就整列遮掉的列（小写，tools/datasource 记下的 masked_at_query）。"""
+    listed = content.get("masked_at_query") if isinstance(content, dict) else None
+    return {c.lower() for c in listed if isinstance(c, str)} if isinstance(listed, list) else set()
+
+
 # --------------------------------------------------------------------------
 # 花费
 # --------------------------------------------------------------------------
@@ -636,6 +663,8 @@ class _State:
    queries: list[dict[str, Any]] = field(default_factory=list)
    ledger: list[dict[str, Any]] = field(default_factory=list)
    model: str | None = None
+    #: 查询工具在工具边界遮掉的值（经 ToolContext.extra 并进来，只在内存里）
+    seen: _Secrets = field(default_factory=_Secrets)

    def outcome(self, **fields: Any) -> VerifyOutcome:
        usage = self.usage
@@ -645,7 +674,8 @@ class _State:
                 for q in self.queries]
        base: dict[str, Any] = {"result": None, "cells": [], "queries": shown, "rationale": None, "usage": usage,
                                "notes": self.notes, "model": self.model,
-                                "duration_ms": int((time.monotonic() - self.started) * 1000)}
+                                "duration_ms": int((time.monotonic() - self.started) * 1000),
+                                "masked_values": self.seen}
        return VerifyOutcome(**{**base, **fields})


@@ -728,12 +758,13 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou
        notes.append(note)
        await _announce(emit, EventType.LOG, level="warn", code="verify_unpriced", message=note)

-    # ---- 工具：只读的 db_schema / db_query
+    # ---- 工具：只读的 db_schema / db_query，查询结果先遮再交给模型（遮掉的值并进 state.seen）
    wanted = list(dict.fromkeys(s for s in sources if isinstance(s, str) and s))
-    ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit)
+    ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit,
+                      extra={MASKED_VALUES_KEY: state.seen})
    async with SessionLocal() as session:
        tools = await build_datasource_tools([f"{p}{s}" for s in wanted for p in (SCHEMA_PREFIX, QUERY_PREFIX)], ctx,
-                                             session, force_readonly=True)
+                                             session, force_readonly=True, mask_results=True)
    tool_map = {t.name: t for t in tools}
    if lost := [s for s in wanted if f"{QUERY_PREFIX}{s}" not in tool_map]:
        notes.append("数据源" + "、".join(f"「{s}」" for s in lost) + "不存在或已停用，核查没有查它")
@@ -53,6 +53,9 @@ RUN_VERSIONS_KEY = "data_versions"
 DATA_UNAVAILABLE = "查询失败（数据不可用）："
 #: 运行中途补固定一个源时发的那条日志事件的 code（_pin_late）
 LATE_PIN_CODE = "data_version_pinned"
+#: ToolContext.extra 里放「查询时遮掉的值」的键（engine/masking 的 _Secrets）：核查 agent（engine/verify_agent.py）放进去，
+#: 打开了遮罩开关的查询工具把每次遮掉的值并进去。只在内存里给调用方按值遮用，不进快照、不交给模型
+MASKED_VALUES_KEY = "masked_values"

 #: 每次查询之后基于数据目录的 SQL 检查最多等这么久。正常一条 SQL 几毫秒；等不到就放弃这一次检查，
 #: 查询结果照常交回——检查是附加的，不能拖慢查询
@@ -199,7 +202,8 @@ def _query_description(source: Any, *, readonly: bool | None = None) -> str:


 async def build_datasource_tools(
-    names: list[str], ctx: ToolContext, session: AsyncSession, *, force_readonly: bool = False
+    names: list[str], ctx: ToolContext, session: AsyncSession, *, force_readonly: bool = False,
+    mask_results: bool = False,
 ) -> list[StructuredTool]:
    """按工具名反查数据源并构造工具。build_tools 的第四个来源。

@@ -209,6 +213,9 @@ async def build_datasource_tools(

    force_readonly：查询工具一律按只读守卫检查，可写源上的写语句也拒绝（见 _make_query_tool）。
    核查（engine/verify_agent.py）建工具时传它：那一路没有审批关卡，绝不能写。
+
+    mask_results：查询结果交给模型、落进快照之前先遮（见 _make_query_tool）。核查建工具时也传它：模型看不到遮罩列的
+    值，就不会把它们写进回答、理由。缺省关着：Agent 节点、调用工具节点的查询照旧原样。
    """
    wanted: dict[str, list[str]] = {}
    for name in names:
@@ -238,7 +245,8 @@ async def build_datasource_tools(
        entries = await _catalog_of(row)
        for prefix in prefixes:
            if prefix == QUERY_PREFIX:
-                tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries, force_readonly=force_readonly))
+                tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries, force_readonly=force_readonly,
+                                              mask_results=mask_results))
            else:
                tools.append(_make_schema_tool(source, fixed=fixed, catalog=entries))
    return tools
@@ -450,7 +458,7 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo

 def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
                     catalog: Mapping[str, data_catalog.CatalogEntry] | None = None,
-                     force_readonly: bool = False) -> StructuredTool:
+                     force_readonly: bool = False, mask_results: bool = False) -> StructuredTool:
    """source 是手工源的 DataSource，或上传源绑定快照的 SourceView（字段同名）。

    fixed：上传源的版本是这次运行固定的（找不到时的说法不同）。
@@ -458,6 +466,8 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
    force_readonly：不管源可不可写，一律按只读守卫检查，写语句交回 FORCED_READONLY、不执行。只靠守卫这一道：
    连接池按源缓存、不分只读与否（engine.cache_key），可写源借不到连接层只读的连接——别为此另起一个只读池，
    缓存键相同，会把这个源平时用的可写连接也换成只读的。
+    mask_results：数据源设了遮罩时，取自遮罩列的格子（按 SQL 认，改了名字的也算）在交给模型、落进快照之前就换成
+    「已遮罩」，见 engine/masking.py 的 mask_query_result；这一次遮掉的值并进 ctx.extra[MASKED_VALUES_KEY]（有的话）。
    """
    # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句，
    # 引擎按它告诉界面上限是多少（metadata["timeout_s"]）
@@ -505,6 +515,17 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,

        payload = result.to_payload()
        payload["source"] = source.name
+        mask = masked_columns(source.options)
+        # 查询时就整列遮掉的结果列；没开遮罩开关、数据源没设遮罩时为 None（快照和原来一字不差）
+        masked_now: list[str] | None = None
+        if mask_results and mask:
+            from app.engine.masking import mask_query_result
+
+            # 按 SQL 认列、逐格比对是纯计算，放进线程（同 _sql_checks）
+            done = await asyncio.to_thread(mask_query_result, payload, mask, source.kind)
+            payload, masked_now = done.payload, done.columns
+            if (sink := ctx.extra.get(MASKED_VALUES_KEY)) is not None:
+                sink.merge(done.secrets)
        # 对照数据目录检查这条 SQL（一对多关联后重复计算、存量跨期求和……）。结果交给模型（Agent 据此改写），
        # 也随查询快照存下：证据面板看得到，口径卡据此给指标标「存疑」、出具降档。没查出问题不加这个键，
        # 快照和以前一字不差
@@ -518,14 +539,19 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
        try:
            from app.core.artifact_store import put_json

-            # 只进快照、不进交给模型的结果的两样（模型看到多余的字段只会误会：以为数据被遮了，
+            # 只进快照、不进交给模型的结果的几样（模型看到多余的字段只会误会：以为数据被遮了，
            # 或者把一串十六进制抄进报告）：
-            # - 查询当时数据源设的遮罩列：数据源事后改名、删掉，证据面板仍按当时的遮；
+            # - 查询当时数据源设的遮罩列：数据源事后改名、删掉，证据面板仍按当时的遮。查询时就遮了的结果列（改了名字的
+            #   也算）一并记进去，证据面板、质询工件按这个键把它们画成遮罩列；
+            # - masked_at_query：开了遮罩开关时，查询时就整列遮掉的结果列（这几列的格子是「已遮罩」，不是数据源里的值）；
            # - data_version：上传表格查的是哪个快照，事后据此认得出这个数出自哪一版数据。
            # 手工源、没设遮罩的，快照和原来一字不差
            extra: dict[str, Any] = {}
-            if mask := masked_columns(source.options):
-                extra["mask_columns"] = mask
+            if mask:
+                own = {m.lower() for m in mask}
+                extra["mask_columns"] = [*mask, *(c for c in masked_now or [] if c.lower() not in own)]
+            if masked_now is not None:
+                extra["masked_at_query"] = masked_now
            if version := getattr(source, "snapshot_id", None):
                extra["data_version"] = version
            # 工件引用的 meta 记下源和 SQL 里用到的表：数据目录按它数每张表被运行查询过几次
@@ -1102,6 +1102,32 @@ async def test_the_task_gives_the_claim_its_evidence_and_only_its_sources(monkey
        await engines.invalidate(source_id)


+async def test_the_challenge_agent_never_sees_masked_values(monkeypatch, tmp_path):
+    """反查 agent 自己的查询选出了遮罩列（起了别名）：交给它的工具结果、它的查询快照里都是「已遮罩」，理由里引用到那一格
+    渲染出来也是「已遮罩」；审稿记录、运行事件里都没有原值。以前查询结果原样交给模型、原样存下，理由也原样渲染。"""
+    name, source_id = await _source(tmp_path, options={"mask_columns": ["buyer_phone"]})
+    try:
+        top = "SELECT buyer_phone AS 电话, amount AS 金额 FROM sales WHERE region = '华东' ORDER BY amount DESC"
+        plays = {"居各区域之首": Play([top], "stands", "华东最大一笔来自 [[v:Q1.r0.电话]]，金额 [[v:Q1.r0.金额]]。[[see:Q1]]")}
+        models = Models(monkeypatch, name, plays=plays)
+        row = await finish(graph(name), run_class="exploratory")
+        assert row.status == "succeeded", row.error
+        seen = [str(m.content) for _, key, messages in models.calls if key == "居各区域之首"
+                for m in messages if isinstance(m, ToolMessage)]
+        assert seen and any("已遮罩" in s for s in seen) and all("1390000000" not in s for s in seen)
+        record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"])
+        [claim] = [c for c in record["claims"] if "居各区域之首" in c["text"]]
+        assert claim["verdict"] == "stands" and claim["rationale"] == "华东最大一笔来自 已遮罩，金额 1,200。"
+        snapshot = artifact_store.load(claim["queries"][0]["artifact"])
+        assert {r[0] for r in snapshot["rows"]} == {"已遮罩"} and snapshot["masked_at_query"] == ["电话"]
+        assert "1390000000" not in json.dumps(record, ensure_ascii=False)
+        async with SessionLocal() as session:
+            evs = (await session.execute(select(RunEvent).where(RunEvent.run_id == row.id))).scalars().all()
+        assert evs and all("1390000000" not in json.dumps(e.data, ensure_ascii=False) for e in evs)
+    finally:
+        await engines.invalidate(source_id)
+
+
 async def test_a_metric_excerpt_marks_the_report_query_numbers_too(monkeypatch, shop):
    """句子挂的是口径卡指标（正式报告的主路径「取数 → 口径卡 → 报告」）：摘录正文里「取自查询 Q1」「查询 Q1 的 SQL」
    也标成原报告的编号，和反查 agent 自己的 Q1 分开；工具按指标的输入取自的查询找源。"""
@@ -1034,6 +1034,27 @@ async def test_a_blind_cell_on_a_masked_column_is_not_used(monkeypatch, masked_s
    assert "30.25" not in json.dumps(out, ensure_ascii=False)


+async def test_a_renamed_masked_column_in_the_blind_query_is_masked_before_the_blind_agent_sees_it(
+        monkeypatch, masked_shop):
+    """盲算的查询选出了改了名字的遮罩列（region AS 区域）：交给盲算模型、落进盲算快照之前就是「已遮罩」，快照记下查询
+    时遮了哪一列；没碰遮罩列的那一列（销售额）照常取值、照常比对。"""
+    name, _ = masked_shop
+    east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
+    renamed = (f"SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} "
+               "GROUP BY region ORDER BY region")
+    Blind(monkeypatch, name, {"华东销售额": (renamed, "gmv")})
+    run = await start(graph(name, BY_REGION, metrics=[east], dual={"definitions": {}}))
+    assert run.status == "succeeded", run.error
+    evs = await events(run.id)
+    out = dual_output(run, evs)
+    assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"
+    [end] = [e.data for e in evs if e.type == "tool.end" and e.node_id == "dual" and e.data.get("query_artifact")]
+    snapshot = artifact_store.load(end["query_artifact"])
+    assert {row[0] for row in snapshot["rows"]} == {"已遮罩"} and snapshot["masked_at_query"] == ["区域"]
+    shown = json.loads(end["preview"].split("\n", 1)[1])
+    assert {row[0] for row in shown["rows"]} == {"已遮罩"}, "交给盲算模型的结果里也是「已遮罩」"
+
+
 async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch, masked_shop):
    name, _ = masked_shop
    first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
@@ -904,6 +904,35 @@ async def test_resuming_after_an_inquiry_does_not_repeat_the_nodes(monkeypatch,
 ALIASED = "SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY amount DESC LIMIT 1"


+async def test_the_inquiry_agent_sees_and_stores_its_queries_masked(monkeypatch, masked_shop, client):
+    """质询 agent 的查询结果在交给模型之前就遮好了（核查 agent 建查询工具时打开遮罩开关）：模型看到的工具结果、落库的
+    查询快照、运行详情「工件」页签走的通用工件接口里，遮罩列（起了别名的也算）都是「已遮罩」。以前快照原样存着，模型
+    看得到原值，通用接口也原样给出。回答照常按值遮（查询时遮掉的值只在内存里交回来）。"""
+    plays = {"买得最多": Play([ALIASED], "这一笔来自 [[v:Q1.r0.手机号]]（13900000001），金额 [[v:Q1.r0.金额]]。[[see:Q1]]")}
+    Models(monkeypatch, masked_shop, plays=plays)
+    seen: list[str] = []
+    answering = MockChatModel._decide
+
+    def spy(model, messages):
+        seen.extend(str(m.content) for m in messages if isinstance(m, ToolMessage))
+        return answering(model, messages)
+
+    monkeypatch.setattr(MockChatModel, "_decide", spy)
+    paused = await pause(masked_shop)
+    [approval] = await pending(paused.id)
+    r = await ask(client, paused.id, report="write", unit=None, question="买得最多的是哪位客户？",
+                  approval_id=approval.id)
+    assert r.status_code == 200, r.text
+    assert r.json()["inquiry"]["answer"]["text"] == "这一笔来自 已遮罩（已遮罩），金额 1,200。"
+    [query] = r.json()["inquiry"]["queries"]
+    assert seen and any("已遮罩" in s for s in seen) and all("1390000000" not in s for s in seen)
+    got = await client.get(f"/api/artifacts/{query['artifact']}")
+    assert got.status_code == 200 and "1390000000" not in got.text
+    content = got.json()["content"]
+    assert content["rows"] == [["已遮罩", 1200]] and content["masked_at_query"] == ["手机号"]
+    assert content == artifact_store.load(query["artifact"])
+
+
 async def test_a_masked_column_under_another_name_is_masked_everywhere(monkeypatch, masked_shop, client):
    """遮罩列起了中文别名（本仓库查询的惯例）：回答里引用到的格子、事件、回答文档、质询工件里都不露出原值。
    报告自己的查询没选手机号，原值哪里都认不出来，只能按 SQL 认出「手机号」就是 buyer_phone。"""
@@ -1197,6 +1197,27 @@ def test_sql_check_excerpts_in_the_artifact_are_masked_like_the_sql():
         "sql_excerpt": "SUM(balance)"}]


+def test_check_excerpts_follow_the_sql_recorded_at_answer_time():
+    """agent 自己的查询快照查询时就遮好了，事后取质询工件时收不到遮罩列里的值：片段、说明照事件里记下的那条 SQL（回答时
+    按内存里的遮罩值换过）换——那条里整个换成「已遮罩」的字面量、换掉的那一截（LIKE 的 %…%），片段、说明里一样换。"""
+    sql = ("SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000001' AND age = 25 "
+           "AND note LIKE '%13900000002%'")
+    recorded = "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩' AND age = 已遮罩 AND note LIKE '%已遮罩%'"
+    assert inquiry_mod._recorded_swaps(sql, recorded, "sqlite") == ({"13900000001", "25"}, {"13900000002"})
+    assert inquiry_mod._recorded_swaps(sql, sql, "sqlite") == (set(), set())
+    assert inquiry_mod._recorded_swaps(sql, recorded + " LIMIT 5", "sqlite") == (set(), set()), "对不上就不认"
+    checks = [{"code": "c1", "level": "warning", "table": "sales", "message": "没有码值「25」",
+               "sql_excerpt": "buyer_phone = '13900000001' AND age = 25"},
+              {"code": "c2", "level": "info", "table": "sales", "message": "按备注筛选",
+               "sql_excerpt": "note LIKE '%13900000002%'"}]
+    whole, pieces = inquiry_mod._recorded_swaps(sql, recorded, "sqlite")
+    assert inquiry_mod._masked_checks(checks, sql, inquiry_mod._Secrets(), set(), "sqlite", swapped=whole,
+                                      pieces=pieces) == [
+        {"code": "c1", "level": "warning", "table": "sales", "message": "没有码值「已遮罩」",
+         "sql_excerpt": "buyer_phone = '已遮罩' AND age = 已遮罩"},
+        {"code": "c2", "level": "info", "table": "sales", "message": "按备注筛选", "sql_excerpt": "note LIKE '%已遮罩%'"}]
+
+
@pytest.mark.parametrize("sql, excerpt, shown", [
    # 片段只是一段表达式，单独拿出来认不出在找哪一个值：按整条 SQL 里换掉的字面量换
    ("SELECT SUM(CASE age WHEN 25 THEN balance END) AS 余额 FROM accounts",
@@ -59,6 +59,17 @@ def test_a_long_number_in_another_shape_is_masked_in_prose(stored, written):
    assert secrets.scrub(f"客户{written}来电", prose=True) == f"客户{MASKED}来电"


+@pytest.mark.parametrize("text, shown", [
+    ("来自（19914742785），共 3 笔", f"来自（{MASKED}），共 3 笔"),
+    ("来自(+86 199 1474 2785)，共 3 笔", f"来自({MASKED})，共 3 笔"),
+    ("电话（+86）199 1474 2785。", f"电话{MASKED}。"),
+    ("（电话 199-1474-2785）", f"（电话 {MASKED}）"),
+])
+def test_brackets_around_a_number_are_kept_and_brackets_inside_it_are_replaced(text, shown):
+    """整个号码外面套着的一对括号是标点，留着；号码里的括号（区号、国家码）和号码一起换掉。"""
+    assert secrets_of([PHONE]).scrub(text, prose=True) == shown
+
+
@pytest.mark.parametrize("text", [
    # 日期、时间、小数、千分位、更长的数里的一截、连字符连着的编号、别的号码
    "2026-10-06 19:14 下单", "金额 19914.74 元", "共 1,991,474 笔", "号码 1991474278512", "批次 19914742785-2",
@@ -0,0 +1,248 @@
+"""核查 agent 看到的查询结果先遮、存下的快照也遮（engine/verify_agent.py 建查询工具时打开 tools/datasource 的遮罩开关）。
+
+双盲复算、反方审稿、审批质询共用的核查 agent 以前把查询结果原样交给模型，遮罩是事后按值做的（审批质询的 _Secrets），
+每一轮复核都找到新的漏法。现在在工具边界先遮：数据源设了遮罩时，按 SQL 认出取自遮罩列的结果列（别名、函数、子查询、
+WITH、SELECT *），连同列名直接对上的，这些格子在交给模型的结果、落库的快照里都是「已遮罩」，快照记下哪些列在查询时
+就遮了；别的文字格里夹着这份结果里遮罩值的那一截同样换掉（换了写法的号码也认）。运行详情「工件」页签走的通用接口
+GET /api/artifacts/{id} 给的也就是遮好的。没碰遮罩列的聚合查询原样。遮掉的值只在内存里交回调用方（审批质询按值遮的
+第二道防线用），不进快照、不交给模型。遮罩开关缺省关着，别的调用方（Agent 节点、调用工具节点）照旧。
+
+模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，手机号、姓名、金额都是虚构的。
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import delete
+
+from app.core import artifact_store
+from app.core.events import EventType
+from app.data import introspect
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Setting
+from app.engine import masking, spend
+from app.engine.verify_agent import run_verify_agent
+from app.main import app
+from app.tools.datasource import build_datasource_tools
+from app.tools.registry import ToolContext, Versions
+from tests.test_verify_agent import SPEC, Script, q
+
+MASKED = "已遮罩"
+#: 遮罩列里的值（和备注里换了写法的号码）：交给模型的、存下的、接口给的都不能出现
+SECRETS = ("1390000000", "张伟", "李娜", "王芳", "刘洋", "139-0000-0003")
+SALES = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, buyer_name TEXT, amount INTEGER,"
+         " note TEXT);"
+         "INSERT INTO sales VALUES (1, '华东', '13900000001', '张伟', 1200, '客户13900000001来电改址'),"
+         " (2, '华东', '13900000002', '李娜', 300, '例行回访'), (3, '华南', '13900000003', '王芳', 900, '回访：联系 139-0000-0003'),"
+         " (4, '华北', '13900000004', '刘洋', 400, '');")
+VALUE_SCHEMA = {"type": "object", "properties": {"value": {"type": "number", "description": "要核对的数"}},
+                "required": ["value"]}
+
+
+@pytest.fixture(autouse=True)
+async def clean_spend():
+    async with SessionLocal() as session:
+        await session.execute(delete(Setting).where(Setting.key.in_([spend.SPEND_KEY, "verify"])))
+        await session.commit()
+    yield
+
+
+async def _source(tmp_path, options: dict | None) -> tuple[str, int]:
+    path = tmp_path / f"{uuid.uuid4().hex[:8]}.db"
+    db = sqlite3.connect(path)
+    db.executescript(SALES)
+    db.commit()
+    db.close()
+    name = f"shop_{uuid.uuid4().hex[:8]}"
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, options=options or {})
+        session.add(row)
+        await session.flush()
+        row.schema_cache = await introspect.introspect(row)
+        await session.commit()
+        return name, row.id
+
+
+@pytest.fixture
+async def masked_shop(tmp_path):
+    """四笔订单，买家手机号、姓名两列设了遮罩；备注没设遮罩，其中两条写着手机号（一条换了写法）。返回源名。"""
+    name, source_id = await _source(tmp_path, {"mask_columns": "buyer_phone, buyer_name"})
+    yield name
+    await engines.invalidate(source_id)
+
+
+@pytest.fixture
+async def plain_shop(tmp_path):
+    name, source_id = await _source(tmp_path, None)
+    yield name
+    await engines.invalidate(source_id)
+
+
+async def ask(name: str, **kwargs):
+    """按剧本跑一次核查 agent（剧本由各个测试先用 Script 装好），交回结果和发出的事件。"""
+    events: list[tuple] = []
+    out = await run_verify_agent(spec=SPEC, system="核对一个数。", task="查一下。", sources=[name],
+                                 output_schema=VALUE_SCHEMA, data_versions=Versions.CURRENT, run_id="run-mask",
+                                 node_id="check", emit=lambda kind, **data: events.append((kind, data)), **kwargs)
+    return out, events
+
+
+def _clean(text: str) -> bool:
+    return not any(s in text for s in SECRETS)
+
+
+def _cells(snapshot: dict, column: str) -> list:
+    i = snapshot["columns"].index(column)
+    return [row[i] for row in snapshot["rows"]]
+
+
+@pytest.mark.parametrize("sql, masked", [
+    pytest.param("SELECT buyer_phone, amount FROM sales ORDER BY id", ["buyer_phone"], id="直接选出"),
+    pytest.param("SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY id", ["手机号"], id="起别名"),
+    pytest.param("SELECT UPPER(buyer_name) AS 姓名, amount FROM sales ORDER BY id", ["姓名"], id="套函数"),
+    pytest.param("SELECT t.p AS 号码, t.amount FROM (SELECT id, buyer_phone AS p, amount FROM sales) t ORDER BY t.id",
+                 ["号码"], id="子查询"),
+    pytest.param("WITH t AS (SELECT id, buyer_phone AS p, amount FROM sales) SELECT p, amount FROM t ORDER BY id",
+                 ["p"], id="WITH"),
+    pytest.param("SELECT * FROM sales ORDER BY id", ["buyer_phone", "buyer_name"], id="SELECT *"),
+])
+async def test_masked_columns_are_masked_before_the_model_sees_them(monkeypatch, masked_shop, sql, masked):
+    name = masked_shop
+    amount = "amount" if "金额" not in sql else "金额"
+    script = Script(monkeypatch, [q(name, sql, "c1")],
+                    {"value": {"alias": "Q1", "row": 0, "column": amount}, "rationale": "见 [[see:Q1]]。"})
+    out, events = await ask(name)
+    assert out.error is None and out.result["value"] == 1200, "没遮的列照常取值"
+
+    # 交给模型的工具结果：遮罩列是「已遮罩」，别的列照原样
+    seen = script.tool_results()["c1"]
+    assert seen.startswith("【证据 Q1】") and MASKED in seen and _clean(seen), seen
+    payload = json.loads(seen.split("\n", 1)[1])
+    for column in masked:
+        assert set(_cells(payload, column)) == {MASKED}
+    assert _cells(payload, amount) == [1200, 300, 900, 400]
+    assert "mask_columns" not in payload and "masked_at_query" not in payload, "交给模型的结果不多出字段"
+
+    # 落库的快照：同样遮好，记下查询时就遮了的列；数据源设的遮罩列照记，换了名字的结果列也记进去（证据面板据此画成遮罩）
+    [query] = out.queries
+    snapshot = artifact_store.load(query["artifact"])
+    assert _clean(json.dumps(snapshot, ensure_ascii=False))
+    for column in masked:
+        assert set(_cells(snapshot, column)) == {MASKED}
+        assert snapshot.get("column_types", {}).get(column, "text") == "text"
+    assert _cells(snapshot, amount) == [1200, 300, 900, 400]
+    assert snapshot["masked_at_query"] == masked
+    assert {c.lower() for c in snapshot["mask_columns"]} == {"buyer_phone", "buyer_name", *(c.lower() for c in masked)}
+
+    # 运行详情「工件」页签走的通用接口：给的就是遮好的
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        got = await client.get(f"/api/artifacts/{query['artifact']}")
+    assert got.status_code == 200 and got.json()["content"] == snapshot and _clean(got.text)
+
+    # 事件（tool.end 的预览）、收尾时交给模型的查询结果里也没有
+    [end] = [d for k, d in events if k == EventType.TOOL_END]
+    assert MASKED in end["preview"] and _clean(end["preview"])
+    final = "\n".join(str(m.content) for m in script.calls[-1][1])
+    assert MASKED in final and _clean(final)
+
+    # 遮掉的值只在内存里交回调用方，不跟着结果往外交
+    assert out.masked_values is not None and out.masked_values.strong
+    assert "masked_values" not in out.as_dict()
+
+
+@pytest.mark.parametrize("sql, value", [
+    ("SELECT SUM(amount) AS total FROM sales", 2800),
+    ("SELECT COUNT(DISTINCT buyer_phone) AS total FROM sales", 4),
+    ("SELECT SUM(amount) AS total FROM sales WHERE buyer_phone = '13900000001'", 1200),
+    ("SELECT region, SUM(amount) AS total FROM sales GROUP BY region ORDER BY total DESC", 1500),
+    ("SELECT region, note, amount AS total FROM sales ORDER BY id", 1200),
+])
+async def test_queries_that_do_not_read_masked_values_are_untouched(monkeypatch, masked_shop, sql, value):
+    """没碰遮罩列的值（汇总、只数个数、只在条件里用到）：结果原样，取值照常，快照记下查询时一列都没遮。只遮这份结果
+    里的遮罩值：备注里写着手机号、这份结果却没有选出手机号的，按值认不出（那是数据源的遮罩没设全）。"""
+    name = masked_shop
+    script = Script(monkeypatch, [q(name, sql, "c1")],
+                    {"value": {"alias": "Q1", "row": 0, "column": "total"}, "rationale": "见 [[see:Q1]]。"})
+    out, _ = await ask(name)
+    assert out.error is None and out.result["value"] == value
+    seen = script.tool_results()["c1"]
+    assert MASKED not in seen
+    snapshot = artifact_store.load(out.queries[0]["artifact"])
+    assert snapshot["masked_at_query"] == [] and MASKED not in json.dumps(snapshot, ensure_ascii=False)
+    assert [c.lower() for c in snapshot["mask_columns"]] == ["buyer_phone", "buyer_name"]
+
+
+async def test_masked_values_inside_other_text_cells_are_cut_out(monkeypatch, masked_shop):
+    """同一份结果里选出了手机号：备注里夹着的同一个号码（换了写法的也算）换成「已遮罩」，备注别的字照原样。"""
+    name = masked_shop
+    sql = "SELECT buyer_phone, note, amount AS total FROM sales ORDER BY id"
+    script = Script(monkeypatch, [q(name, sql, "c1")],
+                    {"value": {"alias": "Q1", "row": 0, "column": "total"}, "rationale": "见 [[see:Q1]]。"})
+    out, _ = await ask(name)
+    seen = script.tool_results()["c1"]
+    assert _clean(seen)
+    snapshot = artifact_store.load(out.queries[0]["artifact"])
+    assert _cells(snapshot, "note") == [f"客户{MASKED}来电改址", "例行回访", f"回访：联系 {MASKED}", ""]
+    assert snapshot["masked_at_query"] == ["buyer_phone"], "备注只换了一截，不算整列遮"
+
+
+async def test_a_cell_reference_into_a_masked_column_is_not_taken(monkeypatch, masked_shop):
+    """模型交的格子引用指到查询时就遮了的那一列：不采信，照实说是遮罩列，不说成「不是数值」。"""
+    name = masked_shop
+    sql = "SELECT buyer_phone AS 手机号, amount FROM sales ORDER BY id"
+    Script(monkeypatch, [q(name, sql, "c1")], {"value": {"alias": "Q1", "row": 0, "column": "手机号"}, "rationale": "x"})
+    out, _ = await ask(name)
+    assert out.result["value"] is None
+    [cell] = out.cells
+    assert cell["status"] == "unresolved" and "1390000000" not in json.dumps(cell)
+    assert cell["reason"] == "Q1 的「手机号」取自设置了遮罩的列，查询时已遮罩，不采信"
+
+
+async def test_sql_that_cannot_be_read_but_names_a_masked_column_masks_the_whole_result(monkeypatch, masked_shop):
+    """SQL 认不出哪一列取自遮罩列、里面又写着遮罩列：整份结果的列都遮（和双盲复算、审批质询同一个保守规则）。"""
+    name = masked_shop
+    monkeypatch.setattr(masking, "_tainted", lambda *args, **kwargs: None)
+    sql = "SELECT buyer_phone, amount FROM sales ORDER BY id"
+    Script(monkeypatch, [q(name, sql, "c1")], {"value": {"alias": "Q1", "row": 0, "column": "amount"}, "rationale": "x"})
+    out, _ = await ask(name)
+    snapshot = artifact_store.load(out.queries[0]["artifact"])
+    assert snapshot["masked_at_query"] == ["buyer_phone", "amount"]
+    assert set(_cells(snapshot, "amount")) == {MASKED} and out.result["value"] is None
+
+
+async def test_the_switch_is_off_by_default_and_a_source_without_masks_is_untouched(masked_shop, plain_shop):
+    """遮罩开关缺省关着：Agent 节点、调用工具节点建的查询工具照旧交回原样的结果。没设遮罩的数据源开着开关也一字不差
+    （快照里不多出键）。"""
+    sql = "SELECT buyer_phone, amount FROM sales ORDER BY id"
+    ctx = ToolContext(run_id="run-switch", node_id="n1", data_versions=Versions.CURRENT)
+    async with SessionLocal() as session:
+        [plain_tool] = await build_datasource_tools([f"db_query__{masked_shop}"], ctx, session)
+        [masking_tool] = await build_datasource_tools([f"db_query__{masked_shop}"], ctx, session, mask_results=True)
+        [unmasked_tool] = await build_datasource_tools([f"db_query__{plain_shop}"], ctx, session, mask_results=True)
+    raw = json.loads(await plain_tool.ainvoke({"sql": sql}))
+    assert _cells(raw, "buyer_phone")[0] == "13900000001"
+    stored = artifact_store.load(raw["artifact"])
+    assert _cells(stored, "buyer_phone")[0] == "13900000001" and "masked_at_query" not in stored
+    shown = json.loads(await masking_tool.ainvoke({"sql": sql}))
+    assert set(_cells(shown, "buyer_phone")) == {MASKED}
+    untouched = json.loads(await unmasked_tool.ainvoke({"sql": sql}))
+    stored = artifact_store.load(untouched["artifact"])
+    assert _cells(stored, "buyer_phone")[0] == "13900000001"
+    assert "masked_at_query" not in stored and "mask_columns" not in stored
+
+
+async def test_column_names_the_model_wrote_are_kept(monkeypatch, masked_shop):
+    """列名是模型自己写在 SQL 里的（哪怕写着一个号码）：工具边界只遮格子、不改列名，它引用的列名才对得上。"""
+    name = masked_shop
+    sql = 'SELECT buyer_phone, amount AS "13900000001 的金额" FROM sales ORDER BY id'
+    Script(monkeypatch, [q(name, sql, "c1")],
+           {"value": {"alias": "Q1", "row": 0, "column": "13900000001 的金额"}, "rationale": "x"})
+    out, _ = await ask(name)
+    snapshot = artifact_store.load(out.queries[0]["artifact"])
+    assert snapshot["columns"] == ["buyer_phone", "13900000001 的金额"]
+    assert set(_cells(snapshot, "buyer_phone")) == {MASKED} and out.result["value"] == 1200