mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
M2 复核第 4 轮：报告自查认遮罩列和证据面板同一个口径，哪张表（含视图）里叫这个名字的列都算；结论句裁判的摘录按快照逐份遮确定取自遮罩列的列；没起别名、盖住全部结果列的表达式算确定取自遮罩列，值照收
第 3 轮的「遮罩表」推断和退回的裁判摘录各开了一个口子，收值时又把按位置认得出的列丢了： - 报告自查去掉「遮罩表」推断（_mask_owners）。数据源的 mask_columns 只认列名、不支持写成「表.列」，按列名在整个数据源上 生效：引用的那一格按 SQL 认出取自任何一张表里叫这个名字的列就记 masked_cell，视图（customer_spend.name）、没有第二个 遮罩列的表（employees.name）、门店名（stores.name）都算，和证据面板的遮法一致。以前经视图引用客户姓名、引用店员姓名， 正式运行照常成功，姓名写进了报告。SQL 血缘只用来点名出处，和排除值确实取自普通表别的列的（SELECT region AS phone FROM stores）；视图、表结构里没有的表、写了别的库名的表、没标 is_view 的老表结构、没有冻结表结构的快照看不出列的出处， 照算（_frozen_schema 只把 is_view 明写为 false 的表当普通表）。违规说明照实写取自哪张表的哪一列：「列「store」按 SQL 认出取自 stores.name，数据源对列名「name」设置了遮罩（按列名对所有表生效）」 - 交给结论句裁判的摘录按快照逐份遮（api/evidence.py 的 _judge_loader）：每份快照里确定取自遮罩列的列（_mask_sets 的 sure）并进交给 judge.prepare 的那一份快照的 mask_columns（工件本身不动），裁判照它自己的办法遮掉，judge.py 不用改。 c.phone AS tel 的号码不再进摘录；不按数据源名合并（另一份快照里同名、取自门店热线的列照给），不用宁可多遮的 hidden （整份遮的快照、只数个数的列照给）。按需裁判和报告撰写节点里当场裁判同一个函数。摘录变了的句子复用键随之变，此前 判过的会重判一次 - 收「这次运行查到过的遮罩值」：_sure 以前凡是要遮的列盖住了全部结果列、名字又对不上，就当认不准、整份算， SELECT UPPER(phone) FROM customers 一个值都收不到，证据面板之后照原样显示 SQL 里写着的号码。现在结果后面垫一列再认一遍 （_whole），垫的那一列也算进来的才是整份算的；按位置认得出的照收，整份遮的照旧只遮显示、不收值 测试：test_report_masked_owners.py 改名 test_report_masked_any_table.py 并按新口径改写（经视图改名 / 不改名、店员、门店名、 客户名真跑判失败且成果里没有值；* 展开、WITH、UNION、视图里改成遮罩列名字的列、没有表结构、没标 is_view、表结构里没有的 表、写了库名的表）；新增 test_judge_excerpt_masks.py（按需、当场两条路的摘录都不给改了名字的号码列，同名的门店热线照给； 没起别名的 UPPER(phone) 摘录也不给）；test_evidence_mask_memo.py 加没起别名的表达式照收值；违规说明的断言跟着改
This commit is contained in:
ThearchyHelios
+146
-58
@@ -31,13 +31,15 @@ options.mask_columns 里的列换成「已遮罩」，SQL 里给它们起了别
|
||||
|
||||
结论句的判定（四期）是模型给的，不是证据：正式运行的判定在报告文档里、随文档封存；探索运行
|
||||
按需裁判（POST …/evidence/judge），判定作为 evidence.judged 事件追加在封存之后，只填补没判过的
|
||||
句子。交给裁判的摘录同样只取封存范围内的工件。
|
||||
句子。交给裁判的摘录同样只取封存范围内的工件，每份查询快照里确定取自遮罩列的列不给（_judge_loader，报告撰写节点
|
||||
当场裁判同一个函数）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import csv
|
||||
import functools
|
||||
import io
|
||||
import logging
|
||||
import re
|
||||
@@ -1474,9 +1476,10 @@ class _Masks:
|
||||
|
||||
def _snapshot_secrets(snap: dict[str, Any], sets: _MaskSets) -> Any:
|
||||
"""一份查询快照里确定取自遮罩列的值（masking._Secrets，一个都没有为 None）：只收 sets.sure 那几列（列名就是遮罩列的、
|
||||
SQL 里认得出取自遮罩列的），只装着分档标签的格子不收，和核查 agent 在工具边界收值同一个口径（masking：整份遮的快照只收
|
||||
确定的那几列）。SQL 认不准、整份遮的快照只遮显示：以前连它的区域名、金额一起收，之后每条 SQL 里的「'华东'」都显示成
|
||||
「已遮罩」。证据面板（_Masks.run_secrets）、推演（api/whatif.py 的 _Hidden.secrets）都按它收。"""
|
||||
SQL 里认得出取自遮罩列的，没起别名、按位置认得出的 SELECT UPPER(phone) 也算），只装着分档标签的格子不收，和核查 agent
|
||||
在工具边界收值同一个口径（masking：认得清的照收，整份遮的快照只收确定的那几列）。SQL 认不准、整份遮的快照只遮显示：
|
||||
以前连它的区域名、金额一起收，之后每条 SQL 里的「'华东'」都显示成「已遮罩」。证据面板（_Masks.run_secrets）、推演
|
||||
（api/whatif.py 的 _Hidden.secrets）都按它收。"""
|
||||
from app.engine.masking import _Secrets
|
||||
|
||||
secrets = _Secrets()
|
||||
@@ -1612,9 +1615,10 @@ class _MaskSets:
|
||||
"""一份查询快照的遮罩（_mask_sets）：
|
||||
|
||||
- hidden：要遮的列（小写）。证据面板、推演、勾行按它遮，宁可多遮：SQL 认不准时整份的列都算
|
||||
- sure：其中能确定取自遮罩列的：列名就是遮罩列的，SQL 里认得出取自遮罩列的（别名、函数、子查询、WITH）。SQL 认不准、
|
||||
整份算进来的那几列不在里面，只装着分档标签的列（CASE 写死的字）也不在。收「这次运行查到过的遮罩值」只收这几列
|
||||
（_snapshot_secrets）；报告自查从这几列起，再按血缘去掉别的表里同名的列（_cited_masks）
|
||||
- sure：其中能确定取自遮罩列的：列名就是遮罩列的，SQL 里认得出取自遮罩列的（别名、函数、子查询、WITH，没起别名
|
||||
却按位置认得出的表达式）。SQL 认不准、整份算进来的那几列不在里面，只装着分档标签的列（CASE 写死的字）也不在。
|
||||
收「这次运行查到过的遮罩值」只收这几列（_snapshot_secrets），交给结论句裁判的摘录逐份遮这几列（_judge_loader），
|
||||
报告自查从这几列起（_cited_masks，只排除血缘证明值取自普通表别的列的）
|
||||
- base：遮罩列本身（查询当时记下的加数据源现在设的，小写）
|
||||
- labels：只装着分档标签的列 → 标签的写法（masking._snapshot_labels），收遮罩值时不收标签
|
||||
- gone：已经不在的数据源
|
||||
@@ -1655,12 +1659,15 @@ def _derived(snap: dict[str, Any], hidden: set[str], kind: str | None,
|
||||
|
||||
def _sure(snap: dict[str, Any], base: set[str], kind: str | None,
|
||||
tables: dict[str, list[str]] | None = None) -> tuple[set[str], dict[str, set[str]]]:
|
||||
"""(能确定取自遮罩列的结果列, 只装着分档标签的列)，都是小写，报告自查按前一项判违规（_MaskSets.sure）。
|
||||
"""(能确定取自遮罩列的结果列, 只装着分档标签的列)，都是小写（_MaskSets.sure）：收遮罩值、交给裁判的摘录、报告自查
|
||||
都从前一项起。
|
||||
|
||||
按数据源自己的方言认（masking._derived_hidden，交回要遮的列和 SQL 里指代遮罩列的名字）：结果列名就是遮罩列、或者是
|
||||
SQL 里指代遮罩列的名字（c.phone AS tel 的 tel、UPPER(phone) AS p 的 p）的算。认不准时那一套把整份的列都算上（派生
|
||||
来源里没起别名的表达式、PIVOT、SQL 解析不了又写着遮罩列……）：要遮的列盖住了全部结果列、名字却对不上这么多的，只认
|
||||
名字对上的那几列——展示上多遮无妨，自查多算就是判失败，写作者也改不掉（最大的一笔订单金额被说成取自手机号）。
|
||||
按数据源自己的方言认（masking._derived_hidden，交回要遮的列和 SQL 里指代遮罩列的名字）：结果列名就是遮罩列、SQL 里
|
||||
指代遮罩列的名字（c.phone AS tel 的 tel、UPPER(phone) AS p 的 p）、没起别名却按位置认得出取自遮罩列的
|
||||
（SELECT UPPER(phone) FROM customers）都算。认不准时那一套把整份的列都算上（派生来源里没起别名的表达式、PIVOT、SQL
|
||||
解析不了又写着遮罩列……）：这样整份算进来的（_whole），只认名字对上的那几列——展示上多遮无妨，这里多算，自查就判
|
||||
失败、写作者也改不掉（最大的一笔订单金额被说成取自手机号），区域名、金额也被收成遮罩值。以前凡是要遮的列盖住了全部
|
||||
结果列、名字又对不上的都当整份算，按位置认得出的那一列也丢了，一个值都收不到（复核第 4 轮）。
|
||||
只装着分档标签的列（CASE WHEN phone IS NULL THEN '未留' ELSE '已留' END）格子里是 SQL 写死的字，不算；只数个数的
|
||||
（COUNT(*) FILTER (WHERE phone IS NULL)、SUM(phone IS NULL)）先改写成那一套认得出的样子再认（_counted）。
|
||||
出了意外只认列名对上的。"""
|
||||
@@ -1671,10 +1678,12 @@ def _sure(snap: dict[str, Any], base: set[str], kind: str | None,
|
||||
try:
|
||||
from app.engine.masking import _derived_hidden, _snapshot_labels
|
||||
|
||||
found, names = _derived_hidden(_counted(snap, kind), set(base), kind, tables)
|
||||
counted = _counted(snap, kind)
|
||||
found, names = _derived_hidden(counted, set(base), kind, tables)
|
||||
found &= columns
|
||||
named = (names & columns) | direct
|
||||
sure = named if columns <= found and not columns <= named else found | direct
|
||||
whole = columns <= found and not columns <= named and _whole(counted, base, kind, tables)
|
||||
sure = named if whole else found | direct
|
||||
labels = _snapshot_labels(snap, kind)
|
||||
return sure - {name for name in labels if name not in base}, labels
|
||||
except Exception: # noqa: BLE001 - 认不准：只认列名对上的
|
||||
@@ -1682,6 +1691,21 @@ def _sure(snap: dict[str, Any], base: set[str], kind: str | None,
|
||||
return direct, {}
|
||||
|
||||
|
||||
#: _whole 垫在结果列后面的一列：SQL 里的名字不会是它
|
||||
_PAD_COLUMN = "\x00pad"
|
||||
|
||||
|
||||
def _whole(snap: dict[str, Any], base: set[str], kind: str | None, tables: dict[str, list[str]] | None) -> bool:
|
||||
"""要遮的列盖住了全部结果列时，是认不准、整份算进来的，还是每一列都真认得出取自遮罩列（_sure）。masking 那一套
|
||||
认不准时交回全部的列，和每一列都按位置认出来的样子一样，所以结果后面垫一列再认一遍：认得出的列按 SQL 里的位置算，
|
||||
垫的那一列没有哪一项落得上；整份算的连它一起算进来。"""
|
||||
from app.engine.masking import _derived_hidden
|
||||
|
||||
padded = [*(str(c) for c in snap.get("columns") or []), _PAD_COLUMN]
|
||||
found, _ = _derived_hidden({**snap, "columns": padded}, set(base), kind, tables)
|
||||
return {c.lower() for c in padded} <= found
|
||||
|
||||
|
||||
def _counted(snap: dict[str, Any], kind: str | None) -> dict[str, Any]:
|
||||
"""报告自查认确定取自遮罩列的列之前（_sure），把只决定算不算进来的写法改写成认遮罩列的那一套认得出的样子：
|
||||
|
||||
@@ -1753,11 +1777,12 @@ async def _mask_sets(snap: dict[str, Any] | None, source: Any, masks: _Masks, *,
|
||||
async def _cited_masks(snap: dict[str, Any], source: Any, masks: _Masks,
|
||||
artifact: str | None = None) -> dict[str, list[str]]:
|
||||
"""报告自查（engine/nodes/report.py 的 _MaskedCells）认一份查询快照里引用了就记 masked_cell 的列：{结果列（小写）:
|
||||
它取自的遮罩列}，违规说明点名后者。
|
||||
它的出处}，违规说明点名出处（认得出表的写「表.列」，见 _owned_columns）。
|
||||
|
||||
从 _mask_sets 的 sure 起（列名就是遮罩列的、SQL 里认得出取自遮罩列的；SQL 认不准、整份算进来的和只装着分档标签的
|
||||
不在里面），普通查询再按血缘核一遍值取自哪张表的哪一列，只认取自遮罩表里那一列的（_owned_columns：stores.name 和设了
|
||||
遮罩的 customers.name 同名不算）。合并查询的值来自它的各个输入、不是数据源里的表，照 sure 算。"""
|
||||
不在里面）。遮罩按列名在整个数据源上生效，和证据面板一样：哪张表（含视图）里叫这个名字的列都算，门店名（stores.name）
|
||||
和客户姓名（customers.name）同叫 name、都设了遮罩时两者都算。普通查询再按血缘点名出处（_owned_columns）；合并查询的值
|
||||
来自它的各个输入、不是数据源里的表，照 sure 算。"""
|
||||
sets = await _mask_sets(snap, source, masks, artifact=artifact)
|
||||
if not sets.sure:
|
||||
return {}
|
||||
@@ -1767,59 +1792,72 @@ async def _cited_masks(snap: dict[str, Any], source: Any, masks: _Masks,
|
||||
|
||||
|
||||
def _owned_columns(snap: dict[str, Any], sure: set[str], base: set[str], kind: str | None) -> dict[str, list[str]]:
|
||||
"""sure 里值确实取自遮罩表里那一列的结果列 → 它取自的遮罩列（按血缘里出现的先后）。报告自查按它判违规。
|
||||
"""sure 里每一列的出处：{结果列: 它取自的遮罩列}，认得出表的写「表.列」（stores.name、customer_spend.name），认不出的
|
||||
写这份快照的全部遮罩列。报告自查按它判违规、写违规说明。
|
||||
|
||||
遮罩按列名设在整个数据源上（data/engine.py 的 masked_columns，不带表名）：设了 name，stores.name 和 customers.name 按
|
||||
列名都算。证据面板、可验证报告、核查 agent 照这样遮（宁可多遮）；自查多算一列就是判失败，所以按 SQL 的血缘
|
||||
（sqlglot.lineage，按查询当时冻结的表结构展开 *、认出不带表名的列出自哪张表）核一遍每一列的值取自哪张表的哪一列：
|
||||
遮罩按列名设在整个数据源上（data/engine.py 的 masked_columns 只认列名，不带表名）：设了 name，stores.name、
|
||||
customers.name、视图里的 name 都算，和证据面板、可验证报告的遮法一致。自查不推断「这项遮罩本意管哪张表」：第 3 轮
|
||||
这样推断过（遮罩列在几张表里都有时只认另有别的遮罩列的那几张），视图、没有第二个遮罩列的表（employees.name）因此被
|
||||
排除，正式运行照常成功、姓名写进了报告。SQL 血缘（sqlglot.lineage，按查询当时冻结的表结构展开 *、认出不带表名的列出自
|
||||
哪张表）只用来：
|
||||
|
||||
- 遮罩列在几张表里都有、其中有的表里另有别的遮罩列（customers 有 name 和 phone，stores 只有 name）：只认那几张表
|
||||
（_mask_owners）。别的情形（只在一张表里有、哪张都没有别的遮罩列）照列名认
|
||||
- 血缘里有认不出出处的（找不到这一列、表值函数、不带表名又有歧义的列、* 展开不了）：排除不了，照算
|
||||
- 认得出、可一处都不取自遮罩表里的遮罩列（门店名、SELECT region AS phone）：不算
|
||||
- 快照里没有冻结的表结构（数据源没探查过）：认不出同名的列还在哪几张表里，照列名认；值根本不取自遮罩列的照样不算
|
||||
- 点名出处：血缘的末端里叫遮罩列名字的列，按出现的先后写「表.列」，读的人看得出是哪一列被这项遮罩管到了
|
||||
- 排除值确实不取自遮罩列的（SELECT region AS phone FROM stores）：每个末端都落在表结构里标明是普通表（is_view 为
|
||||
false）的列上、没有一处叫遮罩列的名字，才排除
|
||||
- 认不出出处的照算（_owned_masks）：血缘认不出的列（没起别名的表达式……）、表值函数、视图（看不出它的列取自哪张表）、
|
||||
表结构里没有的表、没有冻结表结构的快照、表结构没标是不是视图的老快照
|
||||
|
||||
只排除、不新增：sure 里没有的列血缘认出来也不算。分析出了意外，照 sure 算。纯计算，在调用方的线程里算。"""
|
||||
只排除、不新增：sure 里没有的列血缘认出来也不算。分析出了意外，照 sure 算（出处写全部遮罩列）。纯计算，在调用方的
|
||||
线程里算。"""
|
||||
plain = {c: sorted(base) for c in sorted(sure)}
|
||||
sql = snap.get("sql")
|
||||
if not sure or not isinstance(sql, str) or not sql.strip():
|
||||
return plain
|
||||
try:
|
||||
from app.data.sql_semantics import _parse, _read
|
||||
from app.engine.masking import _frozen_columns
|
||||
|
||||
read = _read(kind)
|
||||
root, _ = _parse(sql, read)
|
||||
if root is None:
|
||||
return plain
|
||||
frozen = _frozen_columns(snap["schema_artifact"]) if isinstance(snap.get("schema_artifact"), str) else None
|
||||
# 表结构里一张表按好几个名字记着（带不带库名），这里只要不带库名的那一份
|
||||
tables = {name: cols for name, cols in (frozen or {}).items() if "." not in name}
|
||||
owners = _mask_owners(tables, base)
|
||||
artifact = snap.get("schema_artifact")
|
||||
tables, ordinary = _frozen_schema(artifact) if isinstance(artifact, str) else ({}, frozenset())
|
||||
schema = {name: {c: "TEXT" for c in cols} for name, cols in tables.items()} or None
|
||||
nodes = _lineages(root, schema, read, [str(c) for c in snap.get("columns") or [] if str(c).lower() in sure])
|
||||
out: dict[str, list[str]] = {}
|
||||
for column in sorted(sure):
|
||||
found = _owned_masks(nodes.get(column), base, owners, set(tables))
|
||||
found = _owned_masks(nodes.get(column), base, ordinary)
|
||||
if found:
|
||||
out[column] = found
|
||||
return out
|
||||
except Exception: # noqa: BLE001 - 核对不了出自哪张表：照 sure 算
|
||||
logger.warning("报告自查：按血缘核对遮罩列出自哪张表时出错，照列名认", exc_info=True)
|
||||
except Exception: # noqa: BLE001 - 核对不了出处：照 sure 算
|
||||
logger.warning("报告自查：按血缘核对遮罩列的出处时出错，照列名认", exc_info=True)
|
||||
return plain
|
||||
|
||||
|
||||
def _mask_owners(tables: dict[str, list[str]], base: set[str]) -> dict[str, set[str]]:
|
||||
"""遮罩列 → 它的遮罩表（小写的表名），只记排除得掉一些表的：遮罩列在几张表里都有，其中有的表里另有别的遮罩列，
|
||||
只认那几张（设了 name, phone，customers 有这两列，stores 只有 name：name 指的是客户姓名）。只在一张表里有、几张表
|
||||
里都有可哪张都没有别的遮罩列的不记：照列名认，哪张表的都算。"""
|
||||
out: dict[str, set[str]] = {}
|
||||
for mask in base:
|
||||
having = {t for t, cols in tables.items() if mask in cols}
|
||||
paired = {t for t in having if any(c in base and c != mask for c in tables[t])}
|
||||
if paired and paired != having:
|
||||
out[mask] = paired
|
||||
return out
|
||||
@functools.lru_cache(maxsize=64)
|
||||
def _frozen_schema(artifact: str) -> tuple[dict[str, list[str]], frozenset[str]]:
|
||||
"""表结构快照工件 → ({表名: 按次序的列名}, 标明是普通表的表名)，都是小写（_owned_columns）。
|
||||
|
||||
前一项只要不带库名的那份名字（表结构里一张表按好几个名字记着，masking._schema_keys），血缘按它展开 *、认不带表名的
|
||||
列。后一项是 is_view 明写为 false 的表（带不带库名的名字都记）：视图的列看不出取自哪张表，没标的老快照看不出是表还是
|
||||
视图；几个库里同名的表有一张不是普通表，这个名字就不算。取不到、被改过的两样都是空的。工件按内容寻址、不会变，记下来。"""
|
||||
from app.engine.masking import _schema_columns, _schema_keys
|
||||
|
||||
try:
|
||||
schema = load(artifact)
|
||||
except Exception: # noqa: BLE001 - 取不到就当没有冻结的表结构（出处认不出的照算）
|
||||
return {}, frozenset()
|
||||
entries = schema.get("tables") if isinstance(schema, dict) else None
|
||||
if not isinstance(entries, dict):
|
||||
return {}, frozenset()
|
||||
columns = {name: cols for name, cols in (_schema_columns(schema) or {}).items() if "." not in name}
|
||||
ordinary: dict[str, bool] = {}
|
||||
for key, meta in entries.items():
|
||||
if isinstance(meta, dict):
|
||||
for name in _schema_keys(key, meta):
|
||||
ordinary[name] = ordinary.get(name, True) and meta.get("is_view") is False
|
||||
return columns, frozenset(name for name, yes in ordinary.items() if yes)
|
||||
|
||||
|
||||
def _lineages(root: Any, schema: dict[str, Any] | None, read: str | None, columns: list[str]) -> dict[str, Any]:
|
||||
@@ -1843,10 +1881,12 @@ def _lineages(root: Any, schema: dict[str, Any] | None, read: str | None, column
|
||||
return out
|
||||
|
||||
|
||||
def _owned_masks(node: Any, base: set[str], owners: dict[str, set[str]], known: set[str]) -> list[str]:
|
||||
"""一个结果列的值取自哪几列遮罩列（按血缘里出现的先后），别的表里同名的列不算（owners，_mask_owners）。血缘认不出来
|
||||
（node 为 None）、里面有认不出出处的（_leaf_of 交回 None），有取自遮罩列的就点名它们、没有就照全部遮罩列算：排除不了。
|
||||
known 是冻结的表结构里有的表：不在里面的表认不出有没有别的遮罩列，照算。"""
|
||||
def _owned_masks(node: Any, base: set[str], ordinary: frozenset[str]) -> list[str]:
|
||||
"""一个结果列的出处：血缘的末端里叫遮罩列名字的列（哪张表、哪个视图的都算），按出现的先后写「表.列」。
|
||||
|
||||
一处都没有时：有认不出出处的末端——血缘认不出来（node 为 None）、_leaf_of 交回 None（表值函数、整张表的 *……）、落在
|
||||
不是普通表（ordinary，_frozen_schema）的表上（视图、表结构里没有的表）——排除不了，写全部遮罩列（照算）；每个末端都落在
|
||||
普通表的别的列上（region AS phone）的交回空：不算。"""
|
||||
if node is None:
|
||||
return sorted(base)
|
||||
found: list[str] = []
|
||||
@@ -1861,10 +1901,12 @@ def _owned_masks(node: Any, base: set[str], owners: dict[str, set[str]], known:
|
||||
if place is _NO_COLUMN:
|
||||
continue
|
||||
table, column = place
|
||||
if column not in base or (column in owners and table in known and table not in owners[column]):
|
||||
continue
|
||||
if column not in found:
|
||||
found.append(column)
|
||||
if column in base:
|
||||
if (named := f"{table}.{column}") not in found:
|
||||
found.append(named)
|
||||
elif table not in ordinary:
|
||||
# 视图、表结构里没有的表：看不出这一列取自哪里，排除不了
|
||||
unplaced = True
|
||||
return found or (sorted(base) if unplaced else [])
|
||||
|
||||
|
||||
@@ -1873,8 +1915,9 @@ _NO_COLUMN = object()
|
||||
|
||||
|
||||
def _leaf_of(leaf: Any) -> Any:
|
||||
"""血缘的一个末端出自哪张表的哪一列：(表名, 列名)，都是小写；没有取任何列的（写死的值、COUNT(*)）交回 _NO_COLUMN；
|
||||
认不出出处的（sqlglot 找不到来源、表值函数、整张表的 *、派生来源里的 *）交回 None。"""
|
||||
"""血缘的一个末端出自哪张表的哪一列：(表名, 列名)，都是小写，表名照 SQL 里写的带上库名（archive.sales）；没有取任何列
|
||||
的（写死的值、COUNT(*)）交回 _NO_COLUMN；认不出出处的（sqlglot 找不到来源、表值函数、整张表的 *、派生来源里的 *）交回
|
||||
None。"""
|
||||
from sqlglot import exp
|
||||
|
||||
source = leaf.source
|
||||
@@ -1885,7 +1928,8 @@ def _leaf_of(leaf: Any) -> Any:
|
||||
return None
|
||||
if not isinstance(column, exp.Column) or isinstance(column.this, exp.Star) or not column.name:
|
||||
return None
|
||||
return source.name.lower(), column.name.lower()
|
||||
table = ".".join(part for part in (source.catalog, source.db, source.name) if part)
|
||||
return table.lower(), column.name.lower()
|
||||
if isinstance(source, (exp.Table, exp.Placeholder)) or leaf.expression is source:
|
||||
return None
|
||||
return _NO_COLUMN
|
||||
@@ -3180,6 +3224,49 @@ def _sealed_loader(sealed: _Sealed, *, chain: bool = False) -> Callable[[str], A
|
||||
return loader
|
||||
|
||||
|
||||
async def _judge_loader(doc: dict[str, Any], catalog: dict[str, Any], units: list[str] | None = None,
|
||||
loader: Callable[[str], Any] | None = None) -> Callable[[str], Any]:
|
||||
"""交给结论句裁判的 loader（judge.prepare 的 loader）：摘录里的每份查询快照逐份遮，这份快照里确定取自遮罩列的列
|
||||
（_mask_sets 的 sure，和证据面板同一套认法：c.phone AS tel 的 tel、按位置认得出的 UPPER(phone)）也不给。
|
||||
|
||||
裁判自己只按列名遮（快照记下的 mask_columns 加数据源现在设的，judge._hidden）：改了名字的号码列照样进摘录，裁判理由
|
||||
一引用就在证据面板、审计 CSV 里露出来。这里把那几列并进交给裁判的这一份快照的 mask_columns（只是交给 prepare 的
|
||||
副本，工件本身不动），裁判照它自己的办法遮掉。只并这份快照自己的：不按数据源名合并（另一份快照里同名、却取自别的列
|
||||
的照给），不用证据面板宁可多遮的 hidden（整份遮的快照、只数个数的列照给，被引用的金额、人数裁判要看得到才判得了）。
|
||||
这样摘录变了的句子复用键随之变，此前判过的会重判一次。
|
||||
|
||||
只读要裁判的句子（units，None 是全部结论句，judge.candidates）挂的查询；读出来的快照另记一份交给 prepare 再取，
|
||||
不读第二遍。loader 缺省读工件库；按需裁判传只认封存工件的那个（_sealed_loader）。报告撰写节点当场裁判
|
||||
（engine/nodes/report.py）用同一个函数。"""
|
||||
from app.engine.evidence import _mask_names
|
||||
from app.engine.judge import candidates
|
||||
|
||||
read = loader or load
|
||||
masks = _Masks()
|
||||
shown: dict[str, Any] = {}
|
||||
for alias in dict.fromkeys(a for c in candidates(doc, units) for a in c.aliases):
|
||||
entry = catalog.get(alias)
|
||||
if not isinstance(entry, dict) or entry.get("kind") != "query" or not entry.get("artifact"):
|
||||
continue
|
||||
artifact = str(entry["artifact"])
|
||||
if artifact in shown:
|
||||
continue
|
||||
try:
|
||||
snap = read(artifact)
|
||||
except Exception: # noqa: BLE001 - 读不出来、不在封存范围里：摘录本来就不给它，交给 prepare 照它的办法处理
|
||||
continue
|
||||
if not isinstance(snap, dict):
|
||||
continue
|
||||
recorded = _mask_names(snap.get("mask_columns"))
|
||||
sure = set((await masks.sets(artifact, entry.get("source") or snap.get("source"), snap)).sure)
|
||||
shown[artifact] = {**snap, "mask_columns": sorted(recorded | sure)} if sure - recorded else snap
|
||||
|
||||
def reader(artifact: str) -> Any:
|
||||
key = str(artifact)
|
||||
return shown[key] if key in shown else read(key)
|
||||
return reader
|
||||
|
||||
|
||||
def _chain_set(sealed: _Sealed) -> set[str]:
|
||||
"""封存范围内每份表结构快照经哈希链列出的清单 id 的并集。表结构快照取不回、哈希不符的跳过（不在链上）。"""
|
||||
out: set[str] = set()
|
||||
@@ -3304,7 +3391,8 @@ async def _judge_now(sealed: _Sealed, report: _Report, units: list[str]) -> dict
|
||||
# 带期 4 标记的文档才让裁判取得到哈希链上的清单（摘录的来历行要用）；老文档的 loader 和以前完全相同
|
||||
loader = _sealed_loader(sealed, chain=doc.get("provenance") == DOC_PROVENANCE)
|
||||
masked = await judging.source_masks(catalog, loader=loader)
|
||||
request = judging.prepare(doc, catalog, units=pending, loader=loader, masked=masked)
|
||||
request = judging.prepare(doc, catalog, units=pending, masked=masked,
|
||||
loader=await _judge_loader(doc, catalog, pending, loader))
|
||||
skipped = dict(request.skipped)
|
||||
if request.cands:
|
||||
async with SessionLocal() as session:
|
||||
|
||||
@@ -3131,26 +3131,40 @@ def absence_claims(doc: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
|
||||
|
||||
def _masked_cell_message(ref: str, column: str, sources: list[str]) -> _Reason:
|
||||
"""违规 masked_cell 的说法：列名就是遮罩列的说「在数据源中设置了遮罩」，按 SQL 认出来的点名它取自的那几列遮罩列（sources，
|
||||
按血缘认出来的；认不出时是这份快照的全部遮罩列）——遮罩按列名设在整个数据源上，读的人要看得出是哪一项遮罩设置管到了
|
||||
这里。给写作者的那句说清这一列的值不能以任何写法写进报告：只说「删掉引用」，写作者会把值照抄成文字。"""
|
||||
own = column.lower() in {s.lower() for s in sources} or not sources
|
||||
where = (f"列「{column}」在数据源中设置了遮罩" if own
|
||||
else f"列「{column}」按 SQL 认出取自设置了遮罩的列「{'、'.join(sources[:3])}」")
|
||||
"""违规 masked_cell 的说法，点名这一格的出处（sources）：
|
||||
|
||||
- 认得出表的写「表.列」（api/evidence.py 的 _owned_columns 按血缘认出来的）：照实写取自哪张表的哪一列、数据源对哪个
|
||||
列名设了遮罩，并说明遮罩按列名对所有表生效——设了 name，门店名（stores.name）和客户姓名一样算，读的人要看得出是
|
||||
哪一项遮罩设置管到了这里（遮罩设置只认列名，没有按表写的办法，这里不教人去改设置）
|
||||
- 认不出表的（没有血缘、合并查询）：列名就是遮罩列的说「在数据源中设置了遮罩」，别的点名这份快照的遮罩列
|
||||
|
||||
给写作者的那句说清这一列的值不能以任何写法写进报告：只说「删掉引用」，写作者会把值照抄成文字。"""
|
||||
tabled = [s for s in sources if "." in s]
|
||||
masks = list(dict.fromkeys(s.rsplit(".", 1)[-1] for s in sources))
|
||||
renamed = column.lower() not in {m.lower() for m in masks}
|
||||
if tabled:
|
||||
named = "、".join(f"「{m}」" for m in masks[:3])
|
||||
where = (f"列「{column}」{'按 SQL 认出' if renamed else ''}取自 {'、'.join(tabled[:3])}，"
|
||||
f"数据源对列名{named}设置了遮罩（按列名对所有表生效）")
|
||||
elif sources and renamed:
|
||||
where = f"列「{column}」按 SQL 认出取自设置了遮罩的列「{'、'.join(masks[:3])}」"
|
||||
else:
|
||||
where = f"列「{column}」在数据源中设置了遮罩"
|
||||
return _Reason(
|
||||
f"引用了取自遮罩列的格子 [[{ref}]]：{where}，值不应写进报告",
|
||||
f"[[{ref}]] 引用的{where}（手机号、姓名这类），这一列的值不能以任何形式写进报告，引用、数字、照抄的文字都不行："
|
||||
"删掉这处引用和相应的内容；整表引用时用 cols= 只列别的列",
|
||||
f"[[{ref}]] 引用的{where}。设置了遮罩的列（手机号、姓名这类）的值不能以任何形式写进报告，引用、数字、照抄的文字"
|
||||
"都不行：删掉这处引用和相应的内容；整表引用时用 cols= 只列别的列",
|
||||
)
|
||||
|
||||
|
||||
def masked_cells(doc: dict[str, Any], catalog: dict[str, Any], masked: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||||
"""正文直接引用了取自遮罩列的格子（[[v:Q2.r0.tel]]，整表展开出来的格也算），每格一条违规 masked_cell。
|
||||
|
||||
masked 是 {查询快照工件: 取自遮罩列的结果列}，结果列可以是列名的集合，也可以是 {列名: 它取自的遮罩列}（说法里点名）。
|
||||
由报告节点算好交进来（api/evidence.py 的 _cited_masks：列名就是遮罩列的、SQL 里给遮罩列起了别名套了函数的，按血缘
|
||||
去掉别的表里同名的列；SQL 认不准、整份算进来的不在里面）：遮罩设在数据源上，这里只认文档和目录，查不了库。写作目录
|
||||
不遮罩（写作者要看得到数据），所以拦在自查：和没有出处的数字一样交回写作者重写，仍不改的按「违规处置」办（report.py）。
|
||||
masked 是 {查询快照工件: 取自遮罩列的结果列}，结果列可以是列名的集合，也可以是 {列名: 它的出处}（说法里点名，认得出
|
||||
表的写「表.列」）。由报告节点算好交进来（api/evidence.py 的 _cited_masks：列名就是遮罩列的、SQL 里给遮罩列起了别名
|
||||
套了函数的，哪张表里叫这个名字的列都算；SQL 认不准、整份算进来的不在里面）：遮罩设在数据源上，这里只认文档和目录，
|
||||
查不了库。写作目录不遮罩（写作者要看得到数据），所以拦在自查：和没有出处的数字一样交回写作者重写，仍不改的按
|
||||
「违规处置」办（report.py）。
|
||||
只认引用，不按值在正文里找遮罩值（两个字的姓名会撞上常用词）：照抄成正文的号码由数字核对拦下。
|
||||
|
||||
违规不带那一格的值（不写 text、context）：它会进 report.checked、重写的日志、出具缺口，这些地方不能再露一次。带着
|
||||
|
||||
@@ -324,12 +324,13 @@ class _MaskedCells:
|
||||
重写（evidence.masked_cells）。写作目录本身不遮（写作者要看得到数据），只拦引用。
|
||||
|
||||
哪几列算按 api/evidence.py 的 _cited_masks 认：列名就是遮罩列的，SQL 里给遮罩列起了别名、套了函数、经子查询或 WITH
|
||||
转一道的（_mask_sets 的 sure，和证据面板同一个函数），再按血缘去掉别的表里同名的列（设了 name, phone，门店名不算客户
|
||||
姓名）；合并查询按逐格来历和合并 SQL 认。证据面板为了展示宁可多遮（SQL 认不准时整份的列都遮），自查不跟着多算：多算
|
||||
一列就是一次判失败，写作者也改不掉。
|
||||
转一道的（_mask_sets 的 sure，和证据面板同一个函数）；合并查询按逐格来历和合并 SQL 认。遮罩按列名在整个数据源上
|
||||
生效，和证据面板一样：哪张表（含视图）里叫这个名字的列都算，设了 name 时门店名和客户姓名都算，违规说明点名取自哪张表
|
||||
的哪一列。证据面板为了展示宁可多遮（SQL 认不准时整份的列都遮），自查不跟着多算：多算一列就是一次判失败，写作者也
|
||||
改不掉。
|
||||
|
||||
只查引用，不按值在正文里找遮罩值：两个字的姓名会撞上「销售高峰」这类常用词，同名列的值会撞上门店名，正式运行因此判
|
||||
失败（复核第 3 轮去掉了按值认）；正文里照抄的号码本来就会被「数字没有出处」拦下。
|
||||
只查引用，不按值在正文里找遮罩值：两个字的姓名会撞上「销售高峰」这类常用词，正式运行因此判失败（复核第 3 轮去掉了
|
||||
按值认）；正文里照抄的号码本来就会被「数字没有出处」拦下。
|
||||
|
||||
组装好一稿之后才认：只读正文引用到的那几份快照（读盘复验哈希放进线程），每份只读一次，重写的稿子不再读；没引用格子的
|
||||
一份都不读。"""
|
||||
@@ -821,7 +822,10 @@ async def _judge_inline(ctx: NodeContext, doc: dict[str, Any], catalog: dict[str
|
||||
|
||||
async def judge_round(round_doc: dict[str, Any], limits: judging.Budget, known: dict[str, Any],
|
||||
first: bool) -> dict[str, Any]:
|
||||
request = judging.prepare(round_doc, catalog, masked=masks)
|
||||
# 摘录里每份查询快照逐份遮，改了名字的遮罩列（c.phone AS tel）也不给，和按需裁判同一个函数（_judge_loader）
|
||||
from app.api.evidence import _judge_loader
|
||||
|
||||
request = judging.prepare(round_doc, catalog, masked=masks, loader=await _judge_loader(round_doc, catalog))
|
||||
key = judging.request_key(request, judge_spec, limits, known)
|
||||
if _in_graph():
|
||||
# 写作不进 checkpoint，接着跑时写作者会重写一稿：缓存里先前那一稿的判定对不上指纹，往后翻
|
||||
|
||||
@@ -113,6 +113,23 @@ async def test_values_of_a_snapshot_masked_whole_are_not_collected(engine_up, ma
|
||||
assert shown.text(sql) == sql.replace(PHONES[0], "已遮罩") and shown.changed, shown.text(sql)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, column", [
|
||||
("SELECT UPPER(phone) FROM customers", "UPPER(phone)"),
|
||||
("SELECT TRIM(c.phone) FROM customers c ORDER BY c.id", "TRIM(c.phone)"),
|
||||
])
|
||||
async def test_values_of_an_unaliased_expression_over_every_column_are_collected(engine_up, mall, sql, column):
|
||||
"""复核第 4 轮：没起别名、表达式盖住全部结果列，SQL 认得出这一列取自遮罩列（按位置），不是认不准、整份算进来的：
|
||||
算确定取自遮罩列（sure），值照收，之后别的查询步骤里写着的号码显示成「已遮罩」（第 3 轮一个值都收不到，号码原样显示）。"""
|
||||
from app.api.evidence import _Masks, _Shown, _mask_sets
|
||||
|
||||
art = await put(sql, [column], [[p] for p in PHONES])
|
||||
sets = await _mask_sets(artifact_store.load(art), "mall", _Masks(), artifact=art)
|
||||
assert sets.sure == {column.lower()}, (sets.sure, sets.hidden)
|
||||
secrets = await _Masks(sealed_of(art)).run_secrets()
|
||||
probe = f"SELECT SUM(amount) FROM orders o JOIN customers c ON c.id = o.customer_id WHERE c.phone = '{PHONES[1]}'"
|
||||
assert _Shown(secrets).text(probe) == probe.replace(PHONES[1], "已遮罩")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 按快照工件在进程里记下
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@@ -7,8 +7,8 @@
|
||||
- 查询步骤原样给出 SQL 里写着的遮罩值（按号码筛的 WHERE c.phone IN (…)），合并 SQL、SQL 检查的片段也一样；可验证报告
|
||||
按「这次运行查到过的遮罩值」认出来、整份只给哈希。现在展示时换成「已遮罩」，底下写明只是展示时换的
|
||||
- 第 2 轮把面板那套宁可多遮的列按数据源名并给裁判（裁判按数据源名合起来遮，对同一数据源的每一份快照都生效）：整份遮的
|
||||
快照的全部列、只数个数的列一并从摘录里消失，被引用的金额、人数裁判看不到，只能判证据不足。第 3 轮退回 M2 之前的做法，
|
||||
只按数据源设的列名遮；c.phone AS tel 这种改了名字的号码列仍会进摘录，是裁判摘录已知的缺口，另行处理（judge.py）
|
||||
快照的全部列、只数个数的列一并从摘录里消失，被引用的金额、人数裁判看不到，只能判证据不足。第 4 轮起摘录按快照逐份遮，
|
||||
只遮确定取自遮罩列的列（改了名字的号码列见 test_judge_excerpt_masks.py），这两样照给
|
||||
|
||||
真跑的部分用虚构的 SQLite 商城库（号码是编的），报告由 MockChatModel 照剧本写。
|
||||
"""
|
||||
@@ -198,7 +198,7 @@ def test_sql_check_excerpts_and_merge_parts_are_shown_with_values_masked():
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 交给裁判的摘录：照数据源设的遮罩列（复核第 3 轮退回 M2 之前的做法）
|
||||
# 交给裁判的摘录：整份遮的快照、只数个数的列照给（第 4 轮起逐份只遮确定取自遮罩列的列）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: 派生来源里有没起别名的 UPPER(c.phone)：证据面板整份遮（宁可多遮），可这份快照的列和号码无关
|
||||
@@ -269,9 +269,9 @@ def excerpt_of(request: str) -> str:
|
||||
|
||||
@pytest.mark.parametrize("where", ["on_demand", "inline"])
|
||||
async def test_the_judge_sees_the_cited_columns_of_every_snapshot(client, monkeypatch, mall, where):
|
||||
"""交给裁判的遮罩列只按数据源设的列名（judge.source_masks），不并证据面板那套宁可多遮的列：Q1 整份遮、Q2 只数个数
|
||||
的列，面板都不显示，可它们和号码无关，裁判要看得到被引用的金额、人数才判得了。按需裁判（探索运行）、节点里当场裁判
|
||||
（正式运行）同一个办法。号码列照旧不给（摘录里一个号码都没有）。"""
|
||||
"""交给裁判的摘录不并证据面板那套宁可多遮的列：Q1 整份遮、Q2 只数个数的列，面板都不显示，可它们和号码无关（不在
|
||||
_mask_sets 的 sure 里），裁判要看得到被引用的金额、人数才判得了。按需裁判（探索运行）、节点里当场裁判（正式运行）
|
||||
同一个办法（api/evidence.py 的 _judge_loader）。号码列照旧不给（摘录里一个号码都没有）。"""
|
||||
seen = judge_models(monkeypatch)
|
||||
if where == "on_demand":
|
||||
row = await judged_run("exploratory", claims="judge")
|
||||
|
||||
@@ -0,0 +1,210 @@
|
||||
"""交给结论句裁判的摘录：按快照逐份遮，SQL 认得确定取自遮罩列的列（改了名字的号码列）不给，金额、计数照给。
|
||||
|
||||
复核发现（第 4 轮）：
|
||||
- 第 3 轮把摘录退回只按数据源设的列名遮：c.phone AS tel 这种改了名字的号码列又进了摘录（「列：spend | tel / r0：380 |
|
||||
199…」），裁判理由在证据面板、审计 CSV 里原样显示，裁判一引用号码就露出来。按需裁判（探索运行）和报告撰写节点里当场
|
||||
裁判（正式运行）都是这样
|
||||
- 第 2 轮把面板那套宁可多遮的列按数据源名并给裁判，整份遮的快照、只数个数的列从摘录里消失，被引用的金额、人数裁判看不到
|
||||
|
||||
现在每份快照只遮这份快照里确定取自遮罩列的那几列（_mask_sets 的 sure，和证据面板同一套认法）：不按数据源名合并（另一份
|
||||
快照里同名、却取自别的列的照给），不用宁可多遮的 hidden（整份遮的快照、只数个数的列照给）。摘录变了，复用键随之变，已判过
|
||||
的句子会重判一次。
|
||||
|
||||
真跑的部分用虚构的 SQLite 零售库（号码、热线都是编的），写作者和裁判都是 MockChatModel。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from langchain_core.messages import AIMessage
|
||||
from sqlalchemy import delete, select
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.data.engine import engines
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource, Run, Setting
|
||||
from app.engine.evidence import compose_doc, iter_units, make_eid
|
||||
from app.engine.runner import run_manager
|
||||
from app.main import app
|
||||
|
||||
SOURCE = "judgeshop"
|
||||
CUSTOMERS = [(1, "客户甲", "19900000611"), (2, "客户乙", "19900000612"), (3, "客户丙", "19900000613")]
|
||||
STORES = [(1, "一号店", "0571-88000001"), (2, "二号店", "0571-88000002")]
|
||||
ORDERS = [(1, 1, 1, 120.5), (2, 2, 2, 300.0), (3, 2, 2, 80.0), (4, 2, 3, 210.25)]
|
||||
PHONES = [c[2] for c in CUSTOMERS]
|
||||
|
||||
#: 改了名字的号码列 contact：确定取自遮罩列
|
||||
SQL_CUSTOMER = ("SELECT c.phone AS contact, ROUND(SUM(o.amount), 2) AS spend FROM customers c "
|
||||
"JOIN orders o ON o.customer_id = c.id GROUP BY c.phone ORDER BY spend DESC")
|
||||
#: 同样叫 contact，取自门店热线（没设遮罩）：照给
|
||||
SQL_STORE = ("SELECT s.hotline AS contact, COUNT(*) AS orders FROM stores s JOIN orders o ON o.store_id = s.id "
|
||||
"GROUP BY s.hotline ORDER BY orders DESC")
|
||||
TEXT = "消费最高的一位客户共消费 [[v:Q1.r0.spend]]，订单最多的门店有 [[v:Q2.r0.orders]] 笔订单，热线 [[v:Q2.r0.contact]]。"
|
||||
|
||||
|
||||
def leaks(value) -> list[str]:
|
||||
"""写着的号码，按千分位写的（摘录里数值格按数渲染：19,900,000,611）也算。"""
|
||||
raw = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
|
||||
return [p for p in PHONES if p in raw or p in raw.replace(",", "")]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client(engine_up):
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
async def set_source(path, options=None) -> str:
|
||||
async with SessionLocal() as session:
|
||||
row = (await session.execute(select(DataSource).where(DataSource.name == SOURCE))).scalar_one_or_none()
|
||||
if row is None:
|
||||
row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
|
||||
session.add(row)
|
||||
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
|
||||
row.options = dict(options or {})
|
||||
await session.commit()
|
||||
return row.id
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def shop(tmp_path, engine_up):
|
||||
path = tmp_path / "shop.db"
|
||||
db = sqlite3.connect(path)
|
||||
db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);"
|
||||
"CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT, hotline TEXT);"
|
||||
"CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER, customer_id INTEGER,"
|
||||
" amount REAL);")
|
||||
db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
|
||||
db.executemany("INSERT INTO stores VALUES (?, ?, ?)", STORES)
|
||||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ORDERS)
|
||||
db.commit()
|
||||
db.close()
|
||||
source_id = await set_source(path, {"mask_columns": ["phone"]})
|
||||
await engines.invalidate(source_id)
|
||||
yield path
|
||||
await set_source(path, {})
|
||||
await engines.invalidate(source_id)
|
||||
|
||||
|
||||
def node(nid, ntype, **config):
|
||||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||||
|
||||
|
||||
def judge_models(monkeypatch) -> list[str]:
|
||||
"""写作者回 TEXT；裁判一律判有依据。返回每次交给裁判的请求原文。"""
|
||||
from app.engine import judge as judge_mod
|
||||
from app.engine.judge import JUDGE_ROLE
|
||||
from app.providers.mock_model import MockChatModel
|
||||
|
||||
seen: list[str] = []
|
||||
|
||||
def decide(model, messages):
|
||||
if JUDGE_ROLE in str(messages[0].content):
|
||||
request = str(messages[-1].content)
|
||||
seen.append(request)
|
||||
items = [{"unit": uid, "verdict": "supported", "rationale": "虚构的理由", "used": []}
|
||||
for uid in re.findall(r"^\[(u\d+)\]", request, re.M)]
|
||||
return AIMessage(content=json.dumps({"items": items}, ensure_ascii=False))
|
||||
return AIMessage(content=TEXT)
|
||||
|
||||
monkeypatch.setattr(MockChatModel, "_decide", decide)
|
||||
|
||||
async def judge_model_of(session, spec):
|
||||
return MockChatModel(model_name="claude-sonnet-5"), "claude-sonnet-5"
|
||||
|
||||
monkeypatch.setattr(judge_mod, "get_chat_model", judge_model_of)
|
||||
return seen
|
||||
|
||||
|
||||
async def judged_run(run_class: str) -> Run:
|
||||
from app.engine.judge import SPEND_KEY
|
||||
|
||||
async with SessionLocal() as session:
|
||||
await session.execute(delete(Setting).where(Setting.key.in_([SPEND_KEY, "judge"])))
|
||||
await session.commit()
|
||||
nodes = [node("start", "input"),
|
||||
node("q_customer", "tool", tool=f"db_query__{SOURCE}", args={"sql": SQL_CUSTOMER}),
|
||||
node("q_store", "tool", tool=f"db_query__{SOURCE}", args={"sql": SQL_STORE}),
|
||||
node("write", "report", instructions="写一句", claims="judge", on_violation="flag"),
|
||||
node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}])]
|
||||
spec = {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||||
run_id = (await run_manager.start(graph=spec, input_payload={}, run_class=run_class)).id
|
||||
for _ in range(400):
|
||||
await asyncio.sleep(0.05)
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Run, run_id)
|
||||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||||
assert row.status == "succeeded", row.error
|
||||
return row
|
||||
raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
|
||||
|
||||
|
||||
def excerpt_of(request: str, alias: str) -> str:
|
||||
"""裁判请求里一份查询的那一段摘录。"""
|
||||
return request.split(f"【{alias}】", 1)[1].split("【", 1)[0]
|
||||
|
||||
|
||||
def columns_line(excerpt: str) -> str:
|
||||
return next(line for line in excerpt.splitlines() if line.startswith("列："))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("where", ["on_demand", "inline"])
|
||||
async def test_the_judge_does_not_see_a_renamed_masked_column(client, monkeypatch, shop, where):
|
||||
"""Q1 的 contact 是 c.phone 改了名字：摘录里没有这一列、没有号码，说明有的列遮掉了；被引用的消费额照给。Q2 的 contact
|
||||
取自门店热线：同名不同出处，照给（不按数据源名合并）。按需裁判（探索运行）、节点里当场裁判（正式运行）同一个办法。"""
|
||||
seen = judge_models(monkeypatch)
|
||||
if where == "on_demand":
|
||||
row = await judged_run("exploratory")
|
||||
[report] = (await client.get(f"/api/runs/{row.id}/evidence")).json()["reports"]
|
||||
doc = artifact_store.load(report["doc_artifact"])
|
||||
uid = next(u["id"] for _, u in iter_units(doc)
|
||||
if any(seg.get("ref") == "v:Q1.r0.spend" for seg in u.get("segments") or []))
|
||||
res = await client.post(f"/api/runs/{row.id}/evidence/judge", json={"units": [uid]})
|
||||
assert res.status_code == 200, res.text
|
||||
else:
|
||||
await judged_run("formal")
|
||||
[request] = seen
|
||||
q1, q2 = excerpt_of(request, "Q1"), excerpt_of(request, "Q2")
|
||||
assert "contact" not in columns_line(q1) and "spend" in columns_line(q1) and "380" in q1, q1
|
||||
assert "有的列按数据源的设置遮罩了" in q1, q1
|
||||
assert "contact" in columns_line(q2) and "0571-88000002" in q2 and "orders" in columns_line(q2), q2
|
||||
assert leaks(request) == []
|
||||
|
||||
|
||||
async def test_an_unaliased_expression_over_every_column_is_masked_in_the_excerpt(engine_up, shop):
|
||||
"""SELECT UPPER(phone) FROM customers：没起别名、表达式盖住全部结果列，SQL 认得出取自遮罩列（不是认不准整份算的），
|
||||
摘录不给这一列。派生来源里有没起别名的号码表达式、整份遮的快照，金额照给（裁判要看得到才判得了）。"""
|
||||
from app.api.evidence import _judge_loader
|
||||
from app.engine import judge as judging
|
||||
|
||||
upper = await artifact_store.put_json({"sql": "SELECT UPPER(phone) FROM customers", "columns": ["UPPER(phone)"],
|
||||
"rows": [[p] for p in PHONES], "source": SOURCE}, kind="query")
|
||||
whole = await artifact_store.put_json(
|
||||
{"sql": "SELECT name, amount FROM (SELECT c.name, UPPER(c.phone), o.amount FROM customers c "
|
||||
"JOIN orders o ON o.customer_id = c.id) t ORDER BY amount DESC",
|
||||
"columns": ["name", "amount"], "rows": [["客户乙", 300.0]], "source": SOURCE}, kind="query")
|
||||
listed = (("Q1", upper, ["UPPER(phone)"], 3), ("Q2", whole, ["name", "amount"], 1))
|
||||
catalog = {alias: {"alias": alias, "kind": "query", "artifact": art, "columns": columns, "rows": rows,
|
||||
"node_id": alias, "source": SOURCE, "eid": make_eid("query", art, {})}
|
||||
for alias, art, columns, rows in listed}
|
||||
doc = compose_doc("客户都留了号码，最大一笔订单 [[v:Q2.r0.amount]]，见 [[see:Q1]]。", catalog, node_id="write",
|
||||
run_id="r1")
|
||||
masked = await judging.source_masks(catalog)
|
||||
request = judging.prepare(doc, catalog, masked=masked, loader=await _judge_loader(doc, catalog))
|
||||
assert leaks(request.excerpts["Q1"]) == [] and "UPPER" not in columns_line(request.excerpts["Q1"]), request.excerpts
|
||||
assert "amount" in columns_line(request.excerpts["Q2"]) and "300" in request.excerpts["Q2"], request.excerpts
|
||||
# 不经 _judge_loader（只按数据源设的列名遮）时号码就在摘录里：上面的断言测的是这一层
|
||||
plain = judging.prepare(doc, catalog, masked=masked)
|
||||
assert leaks(plain.excerpts["Q1"]) != []
|
||||
@@ -0,0 +1,329 @@
|
||||
"""报告自查认「遮罩列」和证据面板同一个口径：遮罩按列名设在整个数据源上，哪张表（含视图）里叫这个名字的列都算。
|
||||
|
||||
复核发现（第 4 轮）：
|
||||
- 第 3 轮给自查加了「遮罩表」的推断：遮罩列在几张表里都有、其中有的表另有别的遮罩列时，只认那几张表（customers 有 name
|
||||
和 phone，stores 只有 name，就只认客户表）。视图的列看不出取自哪张表，没有第二个遮罩列的表（employees.name）也被排除：
|
||||
经视图 customer_spend 引用客户姓名、引用店员姓名，正式运行照常成功，姓名写进报告；证据面板照列名把同一格遮成「已遮罩」
|
||||
- 现在不再推断遮罩表：引用的那一格按这份快照的 SQL 认出取自任何一张表里叫这个名字的列，就记 masked_cell，门店名
|
||||
（stores.name）也算，和证据面板的遮法一致。违规说明照实写取自哪张表的哪一列，并说明遮罩按列名对所有表生效
|
||||
- SQL 血缘只用来点名出处，和排除值确实取自别的列的（SELECT region AS phone FROM stores：stores 是表结构里的普通表）；
|
||||
视图、表结构里没有的表、没有冻结表结构的快照看不出列的出处，照算
|
||||
|
||||
正文里写着的字照旧不按值认（第 3 轮的结论不变）：两个字的姓名会撞上「销售高峰」，照抄的号码由「数字没有出处」拦下。
|
||||
|
||||
真跑的部分用虚构的 SQLite 零售库（号码、姓名、门店名都是编的），数据源探查过表结构（查询快照里冻结着表结构，视图标着
|
||||
is_view），写作者是 MockChatModel 照剧本写。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.data.engine import engines
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource, Run, RunEvent
|
||||
from app.engine.evidence import SCHEMA_SNAPSHOT, compose_doc
|
||||
from app.engine.runner import run_manager
|
||||
|
||||
SOURCE = "m2shop"
|
||||
STORES = [(1, "杭州西湖店", "华东"), (2, "广州天河店", "华南")]
|
||||
CUSTOMERS = [(1, "陈思远", "19900000801"), (2, "高峰", "19900000802"), (3, "林若溪", "19900000803")]
|
||||
EMPLOYEES = [(1, "王建国", 1), (2, "赵晓梅", 2)]
|
||||
ORDERS = [(1, 1, 1, 1, 120.5), (2, 2, 2, 2, 300.0), (3, 1, 2, 1, 80.0), (4, 2, 3, 2, 210.25)]
|
||||
PHONES = [c[2] for c in CUSTOMERS]
|
||||
NAMES = [c[1] for c in CUSTOMERS] + [e[1] for e in EMPLOYEES]
|
||||
|
||||
SQL_STORE = ("SELECT s.name AS store, ROUND(SUM(o.amount), 2) AS sales FROM stores s JOIN orders o ON o.store_id = s.id "
|
||||
"GROUP BY s.name ORDER BY sales DESC")
|
||||
SQL_CUSTOMER = ("SELECT c.name AS customer, ROUND(SUM(o.amount), 2) AS spend FROM customers c "
|
||||
"JOIN orders o ON o.customer_id = c.id GROUP BY c.name ORDER BY spend DESC")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
async def set_source(path, options: dict, *, probe: bool = True) -> str:
|
||||
"""数据源 m2shop：probe 时探查表结构（工具查询时把它冻结进快照），不然清空。名字只这个文件用，探查的结构不会
|
||||
冻结进别的测试的快照。"""
|
||||
from app.data.introspect import introspect
|
||||
|
||||
async with SessionLocal() as session:
|
||||
row = (await session.execute(select(DataSource).where(DataSource.name == SOURCE))).scalar_one_or_none()
|
||||
if row is None:
|
||||
row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
|
||||
session.add(row)
|
||||
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
|
||||
row.options = dict(options)
|
||||
row.schema_cache = (await introspect(row)) if probe else None
|
||||
await session.commit()
|
||||
return row.id
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def shop(tmp_path, engine_up):
|
||||
path = tmp_path / "shop.db"
|
||||
db = sqlite3.connect(path)
|
||||
db.executescript(
|
||||
"CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT, region TEXT);"
|
||||
"CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);"
|
||||
"CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT, store_id INTEGER);"
|
||||
"CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER, customer_id INTEGER, clerk_id INTEGER,"
|
||||
" amount REAL);"
|
||||
# 视图里装的就是客户姓名：表结构里只看得到它的列名 name、spend
|
||||
"CREATE VIEW customer_spend AS SELECT c.name AS name, SUM(o.amount) AS spend FROM customers c "
|
||||
"JOIN orders o ON o.customer_id = c.id GROUP BY c.name;")
|
||||
db.executemany("INSERT INTO stores VALUES (?, ?, ?)", STORES)
|
||||
db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
|
||||
db.executemany("INSERT INTO employees VALUES (?, ?, ?)", EMPLOYEES)
|
||||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ORDERS)
|
||||
db.commit()
|
||||
db.close()
|
||||
source_id = await set_source(path, {"mask_columns": ["name", "phone"]})
|
||||
await engines.invalidate(source_id)
|
||||
yield path
|
||||
await set_source(path, {}, probe=False)
|
||||
await engines.invalidate(source_id)
|
||||
|
||||
|
||||
def node(nid, ntype, **config):
|
||||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||||
|
||||
|
||||
def graph(sql: str, **report) -> dict:
|
||||
nodes = [node("start", "input"), node("top", "tool", tool=f"db_query__{SOURCE}", args={"sql": sql}),
|
||||
node("write", "report", instructions="写一句", **report),
|
||||
node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}])]
|
||||
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||||
|
||||
|
||||
async def finish(monkeypatch, spec: dict, text: str) -> Run:
|
||||
from app.providers import mock_model
|
||||
|
||||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
|
||||
run_id = (await run_manager.start(graph=spec, input_payload={})).id
|
||||
for _ in range(400):
|
||||
await asyncio.sleep(0.05)
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Run, run_id)
|
||||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||||
return row
|
||||
raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
|
||||
|
||||
|
||||
async def checked_of(run_id: str) -> dict:
|
||||
async with SessionLocal() as session:
|
||||
q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == "report.checked")
|
||||
return (await session.execute(q)).scalars().one().data
|
||||
|
||||
|
||||
def masked_of(violations: list[dict]) -> list[dict]:
|
||||
return [v for v in violations if v["code"] == "masked_cell"]
|
||||
|
||||
|
||||
def leaks(value) -> list[str]:
|
||||
raw = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
|
||||
return [p for p in PHONES if p in raw or p in raw.replace(",", "")]
|
||||
|
||||
|
||||
def named(value) -> list[str]:
|
||||
raw = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
|
||||
return [n for n in NAMES if n in raw]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 正文里写着的字不按值认
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, text", [
|
||||
# 客户里有一位叫「高峰」：常用词「销售高峰」不是在写这位客户
|
||||
(SQL_CUSTOMER, "本周销售高峰出现在周五，最高一位客户消费 [[v:Q1.r0.spend]]。"),
|
||||
# 正文用文字写门店名：不是引用，自查不认（引用门店名才算，见下）
|
||||
(SQL_STORE, "销售额最高的门店是广州天河店，销售额 [[v:Q1.r0.sales]]。"),
|
||||
])
|
||||
async def test_words_that_happen_to_equal_masked_values_do_not_fail_a_formal_report(monkeypatch, shop, sql, text):
|
||||
"""违规即失败（正式运行的默认值）：正文里的字和遮罩列中的值相同，不判 masked_cell，运行照常成功。"""
|
||||
row = await finish(monkeypatch, graph(sql, on_violation="fail"), text)
|
||||
assert row.status == "succeeded", row.error
|
||||
checked = await checked_of(row.id)
|
||||
assert masked_of(checked["violations"]) == [] and checked["ok"] is True, checked["violations"]
|
||||
|
||||
|
||||
async def test_a_phone_copied_into_the_text_is_stopped_by_the_number_check(monkeypatch, shop):
|
||||
"""照抄成正文的号码不按值认，「数字没有出处」照样拦下：交回重写、仍不改就判失败；违规里没有 masked_cell。"""
|
||||
row = await finish(monkeypatch, graph(SQL_CUSTOMER, on_violation="fail"),
|
||||
f"消费最高的客户电话 {PHONES[1]}，共消费 [[v:Q1.r0.spend]]。")
|
||||
assert row.status == "failed", row.error
|
||||
checked = await checked_of(row.id)
|
||||
assert {v["code"] for v in checked["violations"]} == {"uncited_number"}, checked["violations"]
|
||||
assert "遮罩" not in (row.error or "")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 引用：哪张表（含视图）里叫这个名字的列都算
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: (说明, SQL, 正文, 被引用的那一格的值, 违规说明里点名的出处)
|
||||
CITED = [
|
||||
("经视图引用客户姓名（改了名字）", "SELECT name AS customer, spend FROM customer_spend ORDER BY spend DESC",
|
||||
"消费最高的客户是 [[v:Q1.r0.customer]]，共消费 [[v:Q1.r0.spend]]。", "林若溪",
|
||||
"按 SQL 认出取自 customer_spend.name"),
|
||||
("经视图引用客户姓名（列名照旧）", "SELECT name, spend FROM customer_spend ORDER BY spend DESC",
|
||||
"消费最高的客户是 [[v:Q1.r0.name]]，共消费 [[v:Q1.r0.spend]]。", "林若溪", "列「name」取自 customer_spend.name"),
|
||||
("店员姓名：所在的表里没有第二个遮罩列",
|
||||
"SELECT e.name AS clerk, SUM(o.amount) AS sales FROM employees e JOIN orders o ON o.clerk_id = e.id "
|
||||
"GROUP BY e.name ORDER BY sales DESC",
|
||||
"业绩最高的店员是 [[v:Q1.r0.clerk]]，销售额 [[v:Q1.r0.sales]]。", "赵晓梅", "按 SQL 认出取自 employees.name"),
|
||||
("门店名：和客户姓名同叫 name", SQL_STORE, "销售额最高的门店是 [[v:Q1.r0.store]]，销售额 [[v:Q1.r0.sales]]。",
|
||||
"广州天河店", "按 SQL 认出取自 stores.name"),
|
||||
("客户姓名", SQL_CUSTOMER, "消费最高的客户是 [[v:Q1.r0.customer]]，共消费 [[v:Q1.r0.spend]]。", "林若溪",
|
||||
"按 SQL 认出取自 customers.name"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, sql, text, value, origin", CITED, ids=[c[0] for c in CITED])
|
||||
async def test_a_cited_masked_name_fails_the_run_whatever_table_it_comes_from(monkeypatch, shop, label, sql, text,
|
||||
value, origin):
|
||||
"""违规即失败：引用的那一格取自叫 name 的列（视图、只有一个遮罩列的表、门店表都算），判 masked_cell，运行失败，
|
||||
成果里没有这个值；说明点名取自哪张表的哪一列，说清遮罩按列名对所有表生效。"""
|
||||
row = await finish(monkeypatch, graph(sql, on_violation="fail"), text)
|
||||
assert row.status == "failed", label
|
||||
[v] = masked_of((await checked_of(row.id))["violations"])
|
||||
assert origin in v["message"] and "数据源对列名「name」设置了遮罩" in v["message"], v["message"]
|
||||
assert "按列名对所有表生效" in v["message"] and value not in v["message"], v["message"]
|
||||
assert origin in row.error and value not in row.error, row.error
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, row.id)
|
||||
assert value not in json.dumps(run.output, ensure_ascii=False, default=str), run.output
|
||||
|
||||
|
||||
async def test_flagging_keeps_the_report_and_names_the_view(monkeypatch, shop):
|
||||
"""违规处置为标注：照常产出，违规记着取自视图的哪一列；违规本身不写值。"""
|
||||
row = await finish(monkeypatch, graph(CITED[0][1], on_violation="flag"), CITED[0][2])
|
||||
assert row.status == "succeeded", row.error
|
||||
[v] = masked_of((await checked_of(row.id))["violations"])
|
||||
assert v["ref"] == "v:Q1.r0.customer" and "customer_spend.name" in v["message"] and named(v) == [], v
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 认法：_cited_masks（报告节点经 _MaskedCells 调用）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: 冻结的表结构（introspect 的形状）：普通表标 is_view=False；视图 customer_spend 装着客户姓名，contact_book 的 contact
|
||||
#: 装着号码（视图的列看不出取自哪张表）
|
||||
TABLES = {"stores": ["id", "name", "region"], "customers": ["id", "name", "phone"],
|
||||
"employees": ["id", "name", "store_id"], "orders": ["id", "store_id", "customer_id", "clerk_id", "amount"]}
|
||||
VIEWS = {"customer_spend": ["name", "spend"], "contact_book": ["who", "contact"]}
|
||||
|
||||
|
||||
async def schema_of(tables: dict[str, list[str]], views: dict[str, list[str]] | None = None, *,
|
||||
marked: bool = True) -> str:
|
||||
"""表结构快照。marked=False 时不写 is_view（看不出是表还是视图的老形状）。"""
|
||||
def meta(cols: list[str], view: bool) -> dict:
|
||||
return {"columns": [{"name": c, "type": "TEXT"} for c in cols], **({"is_view": view} if marked else {})}
|
||||
|
||||
return await artifact_store.put_json(
|
||||
{"source": SOURCE, "tables": {**{t: meta(cols, False) for t, cols in tables.items()},
|
||||
**{t: meta(cols, True) for t, cols in (views or {}).items()}}},
|
||||
kind=SCHEMA_SNAPSHOT)
|
||||
|
||||
|
||||
async def cited(sql: str, columns: list[str], *, masks: list[str], schema: bool = True, marked: bool = True,
|
||||
tables: dict[str, list[str]] = TABLES, views: dict[str, list[str]] = VIEWS) -> dict[str, list[str]]:
|
||||
"""一份快照按报告自查认出的列：{结果列: 它的出处（认得出表的写「表.列」，认不出的是全部遮罩列）}。"""
|
||||
from app.api.evidence import _cited_masks, _Masks
|
||||
|
||||
await set_source(":memory:", {"mask_columns": masks}, probe=False)
|
||||
snap = {"sql": sql, "columns": columns, "rows": [], "source": SOURCE,
|
||||
**({"schema_artifact": await schema_of(tables, views, marked=marked)} if schema else {})}
|
||||
art = await artifact_store.put_json(snap, kind="query")
|
||||
return await _cited_masks(snap, SOURCE, _Masks(), art)
|
||||
|
||||
|
||||
MIXED = ("SELECT s.name AS store, c.name AS customer, c.phone AS tel, SUM(o.amount) AS spend FROM orders o "
|
||||
"JOIN stores s ON s.id = o.store_id JOIN customers c ON c.id = o.customer_id GROUP BY s.name, c.name, c.phone")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, sql, columns, masks, expected", [
|
||||
("两张表的同名列在一条查询里：都算，各自点名", MIXED, ["store", "customer", "tel", "spend"], ["name", "phone"],
|
||||
{"store": ["stores.name"], "customer": ["customers.name"], "tel": ["customers.phone"]}),
|
||||
("只设了 name", MIXED, ["store", "customer", "tel", "spend"], ["name"],
|
||||
{"store": ["stores.name"], "customer": ["customers.name"]}),
|
||||
("直接选出门店表的 name", "SELECT name, region FROM stores", ["name", "region"], ["name", "phone"],
|
||||
{"name": ["stores.name"]}),
|
||||
("* 展开门店表", "SELECT * FROM stores", ["id", "name", "region"], ["name", "phone"], {"name": ["stores.name"]}),
|
||||
("* 展开客户表", "SELECT * FROM customers", ["id", "name", "phone"], ["name", "phone"],
|
||||
{"name": ["customers.name"], "phone": ["customers.phone"]}),
|
||||
("店员表只有一个遮罩列", "SELECT e.name AS clerk FROM employees e", ["clerk"], ["name", "phone"],
|
||||
{"clerk": ["employees.name"]}),
|
||||
("经视图（改了名字）", "SELECT name AS customer, spend FROM customer_spend", ["customer", "spend"], ["name", "phone"],
|
||||
{"customer": ["customer_spend.name"]}),
|
||||
("视图的列名不是遮罩列、改成了遮罩列的名字：看不出取自哪里，照算", "SELECT contact AS phone FROM contact_book",
|
||||
["phone"], ["phone"], {"phone": ["phone"]}),
|
||||
("经 WITH 转两道", "WITH a AS (SELECT c.phone AS tel FROM customers c), b AS (SELECT tel AS contact FROM a) "
|
||||
"SELECT contact FROM b", ["contact"], ["phone"], {"contact": ["customers.phone"]}),
|
||||
("列名和遮罩列一样，值却取自普通表的别的列", "SELECT region AS phone FROM stores", ["phone"], ["phone"], {}),
|
||||
("UNION：两支都取自叫 name 的列", "SELECT name FROM stores UNION ALL SELECT name FROM customers", ["name"],
|
||||
["name", "phone"], {"name": ["stores.name", "customers.name"]}),
|
||||
("UNION：一支取自普通表的别的列，另一支经视图", "SELECT region AS name FROM stores UNION ALL "
|
||||
"SELECT name FROM customer_spend", ["name"], ["name"], {"name": ["customer_spend.name"]}),
|
||||
])
|
||||
async def test_cited_masks_count_the_column_name_in_every_table(engine_up, label, sql, columns, masks, expected):
|
||||
assert await cited(sql, columns, masks=masks) == expected, label
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, sql, kwargs", [
|
||||
("没有冻结的表结构：看不出 stores 是表还是视图", "SELECT region AS phone FROM stores", {"schema": False}),
|
||||
("表结构没标 is_view", "SELECT region AS phone FROM stores", {"marked": False}),
|
||||
("表结构里没有这张表", "SELECT region AS phone FROM stores", {"tables": {"customers": ["id", "name", "phone"]}}),
|
||||
("写了库名：表结构探查的是另一个库，同名的表可能是视图", "SELECT s.region AS phone FROM archive.stores s", {}),
|
||||
])
|
||||
async def test_a_column_whose_origin_cannot_be_seen_is_flagged(engine_up, label, sql, kwargs):
|
||||
"""值看起来取自别的列（region AS phone），可这张表看不出是不是视图：排除不了，照列名算（和证据面板一样遮）。"""
|
||||
assert await cited(sql, ["phone"], masks=["phone"], **kwargs) == {"phone": ["phone"]}, label
|
||||
|
||||
|
||||
async def test_without_a_frozen_schema_the_sql_still_names_the_tables(engine_up):
|
||||
"""数据源没探查过表结构：SQL 里写了表的照样点名出处；认不出、排除不了的照算。"""
|
||||
assert await cited(MIXED, ["store", "customer", "tel", "spend"], masks=["name", "phone"], schema=False) == \
|
||||
{"store": ["stores.name"], "customer": ["customers.name"], "tel": ["customers.phone"]}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
# 表值函数：拆的是号码，血缘认不出出处，照遮罩那一套认
|
||||
("SELECT x.value AS p FROM customers c, json_each(c.phone) x", ["p"]),
|
||||
# 没起别名的表达式：血缘按列名找不到这一列
|
||||
("SELECT id, UPPER(phone) FROM customers", ["id", "UPPER(phone)"]),
|
||||
])
|
||||
async def test_columns_the_lineage_cannot_place_stay_flagged(engine_up, sql, columns):
|
||||
assert set(await cited(sql, columns, masks=["name", "phone"])) == {columns[-1].lower()}
|
||||
|
||||
|
||||
async def test_the_report_node_flags_every_cited_name_column(engine_up):
|
||||
"""报告节点的自查（_MaskedCells）按同一个函数认：同一条查询里引用门店名、客户名都算，各自点名出处，不写值。"""
|
||||
from app.engine.nodes.report import _MaskedCells
|
||||
|
||||
await set_source(":memory:", {"mask_columns": ["name", "phone"]}, probe=False)
|
||||
art = await artifact_store.put_json({"sql": MIXED, "columns": ["store", "customer", "tel", "spend"],
|
||||
"rows": [["广州天河店", "高峰", PHONES[1], 300.0]], "source": SOURCE,
|
||||
"schema_artifact": await schema_of(TABLES, VIEWS)}, kind="query")
|
||||
catalog = {"Q1": {"alias": "Q1", "kind": "query", "artifact": art, "columns": ["store", "customer", "tel", "spend"],
|
||||
"rows": 1, "node_id": "top", "source": SOURCE}}
|
||||
doc = compose_doc("门店 [[v:Q1.r0.store]]，客户 [[v:Q1.r0.customer]]，消费 [[v:Q1.r0.spend]]。", catalog,
|
||||
node_id="write", run_id="r1")
|
||||
store, customer = masked_of((await _MaskedCells(catalog)(doc))["violations"])
|
||||
assert store["ref"] == "v:Q1.r0.store" and "stores.name" in store["message"], store
|
||||
assert customer["ref"] == "v:Q1.r0.customer" and "customers.name" in customer["message"], customer
|
||||
for v in (store, customer):
|
||||
assert "按列名对所有表生效" in v["message"] and "按列名对所有表生效" in v["for_model"], v
|
||||
assert leaks(v) == [] and "广州天河店" not in json.dumps(v, ensure_ascii=False) and "高峰" not in v["message"]
|
||||
@@ -1,261 +0,0 @@
|
||||
"""报告自查只查引用，而且只认真取自遮罩表里那一列的格子（复核第 3 轮）。
|
||||
|
||||
复核发现：
|
||||
- 第 2 轮加的按值自查把两个字的遮罩值当子串去认：客户里有一位叫「高峰」，正文写「本周销售高峰」就判 masked_cell，正式
|
||||
运行（默认违规即失败）因此失败，报错还叫人去改写作指令
|
||||
- 按值自查连同名列的值一起收：mask_columns 设了 name（本意是客户姓名），正文提到门店名（stores.name）同样判失败
|
||||
- 引用同理：遮罩按列名设在整个数据源上（不带表名），s.name AS store 按列名认也算取自遮罩列「name」
|
||||
|
||||
现在自查只认写作者用 [[v:Qn.rX.列]] 引用的格子（正文里照抄的号码本来就会被「数字没有出处」拦下），并且按 SQL 的血缘核一遍
|
||||
这一列的值取自哪张表的哪一列：遮罩列在几张表里都有、其中有的表里另有别的遮罩列时（customers 有 name 和 phone，stores 只有
|
||||
name），只算那几张表。证据面板、可验证报告照列名多遮，不受影响。
|
||||
|
||||
真跑的部分用虚构的 SQLite 零售库（号码、姓名、门店名都是编的），数据源探查过表结构（查询快照里冻结着表结构），写作者是
|
||||
MockChatModel 照剧本写。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.data.engine import engines
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource, Run, RunEvent
|
||||
from app.engine.evidence import SCHEMA_SNAPSHOT, compose_doc
|
||||
from app.engine.runner import run_manager
|
||||
|
||||
SOURCE = "m2shop"
|
||||
STORES = [(1, "杭州西湖店", "华东"), (2, "广州天河店", "华南")]
|
||||
CUSTOMERS = [(1, "陈思远", "19900000801"), (2, "高峰", "19900000802"), (3, "林若溪", "19900000803")]
|
||||
ORDERS = [(1, 1, 1, 120.5), (2, 2, 2, 300.0), (3, 1, 2, 80.0), (4, 2, 3, 210.25)]
|
||||
PHONES = [c[2] for c in CUSTOMERS]
|
||||
|
||||
SQL_STORE = ("SELECT s.name AS store, ROUND(SUM(o.amount), 2) AS sales FROM stores s JOIN orders o ON o.store_id = s.id "
|
||||
"GROUP BY s.name ORDER BY sales DESC")
|
||||
SQL_CUSTOMER = ("SELECT c.name AS customer, ROUND(SUM(o.amount), 2) AS spend FROM customers c "
|
||||
"JOIN orders o ON o.customer_id = c.id GROUP BY c.name ORDER BY spend DESC")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
async def set_source(path, options: dict, *, probe: bool = True) -> str:
|
||||
"""数据源 m2shop：probe 时探查表结构（工具查询时把它冻结进快照），不然清空。名字只这个文件用，探查的结构不会
|
||||
冻结进别的测试的快照。"""
|
||||
from app.data.introspect import introspect
|
||||
|
||||
async with SessionLocal() as session:
|
||||
row = (await session.execute(select(DataSource).where(DataSource.name == SOURCE))).scalar_one_or_none()
|
||||
if row is None:
|
||||
row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
|
||||
session.add(row)
|
||||
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
|
||||
row.options = dict(options)
|
||||
row.schema_cache = (await introspect(row)) if probe else None
|
||||
await session.commit()
|
||||
return row.id
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def shop(tmp_path, engine_up):
|
||||
path = tmp_path / "shop.db"
|
||||
db = sqlite3.connect(path)
|
||||
db.executescript("CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT, region TEXT);"
|
||||
"CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);"
|
||||
"CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER, customer_id INTEGER, amount REAL);")
|
||||
db.executemany("INSERT INTO stores VALUES (?, ?, ?)", STORES)
|
||||
db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
|
||||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ORDERS)
|
||||
db.commit()
|
||||
db.close()
|
||||
source_id = await set_source(path, {"mask_columns": ["name", "phone"]})
|
||||
await engines.invalidate(source_id)
|
||||
yield path
|
||||
await set_source(path, {}, probe=False)
|
||||
await engines.invalidate(source_id)
|
||||
|
||||
|
||||
def node(nid, ntype, **config):
|
||||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||||
|
||||
|
||||
def graph(sql: str, **report) -> dict:
|
||||
nodes = [node("start", "input"), node("top", "tool", tool=f"db_query__{SOURCE}", args={"sql": sql}),
|
||||
node("write", "report", instructions="写一句", **report),
|
||||
node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}])]
|
||||
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||||
|
||||
|
||||
async def finish(monkeypatch, spec: dict, text: str) -> Run:
|
||||
from app.providers import mock_model
|
||||
|
||||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
|
||||
run_id = (await run_manager.start(graph=spec, input_payload={})).id
|
||||
for _ in range(400):
|
||||
await asyncio.sleep(0.05)
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Run, run_id)
|
||||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||||
return row
|
||||
raise AssertionError(f"运行 {run_id} 没跑完：{row.status}")
|
||||
|
||||
|
||||
async def checked_of(run_id: str) -> dict:
|
||||
async with SessionLocal() as session:
|
||||
q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == "report.checked")
|
||||
return (await session.execute(q)).scalars().one().data
|
||||
|
||||
|
||||
def masked_of(violations: list[dict]) -> list[dict]:
|
||||
return [v for v in violations if v["code"] == "masked_cell"]
|
||||
|
||||
|
||||
def leaks(value) -> list[str]:
|
||||
raw = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
|
||||
return [p for p in PHONES if p in raw or p in raw.replace(",", "")]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 正文里写着的字不按值认
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, text", [
|
||||
# 客户里有一位叫「高峰」：常用词「销售高峰」不是在写这位客户
|
||||
(SQL_CUSTOMER, "本周销售高峰出现在周五，最高一位客户消费 [[v:Q1.r0.spend]]。"),
|
||||
# 设了 name 本意是客户姓名：正文用文字写门店名照常
|
||||
(SQL_STORE, "销售额最高的门店是广州天河店，销售额 [[v:Q1.r0.sales]]。"),
|
||||
])
|
||||
async def test_words_that_happen_to_equal_masked_values_do_not_fail_a_formal_report(monkeypatch, shop, sql, text):
|
||||
"""违规即失败（正式运行的默认值）：正文里的字和遮罩列中的值相同，不判 masked_cell，运行照常成功。"""
|
||||
row = await finish(monkeypatch, graph(sql, on_violation="fail"), text)
|
||||
assert row.status == "succeeded", row.error
|
||||
checked = await checked_of(row.id)
|
||||
assert masked_of(checked["violations"]) == [] and checked["ok"] is True, checked["violations"]
|
||||
|
||||
|
||||
async def test_a_phone_copied_into_the_text_is_stopped_by_the_number_check(monkeypatch, shop):
|
||||
"""照抄成正文的号码不按值认，「数字没有出处」照样拦下：交回重写、仍不改就判失败；违规里没有 masked_cell。"""
|
||||
row = await finish(monkeypatch, graph(SQL_CUSTOMER, on_violation="fail"),
|
||||
f"消费最高的客户电话 {PHONES[1]}，共消费 [[v:Q1.r0.spend]]。")
|
||||
assert row.status == "failed", row.error
|
||||
checked = await checked_of(row.id)
|
||||
assert {v["code"] for v in checked["violations"]} == {"uncited_number"}, checked["violations"]
|
||||
assert "遮罩" not in (row.error or "")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 引用：只认真取自遮罩表里那一列的
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_citing_a_store_name_is_not_flagged_though_name_is_masked(monkeypatch, shop):
|
||||
"""s.name AS store：门店表只有 name、没有别的遮罩列，customers 里 name 和 phone 都在——设了 name 管的是客户姓名。"""
|
||||
row = await finish(monkeypatch, graph(SQL_STORE, on_violation="fail"),
|
||||
"销售额最高的门店是 [[v:Q1.r0.store]]，销售额 [[v:Q1.r0.sales]]。")
|
||||
assert row.status == "succeeded", row.error
|
||||
assert masked_of((await checked_of(row.id))["violations"]) == []
|
||||
|
||||
|
||||
async def test_citing_a_customer_name_is_flagged_and_names_the_masked_column(monkeypatch, shop):
|
||||
"""c.name AS customer：取自客户表的 name，照旧拦下；说明只点名它取自的那一列遮罩列，不把 phone 也列进来。"""
|
||||
row = await finish(monkeypatch, graph(SQL_CUSTOMER, on_violation="flag"),
|
||||
"消费最高的客户是 [[v:Q1.r0.customer]]，共消费 [[v:Q1.r0.spend]]。")
|
||||
assert row.status == "succeeded", row.error
|
||||
[v] = masked_of((await checked_of(row.id))["violations"])
|
||||
assert v["ref"] == "v:Q1.r0.customer" and "取自设置了遮罩的列「name」" in v["message"], v["message"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 认法：_cited_masks（报告节点经 _MaskedCells 调用）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: 冻结的表结构（introspect 的形状）：stores 只有 name，customers 有 name 和 phone
|
||||
TABLES = {"stores": ["id", "name", "region"], "customers": ["id", "name", "phone"],
|
||||
"orders": ["id", "store_id", "customer_id", "amount"]}
|
||||
|
||||
|
||||
async def schema_of(tables: dict[str, list[str]]) -> str:
|
||||
return await artifact_store.put_json(
|
||||
{"source": SOURCE, "tables": {t: {"columns": [{"name": c, "type": "TEXT"} for c in cols]}
|
||||
for t, cols in tables.items()}}, kind=SCHEMA_SNAPSHOT)
|
||||
|
||||
|
||||
async def cited(sql: str, columns: list[str], *, masks: list[str], schema: bool = True,
|
||||
tables: dict[str, list[str]] = TABLES) -> dict[str, list[str]]:
|
||||
"""一份快照按报告自查认出的列：{结果列: 它取自的遮罩列}。"""
|
||||
from app.api.evidence import _cited_masks, _Masks
|
||||
|
||||
await set_source(":memory:", {"mask_columns": masks}, probe=False)
|
||||
snap = {"sql": sql, "columns": columns, "rows": [], "source": SOURCE,
|
||||
**({"schema_artifact": await schema_of(tables)} if schema else {})}
|
||||
art = await artifact_store.put_json(snap, kind="query")
|
||||
return await _cited_masks(snap, SOURCE, _Masks(), art)
|
||||
|
||||
|
||||
MIXED = ("SELECT s.name AS store, c.name AS customer, c.phone AS tel, SUM(o.amount) AS spend FROM orders o "
|
||||
"JOIN stores s ON s.id = o.store_id JOIN customers c ON c.id = o.customer_id GROUP BY s.name, c.name, c.phone")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, sql, columns, masks, expected", [
|
||||
("两张表的同名列在一条查询里：只认客户表的", MIXED, ["store", "customer", "tel", "spend"], ["name", "phone"],
|
||||
{"customer": ["name"], "tel": ["phone"]}),
|
||||
("只设了 name：几张表都有 name、哪张都没有别的遮罩列，照列名认", MIXED, ["store", "customer", "tel", "spend"],
|
||||
["name"], {"store": ["name"], "customer": ["name"]}),
|
||||
("直接选出门店表的 name（列名对上遮罩列）", "SELECT name, region FROM stores", ["name", "region"],
|
||||
["name", "phone"], {}),
|
||||
("* 展开门店表", "SELECT * FROM stores", ["id", "name", "region"], ["name", "phone"], {}),
|
||||
("* 展开客户表", "SELECT * FROM customers", ["id", "name", "phone"], ["name", "phone"],
|
||||
{"name": ["name"], "phone": ["phone"]}),
|
||||
("经 WITH 转两道", "WITH a AS (SELECT c.phone AS tel FROM customers c), b AS (SELECT tel AS contact FROM a) "
|
||||
"SELECT contact FROM b", ["contact"], ["phone"], {"contact": ["phone"]}),
|
||||
("列名和遮罩列一样，值却取自别的列", "SELECT region AS phone FROM stores", ["phone"], ["phone"], {}),
|
||||
("UNION：有一支取自客户表就算", "SELECT name FROM stores UNION ALL SELECT name FROM customers", ["name"],
|
||||
["name", "phone"], {"name": ["name"]}),
|
||||
])
|
||||
async def test_cited_masks_follow_the_table_the_value_comes_from(engine_up, label, sql, columns, masks, expected):
|
||||
assert await cited(sql, columns, masks=masks) == expected, label
|
||||
|
||||
|
||||
async def test_without_a_frozen_schema_the_mask_is_read_by_column_name(engine_up):
|
||||
"""数据源没探查过表结构（快照里没有冻结的表结构）：认不出同名的列还在哪几张表里，照列名认，门店名也算；值根本不取自
|
||||
遮罩列的（region AS phone）照样不算。"""
|
||||
assert await cited(MIXED, ["store", "customer", "tel", "spend"], masks=["name", "phone"], schema=False) == \
|
||||
{"store": ["name"], "customer": ["name"], "tel": ["phone"]}
|
||||
assert await cited("SELECT region AS phone FROM stores", ["phone"], masks=["phone"], schema=False) == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
# 表值函数：拆的是号码，血缘认不出出处，照遮罩那一套认
|
||||
("SELECT x.value AS p FROM customers c, json_each(c.phone) x", ["p"]),
|
||||
# 没起别名的表达式：血缘按列名找不到这一列
|
||||
("SELECT id, UPPER(phone) FROM customers", ["id", "UPPER(phone)"]),
|
||||
])
|
||||
async def test_columns_the_lineage_cannot_place_stay_flagged(engine_up, sql, columns):
|
||||
assert set(await cited(sql, columns, masks=["name", "phone"])) == {columns[-1].lower()}
|
||||
|
||||
|
||||
async def test_the_report_node_flags_only_the_customer_column(engine_up):
|
||||
"""报告节点的自查（_MaskedCells）按同一个函数认：同一条查询里引用门店名不算、引用客户名算。"""
|
||||
from app.engine.nodes.report import _MaskedCells
|
||||
|
||||
await set_source(":memory:", {"mask_columns": ["name", "phone"]}, probe=False)
|
||||
art = await artifact_store.put_json({"sql": MIXED, "columns": ["store", "customer", "tel", "spend"],
|
||||
"rows": [["广州天河店", "高峰", PHONES[1], 300.0]], "source": SOURCE,
|
||||
"schema_artifact": await schema_of(TABLES)}, kind="query")
|
||||
catalog = {"Q1": {"alias": "Q1", "kind": "query", "artifact": art, "columns": ["store", "customer", "tel", "spend"],
|
||||
"rows": 1, "node_id": "top", "source": SOURCE}}
|
||||
doc = compose_doc("门店 [[v:Q1.r0.store]]，客户 [[v:Q1.r0.customer]]，消费 [[v:Q1.r0.spend]]。", catalog,
|
||||
node_id="write", run_id="r1")
|
||||
[v] = masked_of((await _MaskedCells(catalog)(doc))["violations"])
|
||||
assert v["ref"] == "v:Q1.r0.customer" and "「name」" in v["message"] and leaks(v) == [], v
|
||||
@@ -184,10 +184,11 @@ async def test_the_sure_columns_follow_the_dialect_of_the_source(engine_up):
|
||||
|
||||
|
||||
async def test_a_failure_points_at_the_instructions_not_at_the_model(monkeypatch, mall):
|
||||
"""遮罩列多半是写作指令点名要的：判失败时只说去掉指令里的引用，不建议换模型、不建议先改成标注。说明点名取自哪一列。"""
|
||||
"""遮罩列多半是写作指令点名要的：判失败时只说去掉指令里的引用，不建议换模型、不建议先改成标注。说明点名取自哪张表
|
||||
的哪一列（数据源没探查过表结构，SQL 里写着 c.phone、c 是 customers）。"""
|
||||
row = await finish(monkeypatch, graph(SQL_TOP, on_violation="fail"), "联系电话 [[v:Q1.r0.tel]]。")
|
||||
assert row.status == "failed", row.error
|
||||
assert "指令中去掉对这些列的引用" in row.error and "按 SQL 认出取自设置了遮罩的列「phone」" in row.error, row.error
|
||||
assert "指令中去掉对这些列的引用" in row.error and "按 SQL 认出取自 customers.phone" in row.error, row.error
|
||||
assert "指令遵循能力" not in row.error and "标注" not in row.error and leaks(row.error) == []
|
||||
|
||||
|
||||
|
||||
@@ -248,9 +248,9 @@ export const ABSENCE_CLAIM_TEXT = {
|
||||
}
|
||||
|
||||
/**
|
||||
* 报告直接引用了取自遮罩列的格子（报告自查的 masked_cell：[[v:Q2.r0.tel]]，c.phone AS tel 这种改了名字的也算，别的表里
|
||||
* 同名的列不算），重写后仍在、「违规处置」为标注时照常产出：时间线上那一行（日志 report_masked_cell）和证据面板违规清单
|
||||
* 里的叫法。自查只查引用，不按值在正文里找遮罩值。
|
||||
* 报告直接引用了取自遮罩列的格子（报告自查的 masked_cell：[[v:Q2.r0.tel]]，c.phone AS tel 这种改了名字的也算；遮罩按
|
||||
* 列名对数据源里所有表生效，视图、别的表里同名的列也算），重写后仍在、「违规处置」为标注时照常产出：时间线上那一行（日志
|
||||
* report_masked_cell）和证据面板违规清单里的叫法。自查只查引用，不按值在正文里找遮罩值。
|
||||
* 只说记为缺口，不说出具：探索运行、没设出具契约的成果不出具，报告接在侧枝上时缺口也进不了出具，降不降档由出具那一行说
|
||||
*/
|
||||
export const MASKED_CELL_TEXT = {
|
||||
|
||||
Reference in New Issue
Block a user