From 76fa42849cf1f66558bef135e2f57bc4205e4b48 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 7 Oct 2026 00:13:50 +0800 Subject: [PATCH] =?UTF-8?q?M2=20=E8=AF=81=E6=8D=AE=E9=9D=A2=E6=9D=BF?= =?UTF-8?q?=E5=92=8C=E6=8E=A8=E6=BC=94=E8=AE=A4=E5=BE=97=E5=87=BA=E6=94=B9?= =?UTF-8?q?=E4=BA=86=E5=90=8D=E5=AD=97=E7=9A=84=E9=81=AE=E7=BD=A9=E5=88=97?= =?UTF-8?q?=EF=BC=8C=E6=8C=87=E6=A0=87=E6=9C=AC=E8=BA=AB=E7=9A=84=E5=8E=9F?= =?UTF-8?q?=E5=80=BC=E4=B9=9F=E4=B8=8D=E9=9C=B2=EF=BC=9A=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E6=AD=A5=E9=AA=A4=E3=80=81=E6=8C=87=E6=A0=87=E7=9A=84=E8=BE=93?= =?UTF-8?q?=E5=85=A5=E3=80=81=E6=8E=A8=E6=BC=94=E7=9A=84=E5=8B=BE=E8=A1=8C?= =?UTF-8?q?=E4=B8=8E=E9=81=AE=E7=BD=A9=E3=80=81=E5=90=88=E5=B9=B6=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E6=8C=89=20SQL=20=E8=AE=A4=E5=8F=96=E8=87=AA=E9=81=AE?= =?UTF-8?q?=E7=BD=A9=E5=88=97=E7=9A=84=E7=BB=93=E6=9E=9C=E5=88=97=EF=BC=88?= =?UTF-8?q?c.phone=20AS=20tel=20=E8=BF=99=E7=B1=BB=E5=88=AB=E5=90=8D?= =?UTF-8?q?=E3=80=81=E5=87=BD=E6=95=B0=E3=80=81=E5=AD=90=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E3=80=81WITH=EF=BC=89=EF=BC=8C=E5=92=8C=E5=8F=AF=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=E6=8A=A5=E5=91=8A=E3=80=81=E6=A0=B8=E6=9F=A5=20agent?= =?UTF-8?q?=20=E5=90=8C=E4=B8=80=E5=A5=97=E8=AE=A4=E6=B3=95=EF=BC=9B?= =?UTF-8?q?=E7=94=A8=E5=88=B0=E9=81=AE=E7=BD=A9=E5=88=97=E7=9A=84=E6=8C=87?= =?UTF-8?q?=E6=A0=87=EF=BC=8C=E6=8E=A8=E6=BC=94=E7=AD=94=E5=A4=8D=E9=87=8C?= =?UTF-8?q?=E7=9A=84=20old=E3=80=81rendered=5Fold=20=E5=92=8C=E8=AF=81?= =?UTF-8?q?=E6=8D=AE=E9=9D=A2=E6=9D=BF=E6=8C=87=E6=A0=87=E6=AD=A5=E9=AA=A4?= =?UTF-8?q?=E7=9A=84=20value=20=E5=86=99=E3=80=8C=E5=B7=B2=E9=81=AE?= =?UTF-8?q?=E7=BD=A9=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 验收里证据面板对 c.phone AS tel 的快照给出 masked=[]、三个号码原样交出；推演把 phone_raw 判成 masked、输入置空，指标的 old 却还是那串号码。现在： - 认遮罩列只有一个函数 api/evidence.py 的 _snapshot_masks（查询当时记下的加数据源现在设的，再经 engine/masking.py 的 _derived_hidden 按 SQL 认改了名字的；SQL 解析不了又写着遮罩列的整份遮），查询步骤、_InputMasks、推演的 _mask_of 都调它；_Masks 顺带记数据源类型，按它的方言解析 - 合并查询按逐格来历遮：输入快照里改了名字的遮罩列（输入是合并结果的再往下追）也遮到合并结果里 - 指标的原值按 recompute.hides_value 一个判断：有输入取自遮罩列（含只嵌在下标里的），或者最外层的输入遮没遮核对不了；推演结果标 withheld，报告里引用它的地方 rendered_old 同样写「已遮罩」 - 面板底下的遮罩说明改成「这些列取自数据源中设置了遮罩的列」：遮的可能是别名列 --- backend/app/api/evidence.py | 117 +++++-- backend/app/api/whatif.py | 27 +- backend/app/engine/recompute.py | 54 ++- backend/tests/test_evidence_masked_aliases.py | 309 ++++++++++++++++++ backend/tests/test_whatif_masked_values.py | 268 +++++++++++++++ 5 files changed, 731 insertions(+), 44 deletions(-) create mode 100644 backend/tests/test_evidence_masked_aliases.py create mode 100644 backend/tests/test_whatif_masked_values.py diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index ead0c107..a46e0fb0 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -127,7 +127,8 @@ NONE_NOTE = "本次运行没有报告文档，也没有出具契约，没有可 NO_DOC_NOTE = "本次运行的出具契约采用引用模式核对，但报告没有产出可核对的文档" MASKED = "已遮罩" -MASK_NOTE = "这些列在数据源中设置了遮罩，面板上不显示原值。遮罩只减少暴露，不是安全边界" +#: 面板底下的遮罩说明。遮的列可能是遮罩列本身，也可能是 SQL 里给它起了别名、套了函数得来的结果列（c.phone AS tel） +MASK_NOTE = "这些列取自数据源中设置了遮罩的列，面板上不显示原值。遮罩只减少暴露，不是安全边界" #: 查询条目的数据源现在找不到了（改名或删掉了）：只能按快照里记下的、查询当时的遮罩来遮 SOURCE_GONE = "数据源「{source}」已不存在（可能已改名或删除），按查询时记录的遮罩处理" #: 查询步骤给被引用的行前后各带几行 @@ -878,6 +879,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s wanted: dict[str, list[tuple[Any, Any]]] = {} hide = _InputMasks(sealed, masks, str(entry.get("node_id") or ""), artifact) withheld: set[str] = set() + unverified: set[str] = set() for item in (metric or {}).get("inputs") or []: one = {"step": "input", **item} artifact_of = item.get("artifact") @@ -889,10 +891,16 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s if (hidden := await hide.check(item)) is not None: one = _withheld(one, *hidden) withheld.add(str(item.get("path"))) + if hidden[1] is not None: + unverified.add(str(item.get("path"))) inputs.append(one) if withheld: # 代入式里那几处同样写「已遮罩」：不然标签里遮着、代入式里亮着，同一个值换个地方照样看得到 step["substituted"] = recompute.substituted_without(metric or {}, withheld, MASKED) + if step["value"] is not None and recompute.hides_value(metric or {}, withheld - unverified, unverified): + # 指标的值本身也不给，和推演的原值同一个判断：指标可能就是遮罩列的那一格（cell(nodes.q, 0, 'phone')），输入里 + # 遮着、这里亮着等于没遮。报告上的字（rendered）是读者已经看到的，照旧 + step["value"] = MASKED queries = [s for a, cells in wanted.items() for s in await _query_steps(a, cells, doc, sealed, masks)] return [step, *inputs, *queries] @@ -916,8 +924,8 @@ SNAPSHOT_UNVERIFIED = "查询快照无法读取，核对不了这一格是否取 class _InputMasks: - """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询按它 - 合并的那几个数据源算。 + """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_snapshot_masks）：查询当时记下的加数据源现在设的，按 SQL + 认出改了名字的（c.phone AS tel 的 tel）也算，合并查询按它合并的那几个数据源和逐格来历算。 取自哪份查询结果的哪几列按 recompute.input_origin 认，推演、可验证报告是同一个认法：cell() 的格、Agent 交来的字段按 定位里的列认；vars.res.rows[0][2] 这种按位置取的一格卡上没记定位，顺着取值链认，从写变量的节点在口径卡算这一次之前 @@ -975,13 +983,8 @@ class _InputMasks: async def _hidden(self, artifact: str | None, source: str | None, snap: dict[str, Any] | None) -> set[str]: key = (artifact, source) if key not in self._memo: - if snap is not None and snap.get("source") == MERGE_SOURCE: - hidden, _ = await _merge_masks(snap, self._masks) - else: - info = self._sealed.queries.get(artifact) if artifact else None - current, _ = await self._masks.of((info or {}).get("source") or source or (snap or {}).get("source")) - hidden = set(current) | (_recorded_masks(snap) if snap is not None else set()) - self._memo[key] = hidden + info = self._sealed.queries.get(artifact) if artifact else None + self._memo[key], _ = await _snapshot_masks(snap, (info or {}).get("source") or source, self._masks) return self._memo[key] @@ -1283,10 +1286,12 @@ def _challenge_counts(raw: Any, claims: list[dict[str, Any]]) -> dict[str, int]: class _Masks: - """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。""" + """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。顺带记下数据源的类型（kind）， + 按 SQL 认改了名字的遮罩列时照它的方言解析（_snapshot_masks）；数据源不在了的按通用读法。""" def __init__(self) -> None: self._memo: dict[str, tuple[set[str], bool]] = {} + self._kinds: dict[str, str | None] = {} async def of(self, source: Any) -> tuple[set[str], bool]: name = str(source or "") @@ -1296,8 +1301,56 @@ class _Masks: async with SessionLocal() as session: row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none() self._memo[name] = ({c.lower() for c in masked_columns(row.options if row else None)}, row is not None) + self._kinds[name] = row.kind if row is not None else None return self._memo[name] + async def kind(self, source: Any) -> str | None: + name = str(source or "") + if not name: + return None + await self.of(name) + return self._kinds.get(name) + + +def _derived(snap: dict[str, Any], hidden: set[str], kind: str | None) -> set[str]: + """一份查询快照里值取自遮罩列（hidden，小写）的结果列（小写，含 hidden 本身）：除了列名对上的，SQL 里给遮罩列起了 + 别名、套了函数、经子查询或 WITH 转一道得来的也算。认法是 engine/masking.py 的 _derived_hidden，核查 agent 在工具边界 + 遮、可验证报告判快照附不附带用的同一套；SQL 解析不了、里面又写着遮罩列（或者 * 这类不写列名也会带出遮罩列的写法） + 的，整份快照的列都算。合并查询的快照它不看 SQL，原样交回（合并的遮罩由 _merge_masks 按逐格来历算）。 + + 分析本身出了意外：整份的列都算（宁可多遮）。纯计算，一条 SQL 不到 1 毫秒，在调用方的线程里算：推演在工作线程里 + 等事件循环查库（api/whatif.py 的 _Hidden），这里再开线程会和它争同一个线程池。""" + if not hidden: + return set() + try: + # masking 要用这里的说法（MASKED），也是用到时才导入：两边都不在模块顶上互相导入 + from app.engine.masking import _derived_hidden + + return set(hidden) | _derived_hidden(snap, set(hidden), kind)[0] + except Exception: # noqa: BLE001 - 认不准就整份遮掉 + logger.warning("证据面板：按 SQL 认改了名字的遮罩列时出错，整份快照的列都按遮罩处理", exc_info=True) + return set(hidden) | {str(c).lower() for c in snap.get("columns") or []} + + +async def _snapshot_masks(snap: dict[str, Any] | None, source: Any, masks: _Masks, *, + depth: int = 0) -> tuple[set[str], list[str]]: + """一份查询快照要遮的列（小写）和已经不在的数据源：证据面板的查询步骤、指标的输入、推演、报告自查（engine/nodes/ + report.py）都按这一个函数认。 + + - 遮的是「查询当时记下的」和「数据源现在设的」两者之和：事后加的遮罩照样生效，数据源改名、删掉了也不会把当时遮着 + 的列亮出来；数据源不在了的，它的名字进第二项 + - 列名对上的之外，按 SQL 认出改了名字的遮罩列（_derived） + - 合并查询不是数据源：按它合并的那几个数据源和逐格来历算（_merge_masks） + - snap 为 None（快照读不出来）：只能按 source 现在设的遮罩算，没有 SQL 可认""" + if snap is not None and snap.get("source") == MERGE_SOURCE: + return await _merge_masks(snap, masks, depth=depth) + name = source or (snap or {}).get("source") + hidden, found = await masks.of(name) + hidden = set(hidden) | (_recorded_masks(snap) if snap is not None else set()) + if snap is not None and hidden: + hidden = _derived(snap, hidden, await masks.kind(name)) + return hidden, ([] if found or not name else [str(name)]) + async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str, Any], sealed: _Sealed, masks: _Masks, *, hint: bool = False) -> dict[str, Any]: @@ -1343,22 +1396,15 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str rows = snap["rows"] step["source"] = step["source"] or snap.get("source") merged = snap.get("source") == MERGE_SOURCE + # 遮的列：查询当时记下的加数据源现在设的，按 SQL 认出改了名字的（c.phone AS tel）也遮。合并查询不是数据源，按它 + # 合并的那几个数据源和逐格来历算，不能拿「合并查询」去找数据源，找不到再说它「已不存在」（_snapshot_masks） + hidden, gone = await _snapshot_masks(snap, step["source"], masks) + if gone: + step["mask_note"] = SOURCE_GONE.format(source="、".join(gone)) if merged: - # 合并查询不是数据源：遮罩按它合并的那几个数据源算（_merge_masks），不能拿「合并查询」去找数据源， - # 找不到再说它「已不存在」 - hidden, gone = await _merge_masks(snap, masks) - if gone: - step["mask_note"] = SOURCE_GONE.format(source="、".join(gone)) step["merge"] = _merge_view(snap, cells, catalog, sealed) if hint and _leads_to_upload(snap, cells): step["provenance"] = True - else: - # 遮的是「查询当时记下的」和「数据源现在设的」两者之和：事后加的遮罩照样生效，数据源改名、 - # 删掉了也不会把当时遮着的列亮出来 - hidden, found = await masks.of(step["source"]) - hidden = hidden | _recorded_masks(snap) - if not found: - step["mask_note"] = SOURCE_GONE.format(source=step["source"]) masked = {i for i, c in enumerate(columns) if c.lower() in hidden} index = sorted({i for r in rows_hit for i in range(r - WINDOW, r + WINDOW + 1) if 0 <= i < len(rows)}) step.update(sql=snap.get("sql"), columns=columns, total_rows=len(rows), truncated=bool(snap.get("truncated")), @@ -1497,12 +1543,13 @@ def _input_checks(artifact: str) -> list[dict[str, Any]]: return _query_checks(snap) if isinstance(snap, dict) else [] -async def _merge_masks(snap: dict[str, Any], masks: _Masks) -> tuple[set[str], list[str]]: +async def _merge_masks(snap: dict[str, Any], masks: _Masks, *, depth: int = 0) -> tuple[set[str], list[str]]: """合并结果要遮的列（小写）和已经不在的数据源： - 合并当时记下的（快照的 mask_columns：各输入当时的遮罩列，加上来自遮罩列、换了名字的结果列）； - 它合并的每个数据源现在设的（按列名）； - - 结果列有逐格来历、来自某个数据源现在遮着的列的，按来历遮（换了名字也遮住）。宁可多遮。""" + - 结果列有逐格来历的，按来历遮：来自的那一列在输入里要遮（_merge_input_masks：输入快照按查询步骤同一套认，数据源 + 现在设的、SQL 里改了名字的都算，输入本身是合并结果的再往下追），换了名字也遮住。宁可多遮。""" hidden = _recorded_masks(snap) gone: list[str] = [] for name in snap.get("sources") or []: @@ -1513,17 +1560,33 @@ async def _merge_masks(snap: dict[str, Any], masks: _Masks) -> tuple[set[str], l lineage = snap.get("lineage") if isinstance(snap.get("lineage"), dict) else {} origins = lineage.get("sources") or [] by_alias = {str(i.get("alias")): i for i in snap.get("inputs") or [] if isinstance(i, dict)} + inputs: dict[str, set[str]] = {} for name, target in zip(snap.get("columns") or [], lineage.get("columns") or []): if not isinstance(target, dict) or not isinstance(target.get("source"), int) \ or not 0 <= target["source"] < len(origins): continue origin = origins[target["source"]] if isinstance(origins[target["source"]], dict) else {} - source = (by_alias.get(str(origin.get("input"))) or {}).get("source") - if source and source != MERGE_SOURCE and str(target.get("column") or "").lower() in (await masks.of(source))[0]: + alias = str(origin.get("input")) + if alias not in inputs: + inputs[alias] = await _merge_input_masks(by_alias.get(alias) or {}, masks, depth) + if str(target.get("column") or "").lower() in inputs[alias]: hidden.add(str(name).lower()) return hidden, gone +async def _merge_input_masks(item: dict[str, Any], masks: _Masks, depth: int) -> set[str]: + """合并的一个输入（快照 inputs 里的一项）的结果里要遮的列（小写）。读得出输入快照的按 _snapshot_masks 认（SQL 里给 + 遮罩列起了别名的也算）；读不出来、追得太深的，只能按它的数据源现在设的列名算。""" + source = item.get("source") + artifact = str(item.get("artifact") or "") + content, ok = _fetch(artifact) if artifact and depth < MERGE_DEPTH else (None, None) + if ok is True and isinstance(content, dict): + return (await _snapshot_masks(content, source, masks, depth=depth + 1))[0] + if source and source != MERGE_SOURCE: + return set((await masks.of(source))[0]) + return set() + + def _leads_to_upload(snap: dict[str, Any], cells: list[tuple[Any, Any]], depth: int = 0) -> bool: """被引用的格经合并追下去，最后落在上传表格的查询快照上（有 data_version）：界面才去请求推断来源。""" if depth >= MERGE_DEPTH: diff --git a/backend/app/api/whatif.py b/backend/app/api/whatif.py index a8d96a87..202c67e8 100644 --- a/backend/app/api/whatif.py +++ b/backend/app/api/whatif.py @@ -10,7 +10,8 @@ GET /api/runs/{run_id}/evidence/whatif/rows：推演时勾行用，按页取封 只认这次运行封存范围内的工件，和证据接口同一套白名单（api/evidence.py 的 _sealed / _sealed_loader）：口径卡的指标集 从封存的 node.finished 台账找，排除行只认封存范围内交回过的查询快照，nodes.* 的完整值从封存的节点产出工件取， vars.* 的完整值（value_artifact）只认封存的指标集里写着的。还在跑、停在审批的运行也能推演，结果里 sealed 为假： -数据可能还会变。遮罩列（查询当时记下的加数据源现在设的）不给值、也不能拿来推演，见 engine/recompute.py。 +数据可能还会变。遮罩列（查询当时记下的加数据源现在设的，SQL 里改了名字的也算）不给值、也不能拿来推演，用到它的指标 +原值也不给，见 engine/recompute.py。 重算本身是纯函数（engine/recompute.py）。这里负责：认请求、找卡、准备工件和遮罩、找报告里引用的地方、标出可能受影响 的下游口径卡。 @@ -31,17 +32,15 @@ from app.api.evidence import ( REPORT_NOT_FOUND, TAMPERED_QUERY, _Masks, - _merge_masks, - _recorded_masks, _reports, _Sealed, _sealed, _sealed_cards, _sealed_loader, + _snapshot_masks, ) from app.core.artifact_store import load from app.engine import recompute -from app.engine.merge_query import MERGE_SOURCE router = APIRouter(prefix="/api/runs", tags=["whatif"]) @@ -115,7 +114,8 @@ async def evidence_whatif_rows(run_id: str, artifact: str | None = None, offset: """一页查询快照的行：{artifact, columns, rows, row_offset, row_index, total_rows, masked, hash_ok, sealed, note?}。 形状和证据面板的查询步骤一致（界面用同一个函数读）。只认这次运行封存范围内交回过的快照；遮罩列（查询当时记下的 - 加数据源现在设的）写「已遮罩」；读不出来、与哈希不一致的不给行，note 照实说。只读：不追加事件、不落工件。 + 加数据源现在设的，SQL 里改了名字的也算）写「已遮罩」；读不出来、与哈希不一致的不给行，note 照实说。只读：不追加 + 事件、不落工件。 """ if not artifact or not artifact.strip(): raise CodedHTTPException(422, "请求格式有误：artifact 应为查询快照的工件 ID", BAD_REQUEST) @@ -312,9 +312,9 @@ def _loader(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]]) -> An class _Hidden(Mapping[str, frozenset[str]]): """封存范围内的查询快照 → 要遮的列（小写），推演用到哪份才算哪份。 - 遮的列和证据面板查询步骤同一套（api/evidence.py 的 _query_step）：查询当时记下的加数据源现在设的，合并查询按它 - 合并的那几个数据源算。快照读不出来、被改过时照样按台账记的数据源算：遮罩不能因为快照坏了就失效（那一格的值另由 - 重算不给）。 + 遮的列和证据面板查询步骤同一套（api/evidence.py 的 _snapshot_masks）：查询当时记下的加数据源现在设的，按 SQL 认出 + 改了名字的也算，合并查询按它合并的那几个数据源和逐格来历算。快照读不出来、被改过时照样按台账记的数据源算：遮罩 + 不能因为快照坏了就失效（那一格的值另由重算不给）。 在重算的工作线程里取用：快照经推演用的同一个 loader 读（重算本来也要读这一份，读过的不再读盘）；查数据源现在 设的遮罩要访问库，交回事件循环去查（_Masks 是异步的），线程在这里等它。以前每次请求在事件循环里把这次运行的 @@ -353,9 +353,8 @@ class _Hidden(Mapping[str, frozenset[str]]): async def _mask_of(snap: dict[str, Any] | None, info: dict[str, Any], masks: _Masks) -> set[str]: - """一份查询快照要遮的列（小写）。snap 为 None（读不出来）时只能按台账记的数据源现在设的遮罩算。""" - if snap is not None and snap.get("source") == MERGE_SOURCE: - hidden, _ = await _merge_masks(snap, masks) - return hidden - hidden, _ = await masks.of(info.get("source") or (snap or {}).get("source")) - return hidden | (_recorded_masks(snap) if snap is not None else set()) + """一份查询快照要遮的列（小写），和证据面板的查询步骤同一个函数（api/evidence.py 的 _snapshot_masks）：查询当时记下的 + 加数据源现在设的，SQL 里给遮罩列起了别名、套了函数的结果列也算。snap 为 None（读不出来）时只能按台账记的数据源现在 + 设的遮罩算。""" + hidden, _ = await _snapshot_masks(snap, info.get("source"), masks) + return hidden diff --git a/backend/app/engine/recompute.py b/backend/app/engine/recompute.py index fa3b86d9..0db5a295 100644 --- a/backend/app/engine/recompute.py +++ b/backend/app/engine/recompute.py @@ -15,7 +15,9 @@ - vars.*：有 value_artifact（口径卡当时另存的完整值）才还原得出来；老运行照实说当时没有保存 - 容器本身是一份查询结果（带 artifact 的 {columns, rows}，或者和某份快照的 rows 一模一样的列表）：排除行就去掉那几行 - 遮罩列取来的输入不给值、也不拿来推演：改一个输入、排除一行，新值和原值一比就能算出遮住的那一格。vars.res.rows[0][2] - 这种按位置取自查询结果的标量没有定位，顺着取值链认出是哪一列（positional_cell，证据面板的指标步骤用同一个认法） + 这种按位置取自查询结果的标量没有定位，顺着取值链认出是哪一列（positional_cell，证据面板的指标步骤用同一个认法）。 + 用到了遮罩列的指标，它本身的原值同样写「已遮罩」（hides_value，证据面板指标步骤的值同一个判断）：指标的值可能就是 + 那一格，输入里拿掉了、指标上照给，等于没遮 - 落在读不出来（文件不在、与哈希不一致）的查询快照上的输入同样不给值、不推演：那一格遮没遮核对不了，宁可多遮 有的指标不论改什么都推演不了（_Plan.blocked）：卡上没记全输入、用到了遮罩列、最外层的输入还原不了（容器的完整值 @@ -209,6 +211,13 @@ class _Plan: return "unsupported", f"无法推演：{path}（{problem}）" return None + @property + def hides(self) -> bool: + """指标的原值也不给（hides_value 同一个判断）：用到了遮罩列，或者最外层的输入遮没遮核对不了。成立时一定推演 + 不了（blocked 不为 None），新值本来就是空的。""" + return _hides(self.tops, [p for p, i in self.inputs.items() if i.masked], + [p for p, i in self.inputs.items() if i.unreadable]) + @property def stuck(self) -> str | None: """blocked 的原因换成不说「推演」的说法：recompute 交给双盲复算的预检、盲值，那里说「不能推演」驴唇不对马嘴。 @@ -872,7 +881,7 @@ def _result(plan: _Plan, status: str, new: Any, reason: str | None, values: dict else substitute(plan.tree, values) if status == "ok" and plan.tree is not None else None if any(inp.masked or inp.unreadable for inp in plan.inputs.values()): substituted = None # 代入式里代进去的就是遮罩列（或者遮没遮核对不了）的那一格 - return { + out = { "id": metric.get("id"), "name": metric.get("name") or metric.get("id"), "unit": metric.get("unit") or "", "decimals": metric.get("decimals"), "format": metric.get("format") or DEFAULT_FORMAT, "old": _finite(old), "new": _finite(new), "delta": delta, @@ -882,6 +891,13 @@ def _result(plan: _Plan, status: str, new: Any, reason: str | None, values: dict "status": status, "reason": reason, "substituted": substituted, "inputs": [_input_view(inp, values, notes, blocked=blocked) for inp in plan.inputs.values()], } + if plan.hides: + # 指标的值就是（或者算自）遮罩列的那一格：原值同样写「已遮罩」。以前输入里拿掉了、指标的 old 照给，同一个号码 + # 换个地方照样看得到；原因在 reason 里（推演不了的原因）。本来就没有值的照旧是空的，缺输入还得看得出来 + out["withheld"] = True + if old is not None: + out["old"] = out["rendered_old"] = _masked_word() + return out def _render(metric: dict[str, Any], value: Any, conv: str | None = None) -> str: @@ -1026,6 +1042,35 @@ def _fixed(inp: _Input) -> str | None: return None if value is None or is_number(value) else NOT_NUMBER +def hides_value(metric: dict[str, Any], masked: Iterable[str], unverified: Iterable[str] = ()) -> bool: + """指标本身的值要不要一并不给（推演的原值、证据面板指标步骤的值）：masked、unverified 是取自遮罩列、遮没遮核对不了 + 的输入的路径。 + + - 有输入取自遮罩列：不给，哪怕它只嵌在下标里（推演把整个指标按遮罩处理，_Plan.blocked）。指标的值可能就是那一格 + （cell(nodes.q, 0, 'phone')），也可能是算出来的数，宁可多遮：改一个别的输入、拿指标的值一减，就能算出遮住的那一格 + - 最外层的输入遮没遮核对不了：不给，它的值就在指标值里。只嵌在下标里的不算：它只决定取哪一格，取到的值卡上另记着 + 认不出最外层引用（表达式有误）时，有一个遮没遮核对不了就不给。""" + masked, unverified = set(masked), set(unverified) + if masked or not unverified: + return bool(masked) + try: + tops, _ = _refs(parse_expression(str(metric.get("expression") or ""))[0]) + except ExpressionError: + return True + return _hides(tops, masked, unverified) + + +def _hides(tops: Iterable[str], masked: Iterable[str], unverified: Iterable[str]) -> bool: + return any(True for _ in masked) or bool(set(unverified) & set(tops)) + + +def _masked_word() -> str: + """「已遮罩」：证据面板、推演同一个说法。api.evidence 导入了这个模块，用到时再导入。""" + from app.api.evidence import MASKED + + return MASKED + + def substituted_without(metric: dict[str, Any], paths: Iterable[str], token: str) -> str | None: """卡上的代入式，只把 paths 这几个输入代成 token（证据面板给取自遮罩列的输入写「已遮罩」），别的照卡上代。 @@ -1062,6 +1107,7 @@ def report_segments(reports: Iterable[tuple[str, dict[str, Any]]], *, card_node: 只认指向这张卡这一份指标集的引用（目录条目的 node_id、artifact）：循环里同一张卡算过几次，报告引用的是哪一次 就是哪一次。新值按片段自己的换算（|万、|pct……）渲染，和报告里的字同一套规则。指标推演不了（unsupported、masked） 时 rendered_new、changed 为 None：不知道会变成什么。缺输入、当时就显示「—」的引用也列上：推演里补上了值，它会变。 + 原值不给的指标（withheld），rendered_old 同样写「已遮罩」。 """ by_id = {str(r.get("id")): r for r in results} out: list[dict[str, Any]] = [] @@ -1078,8 +1124,10 @@ def report_segments(reports: Iterable[tuple[str, dict[str, Any]]], *, card_node: if result is None: continue new = None if result.get("rendered_new") is None else _render(result, result.get("new"), cite.get("conv")) + # 原值不给的指标（_Plan.hides）：报告里引用它的地方同样写「已遮罩」，推演的答复里不出现它的值 + old = _masked_word() if result.get("withheld") and result.get("old") is not None else seg.get("text") out.append({"report": report_node, "unit": unit.get("id"), "sid": seg.get("id"), "metric": result["id"], - "rendered_old": seg.get("text"), "rendered_new": new, + "rendered_old": old, "rendered_new": new, "changed": None if new is None else new != seg.get("text")}) return out diff --git a/backend/tests/test_evidence_masked_aliases.py b/backend/tests/test_evidence_masked_aliases.py new file mode 100644 index 00000000..4dd104a2 --- /dev/null +++ b/backend/tests/test_evidence_masked_aliases.py @@ -0,0 +1,309 @@ +"""证据面板认得出改了名字的遮罩列：查询步骤、指标的输入、推演的勾行和遮罩，按 SQL 认。 + +验收发现：正式运行的报告没写号码，可验证报告也把 top 查询（c.phone AS tel）的快照只给了哈希，点开正文里的消费额， +片段接口的查询步骤却是 columns=[customer, tel, spend]、masked=[]，三个号码原样交出来。证据面板以前只遮列名恰好等于 +遮罩列的那几列；可验证报告、核查 agent 早已按 SQL 认出别名、函数、子查询、WITH 转一道得来的结果列。现在证据面板 +按同一套认法（engine/masking.py 的 _derived_hidden）： + +- 查询步骤：c.phone AS tel 的 tel 列遮掉，面板底下的遮罩说明带上它 +- 指标的输入：cell() 取 tel 那一格、按位置取的 vars.res.rows[0][1] 都不给值；指标本身的值也不给 +- 推演：勾行接口遮 tel，用到 tel 的指标推演不了、原值不给 +- SQL 解析不了、里面又写着遮罩列的，整份快照的列都遮（保守规则）；没写着的只遮遮罩列本身 +- 合并查询：输入快照里改了名字的遮罩列，按逐格来历遮到合并结果里 +- 没设遮罩时一切照旧 + +真跑一遍图（虚构的 SQLite 商城库，号码是编的），报告由 MockChatModel 照剧本写。 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.core import artifact_store +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run +from app.engine.evidence import iter_segments +from app.engine.runner import run_manager +from app.main import app + +CUSTOMERS = [(1, "客户甲", "199-0000-0101"), (2, "客户乙", "199-0000-0102"), (3, "客户丙", "199-0000-0103"), + (4, "客户丁", "199-0000-0104")] +ORDERS = [(1, 1, 120.5), (2, 2, 300.0), (3, 2, 80.0), (4, 3, 210.25), (5, 4, 50.0), (6, 1, 99.5)] +PHONES = [c[2] for c in CUSTOMERS] +#: 验收原样的写法：号码起了别名 tel，和消费额排在同一份结果里 +SQL_TOP = ("SELECT c.name AS customer, c.phone AS tel, ROUND(SUM(o.amount), 2) AS spend " + "FROM orders o JOIN customers c ON c.id = o.customer_id GROUP BY c.id, c.name, c.phone ORDER BY spend DESC") +TEL_CELL = "cell(nodes.top, 0, 'tel')" +#: 号码那一格按位置取：vars.res 是数据整形解析好的查询结果，第 1 列是 tel +TEL_POS = "vars.res.rows[0][1]" +METRICS = [ + {"id": "top_spend", "name": "最高消费", "unit": "元", "decimals": 2, "format": "thousands", + "expression": "cell(nodes.top, 0, 'spend')"}, + {"id": "tel_len", "name": "号码位数", "unit": "位", "format": "thousands", "expression": f"len({TEL_CELL})"}, + {"id": "tel_pos2", "name": "号码位数两倍", "unit": "位", "format": "thousands", "expression": f"len({TEL_POS}) * 2"}, +] +TEXT = "消费最高的客户花了 [[v:Q1.r0.spend]]，口径卡记 [[m:top_spend]]；号码 [[m:tel_len]]，两倍 [[m:tel_pos2]]。" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def set_source(path, options=None): + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == "mall"))).scalar_one_or_none() + if row is None: + row = DataSource(name="mall", kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True + row.options = dict(options or {}) + await session.commit() + return row.id + + +@pytest.fixture +async def mall(tmp_path): + path = tmp_path / "mall.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);" + "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL);") + db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS) + db.executemany("INSERT INTO orders VALUES (?, ?, ?)", ORDERS) + db.commit() + db.close() + source_id = await set_source(path) + await engines.invalidate(source_id) + yield path + await engines.invalidate(source_id) + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(*nodes) -> dict: + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def graph() -> dict: + return chain( + node("start", "input"), + node("top", "tool", tool="db_query__mall", args={"sql": SQL_TOP}), + node("parse", "transform", mode="json", template="{{ nodes.top }}", assign_to="res"), + node("card", "metrics", caliber="商城口径", caliber_version="v1", metrics=METRICS), + node("write", "report", instructions="写一句", on_violation="flag"), + node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]), + ) + + +async def run_graph(monkeypatch, spec: dict | None = None, text: str = TEXT) -> Run: + from app.providers import mock_model + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text)) + run_id = (await run_manager.start(graph=spec or graph(), input_payload={})).id + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + assert row.status == "succeeded", row.error + return row + raise AssertionError(f"运行 {run_id} 没跑完：{row.status}") + + +async def doc_of(client, run_id: str) -> dict: + [report] = (await client.get(f"/api/runs/{run_id}/evidence")).json()["reports"] + return artifact_store.load(report["doc_artifact"]) + + +async def segment(client, run_id: str, *, text: str | None = None, ref: str | None = None) -> dict: + """按正文里的字或者引用（v:Q1.r0.spend）找片段，取片段接口的答复。""" + doc = await doc_of(client, run_id) + seg = next(s for s in iter_segments(doc) if (text is not None and s["text"] == text) + or (ref is not None and s.get("ref") == ref)) + res = await client.get(f"/api/runs/{run_id}/evidence/segments/{seg['id']}") + assert res.status_code == 200, res.text + return res.json() + + +def leaks(body) -> list[str]: + raw = json.dumps(body, ensure_ascii=False) + return [p for p in PHONES if p in raw] + + +def steps(body: dict, kind: str) -> list[dict]: + return [s for s in body["chain"] if s["step"] == kind] + + +# -------------------------------------------------------------------------- +# 查询步骤 +# -------------------------------------------------------------------------- + + +async def test_a_renamed_masked_column_is_masked_in_the_query_step(client, monkeypatch, mall): + """验收原样：点开正文里直接引用的消费额，查询步骤里 c.phone AS tel 的 tel 列遮掉，号码一个都不交出来。""" + await set_source(mall, {"mask_columns": ["phone"]}) + row = await run_graph(monkeypatch) + body = await segment(client, row.id, ref="v:Q1.r0.spend") + [query] = steps(body, "query") + assert query["columns"] == ["customer", "tel", "spend"] and query["masked"] == ["tel"], query["masked"] + assert [r[1] for r in query["rows"]] == ["已遮罩"] * len(query["rows"]) and query["rows"][0][2] == 380.0 + assert body["redacted"]["columns"] == ["tel"] and body["redacted"]["note"] + assert leaks(body) == [] + + +async def test_the_metric_chain_masks_the_renamed_column_too(client, monkeypatch, mall): + """口径卡指标 → 输入 → 查询：查询步骤同一套认法，tel 列同样遮掉（验收的 s13）。""" + await set_source(mall, {"mask_columns": ["phone"]}) + row = await run_graph(monkeypatch) + body = await segment(client, row.id, text="380.00元") + [query] = steps(body, "query") + assert query["masked"] == ["tel"] and leaks(body) == [] + [amount] = steps(body, "input") + assert amount["value"] == 380.0 and "masked" not in amount + + +async def test_an_input_taken_from_a_renamed_masked_column_is_withheld(client, monkeypatch, mall): + """cell() 取 tel 那一格：输入不给值、写明取自哪一列，代入式里那一处写「已遮罩」，指标本身的值也不给。""" + await set_source(mall, {"mask_columns": ["phone"]}) + row = await run_graph(monkeypatch) + body = await segment(client, row.id, text="13位") + metric = body["chain"][0] + assert metric["step"] == "metric" and metric["substituted"] == "len(已遮罩)" + assert metric["value"] == "已遮罩" and metric["rendered"] == "13位" + [cell] = steps(body, "input") + assert cell["value"] == "已遮罩" and cell["masked"] == ["tel"] and cell["path"] == TEL_CELL + assert body["redacted"]["columns"] == ["tel"] and leaks(body) == [] + + +async def test_a_scalar_taken_by_position_from_a_renamed_masked_column_is_withheld(client, monkeypatch, mall): + """vars.res.rows[0][1]：没有定位，顺着取值链认出取自查询结果的 tel 列，tel 又是 phone 改的名字。""" + await set_source(mall, {"mask_columns": ["phone"]}) + row = await run_graph(monkeypatch) + body = await segment(client, row.id, text="26位") + assert body["chain"][0]["substituted"] == "len(已遮罩) * 2" and body["chain"][0]["value"] == "已遮罩" + [pos] = steps(body, "input") + assert pos["path"] == TEL_POS and pos["value"] == "已遮罩" and pos["masked"] == ["tel"] + assert leaks(body) == [] + + +async def test_without_masks_nothing_changes(client, monkeypatch, mall): + row = await run_graph(monkeypatch) + body = await segment(client, row.id, ref="v:Q1.r0.spend") + [query] = steps(body, "query") + assert query["masked"] == [] and query["rows"][0][1] == "199-0000-0102" and body["redacted"] == {"columns": []} + plen = await segment(client, row.id, text="13位") + assert plen["chain"][0]["value"] == 13 and plen["chain"][0]["substituted"] == "len('199-0000-0102')" + + +async def test_a_mask_set_after_the_run_also_reaches_renamed_columns(client, monkeypatch, mall): + """事后才给数据源加的遮罩：按数据源现在设的列认 SQL，改了名字的照样遮。""" + row = await run_graph(monkeypatch) + await set_source(mall, {"mask_columns": "phone"}) + body = await segment(client, row.id, ref="v:Q1.r0.spend") + assert steps(body, "query")[0]["masked"] == ["tel"] and leaks(body) == [] + + +# -------------------------------------------------------------------------- +# 推演：勾行接口、用到 tel 的指标 +# -------------------------------------------------------------------------- + + +async def test_the_whatif_masks_the_renamed_column(client, monkeypatch, mall): + await set_source(mall, {"mask_columns": ["phone"]}) + row = await run_graph(monkeypatch) + body = await segment(client, row.id, ref="v:Q1.r0.spend") + snapshot = steps(body, "query")[0]["artifact"] + page = (await client.get(f"/api/runs/{row.id}/evidence/whatif/rows", params={"artifact": snapshot})).json() + assert page["masked"] == ["tel"] and all(r[1] == "已遮罩" for r in page["rows"]) and leaks(page) == [] + res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card"}) + assert res.status_code == 200, res.text + metrics = {m["id"]: m for m in res.json()["metrics"]} + assert metrics["tel_len"]["status"] == "masked" and metrics["tel_pos2"]["status"] == "masked" + assert metrics["top_spend"]["status"] == "unchanged" and metrics["top_spend"]["old"] == 380.0 + assert leaks(res.json()) == [] + res = await client.post(f"/api/runs/{row.id}/evidence/whatif", + json={"card": "card", "overrides": {"inputs": {TEL_CELL: 1}}}) + assert res.status_code == 422 and "遮罩" in res.json()["detail"] + + +# -------------------------------------------------------------------------- +# 合并查询：输入里改了名字的遮罩列，按逐格来历遮到合并结果里 +# -------------------------------------------------------------------------- + + +async def test_a_merged_result_masks_a_column_renamed_in_its_input(client, monkeypatch, mall): + await set_source(mall, {"mask_columns": ["phone"]}) + spec = chain( + node("start", "input"), + node("q_cust", "tool", tool="db_query__mall", args={"sql": "SELECT id, phone AS tel FROM customers ORDER BY id"}), + node("q_ord", "tool", tool="db_query__mall", + args={"sql": "SELECT customer_id AS id, SUM(amount) AS amount FROM orders GROUP BY customer_id ORDER BY id"}), + node("merged", "merge", inputs={"a": "q_cust", "b": "q_ord"}, + sql="SELECT a.id, a.tel, b.amount FROM a JOIN b ON a.id = b.id ORDER BY a.id"), + node("write", "report", instructions="写一句", on_violation="flag"), + node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]), + ) + row = await run_graph(monkeypatch, spec, text="一号客户共消费 [[v:Q3.r0.amount]]。") + body = await segment(client, row.id, ref="v:Q3.r0.amount") + merged = next(s for s in steps(body, "query") if s.get("merge")) + assert "tel" in merged["masked"] and all(r[merged["columns"].index("tel")] == "已遮罩" for r in merged["rows"]) + assert leaks(body) == [] + + +# -------------------------------------------------------------------------- +# 手搭的快照：SQL 的几种写法，解析不了时的保守规则 +# -------------------------------------------------------------------------- + +#: 库里没有这个数据源：遮罩只有查询当时记在快照里的那一份（phone） +GHOST = "ghost_mall" + + +async def query_step(sql: str, columns: list[str], rows: list[list]) -> dict: + """手搭一份封存范围内的查询快照（查询当时记下了遮罩列 phone），直接走查询步骤（evidence._query_step）。""" + from app.api import evidence + + snap = await artifact_store.put_json({"columns": columns, "rows": rows, "source": GHOST, "sql": sql, + "mask_columns": ["phone"]}, kind="query") + sealed = evidence._Sealed(run_id="syn-alias", graph={}, events=[], + seal={"sealed": True, "ok": True, "manifest_seq": 1, "legacy": False}) + sealed.queries = {snap: {"node_id": "q", "tool": f"db_query__{GHOST}", "source": GHOST}} + return await evidence._query_step(snap, [(0, columns[0])], {"catalog": {}}, sealed, evidence._Masks()) + + +@pytest.mark.parametrize("sql, columns, masked", [ + # 函数套一层再经 WITH 转一道 + ("WITH t AS (SELECT name, UPPER(phone) AS p FROM customers) SELECT name AS customer, p AS contact FROM t", + ["customer", "contact"], ["contact"]), + # 子查询里起的别名，外层再选出来 + ("SELECT x.who, x.num FROM (SELECT name AS who, phone AS num FROM customers) x", ["who", "num"], ["num"]), + # 截一段再拼上星号：值仍取自遮罩列 + ("SELECT name, substr(phone, 1, 3) || '****' AS short_tel FROM customers", ["name", "short_tel"], ["short_tel"]), + # 只数个数：带不出值，不遮 + ("SELECT COUNT(DISTINCT phone) AS n FROM customers", ["n"], []), + # 解析不了、又写着遮罩列：整份快照的列都遮 + ("SELECT name, phone AS p FROM customers WHERE ((", ["name", "p"], ["name", "p"]), + # 解析不了、没写着遮罩列：只遮遮罩列本身（结果里没有） + ("SELECT name, id AS p FROM customers WHERE ((", ["name", "p"], []), +]) +async def test_the_query_step_reads_the_sql_like_the_bundle_and_the_verify_agent(sql, columns, masked): + rows = [["客户甲" if i == 0 else "199-0000-0101" for i in range(len(columns))]] + step = await query_step(sql, columns, rows) + assert step["masked"] == masked, step["masked"] + assert [v == "已遮罩" for v in step["rows"][0]] == [c in masked for c in columns] diff --git a/backend/tests/test_whatif_masked_values.py b/backend/tests/test_whatif_masked_values.py new file mode 100644 index 00000000..2f90a258 --- /dev/null +++ b/backend/tests/test_whatif_masked_values.py @@ -0,0 +1,268 @@ +"""推演和证据面板不露指标本身的原值：指标的输入取自遮罩格（或者指标的值本身就是遮罩格）时，原值写「已遮罩」。 + +验收发现：phone_raw = cell(nodes.q_cust, 0, 'phone') 推演时判成 masked，输入里置了空，指标的 old、rendered_old 却还是那串 +号码；按位置取的 phone_pos 也一样。同一份答复里输入拿掉了、指标上照给，等于没遮。现在： + +- 推演（engine/recompute.py）：用到了遮罩列的指标（哪怕只嵌在下标里），old、rendered_old 写「已遮罩」，标 withheld， + 原因照旧写在 reason 里；new、rendered_new 本来就是空的（推演不了）；报告里引用它的地方 rendered_old 同样写「已遮罩」 +- 最外层的输入遮没遮核对不了（快照读不出来）的也一样；只嵌在下标里、遮没遮核对不了的那一格不碍事，原值照给 +- 证据面板的指标步骤：value 按同一个判断写「已遮罩」（recompute.hides_value），报告上的字（rendered）照旧 +- 没用到遮罩列的指标一字不差 + +前半是纯函数（手写的虚构卡和快照），后半真跑一遍图（虚构的 SQLite 客户库，号码是编的）。 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.core import artifact_store +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run +from app.engine import recompute +from app.engine.evidence import iter_segments +from app.engine.recompute import Changes, Sources, whatif +from app.engine.runner import run_manager +from app.main import app + +# -------------------------------------------------------------------------- +# 纯函数：手写的卡和快照 +# -------------------------------------------------------------------------- + +SNAP, OTHER = "a1" * 32, "b2" * 32 +STORE = { + SNAP: {"columns": ["id", "phone", "amount"], "source": "crm", "sql": "SELECT id, phone, amount FROM members", + "rows": [[1, "19900000201", 120.5], [2, "19900000202", 80.0]]}, + OTHER: {"columns": ["n"], "source": "crm", "sql": "SELECT 2 AS n", "rows": [[2]]}, +} +PHONE = "cell(nodes.crm, 0, 'phone')" +AMOUNT = "cell(nodes.crm, 0, 'amount')" + + +def cell(row, column, value, artifact=SNAP): + return {"path": f"cell(nodes.crm, {row}, '{column}')", "value": value, "node_id": "crm", "via": "tool_cell", + "locator": {"row": row, "column": column}, "artifact": artifact, "status": "verified"} + + +def scalar(path, value): + return {"path": path, "value": value, "node_id": "fetch", "via": "transform", "status": "ok"} + + +def card(mid, expression, substituted, value, inputs, unit=""): + return {"kind": "metric_set", "caliber": "会员口径", "caliber_version": "v1", "metrics": [ + {"id": mid, "name": mid, "unit": unit, "value": value, "format": "plain", "expression": expression, + "substituted": substituted, "inputs": inputs, "status": "ok"}]} + + +def sources(**kw): + base = {"load": STORE.get, "queries": (SNAP, OTHER), "masked": {SNAP: frozenset({"phone"})}} + return Sources(**{**base, **kw}) + + +def text_of(result) -> str: + return json.dumps(result, ensure_ascii=False) + + +def test_the_value_of_a_metric_that_is_a_masked_cell_is_withheld(): + """验收的 phone_raw：指标就是遮罩列的那一格。old、rendered_old 都写「已遮罩」，号码不出现在答复的任何地方。""" + result = whatif(card("phone_raw", PHONE, "19900000201", 19900000201, [cell(0, "phone", 19900000201)]), + Changes(), sources()) + [m] = result["metrics"] + assert (m["status"], m["old"], m["rendered_old"], m["withheld"]) == ("masked", "已遮罩", "已遮罩", True) + assert [m[k] for k in ("new", "rendered_new", "delta", "rendered_delta", "changed", "substituted")] == [None] * 6 + assert "遮罩" in m["reason"] and m["inputs"][0]["masked"] is True and m["inputs"][0]["old"] is None + assert "19900000201" not in text_of(result) + + +def test_a_metric_computed_from_a_masked_cell_withholds_its_value_too(): + """号码位数这类算出来的数同样不给：改一个别的输入、拿指标的值一减，就能算出遮住的那一格（宁可多遮）。""" + [m] = whatif(card("mix", f"len({PHONE}) + {AMOUNT}", "len('19900000201') + 120.5", 131.5, + [cell(0, "phone", "19900000201"), cell(0, "amount", 120.5)]), Changes(), sources())["metrics"] + assert (m["status"], m["old"], m["rendered_old"]) == ("masked", "已遮罩", "已遮罩") + inputs = {i["path"]: i for i in m["inputs"]} + assert inputs[AMOUNT]["old"] == 120.5 and inputs[PHONE]["old"] is None + + +def test_a_missing_value_stays_missing(): + """本来就没有值的（缺输入）照旧是空的、显示「—」：缺输入还得看得出来。""" + [m] = whatif(card("gone", PHONE, None, None, [cell(0, "phone", None)]), Changes(), sources())["metrics"] + assert m["status"] == "masked" and m["old"] is None and m["rendered_old"] == "—" and m["withheld"] is True + + +def test_a_top_level_cell_on_an_unreadable_snapshot_withholds_the_value(): + """那一格所在的快照读不出来：遮没遮核对不了，输入不给值，指标的原值同样不给（和代入式一样）。""" + [m] = whatif(card("east", AMOUNT, "120.5", 120.5, [cell(0, "amount", 120.5)]), Changes(), + sources(load=lambda a: None if a == SNAP else STORE.get(a)))["metrics"] + assert m["status"] == "unsupported" and m["old"] == "已遮罩" and m["withheld"] is True + assert "120.5" not in text_of(m) + + +def test_a_cell_only_choosing_a_row_on_an_unreadable_snapshot_does_not_hide_the_value(): + """只嵌在下标里、遮没遮核对不了的那一格：它只决定取哪一项，取到的值卡上另记着，原值照给、照样推演。""" + pick = f"vars.kpi.list[{AMOUNT.replace('amount', 'id')}] + vars.kpi.k" + top = f"vars.kpi.list[{AMOUNT.replace('amount', 'id')}]" + metric = card("pick", pick, "20 + 2", 22, [scalar(top, 20), cell(0, "id", 1), scalar("vars.kpi.k", 2)]) + [m] = whatif(metric, Changes(), sources(load=lambda a: None if a == SNAP else STORE.get(a)))["metrics"] + assert m["status"] == "unchanged" and m["old"] == 22 and "withheld" not in m + [m] = whatif(metric, Changes(inputs={"vars.kpi.k": 5}), sources(load=lambda a: None if a == SNAP else STORE.get(a)))[ + "metrics"] + assert (m["status"], m["old"], m["new"]) == ("ok", 22, 25) + + +def test_an_unmasked_metric_is_unchanged(): + result = whatif(card("east", AMOUNT, "120.5", 120.5, [cell(0, "amount", 120.5)]), Changes(), sources()) + [m] = result["metrics"] + assert (m["status"], m["old"], m["rendered_old"]) == ("unchanged", 120.5, "120.5") and "withheld" not in m + + +def test_report_segments_of_a_withheld_metric_do_not_carry_the_value(): + """报告里引用了原值不给的指标：推演答复里那一处的 rendered_old 同样写「已遮罩」。""" + metric = card("phone_raw", PHONE, "19900000201", 19900000201, [cell(0, "phone", 19900000201)]) + result = whatif(metric, Changes(), sources()) + entry = {"kind": "metric", "node_id": "card", "artifact": "c" * 64, "locator": {"metric": "phone_raw"}} + doc = {"catalog": {"m:phone_raw": entry}, "blocks": [{"units": [{"id": "u0", "segments": [ + {"id": "s0", "kind": "number", "text": "19900000201", "state": "deterministic", + "cite": {"alias": "m:phone_raw", "kind": "metric", "status": "resolved"}}]}]}]} + [seg] = recompute.report_segments([("write", doc)], card_node="card", card_artifact="c" * 64, + results=result["metrics"]) + assert (seg["rendered_old"], seg["rendered_new"], seg["changed"]) == ("已遮罩", None, None) + + +@pytest.mark.parametrize("masked, unverified, expected", [ + ({PHONE}, set(), True), # 取自遮罩列 + ({"vars.kpi.i"}, set(), True), # 遮罩列的那一格只嵌在下标里：推演整个按遮罩处理，原值也不给 + (set(), {AMOUNT}, True), # 最外层的输入遮没遮核对不了 + (set(), {"vars.kpi.i"}, False), # 只嵌在下标里、遮没遮核对不了：不碍事 + (set(), set(), False), +]) +def test_hides_value_is_one_rule_for_the_whatif_and_the_evidence_panel(masked, unverified, expected): + metric = {"expression": f"{AMOUNT} + vars.kpi.list[vars.kpi.i]"} + assert recompute.hides_value(metric, masked, unverified) is expected + # 表达式读不懂：有一个遮没遮核对不了的就不给 + assert recompute.hides_value({"expression": "1 +"}, set(), {AMOUNT}) is True + + +# -------------------------------------------------------------------------- +# 真跑：验收原样的口径卡（单格原值、按位置取、按位置取的位数、编号） +# -------------------------------------------------------------------------- + +CUSTOMERS = [(1, "客户甲", "19900000301"), (2, "客户乙", "19900000302"), (3, "客户丙", "19900000303")] +PHONES = [c[2] for c in CUSTOMERS] +CARD_METRICS = [ + {"id": "phone_raw", "name": "首位客户手机号（单格原值）", "format": "plain", + "expression": "cell(nodes.q_cust, 0, 'phone')"}, + {"id": "phone_pos", "name": "首位客户手机号（按位置）", "format": "plain", "expression": "vars.cust.rows[0][2]"}, + {"id": "phone_pos_len", "name": "首位客户手机号位数", "unit": "位", "format": "plain", + "expression": "len(str(vars.cust.rows[0][2]))"}, + {"id": "id0", "name": "首位客户编号", "format": "plain", "expression": "cell(nodes.q_cust, 0, 'id')"}, +] +TEXT = "首位客户编号 [[m:id0]]，号码位数 [[m:phone_pos_len]]。" + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(engine_up): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +async def members(tmp_path): + path = tmp_path / "members.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);") + db.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS) + db.commit() + db.close() + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == "members"))).scalar_one_or_none() + if row is None: + row = DataSource(name="members", kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True + row.options = {"mask_columns": ["phone"]} + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield path + await engines.invalidate(source_id) + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def graph() -> dict: + nodes = [ + node("start", "input"), + node("q_cust", "tool", tool="db_query__members", + args={"sql": "SELECT id, name, phone FROM customers ORDER BY id LIMIT 5"}), + node("parse_cust", "transform", mode="json", template="{{ nodes.q_cust }}", assign_to="cust"), + node("card_c", "metrics", caliber="遮罩取法对照", caliber_version="v1", metrics=CARD_METRICS), + node("write", "report", instructions="写一句", on_violation="flag"), + node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]), + ] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def run_graph(monkeypatch) -> Run: + from app.providers import mock_model + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=TEXT)) + run_id = (await run_manager.start(graph=graph(), input_payload={})).id + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + assert row.status == "succeeded", row.error + return row + raise AssertionError(f"运行 {run_id} 没跑完：{row.status}") + + +def leaks(raw: str) -> list[str]: + return [p for p in PHONES if p in raw] + + +async def test_the_whatif_answer_carries_no_masked_value(client, monkeypatch, members): + """验收原样：推演答复里号码一个都没有（以前在 phone_raw、phone_pos 的 old、rendered_old 里）。""" + row = await run_graph(monkeypatch) + res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card_c"}) + assert res.status_code == 200, res.text + assert leaks(res.text) == [], leaks(res.text) + metrics = {m["id"]: m for m in res.json()["metrics"]} + for mid in ("phone_raw", "phone_pos", "phone_pos_len"): + m = metrics[mid] + assert (m["status"], m["old"], m["rendered_old"], m["withheld"]) == ("masked", "已遮罩", "已遮罩", True), m + assert m["new"] is None and m["rendered_new"] is None and "遮罩" in m["reason"] + assert (metrics["id0"]["status"], metrics["id0"]["old"]) == ("unchanged", 1) and "withheld" not in metrics["id0"] + # 报告里引用了号码位数：那一处的原值也写「已遮罩」 + [seg] = [s for s in res.json()["segments"] if s["metric"] == "phone_pos_len"] + assert seg["rendered_old"] == "已遮罩" and seg["rendered_new"] is None + + +async def test_the_evidence_panel_withholds_the_metric_value_by_the_same_rule(client, monkeypatch, members): + """证据面板的指标步骤：用到遮罩列的指标 value 写「已遮罩」，报告上的字照旧；没用到的照给。""" + row = await run_graph(monkeypatch) + [report] = (await client.get(f"/api/runs/{row.id}/evidence")).json()["reports"] + doc = artifact_store.load(report["doc_artifact"]) + by_text = {s["text"]: s["id"] for s in iter_segments(doc) if s.get("ref")} + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{by_text['11位']}")).json() + metric = body["chain"][0] + assert (metric["step"], metric["value"], metric["rendered"]) == ("metric", "已遮罩", "11位") + assert leaks(json.dumps(body, ensure_ascii=False)) == [] + plain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{by_text['1']}")).json() + assert plain["chain"][0]["value"] == 1