diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 6d62b2f5..0e53c8e7 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -2497,12 +2497,15 @@ def compose_doc( loader: Loader | None = None, entities: bool = True, absence: bool = False, + masked: dict[str, Any] | None = None, ) -> dict[str, Any]: """模型写的原文（带标记）→ 报告文档：渲染后的 markdown、块、句、片段、统计、违规。 entities=False（报告节点 entities: off）：表名、字段名整层不管——不链接、不核对反引号， [[t:]] / [[c:]] 解析不了，原因是 ENTITIES_OFF_REASON。 absence=True（写作目录有省略）：另查缺失断言（absence_claims），和句中的 [[see:]] 一样只在组装时查。 + masked（{查询快照工件: 取自遮罩列的结果列}，报告节点按数据源的遮罩算好交进来）：另查直接引用了这些列的格子 + （masked_cells），同样只在组装时查。没给的和以前一字不差。 stats / violations 由 verify_doc 算出，和出口契约复核用的是同一个函数。 整表标记先展开成每格一个 [[v:]] 的 Markdown 表，之后和模型自己写的表格走同一条路； @@ -2661,9 +2664,11 @@ def compose_doc( } checked = verify_doc(doc, catalog, allow_numbers=allow_numbers, cells_allowed=cells_allowed, loader=snaps) # 句中的 [[see:]]、目录有省略时的缺失断言只在组装时查（交回写作者重写）：出口复核看的是数字和出处， - # 句子通不通、结论有没有越过证据不影响出具 + # 句子通不通、结论有没有越过证据不影响出具。引用了遮罩列的格子也在这里查：遮罩设在数据源上，出口复核按目录 + # 重建、查不了库；仍不改的由报告节点写进出具缺口 doc["stats"], doc["violations"] = checked["stats"], \ - checked["violations"] + inline_see(source) + (absence_claims(doc) if absence else []) + checked["violations"] + inline_see(source) + (absence_claims(doc) if absence else []) \ + + masked_cells(doc, catalog, masked) return doc @@ -3125,6 +3130,45 @@ def absence_claims(doc: dict[str, Any]) -> list[dict[str, Any]]: return found +def _masked_cell_message(ref: str, column: str) -> _Reason: + return _Reason( + f"引用了取自遮罩列的格子 [[{ref}]]：列「{column}」取自数据源中设置了遮罩的列，值不应写进报告", + f"[[{ref}]] 引用的列「{column}」取自数据源中设置了遮罩的列（手机号、姓名这类），值不能写进报告：删掉这处引用，" + "或者换成不含这一列的写法；整表引用时用 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 的 + _snapshot_masks：查询当时记下的、数据源现在设的、SQL 里给遮罩列起了别名套了函数的都算）：遮罩设在数据源上，这里 + 只认文档和目录，查不了库。写作目录不遮罩（写作者要看得到数据），所以拦在自查：和没有出处的数字一样交回写作者重写， + 仍不改的按「违规处置」办（report.py）。 + + 违规不带那一格的值（不写 text、context）：它会进 report.checked、重写的日志、出具缺口，这些地方不能再露一次。带着 + ref、列名、span 和所在片段：证据面板能定位，写作者知道要去掉哪一处。""" + if not masked: + return [] + hidden = {str(a): {str(c).lower() for c in cols} for a, cols in masked.items() if cols} + found: list[dict[str, Any]] = [] + for _, unit in iter_units(doc): + for seg in unit.get("segments") or []: + cite = seg.get("cite") or {} + if cite.get("kind") != "cell" or cite.get("status") != "resolved": + continue + artifact = str((catalog.get(cite.get("alias")) or {}).get("artifact") or "") + column = str((cite.get("locator") or {}).get("column") or "") + if not column or column.lower() not in hidden.get(artifact, ()): + continue + ref = str(seg.get("ref") or f"v:{cite.get('ref')}") + span = seg.get("span") + found.append(_violation("masked_cell", _masked_cell_message(ref, column), ref=ref, column=column, + span=list(span) if isinstance(span, list) else None, segment=seg.get("id"), + unit=unit.get("id"))) + return found + + #: 引用查询结果的写法。只跟着目录里的查询出现（catalog_prompt）：写作规则本身保持一期那份，升级前 #: 发起的运行跑到报告节点时提示一字不差；只有口径卡的目录也不该教写作者去引用不存在的 Q1 CELL_RULES = ( diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index e6bffc31..dfde4861 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -31,6 +31,7 @@ """ from __future__ import annotations +import asyncio import re import time from collections import Counter @@ -318,6 +319,31 @@ class _RenderedStream: self._ctx.emit(EventType.LLM_TOKEN, delta=tail) +async def _masked_columns(catalog: dict[str, Any]) -> dict[str, set[str]]: + """写作目录里每份查询快照（工件 id）取自遮罩列的结果列（小写），交给 compose_doc 的 masked。认法和证据面板的查询 + 步骤是同一个函数（api/evidence.py 的 _snapshot_masks）：查询当时记下的、数据源现在设的，SQL 里给遮罩列起了别名、 + 套了函数、经子查询转一道的也算，合并查询按逐格来历认。没有一列遮着的不列；读不出来的快照不列：引用它的格子本来就 + 解析不了。快照可能有好几份、每份几 MB，读盘复验哈希放进线程。""" + from app.api.evidence import _Masks, _snapshot_masks + from app.core.artifact_store import load + + masks = _Masks() + out: dict[str, set[str]] = {} + seen: set[str] = set() + for entry in catalog.values(): + artifact = entry.get("artifact") if isinstance(entry, dict) and entry.get("kind") == "query" else None + if not isinstance(artifact, str) or not artifact or artifact in seen: + continue + seen.add(artifact) + try: + snap = await asyncio.to_thread(load, artifact) + except ValueError: + continue + if isinstance(snap, dict) and (hidden := (await _snapshot_masks(snap, entry.get("source"), masks))[0]): + out[artifact] = hidden + return out + + async def _run_class(run_id: str) -> str | None: """按运行记录上的 run_class 取：子图、审批之后续跑都是另起的执行上下文，只有运行记录靠得住。""" from app.db.models import Run @@ -523,6 +549,9 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: spec = ctx.run.spec catalog = report_catalog(state, spec, ctx.node) + # 目录里取自遮罩列的结果列：正文引用了这些列的格子，自查记一条违规（evidence.masked_cells），和没有出处的数字一样 + # 交回写作者重写。写作目录本身不遮（写作者要看得到数据），只拦写进报告的那几格 + masked = await _masked_columns(catalog) allow = report_allowance(spec, ctx.node) cells = True # 结论句策略只对升级后发起的运行生效：升级前的运行（图里也不会有 claims）事件、产出一个字不加 @@ -576,7 +605,7 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 第一稿、重写稿、裁判退回的改写稿都从这里过：整篇包在代码块里的先去掉外层围栏。文档的 source 记的是去掉 # 围栏之后的正文——markdown 由它渲染，出口契约复核取回的也是这份文档 return compose_doc(_unfenced(raw), catalog, node_id=ctx.node.id, run_id=ctx.run.run_id, allow_numbers=allow, - cells_allowed=cells, entities=entities, absence=writer.trimmed) + cells_allowed=cells, entities=entities, absence=writer.trimmed, masked=masked) response, raw = await draft(messages) doc = compose(raw) @@ -616,6 +645,18 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ctx.emit(EventType.LOG, level="warn", code="report_absence_claim", count=len(absent), texts=texts, message=f"报告在证据目录有省略时写了 {len(absent)} 处缺失断言（{_named(absent)}），{tried}：" "被省略的查询仍在证据台账里，这类结论依据不足，应写作本报告未覆盖。报告照常产出，这几处已标注") + gaps: list[str] = [] + hits = [v for v in doc["violations"] if v.get("code") == "masked_cell"] + if hits and not (_fatal(blocking) and on_violation == "fail"): + # 重写之后还引用着遮罩列（「违规处置」为标注，照常产出）：时间线上单独一行说清，节点产出里写一条出具缺口， + # 出口照单收下、降档（io._upstream_issuance_gaps）。只点名引用，不写那几格的值 + tried = f"已要求模型重写 {repairs} 次仍未改掉" if repairs else "未要求重写" + refs = list(dict.fromkeys(str(v.get("ref") or "") for v in hits)) + gaps.append(f"报告引用了 {len(hits)} 处取自遮罩列的格子（{_named(hits)}），{tried}，这些值会随报告出具") + ctx.emit(EventType.LOG, level="warn", code="report_masked_cell", count=len(hits), refs=refs, + columns=list(dict.fromkeys(str(v.get("column") or "") for v in hits)), + message=f"{gaps[0]}。报告照常产出，这几处已标注：数据源设置了遮罩的列不应写进报告，" + "请在报告撰写节点的指令中去掉对这些列的引用") doc_artifact = await _store(doc, ctx) ctx.emit(EventType.REPORT_CHECKED, doc_artifact=doc_artifact, ok=not doc["violations"], stats=doc["stats"], violations=doc["violations"][:20], repairs=repairs, @@ -646,6 +687,8 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "duration_ms": int((time.perf_counter() - started) * 1000), **({"claims": claims} if evidence_on else {}), **({"judge": judgement} if judgement is not None else {}), + # 出口契约照单收下的缺口（只在有的时候带这个键：别的报告产出和以前一字不差） + **({"issuance_gaps": gaps} if gaps else {}), } updates: dict[str, Any] = {"nodes": {ctx.node.id: result}, "usage": usage} if ctx.cfg("emit_message", True): diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py index e56e4713..de87aa31 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -93,6 +93,9 @@ WARN_CODES: dict[str, tuple[str, str]] = { "report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED), # 目录有省略时写了「没有查到」这类缺失断言、重写后仍在：报告照常产出并标注，这几句结论依据不足 "report_absence_claim": ("report_absence_claim", DEGRADED), + # 报告直接引用了取自遮罩列的格子、重写后仍在（「违规处置」为标注）：照常产出并标注，遮罩列的值随报告出具， + # 出具另记缺口降档 + "report_masked_cell": ("report_masked_cell", DEGRADED), # 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据 "evidence_store_failed": ("evidence_store_failed", DEGRADED), # agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了 diff --git a/backend/tests/test_report_masked_cells.py b/backend/tests/test_report_masked_cells.py new file mode 100644 index 00000000..f7f8f297 --- /dev/null +++ b/backend/tests/test_report_masked_cells.py @@ -0,0 +1,267 @@ +"""报告自查认得出引用了取自遮罩列的格子：像没有出处的数字一样记违规、交回写作者重写，仍不改的按「违规处置」办。 + +验收发现：写作要求点名引用 [[v:Q2.r0.tel]]（Q2 的 SQL 是 c.phone AS tel，phone 设了遮罩），报告自查判 ok、violations 为空， +正文写出号码，导出的可验证报告也把它带了出去，从发起到导出没有一步提示。写作目录不遮罩是已知设计（写作者要看得到数据）， +所以拦在自查： + +- 组装文档时（compose_doc 的 masked：{查询快照工件: 取自遮罩列的列}），引用了这些列的格子记一条 masked_cell，列名对上 + 的、按 SQL 认出改了名字的都算（报告节点按证据面板同一个函数 api/evidence.py 的 _snapshot_masks 算） +- 违规里不带那一格的值（不写 text、context）：违规会进 report.checked、重写的日志、出具缺口，这些地方不能再露一次号码 +- 报告节点把它交回写作者重写（不是只标注的那几类）；重写后还在：「违规处置」为失败的判失败，为标注的照常产出， + 时间线上另起一行说清（report_masked_cell），节点产出里写一条出具缺口，出口降档 +- 没设遮罩、引用的是别的列：一切照旧 + +真跑的部分用虚构的 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 import review as rv +from app.engine.evidence import compose_doc, describe_violations, iter_segments +from app.engine.runner import run_manager + +# -------------------------------------------------------------------------- +# 组装文档：compose_doc 的 masked +# -------------------------------------------------------------------------- + +ROWS = [["客户乙", "199-0000-0102", 380.0], ["客户甲", "199-0000-0101", 220.0]] +SECRET = [r[1] for r in ROWS] + + +async def catalog_of() -> tuple[dict, str]: + art = await artifact_store.put_json({"columns": ["customer", "tel", "spend"], "rows": ROWS, "source": "mall", + "sql": "SELECT c.name AS customer, c.phone AS tel, 1 AS spend FROM customers c", + "mask_columns": ["phone"]}, kind="query") + entry = {"alias": "Q1", "kind": "query", "artifact": art, "columns": ["customer", "tel", "spend"], "rows": 2, + "node_id": "top", "source": "mall"} + return {"Q1": entry}, art + + +def masked_of(doc: dict) -> list[dict]: + return [v for v in doc["violations"] if v["code"] == "masked_cell"] + + +async def test_a_cited_masked_cell_is_a_violation_without_its_value(): + catalog, art = await catalog_of() + doc = compose_doc("消费最高的是 [[v:Q1.r0.customer]]，电话 [[v:Q1.r0.tel]]，花了 [[v:Q1.r0.spend]]。", catalog, + node_id="write", run_id="r1", masked={art: {"phone", "tel"}}) + [v] = masked_of(doc) + seg = next(s for s in iter_segments(doc) if s.get("ref") == "v:Q1.r0.tel") + assert (v["ref"], v["column"], v["segment"]) == ("v:Q1.r0.tel", "tel", seg["id"]) and v["unit"] and v["span"] + assert "遮罩" in v["message"] and "「tel」" in v["message"] and "删掉" in v["for_model"] + # 违规本身不再写一次号码：它会进事件、日志和出具缺口 + assert not [p for p in SECRET if p in json.dumps(v, ensure_ascii=False)] + assert "text" not in v and "context" not in v + assert not [p for p in SECRET if p in describe_violations(masked_of(doc))] + + +async def test_without_masks_or_for_other_columns_nothing_is_flagged(): + catalog, art = await catalog_of() + text = "电话 [[v:Q1.r0.tel]]，花了 [[v:Q1.r0.spend]]。" + assert masked_of(compose_doc(text, catalog, node_id="write", run_id="r1")) == [] + assert masked_of(compose_doc(text, catalog, node_id="write", run_id="r1", masked={})) == [] + assert masked_of(compose_doc("花了 [[v:Q1.r0.spend]]。", catalog, node_id="write", run_id="r1", + masked={art: {"phone", "tel"}})) == [] + + +async def test_a_whole_table_reports_every_masked_cell(): + """整表引用展开成每格一个 [[v:]]：遮罩列的每一格各记一条，写作者看得到要去掉哪一列。""" + catalog, art = await catalog_of() + doc = compose_doc("前两名：\n\n[[table:Q1]]\n", catalog, node_id="write", run_id="r1", masked={art: {"tel"}}) + assert [v["ref"] for v in masked_of(doc)] == ["v:Q1.r0.tel", "v:Q1.r1.tel"] + assert all("cols=" in v["for_model"] for v in masked_of(doc)) + + +def test_the_masked_cell_code_is_a_degraded_signal_of_its_own_kind(): + """复核层认得这条日志：报告照常产出、有缺口，按它自己的 kind 记，不落成笼统的 warn。""" + assert rv.WARN_CODES["report_masked_cell"] == ("report_masked_cell", rv.DEGRADED) + warn = {"type": "log", "node_id": "write", + "data": {"level": "warn", "code": "report_masked_cell", "message": "报告引用了取自遮罩列的格子"}} + [signal] = rv.scan([warn], {"报告": "首位客户已联系"}) + assert (signal.kind, signal.severity) == ("report_masked_cell", rv.DEGRADED) + + +# -------------------------------------------------------------------------- +# 真跑：报告节点交回重写、按违规处置办、时间线和出具缺口说清 +# -------------------------------------------------------------------------- + +CUSTOMERS = [(1, "客户甲", "199-0000-0101"), (2, "客户乙", "199-0000-0102"), (3, "客户丙", "199-0000-0103")] +ORDERS = [(1, 1, 120.5), (2, 2, 300.0), (3, 2, 80.0), (4, 3, 210.25), (5, 1, 99.5)] +PHONES = [c[2] for c in CUSTOMERS] +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") +CITES_TEL = "消费最高的客户是 [[v:Q1.r0.customer]]，联系电话 [[v:Q1.r0.tel]]，共消费 [[v:Q1.r0.spend]]。" +CLEAN = "消费最高的客户是 [[v:Q1.r0.customer]]，共消费 [[v:Q1.r0.spend]]。" + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +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, engine_up): + 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, {"mask_columns": ["phone"]}) + 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(*, sql: str = SQL_TOP, contract: dict | None = None, **report) -> dict: + out = {"fields": [{"name": "报告", "value": "{{ nodes.write.text }}"}]} + if contract is not None: + out["contract"] = contract + nodes = [ + node("start", "input"), + node("top", "tool", tool="db_query__mall", args={"sql": sql}), + node("write", "report", instructions="写一句，联系电话写 [[v:Q1.r0.tel]]", **report), + node("out", "output", **out), + ] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def script(monkeypatch, *replies) -> list[list]: + """写作者依次回这几段，用完了一直回最后一段。返回每次调用收到的消息。""" + from app.providers import mock_model + + seen: list[list] = [] + + def _decide(self, messages): + seen.append(list(messages)) + return AIMessage(content=replies[min(len(seen), len(replies)) - 1]) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + return seen + + +async def finish(spec: dict) -> Run: + 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 events(run_id: str, etype: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id) + if etype: + q = q.where(RunEvent.type == etype) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +def logs(rows: list[RunEvent], code: str) -> list[dict]: + return [e.data for e in rows if e.type == "log" and e.data.get("code") == code] + + +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] + + +async def test_the_writer_is_asked_to_rewrite_and_the_rewrite_passes(monkeypatch, mall): + """第一稿引用了 tel（phone 改的名字）：自查记违规、交回重写，重写稿不引用就通过。重写请求和日志里不再写号码。""" + seen = script(monkeypatch, CITES_TEL, CLEAN) + row = await finish(graph()) + assert row.status == "succeeded", row.error + [checked] = await events(row.id, "report.checked") + assert checked.data["ok"] is True and checked.data["repairs"] == 1, checked.data + repair = str(seen[1][-1].content) + assert "tel" in repair and "遮罩" in repair and leaks(repair) == [] + [log] = logs(await events(row.id), "report_repair") + assert "v:Q1.r0.tel" in log["message"] and leaks(log) == [] + + +async def test_a_writer_that_insists_is_flagged_and_the_issuance_says_why(monkeypatch, mall): + """重写后还引用：「违规处置」为标注时照常产出，自查结果、时间线、出具缺口都说清，这几处都不再写一次号码。""" + script(monkeypatch, CITES_TEL) + row = await finish(graph(on_violation="flag", contract={"report_from": "write"})) + assert row.status == "succeeded", row.error + rows = await events(row.id) + [checked] = [e.data for e in rows if e.type == "report.checked"] + assert checked["ok"] is False and checked["repairs"] == 1 and not checked["failed"] + [v] = [v for v in checked["violations"] if v["code"] == "masked_cell"] + assert v["ref"] == "v:Q1.r0.tel" and v["column"] == "tel" and leaks(v) == [] + [log] = logs(rows, "report_masked_cell") + assert log["count"] == 1 and log["refs"] == ["v:Q1.r0.tel"] and log["columns"] == ["tel"] and log["level"] == "warn" + assert "重写 1 次" in log["message"] and leaks(log) == [] + finished = next(e.data for e in rows if e.type == "node.finished" and e.node_id == "write") + payload = artifact_store.load(finished["artifact"]) + [gap] = payload["issuance_gaps"] + assert "遮罩" in gap and "v:Q1.r0.tel" in gap and leaks(gap) == [] + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded" and any(g.startswith("报告撰写「write」：") and "遮罩" in g + for g in issuance["gaps"]), issuance["gaps"] + assert leaks(issuance["gaps"]) == [] + + +async def test_a_writer_that_insists_fails_the_node_when_violations_fail(monkeypatch, mall): + """「违规处置」为失败：重写后还引用就判失败，原因说是遮罩列，不写号码。""" + script(monkeypatch, CITES_TEL) + row = await finish(graph(on_violation="fail")) + assert row.status == "failed" and "遮罩" in (row.error or "") and "tel" in row.error, row.error + assert leaks(row.error) == [] + [checked] = await events(row.id, "report.checked") + assert checked.data["failed"] is True + assert logs(await events(row.id), "report_masked_cell") == [], "节点判失败时不另起「照常产出」的那一行" + + +async def test_the_masked_column_itself_is_flagged_too(monkeypatch, mall): + """列名直接就是遮罩列（没起别名）的也算。""" + script(monkeypatch, "首位客户 [[v:Q1.r0.name]]，电话 [[v:Q1.r0.phone]]。") + row = await finish(graph(sql="SELECT name, phone FROM customers ORDER BY id", on_violation="flag")) + [checked] = await events(row.id, "report.checked") + assert [v["ref"] for v in checked.data["violations"] if v["code"] == "masked_cell"] == ["v:Q1.r0.phone"] + + +async def test_without_masks_the_report_is_unchanged(monkeypatch, mall): + await set_source(mall, {}) + script(monkeypatch, CITES_TEL) + row = await finish(graph(on_violation="flag", contract={"report_from": "write"})) + assert row.status == "succeeded", row.error + [checked] = await events(row.id, "report.checked") + assert checked.data["ok"] is True and checked.data["repairs"] == 0 + assert logs(await events(row.id), "report_masked_cell") == [] + finished = next(e.data for e in await events(row.id, "node.finished") if e.node_id == "write") + assert "issuance_gaps" not in artifact_store.load(finished["artifact"]) + assert row.output["_issuance"]["tier"] == "formal" or not any("遮罩" in g for g in row.output["_issuance"]["gaps"])