M2 报告自查认得出引用了取自遮罩列的格子：写作者写了 [[v:Qn.rX.列]]、而这一列是遮罩列或按 SQL 认出取自遮罩列（c.phone AS tel）时，自查记一条 masked_cell，像没有出处的数字一样交回写作者重写；仍不改的按「违规处置」办——失败的判失败，标注的照常产出，时间线另起一行 report_masked_cell，节点产出写一条 issuance_gaps，出口照单收下、降档

验收里写作要求点名引用 tel，自查判 ok、violations 为空，号码随报告和可验证报告出去，一步提示都没有。写作目录不遮罩是已知设计，所以拦在自查：
- compose_doc 新增 masked（{查询快照工件: 取自遮罩列的结果列}），engine/evidence.py 的 masked_cells 按它查，和句中的 [[see:]]、缺失断言一样只在组装时查；整表展开出来的格也算
- 报告节点按证据面板同一个函数（api/evidence.py 的 _snapshot_masks）算目录里每份快照的遮罩列，交给 compose_doc
- 违规、日志、出具缺口只点名引用和列名，不写那一格的值：它们会进 report.checked、事件和可验证报告
- review.py 的码表登记 report_masked_cell（degraded）

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 = (

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):

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 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了

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"])