审批质询复核修正（第 3 轮，后端）：SQL 里的字面量按语法树认用在哪，数字照原样留着的只有没夹住某一个值的范围边界和 LIMIT 行数；SQL 里留着的阈值只在回答正文里不遮，格子照旧按值遮；质询工件的免遮接线补上端到端核对

复核提出的问题和改法：
- 第 2 轮为了不误遮阈值，把范围边界、字符串的范围边界和选出列、函数参数里的数从正文和格子的按值遮里一并免掉；而按词认用法时，
  简单 CASE 的 WHEN、IN (VALUES …)、行值、= ANY(ARRAY[…])、IS NOT DISTINCT FROM、套了括号的值这几种在找某一个值的写法被当成「别处」，
  两头相同的范围（BETWEEN v AND v、>= v AND <= v）被当成阈值，遮罩值就留在封存的 SQL、回答正文和质询工件里，最后一道关也放行：
  - 字面量改按语法树认用在哪：越过括号、负号、CAST、行值、VALUES、ARRAY、ANY / ALL 这些外壳再看上一层，上面几种都认作在找某一个值，
    被比较的是哪一列也认得出（短的数照旧只在和遮罩列比较时换）
  - 同一列上同一个值两边都被隔开的范围按在找这个值换：BETWEEN v AND v、>= v AND <= v、> v OR < v、整数的 >= v AND < v + 1，一头是几个数
    算出来的（13900000002 - 1）也算；只从一边隔开的（>= v，或按 v 分成两段）照旧是阈值
  - 数字照原样留着的只剩没夹住某一个值的范围边界和 LIMIT / OFFSET 的行数，而且同一个数在这条 SQL 别的字面量里没被换掉；选出列、函数参数、
    算式里够长的遮罩值照第 1 轮换；字符串的范围边界照第 1 轮按值换。名字、注释里写着阈值不牵连阈值本身（AS 超过30000的笔数）
  - SQL 里照原样留着的数只在回答正文里不遮，格子、列名、SQL 里照旧按值遮；同一个数在别的查询里被换掉了，正文里也遮。组装后的最后一道关
    只放过正文里的这些数，渲染进单元格引用里的照旧拦下
  - SQL 解析不了时认不准哪个数是阈值，一律不免，只按词认简单的等值比较
- 质询工件里两处免遮接线没有测试守住：报告指标、运行输入的值那一处，单笔目标那条端到端测试补上取质询工件，核对别的列里正好是这个数的
  格子在回答、工件里都照常显示；SQL 里阈值那一处随上面的改法拿掉了（格子不再免阈值），补上「格子正好等于阈值」的端到端测试，核对回答和
  工件里这一格一样遮住、两边对得上；阈值那条端到端测试补上取等值查询的质询工件，核对 SQL 和事件里一字不差
- 上面每种写法都补了单元测试和端到端测试（手机号存成整数、文字两种），另补开头带换行的 SQL 位置对得上、名字里的阈值不牵连阈值的测试。
  查询清单的 SQL 现在要解析，改在线程里换（同 _plan）。每处改动都在副本上做过去掉实现、测试变红的验证

backend/app/engine/inquiry.py

@@ -15,9 +15,10 @@
 3. 遮罩（_plan）：哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列
    （起别名、套函数、经子查询转一道、和 * 写在同一支里，_derived_hidden；认不出落在哪一列的，整份快照遮掉）；快照按列
    遮，再按值遮（别的列里写着遮罩值的格子，_masked_view）。回答正文、查询清单里的遮罩值换成「已遮罩」（_Secrets、
-   _scrub_sql）：按值认有门槛；SQL 只换在找某一个值的字面量（等值、IN、LIKE），范围比较的边界、LIMIT 照原样，封存的是
-   实际执行的语句。已经摆在读者眼前的值不换：问题和选中原文里的、报告列出的指标和运行输入的值、SQL 里照原样留着的
-   阈值。组装之后正文里还认得出遮罩值的，这份回答不交出去
+   _scrub_sql）：按值认有门槛；SQL 里的字面量按语法树认用在哪（_literal_uses），数字照原样留着的只有范围比较的边界
+   （没把某一个值夹住的）和 LIMIT 的行数，封存的是实际执行的语句；在找某一个值的（等值、IN、简单 CASE……）和别处
+   够长的遮罩值都换。已经摆在读者眼前的值不换：问题和选中原文里的、报告列出的指标和运行输入的值；SQL 里照原样留着
+   的阈值只在回答正文里不换，格子照旧换。组装之后还认得出遮罩值的，这份回答不交出去
 4. 写入（_record）：一条 approval.inquiry 事件，node_id 为空，用量写在数据的 usage 里（计进运行用量，见 runner 的
    _usage_of_events）。质询的查询快照只放在 inquiry_artifacts / queries 里，绝不放在 query_artifact 下：证据接口按那个键
    收集封存的查询，会把质询的查询当成报告的证据。回答期间运行状态变了（被批准、被放弃、继续运行之后重新封存），
@@ -41,6 +42,7 @@ from collections import OrderedDict
 from collections.abc import Callable, Iterable, Iterator
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
+from decimal import Decimal
 from typing import Any
 
 from sqlalchemy import func, select
@@ -98,7 +100,7 @@ REFS_MAX = 40
 TABLE_ROWS = 20
 #: 按值认遮罩值的门槛：文字至少这么多字（一个字的「是」「男」换掉，会把不相干的词也换得面目全非）；纯数字至少这么多位
 #: （年龄、等级、小额金额这类短的数和阈值、天数、日期、LIMIT 里的数太容易撞上，按值换会把实际执行的 SQL、提问人自己
-#: 的条件改得面目全非。它们只在 SQL 里和遮罩列直接比较时才换，见 _scrub_sql；引用到的格子照旧按列遮）
+#: 的条件改得面目全非。它们只在 SQL 里和遮罩列做等值一类比较时才换，见 _scrub_sql；引用到的格子照旧按列遮）
 SECRET_MIN = 2
 SECRET_DIGITS = 5
 #: 遮罩列的值最多收这么多种写法（一份快照最多上千行）：再多就按已经收到的换
@@ -904,12 +906,14 @@ class _Secrets:
     """遮罩列里出现过的值。模型看得到原始的查询结果（工具交回的是原样），回答里照着复述、或者把值写进查询条件，
     都会把它带进回答正文和查询清单；按值找出来换成「已遮罩」。这只减少暴露，不是安全边界（同证据面板的遮罩）。
 
-    strong 是按值认得出的写法（够长的数、两个字以上的文字），正文、SQL 里的字符串、名字和注释、别的列里的格子、列名都按
-    它换；weak 是遮罩列里的全部写法（含短的数），只用在 SQL 里和遮罩列做等值比较的字面量上。换的时候先按词、按开头两个
+    strong 是按值认得出的写法（够长的数、两个字以上的文字），正文、SQL 里的字面量、名字和注释、别的列里的格子、列名都按
+    它换；weak 是遮罩列里的全部写法（含短的数），只用在 SQL 里和遮罩列做等值比较的字面量上；shown 是 SQL 里照原样
+    留着的阈值、行数（去掉千分位、正负号的数字），只在回答正文里不换（scrub 的 prose）。换的时候先按词、按开头两个
     字筛出正文里真有的那几个值，再拿它们现编正则：遮罩值上万个时，现编一个上万分支的正则要一秒上下，每次都卡住整个进程。"""
 
     strong: set[str] = field(default_factory=set)
     weak: set[str] = field(default_factory=set)
+    shown: set[str] = field(default_factory=set)
     _index: Any = field(default=None, repr=False, compare=False)
     _compiled: dict[frozenset[str], re.Pattern[str]] = field(default_factory=dict, repr=False, compare=False)
 
@@ -936,12 +940,9 @@ class _Secrets:
                         self.strong.add(form)
         self._index = None
 
-    def protect(self, texts: Iterable[str | None], *, strong_only: bool = False) -> None:
+    def protect(self, texts: Iterable[str | None]) -> None:
         """已经摆在读者眼前的值不换：提问人自己写进问题、选中原文里的（「13900000001 这位客户」「排除 300 以下的」），
-        报告里列出的指标、运行输入的值——换了只会让回答、查询对不上问题和报告。
-
-        strong_only：只免正文、快照里按值认的（SQL 里照原样留着的阈值、LIMIT，见 _shown_literals）；SQL 里和遮罩列
-        做等值比较的照旧换（weak 不动）。"""
+        报告里列出的指标、运行输入的值——换了只会让回答、查询对不上问题和报告。正文、格子、SQL 里都不换。"""
         given = " ".join(t for t in texts if t)
         if not given:
             return
@@ -951,10 +952,18 @@ class _Secrets:
             return _digits_of(form) in numbers if _NUMERIC.fullmatch(form) else form in given
 
         self.strong = {f for f in self.strong if not asked(f)}
-        if not strong_only:
-            self.weak = {f for f in self.weak if not asked(f)}
+        self.weak = {f for f in self.weak if not asked(f)}
         self._index, self._compiled = None, {}
 
+    def show(self, numbers: Iterable[str]) -> None:
+        """SQL 里照原样留着的阈值、行数（_shown_literals）：回答正文里复述它们不换（「金额不低于 30000 元」），换了只会
+        让回答和 SQL 对不上。只免正文：格子、列名、SQL 里照旧按值换——格子里的值正好是这个数，可能就是那个人
+        （col >= v 再按 col 排序取第一行，查出来的就是 v）。"""
+        self.shown = set(numbers)
+
+    def is_shown(self, form: str) -> bool:
+        return bool(_NUMERIC.fullmatch(form)) and _digits_of(form) in self.shown
+
     def hit(self, value: Any) -> bool:
         """一格（或 SQL 里一个字面量）的值是不是按值认得出的遮罩值。短的数不算（见 SECRET_DIGITS）。"""
         if value is None or isinstance(value, bool) or not self.strong:
@@ -971,15 +980,18 @@ class _Secrets:
         """SQL 里一个字面量是不是遮罩列里出现过的值（短的数也算）。"""
         return _plain_text(text) in self.weak
 
-    def scrub(self, text: str) -> str:
+    def scrub(self, text: str, *, prose: bool = False) -> str:
+        """text 里的遮罩值换成「已遮罩」。prose：这是回答正文，SQL 里照原样留着的阈值、行数不换（show）。"""
         if not text or not self.strong:
             return text
-        pattern = self.pattern(text)
+        pattern = self.pattern(text, prose=prose)
         return pattern.sub(_masked_word(), text) if pattern is not None else text
 
-    def pattern(self, text: str) -> re.Pattern[str] | None:
-        """认 text 里出现的遮罩值的正则；一个都没有时为 None。"""
+    def pattern(self, text: str, *, prose: bool = False) -> re.Pattern[str] | None:
+        """认 text 里出现的遮罩值的正则；一个都没有时为 None。prose 同 scrub。"""
         found = self._found(text) if text and self.strong else set()
+        if prose and self.shown:
+            found = {v for v in found if not self.is_shown(v)}
         if not found:
             return None
         key = frozenset(found)
@@ -1033,8 +1045,9 @@ _CELL_HEAD = re.compile(r"^\s*[QK]\d+\.r\d+\.")
 
 
 def _scrubbed_source(text: str, secrets: _Secrets) -> str:
-    """回答原文里的遮罩值换成「已遮罩」：标记之外按值换；标记里的按段换（| 隔开的几段各自换），单元格引用只换列名那
-    一截——编号、行号换不出东西，整段换的话「r0.」会让列名开头的长号码看着像小数的一截，反而认不出来。
+    """回答原文里的遮罩值换成「已遮罩」：标记之外按正文的办法换（SQL 里照原样留着的阈值不换）；标记里的按段换（| 隔开
+    的几段各自换），单元格引用只换列名那一截——编号、行号换不出东西，整段换的话「r0.」会让列名开头的长号码看着像小数
+    的一截，反而认不出来。列名照快照、SQL 里的名字那样换（阈值也换），换得一样才对得上。
     标记里写着遮罩值的引用换完多半解析不了（列名里写着遮罩值的除外：快照的列名也换过了，正好对上），宁可弄坏一个
     引用，也不让值留在存下的原文、片段里，或者从解析不了的占位里露出来。"""
     if not text or not secrets.strong:
@@ -1042,15 +1055,16 @@ def _scrubbed_source(text: str, secrets: _Secrets) -> str:
     out: list[str] = []
     last = 0
     for m in MARKER_RE.finditer(text):
-        out.append(secrets.scrub(text[last:m.start()]))
+        out.append(secrets.scrub(text[last:m.start()], prose=True))
         start, end = m.span("body")
         parts = text[start:end].split("|")
         head = _CELL_HEAD.match(parts[0]) if m.group("kind").strip() == "v" else None
-        cut = head.end() if head else 0
-        parts = [parts[0][:cut] + secrets.scrub(parts[0][cut:]), *(secrets.scrub(p) for p in parts[1:])]
+        first = parts[0][:head.end()] + secrets.scrub(parts[0][head.end():]) if head \
+            else secrets.scrub(parts[0], prose=True)
+        parts = [first, *(secrets.scrub(p, prose=True) for p in parts[1:])]
         out.append(text[m.start():start] + "|".join(parts) + text[end:m.end()])
         last = m.end()
-    out.append(secrets.scrub(text[last:]))
+    out.append(secrets.scrub(text[last:], prose=True))
     return "".join(out)
 
 
@@ -1101,15 +1115,22 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets)
     return view, [names[i] for i in sorted(touched)]
 
 
-#: 在找某一个值的比较：col = 'x'、col <> 'x'、col LIKE 'x'、SQLite 的 col IS 'x'（IN 列表另外认）
+#: 按词认在找某一个值的比较（SQL 解析不了时的退路，见 _context）：col = 'x'、col <> 'x'、col LIKE 'x'、SQLite 的
+#: col IS 'x'（IN 列表另外认）
 _EQUAL_TOKENS = frozenset({"EQ", "NEQ", "NULLSAFE_EQ", "LIKE", "ILIKE", "GLOB", "RLIKE", "IS"})
-#: 范围比较：边界是阈值，不是在找某一个值（BETWEEN 另外认）
+#: 范围比较（BETWEEN 另外认）
 _RANGE_TOKENS = frozenset({"GT", "GTE", "LT", "LTE"})
 _NAME_TOKENS = frozenset({"VAR", "IDENTIFIER"})
 #: 字面量前面可能连着的：正负号、MySQL 的字符集前缀（_utf8mb4'x'）
 _LEADS = frozenset({"DASH", "PLUS", "INTRODUCER"})
 #: 能结束一个值的词：减号跟在它们后面是两数相减，不是负号
 _VALUE_ENDS = frozenset({"NUMBER", "VAR", "IDENTIFIER", "R_PAREN", "R_BRACKET", "NULL", "TRUE", "FALSE"})
+#: 数字照原样留在 SQL 里的两种用法（_literal_uses）：范围比较的边界、LIMIT 这类的行数
+_KEPT = frozenset({"range", "limit"})
+#: 字面量写在左边时运算符翻过来：25 < age 就是 age > 25
+_FLIP = {">": "<", ">=": "<=", "<": ">", "<=": ">="}
+#: 一个纯粹的数（可带正负号、小数）：范围边界按数比大小，写成字符串的也一样
+_PLAIN_NUMBER = re.compile(r"[-+]?\d+(?:\.\d+)?")
 
 
 def _is_literal(kind: str) -> bool:
@@ -1130,10 +1151,10 @@ def _name_at(tokens: list[Any], kinds: list[str], k: int) -> str | None:
 
 
 def _context(tokens: list[Any], kinds: list[str], i: int) -> tuple[str, str | None]:
-    """第 i 个字面量用在哪：("equal", 列名) 在找某一个值——col = 'x'、col NOT LIKE 'x'、col IS NOT 25、
-    col IN ('x', …)、'x' = col；("range", None) 范围比较的边界——col > 25、col BETWEEN a AND b；("other", None) 别的
-    （LIMIT / OFFSET、选出列、函数参数、算式）。列名是比较另一边的那个名字（s.buyer_phone 取 buyer_phone，小写），
-    另一边不是单独一个名字（函数、算式）时为 None。"""
+    """按词认第 i 个字面量用在哪，只在 SQL 解析不了时用（_literals），只认得出简单的写法：("equal", 列名) 在找某一个
+    值——col = 'x'、col NOT LIKE 'x'、col IS NOT 25、col IN ('x', …)、'x' = col；("range", None) 范围比较的边界——
+    col > 25、col BETWEEN a AND b；("other", None) 别的。列名是比较另一边的那个名字（s.buyer_phone 取 buyer_phone，
+    小写），另一边不是单独一个名字（函数、算式）时为 None。"""
     j = _operand_start(kinds, i) - 1
     if j >= 1 and kinds[j] == "NOT" and kinds[j - 1] == "IS":
         return "equal", _name_at(tokens, kinds, j - 2)
@@ -1164,6 +1185,205 @@ def _context(tokens: list[Any], kinds: list[str], i: int) -> tuple[str, str | No
     return "other", None
 
 
+def _column_of(node: Any, index: int | None) -> str | None:
+    """比较另一边的列名（小写、不带表名）；另一边是行值时取第 index 项。不是单独一列（函数、算式、子查询）时为 None。"""
+    from sqlglot import exp
+
+    if index is not None and isinstance(node, exp.Tuple):
+        items = node.expressions
+        node = items[index] if index < len(items) else None
+    while isinstance(node, exp.Paren):
+        node = node.this
+    if isinstance(node, exp.Column) and not isinstance(node.this, exp.Star):
+        return node.name.lower() or None
+    return None
+
+
+def _use_of(node: Any) -> tuple[str, str | None]:
+    """语法树里一个字面量用在哪：(用法, 列名)，用法见 _literal_uses。
+
+    先往上越过不改变「找的是哪个值」的外壳：括号、负号、CAST、字符集前缀、COLLATE；行值、VALUES、ARRAY、ANY / ALL
+    也越过，记下「在一组值里」（行值另记下是第几项）。再看外壳的上一层：等值一类、IN、简单 CASE 的 WHEN 是在找这个值；
+    范围比较、LIMIT 一类只认单独一个值（一组值里的、算式里的都不算）。"""
+    from sqlglot import exp
+
+    child, parent = node, node.parent
+    index, grouped = None, False
+    while parent is not None:
+        if isinstance(parent, exp.Tuple):
+            if index is None:
+                index = next((k for k, item in enumerate(parent.expressions) if item is child), None)
+            grouped = True
+        elif isinstance(parent, (exp.Array, exp.Values, exp.Any, exp.All)) or (
+                isinstance(parent, exp.Subquery) and isinstance(child, exp.Values)) or (
+                isinstance(parent, exp.Anonymous) and str(parent.this).upper() in ("ANY", "ALL", "SOME")):
+            # = ANY(ARRAY[…])、= ANY(VALUES …)；Postgres 的 <> ALL(…) 解析成名叫 ALL 的函数
+            grouped = True
+        elif not _wraps(parent, child):
+            break
+        child, parent = parent, parent.parent
+    if isinstance(parent, (exp.EQ, exp.NEQ, exp.NullSafeEQ, exp.NullSafeNEQ, exp.Is, exp.Like, exp.ILike, exp.Glob,
+                           exp.RegexpLike, exp.SimilarTo)):
+        return "equal", _column_of(parent.expression if child is parent.this else parent.this, index)
+    if isinstance(parent, exp.In):
+        return "equal", None if child is parent.this else _column_of(parent.this, index)
+    if isinstance(parent, exp.If) and child is parent.this and isinstance(parent.parent, exp.Case) \
+            and parent.parent.this is not None:
+        # 简单 CASE：CASE buyer_phone WHEN 13900000001 THEN …（CASE WHEN col = v 的写法走上面的等值）
+        return "equal", _column_of(parent.parent.this, index)
+    if grouped:
+        return "other", None
+    if type(parent) in _range_ops() and (child is parent.this or child is parent.expression):
+        return "range", _column_of(parent.expression if child is parent.this else parent.this, None)
+    if isinstance(parent, exp.Between) and child is not parent.this:
+        return "range", _column_of(parent.this, None)
+    if (isinstance(parent, (exp.Limit, exp.Offset)) and child is parent.expression) or (
+            isinstance(parent, exp.Fetch) and child is parent.args.get("count")):
+        return "limit", None
+    return "other", None
+
+
+def _wraps(parent: Any, child: Any) -> bool:
+    """parent 只是 child 的外壳，不改变它是哪个值：括号、负号、CAST、字符集前缀（_utf8mb4'x'）、COLLATE。"""
+    from sqlglot import exp
+
+    return isinstance(parent, (exp.Paren, exp.Neg, exp.Cast)) or (
+        isinstance(parent, exp.Introducer) and child is parent.expression) or (
+        isinstance(parent, exp.Collate) and child is parent.this)
+
+
+def _range_ops() -> dict[Any, str]:
+    """范围比较的语法树节点 → 「列 运算符 值」方向的运算符。"""
+    from sqlglot import exp
+
+    return {exp.GT: ">", exp.GTE: ">=", exp.LT: "<", exp.LTE: "<="}
+
+
+def _constant(node: Any) -> Any:
+    """写死的值：数按数（几个数加减乘除算出来的也算，13900000002 - 1），字符串按原文（纯粹的数照样按数）；别的为 None。"""
+    from sqlglot import exp
+
+    if isinstance(node, exp.Neg):
+        value = _constant(node.this)
+        return -value if isinstance(value, Decimal) else None
+    if isinstance(node, (exp.Paren, exp.Cast, exp.Collate)):
+        return _constant(node.this)
+    if isinstance(node, exp.Introducer):
+        return _constant(node.expression)
+    if isinstance(node, _literal_types()):
+        text = str(node.this).strip()
+        return Decimal(text) if _PLAIN_NUMBER.fullmatch(text) else text
+    if isinstance(node, (exp.Add, exp.Sub, exp.Mul, exp.Div)):
+        a, b = _constant(node.this), _constant(node.expression)
+        if not (isinstance(a, Decimal) and isinstance(b, Decimal)):
+            return None
+        if isinstance(node, exp.Div):
+            return a / b if b else None
+        return a + b if isinstance(node, exp.Add) else a - b if isinstance(node, exp.Sub) else a * b
+    return None
+
+
+def _bare_literal(node: Any) -> Any:
+    """边界本身就是一个字面量（套了括号、负号、CAST 的也算）时交回那个字面量，否则 None。"""
+    from sqlglot import exp
+
+    while node is not None and not isinstance(node, _literal_types()):
+        if isinstance(node, (exp.Paren, exp.Neg, exp.Cast, exp.Collate)):
+            node = node.this
+        elif isinstance(node, exp.Introducer):
+            node = node.expression
+        else:
+            return None
+    return node
+
+
+def _bounds_of(node: Any) -> list[tuple[Any, str | None, Any, str]]:
+    """一个范围比较里写死的边界：(字面量节点, 列名, 值, 「列 运算符 值」的运算符)。边界是几个数算出来的也算，
+    字面量节点记 None：它照样能夹住另一头的值，换的是另一头的字面量。"""
+    from sqlglot import exp
+
+    if isinstance(node, exp.Between):
+        sides = [(node.args.get("low"), node.this, ">="), (node.args.get("high"), node.this, "<=")]
+    elif (op := _range_ops().get(type(node))) is not None:
+        sides = [(node.expression, node.this, op), (node.this, node.expression, _FLIP[op])]
+    else:
+        return []
+    out = []
+    for operand, other, op in sides:
+        value = _constant(operand) if operand is not None else None
+        if value is not None:
+            out.append((_bare_literal(operand), _column_of(other, None), value, op))
+    return out
+
+
+def _literal_types() -> tuple[Any, ...]:
+    """语法树里的字面量：数字、字符串，以及字符串带前缀的几种变体（N'…'、$$…$$、U&'…'、E'…'、X'…'、B'…'）。"""
+    from sqlglot import exp
+
+    return (exp.Literal, exp.National, exp.RawString, exp.UnicodeString, exp.ByteString, exp.HexString, exp.BitString)
+
+
+def _pinned(bounds: list[tuple[Any, str | None, Any, str]]) -> set[int]:
+    """范围比较里把某一个值夹住了的边界字面量（交回 id(字面量节点)）。bounds 见 _bounds_of。
+
+    一个值 v 被从下面隔开，是说 v 和比它小一点的值结果不同（col >= v、col < v）；被从上面隔开，是和比它大一点的值结果
+    不同（col <= v、col > v）。同一列上 v 两边都被隔开，条件就只认 v 这一个值：col >= v AND col <= v、BETWEEN v AND v、
+    col > v OR col < v（就是 col <> v）——这是在找 v，和等值比较一样换。整数的 col < v + 1 就是 col <= v、col > v - 1
+    就是 col >= v，所以 col >= v AND col < v + 1 也算夹住了 v。只从一边隔开的是阈值：col >= v 和 col < v 各数一遍，
+    是按 v 分成两段，照原样。另一边不是单独一列的（SUM(amount) >= 900）和哪一列都算同一列（宁可多换）。"""
+    sides: dict[Any, list[tuple[str, str | None, Any]]] = {}
+    for literal, column, value, op in bounds:
+        whole = isinstance(value, Decimal) and value == value.to_integral_value()
+        marks = {">=": [(value, "below")], "<=": [(value, "above")],
+                 ">": [(value, "above"), *([(value + 1, "below")] if whole else [])],
+                 "<": [(value, "below"), *([(value - 1, "above")] if whole else [])]}[op]
+        for point, side in marks:
+            sides.setdefault(point, []).append((side, column, literal))
+    out: set[int] = set()
+    for marks in sides.values():
+        for side, column, literal in marks:
+            for other, column2, literal2 in marks:
+                if side == "below" and other == "above" and (column is None or column2 is None or column == column2):
+                    out |= {id(n) for n in (literal, literal2) if n is not None}
+    return out
+
+
+def _literal_uses(sql: str, kind: str | None) -> dict[int, tuple[str, str | None]] | None:
+    """SQL 里每个字面量（按它在原文里的起始位置）用在哪，按语法树认（写法怎么变都认得出）：
+    - ("equal", 列名)：在找某一个值——等值、不等、IN（含 VALUES、行值）、LIKE 一类、IS [NOT] DISTINCT FROM、简单 CASE 的
+      WHEN、= ANY(ARRAY[…])，套了括号、负号、CAST 的照认；把某一个值夹住的范围（_pinned）也算。列名是被比较的那一列
+      （小写、不带表名），另一边不是单独一列时为 None；
+    - ("range", 列名)：范围比较（> >= < <=、BETWEEN）的边界，是 agent 定的阈值；
+    - ("limit", None)：LIMIT / OFFSET / FETCH / TOP 的行数；
+    - ("other", None)：别的——选出列、函数参数、算式、THEN 后面的值、子查询里选出的值。
+    解析不了返回 None。"""
+    from sqlglot import exp
+
+    from app.data.sql_semantics import _parse, _read
+
+    try:
+        root, _ = _parse(sql, _read(kind))
+    except Exception:  # noqa: BLE001 - 解析器的意外按解析不了处理
+        root = None
+    if root is None:
+        return None
+    # 解析的是去掉头尾空白的 SQL：位置补上开头的空白，才对得上原文里的词
+    offset = len(sql) - len(sql.lstrip())
+    uses: dict[int, tuple[str, str | None]] = {}
+    try:
+        literals = {id(n): n for n in root.walk() if isinstance(n, _literal_types())}
+        pinned = _pinned([b for n in root.find_all(exp.Between, *_range_ops()) for b in _bounds_of(n)])
+        for key, node in literals.items():
+            start = (node.meta or {}).get("start")
+            if isinstance(start, int):
+                use, column = _use_of(node)
+                uses[start + offset] = ("equal", column) if key in pinned and use == "range" else (use, column)
+    except Exception:  # noqa: BLE001 - 分析本身的缺陷不能让回答出错：按解析不了处理（认不准的地方不免）
+        logger.warning("质询的 SQL 字面量分析失败", exc_info=True)
+        return None
+    return uses
+
+
 def _requote(raw: str, change: Callable[[str], str]) -> str:
     """一个字面量或名字的原文 → 只换里面的字：引号、N'' E'' U&'' 这类前缀、$tag$ 照留；不带引号的整个交给 change。"""
     if raw.startswith("$") and (end := raw.find("$", 1)) > 0:
@@ -1187,23 +1407,58 @@ def _sql_tokens(sql: str, kind: str | None) -> tuple[list[Any], list[str]] | Non
     return tokens, [t.token_type.name for t in tokens]
 
 
-def _scrub_sql(sql: str, secrets: _Secrets, compared: set[str], kind: str | None) -> str:
-    """SQL 里遮罩列的值换成「已遮罩」，别的一个字不动：查询清单随运行封存，它得是实际执行的那条，审批人看 what-if 的
-    条件（阈值、日期、LIMIT、别的列的条件）看的就是它。只换在找某一个值的字面量（_context 的 equal）：
+def _literals(sql: str, kind: str | None) -> tuple[list[Any], list[str], dict[int, tuple[str, str | None]]] | None:
+    """分词，加上每个字面量（按词的下标）用在哪（_literal_uses）。语法树里对不上位置的字面量记 "other"；SQL 解析
+    不了时只按词认简单的等值比较（_context），别的一律 "other"——认不准的地方不免。分词出错返回 None。
 
-    - 和遮罩列（或它的别名，compared）做等值、IN、LIKE 比较的，是遮罩列里出现过的值就换，短的数也换（age = 25）；
-      和别的列做等值比较的，够长的遮罩值也换（另一张表里没设遮罩的同一个手机号）；
-    - 范围比较的边界（age > 31、BETWEEN 10000 AND 50000）、LIMIT / OFFSET、选出列和算式里的数照原样：它们是 agent
-      自己定的阈值，不是在找某一个人，换了封存的就不是实际执行的语句；
-    - 别处的字符串（LIKE 别的列、函数参数）、注释、名字（AS "13900000001 的单"、张伟的单数）按值换（同正文：够长的数、
-      两个字以上的文字）。
-    分词出错就整条按正文的办法换。"""
-    if not sql or not (secrets.strong or secrets.weak):
-        return sql
+    照原样留着的数还得在这条 SQL 的字面量里处处都是阈值、行数：同一个数在别的字面量里（等值比较、算式、字符串）也
+    出现过，那里会被换掉，这里留着就把它露了（BETWEEN 13900000001 AND 13900000001 + 0），所以这里也改记成在找这个值。
+    名字、注释里的不算：那是 agent 给阈值起的说法（AS 超过30000的笔数），照常按值换，留着的阈值不必跟着换。"""
     split = _sql_tokens(sql, kind)
     if split is None:
-        return secrets.scrub(sql)
+        return None
     tokens, kinds = split
+    uses = _literal_uses(sql, kind)
+    found: dict[int, tuple[str, str | None]] = {}
+    for i, token in enumerate(tokens):
+        if not _is_literal(kinds[i]):
+            continue
+        if uses is not None:
+            found[i] = uses.get(token.start, ("other", None))
+        else:
+            use, column = _context(tokens, kinds, i)
+            found[i] = (use, column) if use == "equal" else ("other", None)
+    kept = {i for i, (use, _) in found.items() if kinds[i] == "NUMBER" and use in _KEPT}
+    elsewhere = _numbers_in(tokens[i].text for i in found if i not in kept)
+    for i in kept:
+        if _digits_of(tokens[i].text) in elsewhere:
+            found[i] = ("equal", found[i][1])
+    return tokens, kinds, found
+
+
+def _numbers_in(texts: Iterable[str]) -> set[str]:
+    """这几段字里写着的数（去掉千分位、正负号）。"""
+    return {_digits_of(n) for text in texts for n in _DIGITS.findall(text)}
+
+
+def _scrub_sql(sql: str, secrets: _Secrets, compared: set[str], kind: str | None) -> str:
+    """SQL 里遮罩列的值换成「已遮罩」，别的一个字不动：查询清单随运行封存，它得是实际执行的那条，审批人看 what-if 的
+    条件（阈值、日期、LIMIT、别的列的条件）看的就是它。字面量按用在哪换（_literals）：
+
+    - 在找某一个值的（equal）：和遮罩列（或它的别名，compared）比较的，是遮罩列里出现过的值就换，短的数也换
+      （age = 25）；和别的列比较的，够长的遮罩值也换（另一张表里没设遮罩的同一个手机号）；
+    - 范围比较的边界（age > 31、BETWEEN 10000 AND 50000）、LIMIT / OFFSET 的行数，数字照原样：它们是 agent 自己定的
+      阈值，不是在找某一个人，换了封存的就不是实际执行的语句；
+    - 别处的数（选出列、函数参数、算式）认不准是不是在找某一个值，够长的遮罩值换；
+    - 字符串在哪都按值换（范围边界也一样：写成字符串的长号码做边界，多半是在找那一个人），注释、名字（AS
+      "13900000001 的单"、张伟的单数）也按值换（同正文：够长的数、两个字以上的文字）。
+    分词出错就整条按正文的办法换；解析不了的认不准哪个数是阈值，范围边界、行数也不免（_literals）。"""
+    if not sql or not (secrets.strong or secrets.weak):
+        return sql
+    split = _literals(sql, kind)
+    if split is None:
+        return secrets.scrub(sql)
+    tokens, kinds, uses = split
     word = _masked_word()
     out: list[str] = []
     last = 0
@@ -1214,12 +1469,14 @@ def _scrub_sql(sql: str, secrets: _Secrets, compared: set[str], kind: str | None
         gap = sql[last:start]
         out.append(secrets.scrub(gap) if gap.strip() else gap)
         raw = sql[start:end]
-        if _is_literal(kinds[i]):
-            use, column = _context(tokens, kinds, i)
+        if i in uses:
+            use, column = uses[i]
             if use == "equal" and ((column in compared and secrets.weak_hit(token.text)) or secrets.hit(token.text)):
                 raw = _requote(raw, lambda _: word)
-            elif use != "range" and kinds[i] != "NUMBER":
+            elif kinds[i] != "NUMBER":
                 raw = _requote(raw, secrets.scrub)
+            elif use not in _KEPT and secrets.hit(token.text):
+                raw = word
         elif kinds[i] in _NAME_TOKENS:
             raw = _requote(raw, secrets.scrub)
         out.append(raw)
@@ -1229,20 +1486,24 @@ def _scrub_sql(sql: str, secrets: _Secrets, compared: set[str], kind: str | None
     return "".join(out)
 
 
-def _shown_literals(sql: str, kind: str | None) -> list[str]:
-    """SQL 里不管是什么值都照原样留着的字面量（_scrub_sql 不换的两种）：范围比较的边界，LIMIT、选出列这类地方的数。
-    它们随 SQL 摆在读者眼前，回答正文、快照里同样的值再遮也遮不住什么，只会让回答和 SQL 对不上。"""
-    split = _sql_tokens(sql, kind) if sql else None
-    if split is None:
-        return []
-    tokens, kinds = split
-    out: list[str] = []
-    for i, token in enumerate(tokens):
-        if _is_literal(kinds[i]):
-            use = _context(tokens, kinds, i)[0]
-            if use == "range" or (use == "other" and kinds[i] == "NUMBER"):
-                out.append(token.text)
-    return out
+def _shown_literals(sqls: Iterable[tuple[str, str | None]]) -> set[str]:
+    """几条 SQL（和各自数据源的类型）里照原样留着的数，去掉千分位、正负号：范围比较的边界、LIMIT / OFFSET 的行数
+    （_scrub_sql 不换的那两种）。它们随 SQL 摆在读者眼前，回答正文里复述（「金额不低于 30000 元」）再遮也遮不住
+    什么，只会让回答和 SQL 对不上（_Secrets.show）。同一个数在别的查询里出现在别的字面量里也不算（等值比较、选出列、
+    函数参数、字符串；同一条里的 _literals 已经不留）：那里它会被换掉，正文写出来就把那里换掉的值露了。SQL 解析
+    不了的一个都不算，里面写着的数也都不算。"""
+    kept: set[str] = set()
+    elsewhere: set[str] = set()
+    for sql, kind in sqls:
+        split = _literals(sql, kind) if sql else None
+        if split is None:
+            elsewhere |= _numbers_in([sql or ""])
+            continue
+        tokens, kinds, uses = split
+        keep = {i for i, (use, _) in uses.items() if kinds[i] == "NUMBER" and use in _KEPT}
+        kept |= {_digits_of(tokens[i].text) for i in keep}
+        elsewhere |= _numbers_in(tokens[i].text for i in uses if i not in keep)
+    return kept - elsewhere
 
 
 @dataclass
@@ -1270,16 +1531,18 @@ 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]]) -> _Plan:
+                shown: list[tuple[str, str]] | None = None) -> _Plan:
     """own：agent 自己的查询快照（工件 id → 内容）；others：报告自己的查询快照（被问的那句可能直接写着遮罩列的值，
     agent 会照着复述、写进查询条件）；given：问题、选中的原文、报告里列出的指标和运行输入的值（都已经摆在读者眼前，
-    里面的值不换）；shown：agent 执行的 SQL 和它查的数据源名（照原样留在 SQL 里的阈值、LIMIT，正文里同样不换）。
+    里面的值哪里都不换）；shown：agent 执行的 SQL 和它查的数据源名（照原样留在 SQL 里的阈值、LIMIT，回答正文里
+    不换，见 _Secrets.show；取质询工件不用：快照里的格子照旧按值遮）。
 
     查数据源的遮罩要读库，在这里读；解析 SQL、收上万个值、逐格比对是纯计算，放进线程：放在事件循环里会挡住同一进程
     里别的运行（同 tools/datasource 的 _sql_checks）。"""
     from app.api.evidence import _Masks
 
     masks = _Masks()
+    shown = shown or []
     named = {artifact: await _hidden_columns(snapshot, masks) for artifact, snapshot in own.items()}
     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]),
@@ -1300,8 +1563,7 @@ def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str]
     for snapshot, base in based:
         secrets.add(snapshot, _derived_hidden(snapshot, base, kinds.get(str(snapshot.get("source") or "")))[0])
     secrets.protect(given)
-    secrets.protect([value for sql, name in shown for value in _shown_literals(sql, kinds.get(name))],
-                    strong_only=True)
+    secrets.show(_shown_literals((sql, kinds.get(name)) for sql, name in shown))
     secrets.prepare()
     views: dict[str, dict[str, Any]] = {}
     masked: dict[str, list[str]] = {}
@@ -1426,13 +1688,7 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict
         error = _error_words(outcome, timeout_s=part.get("timeout_s"), daily_max_usd=settings.get("daily_max_usd"))
         if outcome.error:
             logger.info("质询没能给出回答（运行 %s）：%s", op.run_id, outcome.error)
-    queries = []
-    for q in outcome.queries:
-        artifact, name = str(q.get("artifact") or ""), str(q.get("source") or "")
-        # 快照没取到的（存不下、被删了），SQL 里按这个源设的遮罩列认比较
-        compared = plan.tainted.get(artifact) or {c.lower() for c in live.get(name, [])}
-        queries.append({**{k: q.get(k) for k in QUERY_KEYS},
-                        "sql": _scrub_sql(str(q.get("sql") or ""), plan.secrets, compared, plan.kinds.get(name))})
+    queries = await asyncio.to_thread(_sealed_queries, outcome.queries, plan, live)
     return {
         "answer_text": text, "answer_doc": doc_id, "ok": ok, "limit": _limit(outcome),
         "error": None if ok else (error or failed or NO_ANSWER),
@@ -1443,6 +1699,18 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict
     }
 
 
+def _sealed_queries(queries: list[dict[str, Any]], plan: _Plan, live: dict[str, list[str]]) -> list[dict[str, Any]]:
+    """随运行封存的查询清单：每条只留约定的几项，SQL 里的遮罩值换掉。_scrub_sql 要解析 SQL，放在线程里（同 _plan）。"""
+    out: list[dict[str, Any]] = []
+    for q in queries:
+        artifact, name = str(q.get("artifact") or ""), str(q.get("source") or "")
+        # 快照没取到的（存不下、被删了），SQL 里按这个源设的遮罩列认比较
+        compared = plan.tainted.get(artifact) or {c.lower() for c in live.get(name, [])}
+        out.append({**{k: q.get(k) for k in QUERY_KEYS},
+                    "sql": _scrub_sql(str(q.get("sql") or ""), plan.secrets, compared, plan.kinds.get(name))})
+    return out
+
+
 def _quiet_load(loader: Callable[[str], Any], artifact: Any) -> Any:
     """取一份快照；取不到、哈希不符都当没有（这里只是为了找遮罩列的值，取不到的那份本来也渲染不出来）。"""
     if not artifact:
@@ -1456,21 +1724,31 @@ def _quiet_load(loader: Callable[[str], Any], artifact: Any) -> Any:
 def _compose(source: str, catalog: dict[str, Any], *, run_id: str, views: dict[str, dict[str, Any]],
              secrets: _Secrets, allow: list[str]) -> dict[str, Any]:
     """agent 写的回答原文（带引用标记）→ 回答文档：目录是 agent 自己的查询，加上报告里的指标和运行输入；快照按遮罩
-    遮好再渲染。遮罩值在组装之前换掉（_scrubbed_source）；组装出来正文里还认得出遮罩值（渲染出来才露的，比如别的列
-    里一格文字夹着手机号），抛 _Leak，不交出去。allow 是提问人自己写的数（_given_numbers），回答里复述它们不标
-    「无出处」。"""
+    遮好再渲染。遮罩值在组装之前换掉（_scrubbed_source）；组装出来还认得出遮罩值（渲染出来才露的，比如别的列里一格
+    文字夹着手机号），抛 _Leak，不交出去。allow 是提问人自己写的数（_given_numbers），回答里复述它们不标「无出处」。"""
     loader = _masking_loader(views)
     allowed = list(allow) or None
     # 列名里写着遮罩值的查询：目录里的列名照快照换（_masked_view），整表展开、表头核对、存下的目录都按换过的名字
     catalog = {alias: _renamed_entry(entry, secrets) for alias, entry in catalog.items()}
     doc = compose_doc(_scrubbed_source(source, secrets), catalog, run_id=run_id, loader=loader,
                       allow_numbers=allowed)
-    markdown = str(doc.get("markdown") or "")
-    if secrets.scrub(markdown) != markdown:
+    if _leaked(doc, secrets):
         raise _Leak()
     return doc
 
 
+def _leaked(doc: dict[str, Any], secrets: _Secrets) -> bool:
+    """组装出来的正文里还认得出遮罩值。SQL 里照原样留着的阈值写在正文里不算（_Secrets.show），渲染进单元格引用里的
+    算：格子已经按值遮过，遮完还认得出，就是一格文字里夹着这个数，渲染出来才露。"""
+    markdown = str(doc.get("markdown") or "")
+    pattern = secrets.pattern(markdown)
+    if pattern is None:
+        return False
+    cells = [seg["span"] for seg in iter_segments(doc) if str(seg.get("ref") or "").startswith("v:")]
+    return any(not secrets.is_shown(m.group()) or any(a < m.end() and m.start() < b for a, b in cells)
+               for m in pattern.finditer(markdown))
+
+
 def _renamed_entry(entry: Any, secrets: _Secrets) -> Any:
     if not (isinstance(entry, dict) and entry.get("kind") == "query" and isinstance(entry.get("columns"), list)):
         return entry
@@ -1657,7 +1935,8 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]:
     if not _query_shaped(content):
         return {"id": artifact_id, "content": content, "masked": []}
     # 遮罩值照回答时收：引用它的那几条质询各自查到的全部快照（查询条件里写着的值，不一定在这一份的结果里），加上报告
-    # 自己的查询快照；免遮的也照回答时：提问人自己写的、报告里列出的指标和运行输入的值、SQL 里照原样留着的阈值
+    # 自己的查询快照；免遮的也照回答时：提问人自己写的、报告里列出的指标和运行输入的值。SQL 里照原样留着的阈值只在
+    # 回答正文里不遮，格子照旧按值遮，这里不用
     own: dict[str, dict[str, Any]] = {artifact_id: content}
     for data in referenced:
         for other in dict.fromkeys(str(a) for a in data.get("inquiry_artifacts") or []):
@@ -1665,8 +1944,7 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]:
                 own[other] = snapshot
     others, values = await _report_material(run_id, (d.get("report") for d in referenced))
     asked = [t for d in referenced for t in (d.get("question"), d.get("quote")) if isinstance(t, str)]
-    plan = await _plan(own, others, [*asked, *values],
-                       [(str(s.get("sql") or ""), str(s.get("source") or "")) for s in own.values()])
+    plan = await _plan(own, others, [*asked, *values])
     shown = plan.views[artifact_id]
     recorded = next((q["sql"] for d in referenced for q in d.get("queries") or []
                      if isinstance(q, dict) and q.get("artifact") == artifact_id and isinstance(q.get("sql"), str)),

backend/tests/test_inquiry_api.py

@@ -12,8 +12,9 @@
   事件里记的报告文档不是这次运行的 422；输入不合法 422，指明是哪一项
 - 回答：先给结论，数字只能引用自己查询的格子或报告已有的指标，裸数字被标出（提问人自己写的数不标）；遮罩列（起了
   别名的、和 * 写在一起没起别名的表达式也算）的格子渲染成「已遮罩」，回答正文、查询清单、事件、质询工件里都不出现
-  遮罩列的值；遮罩值不误伤阈值、天数、LIMIT（长的数也一样）、提问人自己写的数、报告口径卡里的指标；质询工件按值遮
-  只出现在报告快照里的遮罩值
+  遮罩列的值；换着写法找某一个客户（简单 CASE、IN (VALUES …)、两头相同的范围、行值、选出列里写着同一个号码……）
+  也一样；遮罩值不误伤阈值、天数、LIMIT（长的数也一样）、提问人自己写的数、报告口径卡里的指标；SQL 里照原样留着的
+  阈值只在回答正文里不遮，格子照旧按值遮，回答和质询工件对得上；质询工件按值遮只出现在报告快照里的遮罩值
 - 接线：设置里的上限、运行固定的数据版本交到 agent 手里（上传表格重传之后查的仍是运行固定的那一版）；同名重建的
   数据源、没固定过版本的上传表格不查
 - 触顶、失败照样记一条事件，说法给提问人看得懂、说下一步；轮数用完不算查询次数到顶，查询失败、引用解析不了在末尾注明
@@ -1016,9 +1017,9 @@ async def test_an_unnamed_expression_next_to_a_star_is_masked_everywhere(monkeyp
     assert got.json()["masked"] == ["buyer_phone", "SUBSTR(buyer_phone, 4)"]
 
 
-#: 金额设了遮罩的大单：金额都是五位数，和阈值、LIMIT、访问次数的条件撞在一起
+#: 金额设了遮罩的大单：金额都是五位数，和阈值、LIMIT、访问次数撞在一起（第二单的访问次数正好等于第三单的金额）
 BIG = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, amount INTEGER, visits INTEGER);"
-       "INSERT INTO sales VALUES (1, '华东', 10000, 25000), (2, '华东', 12000, 18000), (3, '华南', 20000, 30000),"
+       "INSERT INTO sales VALUES (1, '华东', 10000, 25000), (2, '华东', 12000, 20000), (3, '华南', 20000, 30000),"
        " (4, '华南', 30000, 22000), (5, '华北', 50000, 9000);")
 #: 报告取数选了金额：金额列里的值都算遮罩值
 AMOUNTS = "SELECT region AS 区域, amount AS 金额 FROM sales ORDER BY id"
@@ -1053,14 +1054,39 @@ async def test_thresholds_and_limits_are_sealed_as_written_even_when_they_match_
     assert [q["sql"] for q in event.data["queries"]] == sealed
     got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{item['queries'][0]['artifact']}")
     assert got.status_code == 200 and got.json()["content"]["sql"] == large
+    got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{item['queries'][1]['artifact']}")
+    assert got.status_code == 200 and got.json()["content"]["sql"] == sealed[1], "质询工件的 SQL 和事件里的一字不差"
     doc = artifact_store.load(item["answer"]["doc_artifact"])
     assert [s["text"] for s in iter_segments(doc) if s.get("issue") == "uncited_number"] == ["30000", "20000"], \
         "不遮不等于有出处：agent 自己定的阈值照样标「无出处」"
 
 
+async def test_a_cell_equal_to_a_threshold_in_the_sql_is_masked_alike_in_the_answer_and_the_artifact(
+        monkeypatch, big_shop, client):
+    """访问次数正好等于 SQL 里照原样留着的阈值（20000，也是某一笔的金额）：正文里复述的阈值不遮，格子里的照旧按值遮
+    （col >= v 再取第一行，查出来的可能就是那一笔）。回答和质询工件里这一格一样是「已遮罩」，两边对得上。"""
+    least = "SELECT region AS 区域, visits AS 访问 FROM sales WHERE visits >= 20000 ORDER BY visits LIMIT 1"
+    plays = {"访问": Play([least], "访问不少于 20000 次的区域里，最少的是 [[v:Q1.r0.区域]]，"
+                                   "访问 [[v:Q1.r0.访问]] 次。[[see:Q1]]")}
+    Models(monkeypatch, big_shop, writer="华东首单金额 [[v:Q1.r0.金额]]。[[see:Q1]]", plays=plays)
+    paused = await pause(big_shop, graph=fetching(big_shop, AMOUNTS))
+    [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
+    item = r.json()["inquiry"]
+    assert item["answer"]["text"] == "访问不少于 20000 次的区域里，最少的是 华东，访问 已遮罩 次。"
+    [query] = item["queries"]
+    assert query["sql"] == least, "阈值、LIMIT 照原样封存"
+    got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}")
+    assert got.status_code == 200, got.text
+    assert got.json()["content"]["rows"] == [["华东", "已遮罩"]] and got.json()["masked"] == ["访问"]
+    assert got.json()["content"]["sql"] == least
+
+
 async def test_a_metric_from_the_report_that_equals_a_masked_value_can_be_cited(monkeypatch, big_shop, client):
     """报告口径卡里的指标（单笔目标 20000）恰好和遮罩列里的某一笔一样：报告里本来就看得到，引用它的回答照常交出去，
-    换个问法也不会被拦。"""
+    换个问法也不会被拦；别的列里正好是这个数的格子（第二单访问 20000 次）在回答和质询工件里都照常显示，两边对得上。"""
     metrics = [{"id": "target", "name": "单笔目标", "unit": "元", "expression": "vars.kpi.target"}]
     graph = chain(node("start", "input"),
                   node("fetch", "tool", tool=f"db_query__{big_shop}", args={"sql": AMOUNTS}),
@@ -1070,7 +1096,9 @@ async def test_a_metric_from_the_report_that_equals_a_masked_value_can_be_cited(
                   node("gate", "human", mode="approve", title="周报可以出具吗"),
                   node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]))
     smallest = "SELECT amount AS 金额 FROM sales WHERE amount > 15000 ORDER BY amount LIMIT 1"
-    plays = {"定得高不高": Play([smallest], "单笔目标为 [[m:target]]，达标的最小一笔是 [[v:Q1.r0.金额]]。[[see:Q1]]")}
+    second = "SELECT region AS 区域, visits AS 访问 FROM sales WHERE id = 2"
+    plays = {"定得高不高": Play([smallest, second], "单笔目标为 [[m:target]]，达标的最小一笔是 [[v:Q1.r0.金额]]；"
+                                                  "[[v:Q2.r0.区域]]第二单访问 [[v:Q2.r0.访问]] 次。[[see:Q1]]")}
     Models(monkeypatch, big_shop, writer="单笔目标 [[m:target]]。[[see:m:target]]", plays=plays)
     paused = await pause(big_shop, graph=graph)
     [approval] = await pending(paused.id)
@@ -1079,7 +1107,13 @@ async def test_a_metric_from_the_report_that_equals_a_masked_value_can_be_cited(
     assert r.status_code == 200, r.text
     answer = r.json()["inquiry"]["answer"]
     assert (answer["ok"], answer["error"]) == (True, None)
-    assert answer["text"] == "单笔目标为 20,000元，达标的最小一笔是 已遮罩。"
+    assert answer["text"] == "单笔目标为 20,000元，达标的最小一笔是 已遮罩；华东第二单访问 20,000 次。"
+    first, other = r.json()["inquiry"]["queries"]
+    got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{other['artifact']}")
+    assert got.status_code == 200, got.text
+    assert got.json()["content"]["rows"] == [["华东", 20000]] and got.json()["masked"] == []
+    got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{first['artifact']}")
+    assert got.json()["content"]["rows"] == [["已遮罩"]] and got.json()["masked"] == ["金额"]
 
 
 async def test_a_masked_value_the_asker_wrote_stays_in_the_sql(monkeypatch, tmp_path, client):
@@ -1136,6 +1170,71 @@ async def test_a_value_seen_only_in_the_reports_snapshot_is_masked_in_the_artifa
         await engines.invalidate(source_id)
 
 
+#: 同样的四笔订单，手机号存成整数：遮罩值在 SQL 里写成数字
+PHONES = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone INTEGER, amount INTEGER, is_test INTEGER);"
+          "INSERT INTO sales VALUES (1, '华东', 13900000001, 1200, 0), (2, '华东', 13900000002, 300, 1),"
+          " (3, '华南', 13900000003, 900, 0), (4, '华北', 13900000004, 400, 0);")
+#: 报告取数选了手机号：遮罩值从报告自己的查询快照里收
+WITH_PHONE = "SELECT region AS 区域, buyer_phone, amount AS 金额 FROM sales ORDER BY id"
+_COUNT = "SELECT COUNT(*) AS 笔数 FROM sales WHERE "
+
+
+@pytest.mark.parametrize("script, sql, sealed", [
+    # 数个数用 COUNT：SUM(CASE buyer_phone …) 的结果取自遮罩列，整列按列遮
+    pytest.param(PHONES, "SELECT COUNT(CASE buyer_phone WHEN 13900000001 THEN 1 END) AS 笔数 FROM sales",
+                 "SELECT COUNT(CASE buyer_phone WHEN 已遮罩 THEN 1 END) AS 笔数 FROM sales", id="简单 CASE"),
+    pytest.param(PHONES, _COUNT + "buyer_phone IN (VALUES (13900000001))", _COUNT + "buyer_phone IN (VALUES (已遮罩))",
+                 id="IN (VALUES)"),
+    pytest.param(PHONES, _COUNT + "buyer_phone BETWEEN 13900000001 AND 13900000001",
+                 _COUNT + "buyer_phone BETWEEN 已遮罩 AND 已遮罩", id="BETWEEN 两头相同"),
+    pytest.param(PHONES, _COUNT + "buyer_phone >= 13900000001 AND buyer_phone <= 13900000001",
+                 _COUNT + "buyer_phone >= 已遮罩 AND buyer_phone <= 已遮罩", id="不低于且不高于同一个值"),
+    pytest.param(PHONES, _COUNT + "NOT (buyer_phone > 13900000001 OR buyer_phone < 13900000001)",
+                 _COUNT + "NOT (buyer_phone > 已遮罩 OR buyer_phone < 已遮罩)", id="既不大于也不小于"),
+    pytest.param(PHONES, _COUNT + "buyer_phone >= 13900000001 AND buyer_phone < 13900000002",
+                 _COUNT + "buyer_phone >= 已遮罩 AND buyer_phone < 已遮罩", id="整数的半开区间"),
+    pytest.param(PHONES, _COUNT + "(buyer_phone, region) = (13900000001, '华东')",
+                 _COUNT + "(buyer_phone, region) = (已遮罩, '华东')", id="行值"),
+    pytest.param(PHONES, _COUNT + "buyer_phone IS NOT DISTINCT FROM 13900000001",
+                 _COUNT + "buyer_phone IS NOT DISTINCT FROM 已遮罩", id="IS NOT DISTINCT FROM"),
+    pytest.param(PHONES, _COUNT + "buyer_phone = (13900000001)", _COUNT + "buyer_phone = (已遮罩)", id="套了括号"),
+    pytest.param(PHONES, "SELECT COUNT(*) AS 笔数, 13900000001 AS 号码 FROM sales WHERE buyer_phone = 13900000001",
+                 "SELECT COUNT(*) AS 笔数, 已遮罩 AS 号码 FROM sales WHERE buyer_phone = 已遮罩", id="选出列里的字面量"),
+    pytest.param(SALES, _COUNT + "buyer_phone >= '13900000001' AND buyer_phone <= '13900000001'",
+                 _COUNT + "buyer_phone >= '已遮罩' AND buyer_phone <= '已遮罩'", id="字符串范围两头相同"),
+    pytest.param(SALES, _COUNT + "buyer_phone BETWEEN '13900000001' AND '13900000001'",
+                 _COUNT + "buyer_phone BETWEEN '已遮罩' AND '已遮罩'", id="字符串 BETWEEN 两头相同"),
+])
+async def test_a_customer_looked_up_in_any_shape_never_shows(monkeypatch, tmp_path, client, script, sql, sealed):
+    """agent 换着写法找某一个客户（简单 CASE、IN (VALUES …)、两头相同的范围、行值、选出列里写着同一个号码……），回答里
+    复述了号码：回答正文、封存的 SQL、事件、回答文档、质询工件里都不出现这个号码。报告取数选了手机号，号码是从报告
+    自己的查询快照里认出来的。"""
+    name, source_id = await _source(tmp_path, options={"mask_columns": "buyer_phone"}, script=script)
+    plays = {"这位客户": Play([sql], "客户 13900000001 共下了 [[v:Q1.r0.笔数]] 笔。[[see:Q1]]")}
+    Models(monkeypatch, name, writer="华东首单金额 [[v:Q1.r0.金额]]。[[see:Q1]]", plays=plays)
+    try:
+        paused = await pause(name, graph=fetching(name, WITH_PHONE))
+        [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
+        item = r.json()["inquiry"]
+        assert (item["answer"]["ok"], item["answer"]["text"]) == (True, "客户 已遮罩 共下了 1 笔。")
+        [query] = item["queries"]
+        assert query["sql"] == sealed
+        [event] = await events(paused.id, "approval.inquiry")
+        got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}")
+        assert got.status_code == 200, got.text
+        assert got.json()["content"]["sql"] == sealed, "质询工件的 SQL 和事件里的一字不差"
+        doc = artifact_store.load(item["answer"]["doc_artifact"])
+        for shown in (r.text, json.dumps(event.data, ensure_ascii=False), json.dumps(doc, ensure_ascii=False), got.text):
+            assert "1390000000" not in shown
+        if "号码" in sql:
+            assert got.json()["content"]["rows"] == [[1, "已遮罩"]] and got.json()["masked"] == ["号码"]
+    finally:
+        await engines.invalidate(source_id)
+
+
 # --------------------------------------------------------------------------
 # 接线：两道审批的竞态、上传表格的版本、设置、开放之后状态又变了、张冠李戴的报告文档
 # --------------------------------------------------------------------------

backend/tests/test_inquiry_masking.py

@@ -4,11 +4,14 @@
   照样遮；COUNT 只数个数、窗口的分区键只决定次序，不遮；认不出落在哪一列的（两个 * 之间、子查询里没起别名的表达式）、
   解析不了又写着遮罩列的，整份快照遮掉
 - 按值认的遮罩值有门槛：纯数字够长（SECRET_DIGITS 位）、文字至少两个字；日期、小数、千分位里的一截不算；提问人自己
-  写进问题、选中原文里的，报告列出的指标和运行输入的值，SQL 里照原样留着的阈值，都不换
-- SQL 只换在找某一个值的字面量（等值、IN、LIKE，带前缀的字符串也算）和注释、名字里的遮罩值：和遮罩列比较的字面量是
-  遮罩列出现过的值就换，短的数也换；范围比较的边界、BETWEEN、LIMIT、日期、别的列的范围条件照原样
+  写进问题、选中原文里的，报告列出的指标和运行输入的值，都不换；SQL 里照原样留着的阈值、行数只在回答正文里不换，
+  格子里的同一个数照旧换
+- SQL 里数字照原样留着的只有范围比较的边界（没把某一个值夹住的）和 LIMIT / OFFSET 的行数；在找某一个值的（等值、IN、
+  LIKE、简单 CASE、IS [NOT] DISTINCT FROM，套了括号、写成行值、VALUES、ARRAY 的也算，还有 BETWEEN v AND v、
+  >= v AND <= v 这类夹住某一个值的范围）和遮罩列比较时短的数也换；选出列、函数参数、算式里够长的遮罩值也换；字符串
+  （范围边界也一样）、注释、名字按值换；SQL 解析不了时认不准哪个是阈值，一律不免
 - 快照按列遮，另按值遮：别的列里写着遮罩值的格子也换成「已遮罩」；列名里写着遮罩值的，列名也换
-- 组装回答之后正文里还认得出遮罩值的，不交出去
+- 组装回答之后还认得出遮罩值的（正文里复述的阈值除外），不交出去
 - 遮罩值上万个时换一遍也不卡：不现编上万个分支的正则
 
 数据都是虚构的：手机号、姓名、年龄、区域、金额。
@@ -213,9 +216,9 @@ def people() -> inquiry_mod._Secrets:
     ("SELECT region FROM sales WHERE buyer_phone NOT IN ('13900000009') AND age <> 40",
      "SELECT region FROM sales WHERE buyer_phone NOT IN ('13900000009') AND age <> 40"),
     ("SELECT amount FROM sales WHERE amount = 13900000001", "SELECT amount FROM sales WHERE amount = 已遮罩"),
-    # 和遮罩列做范围比较：边界是阈值，不是在找某个人，照原样
-    ("SELECT COUNT(*) FROM sales WHERE age > 31 AND age <= 25 AND 25 < age AND age NOT BETWEEN 25 AND 31",
-     "SELECT COUNT(*) FROM sales WHERE age > 31 AND age <= 25 AND 25 < age AND age NOT BETWEEN 25 AND 31"),
+    # 和遮罩列做范围比较：边界是阈值，不是在找某个人，照原样（同一个数只从一边隔开：25 的几处都是「不超过 25」那一侧）
+    ("SELECT COUNT(*) FROM sales WHERE age > 31 AND age <= 25 AND 25 < age AND age NOT BETWEEN 20 AND 25",
+     "SELECT COUNT(*) FROM sales WHERE age > 31 AND age <= 25 AND 25 < age AND age NOT BETWEEN 20 AND 25"),
     # 不等、SQLite 的 IS、NOT LIKE 也是在找某一个值
     ("SELECT COUNT(*) FROM sales WHERE age != 25 AND age IS NOT 31 AND buyer_phone NOT LIKE '13900000001'",
      "SELECT COUNT(*) FROM sales WHERE age != 已遮罩 AND age IS NOT 已遮罩 AND buyer_phone NOT LIKE '已遮罩'"),
@@ -231,20 +234,160 @@ def amounts() -> inquiry_mod._Secrets:
 
 
 @pytest.mark.parametrize("sql, shown", [
-    # 范围比较、LIMIT / OFFSET、选出列里的数：长的数也照原样
+    # 范围比较、LIMIT / OFFSET：长的数也照原样
     ("SELECT COUNT(*) FROM sales WHERE visits > 20000 AND amount >= 30000 AND amount BETWEEN 10000 AND 50000 "
      "LIMIT 10000 OFFSET 12000",
      "SELECT COUNT(*) FROM sales WHERE visits > 20000 AND amount >= 30000 AND amount BETWEEN 10000 AND 50000 "
      "LIMIT 10000 OFFSET 12000"),
-    ("SELECT 20000 AS 门槛, amount - 10000 AS 差额 FROM sales", "SELECT 20000 AS 门槛, amount - 10000 AS 差额 FROM sales"),
+    # 按同一个数分成两段（不低于、低于）：只从一边隔开，是阈值
+    ("SELECT SUM(amount >= 30000) AS 大单, SUM(amount < 30000) AS 小单 FROM sales",
+     "SELECT SUM(amount >= 30000) AS 大单, SUM(amount < 30000) AS 小单 FROM sales"),
+    # 选出列、算式、函数参数里的数：认不准是不是在找某一个值，够长的遮罩值换（短的数照原样）
+    ("SELECT 20000 AS 门槛, amount - 10000 AS 差额, ROUND(amount, 2) FROM sales",
+     "SELECT 已遮罩 AS 门槛, amount - 已遮罩 AS 差额, ROUND(amount, 2) FROM sales"),
+    ("SELECT COUNT(*) FROM sales WHERE amount - 30000 >= 0", "SELECT COUNT(*) FROM sales WHERE amount - 已遮罩 >= 0"),
     # 等值、IN：和遮罩列比较的换，和别的列比较的、够长的遮罩值也换
     ("SELECT COUNT(*) FROM sales WHERE amount = 30000 OR amount IN (10000, -12000) OR visits = 20000",
      "SELECT COUNT(*) FROM sales WHERE amount = 已遮罩 OR amount IN (已遮罩, -已遮罩) OR visits = 已遮罩"),
 ])
-def test_long_masked_numbers_are_only_masked_where_a_value_is_looked_up(amounts, sql, shown):
+def test_long_masked_numbers_are_kept_only_as_thresholds_and_row_counts(amounts, sql, shown):
     assert inquiry_mod._scrub_sql(sql, amounts, {"amount"}, "sqlite") == shown
 
 
+@pytest.fixture
+def phones() -> inquiry_mod._Secrets:
+    """手机号存成整数（遮罩值在 SQL 里写成数字），年龄是短的数。"""
+    return secrets_of({"columns": ["buyer_phone", "age"], "rows": [[13900000001, 25], [13900000002, 31]]},
+                      {"buyer_phone", "age"})
+
+
+@pytest.mark.parametrize("kind, sql, shown", [
+    ("sqlite", "SELECT SUM(CASE buyer_phone WHEN 13900000001 THEN 1 ELSE 0 END) AS 笔数 FROM sales",
+     "SELECT SUM(CASE buyer_phone WHEN 已遮罩 THEN 1 ELSE 0 END) AS 笔数 FROM sales"),
+    ("sqlite", "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone IN (VALUES (13900000001), (13900000002))",
+     "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone IN (VALUES (已遮罩), (已遮罩))"),
+    ("sqlite", "SELECT COUNT(*) AS 笔数 FROM sales WHERE (buyer_phone, region) = (13900000001, '华东')",
+     "SELECT COUNT(*) AS 笔数 FROM sales WHERE (buyer_phone, region) = (已遮罩, '华东')"),
+    ("sqlite", "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone IS NOT DISTINCT FROM 13900000001",
+     "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone IS NOT DISTINCT FROM 已遮罩"),
+    ("sqlite", "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = (13900000001)",
+     "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = (已遮罩)"),
+    # 选出列里的同一个号码：认不准是不是在找某一个值，够长的遮罩值照样换，不留下「13900000001 AS 号码 … = 已遮罩」
+    ("sqlite", "SELECT COUNT(*) AS 笔数, 13900000001 AS 号码 FROM sales WHERE buyer_phone = 13900000001",
+     "SELECT COUNT(*) AS 笔数, 已遮罩 AS 号码 FROM sales WHERE buyer_phone = 已遮罩"),
+    ("sqlite", "SELECT COUNT(*) FROM sales WHERE COALESCE(buyer_phone, 0) = 13900000001",
+     "SELECT COUNT(*) FROM sales WHERE COALESCE(buyer_phone, 0) = 已遮罩"),
+    # 短的数：认得出被比较的是遮罩列才换
+    ("sqlite", "SELECT COUNT(*) FROM sales WHERE age = (25) OR (age, region) = (31, '华东') OR age IN (VALUES (25)) "
+     "OR age IS NOT DISTINCT FROM 31 OR age IS DISTINCT FROM 25",
+     "SELECT COUNT(*) FROM sales WHERE age = (已遮罩) OR (age, region) = (已遮罩, '华东') OR age IN (VALUES (已遮罩)) "
+     "OR age IS NOT DISTINCT FROM 已遮罩 OR age IS DISTINCT FROM 已遮罩"),
+    ("sqlite", "SELECT SUM(CASE age WHEN 25 THEN 1 END) AS 人数, SUM(CASE WHEN age = 31 THEN 1 END) FROM sales",
+     "SELECT SUM(CASE age WHEN 已遮罩 THEN 1 END) AS 人数, SUM(CASE WHEN age = 已遮罩 THEN 1 END) FROM sales"),
+    ("postgresql", "SELECT COUNT(*) FROM sales WHERE buyer_phone = ANY(ARRAY[13900000001, 13900000002]) "
+     "AND age <> ALL(ARRAY[25]) AND age = ANY(VALUES (31))",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone = ANY(ARRAY[已遮罩, 已遮罩]) "
+     "AND age <> ALL(ARRAY[已遮罩]) AND age = ANY(VALUES (已遮罩))"),
+])
+def test_a_value_looked_up_in_any_shape_is_masked_in_the_sql(phones, kind, sql, shown):
+    """在找某一个值的写法不止「col = 值」：简单 CASE、IN (VALUES …)、行值、IS NOT DISTINCT FROM、套了括号、
+    = ANY(ARRAY[…])，都按语法树认出来；THEN 后面的 1、0 照原样。"""
+    assert inquiry_mod._scrub_sql(sql, phones, {"buyer_phone", "age"}, kind) == shown
+
+
+@pytest.mark.parametrize("sql, shown", [
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 13900000001 AND 13900000001",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 已遮罩 AND 已遮罩"),
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone >= 13900000001 AND buyer_phone <= 13900000001",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone >= 已遮罩 AND buyer_phone <= 已遮罩"),
+    ("SELECT COUNT(*) FROM sales WHERE 13900000001 <= buyer_phone AND buyer_phone <= 13900000001",
+     "SELECT COUNT(*) FROM sales WHERE 已遮罩 <= buyer_phone AND buyer_phone <= 已遮罩"),
+    # 大于或小于就是不等于
+    ("SELECT COUNT(*) FROM sales WHERE NOT (buyer_phone > 13900000001 OR buyer_phone < 13900000001)",
+     "SELECT COUNT(*) FROM sales WHERE NOT (buyer_phone > 已遮罩 OR buyer_phone < 已遮罩)"),
+    # 整数的 < v + 1 就是 <= v
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone >= 13900000001 AND buyer_phone < 13900000002",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone >= 已遮罩 AND buyer_phone < 已遮罩"),
+    ("SELECT COUNT(*) FROM sales WHERE age >= 25 AND age <= 25",
+     "SELECT COUNT(*) FROM sales WHERE age >= 已遮罩 AND age <= 已遮罩"),
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone >= '13900000001' AND buyer_phone <= '13900000001'",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone >= '已遮罩' AND buyer_phone <= '已遮罩'"),
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN '13900000001' AND '13900000001'",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN '已遮罩' AND '已遮罩'"),
+    # 一头是几个数算出来的（13900000002 - 1 就是 13900000001）：照样夹住了
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 13900000001 AND 13900000002 - 1",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 已遮罩 AND 已遮罩 - 1"),
+    # 同一个数在算式里被换掉了，边界上的也不留：留着就把算式里换掉的露了
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 13900000001 AND 13900000001 + 0",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone BETWEEN 已遮罩 AND 已遮罩 + 0"),
+])
+def test_a_range_that_pins_a_single_value_is_a_lookup(phones, sql, shown):
+    """两头是同一个值的范围（BETWEEN v AND v、>= v AND <= v）、大于或小于同一个值、整数的半开区间，都只夹住了一个值，
+    是在找它，和等值比较一样换。"""
+    assert inquiry_mod._scrub_sql(sql, phones, {"buyer_phone", "age"}, "sqlite") == shown
+
+
+@pytest.mark.parametrize("sql, shown", [
+    ("SELECT COUNT(*) AS 笔数, 13900000001 AS 号码 FROM sales WHERE buyer_phone >= 13900000001",
+     "SELECT COUNT(*) AS 笔数, 已遮罩 AS 号码 FROM sales WHERE buyer_phone >= 已遮罩"),
+    ("SELECT COUNT(*) FROM sales WHERE buyer_phone >= 13900000001 AND buyer_phone - 13900000001 < 5",
+     "SELECT COUNT(*) FROM sales WHERE buyer_phone >= 已遮罩 AND buyer_phone - 已遮罩 < 5"),
+])
+def test_a_number_masked_elsewhere_in_the_sql_is_not_kept_as_a_threshold(phones, sql, shown):
+    """同一个号码在选出列、算式里被换掉了，条件里的单边范围也不留：留着就把那里换掉的露了。"""
+    assert inquiry_mod._scrub_sql(sql, phones, {"buyer_phone", "age"}, "sqlite") == shown
+    assert inquiry_mod._shown_literals([(sql, "sqlite")]) <= {"5"}
+
+
+def test_a_threshold_named_in_an_alias_or_a_comment_stays(amounts):
+    """agent 把阈值写进别名、注释（AS 超过30000的笔数）：名字、注释照常按值换，阈值本身照原样，回答正文里复述也不遮。"""
+    sql = "SELECT SUM(amount >= 30000) AS 超过30000的笔数 FROM sales -- 30000 以上算大单"
+    assert inquiry_mod._scrub_sql(sql, amounts, {"amount"}, "sqlite") == (
+        "SELECT SUM(amount >= 30000) AS 超过已遮罩的笔数 FROM sales -- 已遮罩 以上算大单")
+    assert inquiry_mod._shown_literals([(sql, "sqlite")]) == {"30000"}
+
+
+def test_sql_starting_with_blank_lines_is_read_at_the_right_places():
+    """agent 写的 SQL 常以换行开头：解析时去掉了开头的空白，按语法树认出的位置要补回来，才对得上原文里的词——阈值、
+    行数照原样，找的那个号码换掉。"""
+    secrets = secrets_of({"columns": ["buyer_phone", "amount"], "rows": [[13900000001, 30000]]}, {"buyer_phone", "amount"})
+    sql = "\n  SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND buyer_phone IN (13900000001, 1) LIMIT 30000"
+    assert inquiry_mod._scrub_sql(sql, secrets, {"buyer_phone", "amount"}, "sqlite") == (
+        "\n  SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND buyer_phone IN (已遮罩, 1) LIMIT 30000")
+    assert inquiry_mod._shown_literals([(sql, "sqlite")]) == {"30000"}
+
+
+def test_string_range_bounds_are_masked_by_value(people):
+    """字符串的范围边界按值换：写成字符串的长号码做边界，多半是在找那一个人；日期边界不是遮罩值，照原样。"""
+    sql = "SELECT COUNT(*) FROM sales WHERE buyer_phone >= '13900000001' AND buyer_phone < '13900000003' " \
+          "AND paid_on >= '2026-01-25'"
+    assert inquiry_mod._scrub_sql(sql, people, {"buyer_phone", "age"}, "sqlite") == (
+        "SELECT COUNT(*) FROM sales WHERE buyer_phone >= '已遮罩' AND buyer_phone < '已遮罩' "
+        "AND paid_on >= '2026-01-25'")
+
+
+def test_sql_that_cannot_be_parsed_keeps_no_thresholds():
+    """SQL 解析不了：认不准哪个数是阈值，一律不免——够长的遮罩值在哪都换，简单的等值比较照样按词认（短的数也换）。"""
+    secrets = secrets_of({"columns": ["age", "amount"], "rows": [[25, 30000], [31, 10000]]}, {"age", "amount"})
+    sql = "SELECT COUNT(*) FROM sales WHERE age = 25 AND amount >= 30000 AND visits > 31 LIMIT 10000 GROUP"
+    assert inquiry_mod._scrub_sql(sql, secrets, {"age", "amount"}, "sqlite") == (
+        "SELECT COUNT(*) FROM sales WHERE age = 已遮罩 AND amount >= 已遮罩 AND visits > 31 LIMIT 已遮罩 GROUP")
+    assert inquiry_mod._shown_literals([(sql, "sqlite")]) == set()
+
+
+def test_only_thresholds_and_row_counts_count_as_shown():
+    """照原样留在 SQL 里的只有范围边界（没夹住某一个值的）和 LIMIT / OFFSET 的行数；选出列、函数参数、CASE、夹住
+    某一个值的范围里的数不算，在别处（这一条或别的查询里）被换掉的数也不算。"""
+    shown = inquiry_mod._shown_literals
+    assert shown([("SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND visits > 20000 LIMIT 10000 OFFSET 12000",
+                   "sqlite")]) == {"30000", "20000", "10000", "12000"}
+    assert shown([("SELECT 50000 AS 门槛, SUM(CASE amount WHEN 12000 THEN 1 END), ROUND(amount, 10000) FROM sales "
+                   "WHERE amount BETWEEN 20000 AND 20000", "sqlite")]) == set()
+    assert shown([("SELECT COUNT(*) FROM sales WHERE amount >= 30000", "sqlite"),
+                  ("SELECT COUNT(*) FROM sales WHERE amount = 30000", "sqlite")]) == set()
+    assert shown([("SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND note LIKE '%30000%'", "sqlite")]) == set()
+
+
 @pytest.mark.parametrize("kind, sql, shown", [
     ("oracle", "SELECT COUNT(*) FROM sales WHERE buyer_phone = N'13900000001'",
      "SELECT COUNT(*) FROM sales WHERE buyer_phone = N'已遮罩'"),
@@ -272,9 +415,9 @@ def test_numbers_left_as_written_in_the_sql_are_not_masked_in_the_answer(amounts
     """范围比较的边界、LIMIT 照原样留在 SQL 里，回答正文里同样的数再遮也遮不住什么，只会和 SQL 对不上；等值比较
     里换掉的照旧遮。"""
     sql = "SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND amount <> 50000 LIMIT 10000"
-    assert inquiry_mod._shown_literals(sql, "sqlite") == ["30000", "10000"]
-    amounts.protect(inquiry_mod._shown_literals(sql, "sqlite"), strong_only=True)
-    assert amounts.scrub("金额不低于 30000 元、不是 50000 元的，取前 10000 条") == (
+    assert inquiry_mod._shown_literals([(sql, "sqlite")]) == {"30000", "10000"}
+    amounts.show(inquiry_mod._shown_literals([(sql, "sqlite")]))
+    assert amounts.scrub("金额不低于 30000 元、不是 50000 元的，取前 10000 条", prose=True) == (
         "金额不低于 30000 元、不是 已遮罩 元的，取前 10000 条")
     assert inquiry_mod._scrub_sql(sql, amounts, {"amount"}, "sqlite") == (
         "SELECT COUNT(*) FROM sales WHERE amount >= 30000 AND amount <> 已遮罩 LIMIT 10000")
@@ -283,6 +426,35 @@ def test_numbers_left_as_written_in_the_sql_are_not_masked_in_the_answer(amounts
         "只免了正文里的：和遮罩列做等值比较的照旧换"
 
 
+def test_a_threshold_left_in_the_sql_is_shown_in_the_prose_but_not_in_the_cells(amounts):
+    """SQL 里照原样留着的阈值只在回答正文里不遮：格子里的同一个数照旧按值遮（col >= v 再取第一行，查出来的可能就是
+    那一笔），列名、SQL 里的名字也照旧换。组装出来正文里的不算露出，格子里渲染出来的算。"""
+    sql = "SELECT region AS 区域, visits AS 访问 FROM sales WHERE visits >= 20000 ORDER BY visits LIMIT 1"
+    amounts.show(inquiry_mod._shown_literals([(sql, "sqlite")]))
+    assert amounts.scrub("访问不少于 20000 次", prose=True) == "访问不少于 20000 次"
+    assert amounts.scrub("访问不少于 20000 次") == "访问不少于 已遮罩 次"
+    snapshot = {"columns": ["区域", "访问"], "rows": [["华东", 20000]]}
+    view, masked = inquiry_mod._masked_view(snapshot, set(), amounts)
+    assert view["rows"] == [["华东", MASKED]] and masked == ["访问"]
+    doc = inquiry_mod._compose("访问不少于 20000 次的区域里最少的是 [[v:Q1.r0.区域]]，访问 [[v:Q1.r0.访问]] 次。",
+                               {"Q1": query_entry(["区域", "访问"])}, run_id="r1", views={"a1": view},
+                               secrets=amounts, allow=[])
+    assert doc["markdown"] == "访问不少于 20000 次的区域里最少的是 华东，访问 已遮罩 次。"
+    # 一格文字里夹着这个数：按整格比认不出来，渲染出来才露，照旧不交出去
+    noted = {"columns": ["备注"], "rows": [["回访 20000 次"]]}
+    views = {"a1": inquiry_mod._masked_view(noted, set(), amounts)[0]}
+    with pytest.raises(inquiry_mod._Leak):
+        inquiry_mod._compose("备注是 [[v:Q1.r0.备注]]。", {"Q1": query_entry(["备注"])}, run_id="r1", views=views,
+                             secrets=amounts, allow=[])
+    # 引用的列名里写着这个数：列名在快照、SQL 里都照旧换，标记里的列名换得一样，引用照样解析得了
+    named = {"columns": ["超过20000次"], "rows": [[3]]}
+    shown, _ = inquiry_mod._masked_view(named, set(), amounts)
+    assert shown["columns"] == ["超过已遮罩次"]
+    doc = inquiry_mod._compose("访问超过 20000 次的有 [[v:Q1.r0.超过20000次]] 笔。", {"Q1": query_entry(named["columns"])},
+                               run_id="r1", views={"a1": shown}, secrets=amounts, allow=[])
+    assert doc["markdown"] == "访问超过 20000 次的有 3 笔。"
+
+
 def test_a_comparison_with_a_renamed_masked_column_is_masked_too(people):
     sql = "SELECT * FROM (SELECT buyer_phone AS p FROM sales) t WHERE p = '13900000003' AND 1 = 1"
     assert inquiry_mod._scrub_sql(sql, people, {"buyer_phone", "p"}, "sqlite") == (