From 1cfa5a524d7e23a99a730a922d4e81f0605c27e2 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 12 Oct 2026 13:50:31 +0800 Subject: [PATCH] =?UTF-8?q?M2=20=E5=A4=8D=E6=A0=B8=E7=AC=AC=204=20?= =?UTF-8?q?=E8=BD=AE=EF=BC=9A=E6=8A=A5=E5=91=8A=E8=87=AA=E6=9F=A5=E8=AE=A4?= =?UTF-8?q?=E9=81=AE=E7=BD=A9=E5=88=97=E5=92=8C=E8=AF=81=E6=8D=AE=E9=9D=A2?= =?UTF-8?q?=E6=9D=BF=E5=90=8C=E4=B8=80=E4=B8=AA=E5=8F=A3=E5=BE=84=EF=BC=8C?= =?UTF-8?q?=E5=93=AA=E5=BC=A0=E8=A1=A8=EF=BC=88=E5=90=AB=E8=A7=86=E5=9B=BE?= =?UTF-8?q?=EF=BC=89=E9=87=8C=E5=8F=AB=E8=BF=99=E4=B8=AA=E5=90=8D=E5=AD=97?= =?UTF-8?q?=E7=9A=84=E5=88=97=E9=83=BD=E7=AE=97=EF=BC=9B=E7=BB=93=E8=AE=BA?= =?UTF-8?q?=E5=8F=A5=E8=A3=81=E5=88=A4=E7=9A=84=E6=91=98=E5=BD=95=E6=8C=89?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=E9=80=90=E4=BB=BD=E9=81=AE=E7=A1=AE=E5=AE=9A?= =?UTF-8?q?=E5=8F=96=E8=87=AA=E9=81=AE=E7=BD=A9=E5=88=97=E7=9A=84=E5=88=97?= =?UTF-8?q?=EF=BC=9B=E6=B2=A1=E8=B5=B7=E5=88=AB=E5=90=8D=E3=80=81=E7=9B=96?= =?UTF-8?q?=E4=BD=8F=E5=85=A8=E9=83=A8=E7=BB=93=E6=9E=9C=E5=88=97=E7=9A=84?= =?UTF-8?q?=E8=A1=A8=E8=BE=BE=E5=BC=8F=E7=AE=97=E7=A1=AE=E5=AE=9A=E5=8F=96?= =?UTF-8?q?=E8=87=AA=E9=81=AE=E7=BD=A9=E5=88=97=EF=BC=8C=E5=80=BC=E7=85=A7?= =?UTF-8?q?=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第 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 加没起别名的表达式照收值；违规说明的断言跟着改 --- backend/app/api/evidence.py | 204 ++++++++--- backend/app/engine/evidence.py | 38 +- backend/app/engine/nodes/report.py | 16 +- backend/tests/test_evidence_mask_memo.py | 17 + .../tests/test_evidence_masks_like_bundle.py | 12 +- backend/tests/test_judge_excerpt_masks.py | 210 +++++++++++ backend/tests/test_report_masked_any_table.py | 329 ++++++++++++++++++ backend/tests/test_report_masked_owners.py | 261 -------------- backend/tests/test_report_masked_precision.py | 5 +- frontend/src/lib/terms.ts | 6 +- 10 files changed, 750 insertions(+), 348 deletions(-) create mode 100644 backend/tests/test_judge_excerpt_masks.py create mode 100644 backend/tests/test_report_masked_any_table.py delete mode 100644 backend/tests/test_report_masked_owners.py diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index bbed0926..0c31b5e7 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -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: diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 3fd0eb6f..757edfa6 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -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、重写的日志、出具缺口，这些地方不能再露一次。带着 diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index 5febc3eb..3f3891f7 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -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，接着跑时写作者会重写一稿：缓存里先前那一稿的判定对不上指纹，往后翻 diff --git a/backend/tests/test_evidence_mask_memo.py b/backend/tests/test_evidence_mask_memo.py index 86bf6758..a23f1e18 100644 --- a/backend/tests/test_evidence_mask_memo.py +++ b/backend/tests/test_evidence_mask_memo.py @@ -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], "已遮罩") + + # -------------------------------------------------------------------------- # 按快照工件在进程里记下 # -------------------------------------------------------------------------- diff --git a/backend/tests/test_evidence_masks_like_bundle.py b/backend/tests/test_evidence_masks_like_bundle.py index e26f2565..de11723f 100644 --- a/backend/tests/test_evidence_masks_like_bundle.py +++ b/backend/tests/test_evidence_masks_like_bundle.py @@ -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") diff --git a/backend/tests/test_judge_excerpt_masks.py b/backend/tests/test_judge_excerpt_masks.py new file mode 100644 index 00000000..3b3dd1a8 --- /dev/null +++ b/backend/tests/test_judge_excerpt_masks.py @@ -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"]) != [] diff --git a/backend/tests/test_report_masked_any_table.py b/backend/tests/test_report_masked_any_table.py new file mode 100644 index 00000000..ffa34620 --- /dev/null +++ b/backend/tests/test_report_masked_any_table.py @@ -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"] diff --git a/backend/tests/test_report_masked_owners.py b/backend/tests/test_report_masked_owners.py deleted file mode 100644 index 6f5e7e6c..00000000 --- a/backend/tests/test_report_masked_owners.py +++ /dev/null @@ -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 diff --git a/backend/tests/test_report_masked_precision.py b/backend/tests/test_report_masked_precision.py index 25e6e024..1e0353cb 100644 --- a/backend/tests/test_report_masked_precision.py +++ b/backend/tests/test_report_masked_precision.py @@ -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) == [] diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index c438fcfb..59233774 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -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 = {