清单有省略时不许断言「没有数据」：写作规则加一条，核对认出缺失断言交回重写一次，仍有的只标注

一次真实运行里写作者只看到上百次查询里的前二十来次，却写出「某类数据缺失、本轮没有查到任何结果」这样的断言，可那个 Agent 实际查到了上百条记录。

- 写作规则：清单有省略、或有来源的结构化结论为空时，不得断言没有查到、不存在、缺失、没有数据，只能写「本报告未覆盖」并说明依据不足
- 组装文档：清单有省略时按一组明确的说法认出缺失断言（没有查到、未查询到、查不到、没有任何记录、数据为空、数据缺失、不存在相关记录、暂无数据等；「不存在明显差异」「不代表没有数据」这类不算，代码块不查），记新违规 absence_claim，带 span、所在片段和原句
- 和句中的 [[see:]] 一样归入 REWRITE_ONLY：交回重写一次，重写后仍有的只标注、不判失败；只剩这几类时不再按最多重写次数反复重写。仍有时报告节点发 report_absence_claim（带 count、texts），运行复核算作缺口
- 证据面板的违规清单标「缺失断言」并写明为什么依据不足，时间线给 report_absence_claim 专门的一行，文案在 terms.ts；check-decode、check-evidence 补断言
- 没展开的查询的摘要在没有表结构快照时从查询快照里的 SQL 认表名
- 回归测试：合成的门店、会员、卡券、活动库真跑「4 个 Agent 并行查库 → 报告」（查询次数虚构、分布悬殊），覆盖抽取截断重试、两次截断记空值、清单均衡和缺失断言的重写

backend/app/engine/evidence.py

@@ -2474,11 +2474,13 @@ def compose_doc(
     cells_allowed: bool = True,
     loader: Loader | None = None,
     entities: bool = True,
+    absence: bool = False,
 ) -> dict[str, Any]:
     """模型写的原文（带标记）→ 报告文档：渲染后的 markdown、块、句、片段、统计、违规。
 
     entities=False（报告节点 entities: off）：表名、字段名整层不管——不链接、不核对反引号，
     [[t:]] / [[c:]] 解析不了，原因是 ENTITIES_OFF_REASON。
+    absence=True（写作目录有省略）：另查缺失断言（absence_claims），和句中的 [[see:]] 一样只在组装时查。
 
     stats / violations 由 verify_doc 算出，和出口契约复核用的是同一个函数。
     整表标记先展开成每格一个 [[v:]] 的 Markdown 表，之后和模型自己写的表格走同一条路；
@@ -2634,8 +2636,10 @@ def compose_doc(
         "markdown": markdown, "source": source, "catalog": kept, "blocks": out_blocks,
     }
     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)
+    # 句中的 [[see:]]、目录有省略时的缺失断言只在组装时查（交回写作者重写）：出口复核看的是数字和出处，
+    # 句子通不通、结论有没有越过证据不影响出具
+    doc["stats"], doc["violations"] = checked["stats"], \
+        checked["violations"] + inline_see(source) + (absence_claims(doc) if absence else [])
     return doc
 
 
@@ -3035,6 +3039,58 @@ def inline_see(source: str) -> list[dict[str, Any]]:
     return [_violation("inline_see", INLINE_SEE_MESSAGE, text=m.group(0)) for m in _INLINE_SEE.finditer(source or "")]
 
 
+#: 写作目录有省略、或者有来源的结构化结论为空时，拼在写作规则后面（给模型看）
+ABSENCE_RULE = (
+    "这次你看不到全部证据（证据目录有省略，或者有来源的结构化结论为空）：不得断言「没有查到」「未查到」「不存在」"
+    "「缺失」「没有数据」「无数据」这类结论。某个方向的证据没有展开或为空时，只能写「本报告未覆盖……」，并说明依据不足"
+    "（比如「目录有省略，相关查询未展开」），句末可以挂 [[see:Qn]] 指向相关查询。"
+)
+
+#: 缺失断言：断言「没有数据」的几种说法。只认拿得准的，不泛化：「不存在明显差异」「缺少口径说明」这类说的是看到的
+#: 数据里的情况，不算；「不存在」「为空」「缺失」只认跟在数据、记录、结果后面（或者前面）的
+_ABSENCE = re.compile(
+    r"(?:没有|没|未|未能|没能)(?:查询|查)到|查不到|查无"
+    r"|(?:没有|暂无|并无)(?:任何)?(?:相关的?)?(?:数据|记录|结果)"
+    r"|无(?:任何)?(?:相关的?)?(?:数据|记录)(?![库源集表权字项列])"
+    r"|不存在(?:任何)?(?:相关的?)?(?:数据|记录|结果)|(?:数据|记录|结果)(?:均|都|全部)?(?:不存在|为空|缺失|空缺)"
+)
+#: 前面是这些词的不算：「不代表没有数据」「并非查不到」说的恰恰是不能下这个结论
+_ABSENCE_NEGATED = re.compile(r"(?:不代表|并不代表|不等于|不意味着|并非|不是|不能说明|不能断言|不得断言|不应断言)[^。！？\n]{0,6}$")
+ABSENCE_MESSAGE = _Reason(
+    "证据目录有省略时写了缺失断言：写作模型没有看到全部证据，「没有查到」「不存在」「缺失」这类结论依据不足",
+    "证据目录有省略（没展开的查询只列了编号和摘要），这一句却断言了没有数据。看不到的证据不能当成不存在："
+    "改成「本报告未覆盖……」，并说明依据不足（目录有省略、相关查询未展开），不要写「没有查到」「不存在」「缺失」「没有数据」",
+)
+
+
+def absence_claims(doc: dict[str, Any]) -> list[dict[str, Any]]:
+    """正文里的缺失断言（_ABSENCE），每处一条违规 absence_claim。只在写作目录有省略时由报告节点要求查（compose_doc 的
+    absence）：目录全给了的时候，「本周没有退款记录」可能就是看了数据说的。
+
+    在渲染后的正文里找（标记已经换成了字），代码块不查。带着 span、所在片段和句子：证据面板能定位、能显示原句。
+    """
+    markdown = str(doc.get("markdown") or "")
+    found: list[dict[str, Any]] = []
+    for block in doc.get("blocks") or []:
+        if block.get("type") == "code":
+            continue
+        for unit in block.get("units") or []:
+            span = unit.get("span")
+            if not (isinstance(span, list) and len(span) == 2):
+                continue
+            sentence = markdown[span[0]:span[1]]
+            for m in _ABSENCE.finditer(sentence):
+                if _ABSENCE_NEGATED.search(sentence[:m.start()]):
+                    continue
+                at = span[0] + m.start()
+                seg = next((g for g in unit.get("segments") or [] if g.get("kind") != "structural"
+                            and isinstance(g.get("span"), list) and g["span"][0] <= at < g["span"][1]), None)
+                found.append(_violation("absence_claim", ABSENCE_MESSAGE, text=m.group(0),
+                                        span=[at, span[0] + m.end()], context=sentence.strip()[:120],
+                                        segment=seg.get("id") if seg else None, unit=unit.get("id")))
+    return found
+
+
 #: 引用查询结果的写法。只跟着目录里的查询出现（catalog_prompt）：写作规则本身保持一期那份，升级前
 #: 发起的运行跑到报告节点时提示一字不差；只有口径卡的目录也不该教写作者去引用不存在的 Q1
 CELL_RULES = (
@@ -3320,8 +3376,16 @@ def _fields_prompt(sources: list[dict[str, Any]], catalog: dict[str, Any], snaps
             empty, list(dict.fromkeys(cited)))
 
 
-def _query_brief(entry: dict[str, Any], tables: list[str]) -> str:
-    """没展开的查询在目录里的一行摘要：编号、查的表（没有表结构时写工具名）、行数。"""
+def _query_brief(entry: dict[str, Any], tables: list[str], snaps: _Snapshots) -> str:
+    """没展开的查询在目录里的一行摘要：编号、查的表、行数。
+
+    表先看目录里的表条目（带表结构快照的查询才有），没有就从查询快照里存的 SQL 认（sql_tables），
+    再没有写工具名。
+    """
+    if not tables:
+        snapshot, why = snaps.get(entry["alias"], entry.get("artifact"))
+        if not why and isinstance(snapshot.get("sql"), str):
+            tables = sql_tables(snapshot["sql"])
     rows = entry.get("rows")
     what = ("、".join(tables[:3]) + (f" 等 {len(tables)} 张表" if len(tables) > 3 else "")) if tables \
         else str(entry.get("tool") or entry.get("source") or "查询")
@@ -3401,7 +3465,7 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str],
     for e in queries:
         groups.setdefault(str(e.get("node_id") or ""), []).append(e)
     compact = {e["alias"]: _query_prompt(e, snaps, cells_allowed, _PROMPT_COMPACT_ROWS) for e in queries}
-    brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []))] for e in queries}
+    brief = {e["alias"]: [_query_brief(e, tables.get(e["alias"], []), snaps)] for e in queries}
     shapes = {"brief": brief, "compact": compact, "full": full}
     tier = {e["alias"]: "brief" for e in queries}
 

backend/app/engine/nodes/report.py

@@ -44,6 +44,7 @@ from app.engine import judge as judging
 from app.engine.context import NodeContext, NodeError
 from app.engine.labels import field_label, option_label
 from app.engine.evidence import (
+    ABSENCE_RULE,
     CELLS_REASON,
     CLAIMS_RULE,
     ENTITY_KINDS,
@@ -84,8 +85,10 @@ CLAIMS = CLAIMS_VALUES
 #: 只标注、不让写作者重写也不判失败的违规：可疑实体（受管的正式出具按缺口降档）、表结构快照不全时
 #: 核对不了的名字（出口只标注）
 SOFT = frozenset({"unknown_entity", "unverified_entity"})
-#: 交回写作者重写、重写不好也不判失败的违规：句中的 [[see:]]（渲染后留下断句，数字和出处都没问题）
-REWRITE_ONLY = frozenset({"inline_see"})
+#: 交回写作者重写一次、重写不好也不判失败的违规：句中的 [[see:]]（渲染后留下断句，数字和出处都没问题）；
+#: 写作目录有省略时的缺失断言（「没有查到」「数据缺失」：写作者没看全证据，结论依据不足，但数字和出处都核对过）。
+#: 只剩这几类时不按 max_repairs 反复重写：重写一次没改掉，再写也多半一样，留在核对结果里标注
+REWRITE_ONLY = frozenset({"inline_see", "absence_claim"})
 #: 重写一次就是多一次整篇的模型调用，写不对的模型多给几次也大多写不对
 MAX_REPAIRS = 3
 
@@ -410,6 +413,9 @@ def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims
     instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \
         or "根据下面的证据写一份简洁的报告，先总后分。"
     rules = f"{MARKER_RULES}\n{SEE_AND_CODE_RULES}"
+    if writer.guarded:
+        # 写作者看不全证据（目录有省略、有来源的结构化结论为空）：「没有查到」这类结论没有依据
+        rules = f"{rules}\n{ABSENCE_RULE}"
     if claims in ("require_citation", CLAIMS_JUDGE):
         # 裁判只看写作者挂的依据：要判，就更得每句都挂上
         rules = f"{rules}\n{CLAIMS_RULE}" + (f"\n{judging.JUDGE_RULE}" if claims == CLAIMS_JUDGE else "")
@@ -500,12 +506,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
 
     def compose(raw: str) -> dict[str, Any]:
         return compose_doc(raw, catalog, node_id=ctx.node.id, run_id=ctx.run.run_id, allow_numbers=allow,
-                           cells_allowed=cells, entities=entities)
+                           cells_allowed=cells, entities=entities, absence=writer.trimmed)
 
     response, raw = await draft(messages)
     doc = compose(raw)
     repairs = 0
-    while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs:
+    while (blocking := _blocking(doc["violations"], numbers)) and repairs < max_repairs \
+            and not (repairs and not _fatal(blocking)):
         repairs += 1
         ctx.emit(EventType.LOG, level="warn", code="report_repair",
                  message=f"报告有 {len(blocking)} 处未通过核对（{_named(blocking)}），"
@@ -528,6 +535,13 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 draft=draft, compose=compose, numbers=numbers, blocking=blocking)
             spent.extend(usages)
 
+    if absent := [v for v in doc["violations"] if v.get("code") == "absence_claim"]:
+        # 重写之后还在（或者没让重写）：报告照常产出，这几处标注出来，时间线上单独说清为什么依据不足
+        texts = list(dict.fromkeys(str(v.get("text") or "") for v in absent))
+        tried = f"已要求模型重写 {repairs} 次仍未改掉" if repairs else "未要求重写"
+        ctx.emit(EventType.LOG, level="warn", code="report_absence_claim", count=len(absent), texts=texts,
+                 message=f"报告在证据目录有省略时写了 {len(absent)} 处缺失断言（{_named(absent)}），{tried}："
+                         "被省略的查询仍在证据台账里，这类结论依据不足，应写作本报告未覆盖。报告照常产出，这几处已标注")
     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,

backend/app/engine/review.py

@@ -91,6 +91,8 @@ WARN_CODES: dict[str, tuple[str, str]] = {
     "report_no_evidence": ("report_no_evidence", DEGRADED),
     # 证据目录超出篇幅，有的查询只列了编号和摘要：写作模型没看全证据，报告的覆盖面有缺口
     "report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED),
+    # 目录有省略时写了「没有查到」这类缺失断言、重写后仍在：报告照常产出并标注，这几句结论依据不足
+    "report_absence_claim": ("report_absence_claim", DEGRADED),
     # 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据
     "evidence_store_failed": ("evidence_store_failed", DEGRADED),
     # agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了

backend/tests/test_report_absence_claim.py

@@ -0,0 +1,82 @@
+"""缺失断言（absence_claim）：写作目录有省略时，「没有查到」「不存在」「缺失」「没有数据」这类结论依据不足。
+
+2026-10 一次运行，写作者只看到上百次查询里的前二十来次，写出「某类数据缺失：本轮没有查到任何结果」这样的断言，
+可那个 Agent 实际查到了上百条记录。
+
+- 写作规则：清单有省略、或有来源的结构化结论为空时，不得断言没有数据，只能写「本报告未覆盖」并说明依据不足
+- 组装文档：清单有省略时按一组明确的说法认出缺失断言，记 absence_claim，交回重写一次；重写后仍有的只标注、不判失败
+  （report.REWRITE_ONLY，和句中的 [[see:]] 同一种处理）
+- 只认拿得准的说法：「不存在明显差异」「本报告未覆盖」「不代表没有数据」都不算
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.engine.evidence import ABSENCE_RULE, build_catalog, compose_doc
+from app.engine.nodes.report import REWRITE_ONLY
+
+
+@pytest.fixture
+def catalog():
+    card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1",
+            "metrics": [{"id": "gmv", "name": "销售额", "value": 45678.5, "unit": "元", "format": "thousands"}]}
+    return build_catalog(nodes={"card": card}, ledger=[{"kind": "metric_set", "node_id": "card", "artifact": "c" * 64}])
+
+
+def claims(doc):
+    return [v for v in doc["violations"] if v["code"] == "absence_claim"]
+
+
+def test_the_rule_tells_the_writer_what_to_write_instead():
+    for phrase in ("没有查到", "不存在", "缺失", "没有数据", "本报告未覆盖", "依据不足"):
+        assert phrase in ABSENCE_RULE
+    assert "absence_claim" in REWRITE_ONLY and "inline_see" in REWRITE_ONLY
+
+
+@pytest.mark.parametrize("text, found", [
+    ("卡券数据缺失：本轮没有查到核销、兑换、退券的任何结果。", ["数据缺失", "没有查到"]),
+    ("卡券核销没有任何记录。", ["没有任何记录"]),
+    ("本周门店客流数据为空。", ["数据为空"]),
+    ("活动内容无数据。", ["无数据"]),
+    ("未查询到会员流失的情况。", ["未查询到"]),
+    ("库里不存在相关记录。", ["不存在相关记录"]),
+    ("兑换记录均不存在。", ["记录均不存在"]),
+    ("暂无数据，查不到核销明细。", ["暂无数据", "查不到"]),
+    ("| 方向 | 结论 |\n| --- | --- |\n| 卡券 | 没有查到 |", ["没有查到"]),
+])
+def test_absence_claims_are_recognised(catalog, text, found):
+    doc = compose_doc(text, catalog, absence=True)
+    got = claims(doc)
+    assert [v["text"] for v in got] == found, doc["violations"]
+    for v in got:
+        assert doc["markdown"][v["span"][0]:v["span"][1]] == v["text"], "span 指着正文里的那几个字"
+        assert v.get("unit") and v.get("context")
+        assert "依据不足" in v["message"] and "本报告未覆盖" in v["for_model"]
+
+
+@pytest.mark.parametrize("text", [
+    "本报告未覆盖卡券核销的明细：证据目录有省略，相关查询未展开，依据不足。",
+    "两家门店的客单价不存在明显差异。",
+    "目录里没展开的查询不代表没有数据。",
+    "这并非没有查到，而是没有展开。",
+    "本周无数据库变更，无数据源切换。",
+    "本周销售额 [[m:gmv]]，缺少口径说明的部分已注明。",
+    "```sql\nSELECT * FROM t -- 没有查到\n```",
+])
+def test_other_wording_is_left_alone(catalog, text):
+    assert not claims(compose_doc(text, catalog, absence=True))
+
+
+def test_only_checked_when_the_catalog_was_trimmed(catalog):
+    text = "卡券核销数据缺失：没有查到任何记录。"
+    assert not claims(compose_doc(text, catalog))
+    assert claims(compose_doc(text, catalog, absence=True))
+
+
+def test_the_claim_can_be_located_in_the_document(catalog):
+    doc = compose_doc("## 卡券\n\n本周销售额 [[m:gmv]]。核销数据缺失。", catalog, absence=True)
+    [v] = claims(doc)
+    segments = {s["id"]: s for b in doc["blocks"] for u in b["units"] for s in u["segments"]}
+    seg = segments[v["segment"]]
+    assert seg["kind"] == "text" and seg["span"][0] <= v["span"][0] < seg["span"][1]
+    assert v["context"] == "核销数据缺失。"

backend/tests/test_report_evidence_regression.py

@@ -0,0 +1,316 @@
+"""回归：「4 个专项 Agent 并行查库 → 报告」整条跑一遍，复现 2026-10 那次运行的形状。
+
+那次运行状态是 succeeded，报告却不可靠：4 个 Agent 的结构化抽取全部被截断（推理模型的额度在推理阶段就用完了）；
+写作者的出处清单写死 12000 字，上百次查询只有前二十来次露了面；写作者于是在清单被截断的情况下写了「数据缺失：
+本轮没有查到任何核销结果」，可那个 Agent 查到了上百条。
+
+这里用合成的门店、会员、卡券、活动库（SQLite）真跑：4 个 Agent 分别查 26、17、79、42 次（虚构的次数，一次并行发完），都配了
+output_schema 和按出处核对字段；写作模型、抽取模型都是脚本。验证：
+
+- 抽取被截断后加大额度重试成功；两次都截断的记 agent_field_truncated，字段为空值
+- 出处清单在预算内，每个 Agent 都有代表，被省略的查询有编号和摘要，开头说明各来源省略了几次
+- 清单有省略时的缺失断言被认出来（absence_claim）、交回重写；重写后仍有的只标注、不判失败
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import sqlite3
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+from openai import LengthFinishReasonError
+from openai.types import CompletionUsage
+from openai.types.chat import ChatCompletion
+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 ABSENCE_RULE, DEFAULT_CATALOG_BUDGET, MARKER_RULES
+from app.engine.nodes import llm as llm_mod
+from app.engine.nodes import report as report_mod
+from app.engine.runner import run_manager
+from tests.fixtures.sources import drop_source
+
+SOURCE = "fa_store"
+TOOL = f"db_query__{SOURCE}"
+#: (节点 id, 画布上的名字, 记号, 查询次数, 表, 列)
+AGENTS = [
+    ("agent_member", "会员专项", "member", 26, "member_daily", ["store_name", "biz_date", "new_members", "active_members"]),
+    ("agent_store", "门店专项", "store", 17, "store_daily", ["store_name", "biz_date", "visits", "sales"]),
+    ("agent_coupon", "卡券核销", "coupon", 79, "coupon_redemption", ["store_name", "redeemed_at", "coupon_type", "amount"]),
+    ("agent_campaign", "活动内容", "campaign", 42, "campaign_view", ["campaign", "biz_date", "views", "clicks"]),
+]
+SCHEMA = {"type": "object", "properties": {
+    "first_store": {"type": "string", "description": "第一次查询第一行的门店或活动"},
+    "first_value": {"type": "number", "description": "第一次查询第一行的第三列"},
+    "churn_rate": {"type": "number", "description": "流失率（要计算，填 null）"},
+}}
+ROWS_PER_QUERY = 5
+
+ABSENT = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n"
+          "## 卡券核销\n\n卡券数据缺失：本轮没有查到核销、兑换的任何结果。")
+COVERED = ("## 概览\n\n各专项都已完成查询。[[see:Q1]]\n\n"
+           "## 卡券核销\n\n本报告未覆盖卡券核销的明细：证据目录有省略，相关查询未展开，依据不足。[[see:Q1]]")
+
+
+def length_error(completion: int) -> LengthFinishReasonError:
+    usage = CompletionUsage(completion_tokens=completion, prompt_tokens=14000, total_tokens=14000 + completion)
+    return LengthFinishReasonError(completion=ChatCompletion.model_construct(
+        id="x", choices=[], created=0, model="m", object="chat.completion", usage=usage))
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def store_db(tmp_path):
+    """合成库：每张表 120 行，第 i 次查询取 OFFSET i 的 5 行，次次结果不同。"""
+    path = tmp_path / "store.db"
+    db = sqlite3.connect(path)
+    for _, _, _, _, table, cols in AGENTS:
+        db.execute(f"CREATE TABLE {table} ({cols[0]} TEXT, {cols[1]} TEXT, {cols[2]} {'TEXT' if table == 'coupon_redemption' else 'INTEGER'}, {cols[3]} REAL)")
+        rows = []
+        for r in range(120):
+            when = f"2026-06-{r % 28 + 1:02d}" + (f"T08:{r % 60:02d}:39" if table == "coupon_redemption" else "")
+            third = f"券型{r % 4 + 1}" if table == "coupon_redemption" else (r * 7) % 300 + 5
+            rows.append((f"门店{r % 30 + 1:02d}", when, third, round((r * 13) % 900 + 0.5, 1)))
+        db.executemany(f"INSERT INTO {table} VALUES (?, ?, ?, ?)", rows)
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = DataSource(name=SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
+        session.add(row)
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield path
+    await drop_source(source_id)
+
+
+def sql_of(table: str, cols: list[str], i: int) -> str:
+    return f"SELECT {', '.join(cols)} FROM {table} ORDER BY rowid LIMIT {ROWS_PER_QUERY} OFFSET {i}"
+
+
+def node(nid, ntype, label, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}}
+
+
+def graph(counts: dict[str, int] | None = None, **report):
+    nodes = [node("start", "input", "开始")]
+    for nid, title, key, count, table, cols in AGENTS:
+        nodes.append(node(nid, "agent", title, prompt=f"【{key}】按{title}的口径查数，每次查一段", tools=[TOOL],
+                          approval="never", parallel_tools=True, output_schema=SCHEMA, cite_fields=True,
+                          assign_to=f"{key}_findings"))
+    nodes.append(node("write", "report", "专项报告", instructions="汇总各专项的发现", **report))
+    nodes.append(node("out", "output", "成果", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]))
+    edges = [{"source": "start", "target": a[0]} for a in AGENTS] + [{"source": a[0], "target": "write"} for a in AGENTS]
+    edges.append({"source": "write", "target": "out"})
+    return {"nodes": nodes, "edges": edges}
+
+
+class World:
+    """脚本化的模型：Agent 一步并行发完自己的查询再给结论；抽取按 outcome 决定；写作者依次回 drafts。"""
+
+    def __init__(self, monkeypatch, db_path, *, counts: dict[str, int], drafts: list[str], outcome):
+        from app.providers import mock_model
+
+        self.counts = counts
+        self.drafts = drafts
+        self.writer: list[list] = []
+        self.extract_calls: dict[str, list[int | None]] = {a[2]: [] for a in AGENTS}
+        self.budgets: list[int | None] = []
+        db = sqlite3.connect(db_path)
+        self.first_rows = {key: db.execute(sql_of(table, cols, 1)).fetchone() for _, _, key, _, table, cols in AGENTS}
+        db.close()
+        real = llm_mod.get_chat_model
+
+        async def spy(session, spec):
+            if spec.thinking == "off":
+                self.budgets.append(spec.max_tokens)
+            return await real(session, spec)
+
+        monkeypatch.setattr(llm_mod, "get_chat_model", spy)
+        world = self
+
+        def text(messages) -> str:
+            return "\n".join(m.content if isinstance(m.content, str) else json.dumps(m.content, ensure_ascii=False)
+                             for m in messages)
+
+        def who(messages) -> str:
+            found = re.search(r"【(member|store|coupon|campaign)】", text(messages))
+            assert found, "认不出是哪个 Agent"
+            return found.group(1)
+
+        def decide(model, messages):
+            if model.response_format:
+                key = who(messages)
+                world.extract_calls[key].append(world.budgets[-1])
+                outcome(key, len(world.extract_calls[key]), world.budgets[-1])
+                return AIMessage(content=json.dumps(world.extracted(key), ensure_ascii=False))
+            if any(isinstance(m, SystemMessage) and "报告撰写人" in str(m.content) for m in messages):
+                world.writer.append(list(messages))
+                return AIMessage(content=world.drafts[min(len(world.writer), len(world.drafts)) - 1])
+            key = who(messages)
+            if model.tools and not any(isinstance(m, ToolMessage) for m in messages):
+                _, _, _, _, table, cols = next(a for a in AGENTS if a[2] == key)
+                return AIMessage(content="分段查询。", tool_calls=[
+                    {"name": TOOL, "args": {"sql": sql_of(table, cols, i)}, "id": f"{key}_{i}"}
+                    for i in range(1, world.counts[key] + 1)])
+            return AIMessage(content=f"【{key}】查完了，结论见各次查询。")
+
+        monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
+
+    def extracted(self, key: str) -> dict:
+        first = self.first_rows[key]
+        _, _, _, _, _, cols = next(a for a in AGENTS if a[2] == key)
+        return {"first_store": {"value": first[0], "from": {"call": "Q1", "row": 0, "column": cols[0]}},
+                "first_value": {"value": first[3], "from": {"call": "Q1", "row": 0, "column": cols[3]}},
+                "churn_rate": {"value": None, "from": None}}
+
+
+async def finish(graph_json) -> Run:
+    run = await run_manager.start(graph=graph_json, input_payload={})
+    for _ in range(1200):
+        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 {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(evts, code):
+    return [e.data for e in evts if e.type == "log" and e.data.get("code") == code]
+
+
+async def output_of(run_id: str, node_id: str) -> dict:
+    [finished] = [e for e in await events(run_id, "node.finished") if e.node_id == node_id]
+    return artifact_store.load(finished.data["artifact"])
+
+
+def catalog_text(messages) -> str:
+    """写作者第一次收到的提示里出处清单那一段（从「可引用的证据」到写作规则之前）。"""
+    prompt = next(m.content for m in messages if isinstance(m, HumanMessage))
+    return prompt[prompt.index("可引用的证据"):prompt.index(MARKER_RULES)].rstrip()
+
+
+# --------------------------------------------------------------------------
+# 和那次运行同样的形状：上百次查询、分布不均
+# --------------------------------------------------------------------------
+
+
+async def test_the_reported_run_shape_end_to_end(monkeypatch, store_db):
+    counts = {key: count for _, _, key, count, _, _ in AGENTS}
+
+    def outcome(key, attempt, budget):
+        if key == "coupon":
+            raise length_error(budget)                  # 两次都截断
+        if key == "member" and attempt == 1:
+            raise length_error(budget)                  # 截断一次，加大额度后成功
+
+    world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT, COVERED], outcome=outcome)
+    row = await finish(graph())
+    assert row.status == "succeeded", row.error
+    evts = await events(row.id)
+
+    # 1. 抽取：会员专项截断后重试成功；卡券核销两次都截断，字段为空值
+    assert world.extract_calls["member"] == [16384, 32768]
+    assert world.extract_calls["coupon"] == [16384, 32768]
+    assert world.extract_calls["store"] == world.extract_calls["campaign"] == [16384]
+    member = await output_of(row.id, "agent_member")
+    assert member["data"]["first_store"] == world.first_rows["member"][0]
+    assert member["data_evidence"]["first_store"]["status"] == "verified"
+    coupon = await output_of(row.id, "agent_coupon")
+    assert coupon["data"] == {"first_store": None, "first_value": None, "churn_rate": None}
+    [cut] = logs(evts, "agent_field_truncated")
+    assert cut["fields"] == ["first_store", "first_value", "churn_rate"] and "推理阶段就用完了" in cut["message"]
+    assert not logs(evts, "agent_field_unverified")
+
+    # 2. 出处清单：预算内，每个 Agent 都有代表，被省略的有编号和摘要，开头说明各来源省略了几次
+    first = world.writer[0]
+    text = catalog_text(first)
+    assert len(text) <= DEFAULT_CATALOG_BUDGET, len(text)
+    head = text[:900]
+    assert "目录有省略" in head
+    for _, title, _, count, table, _ in AGENTS:
+        assert f"{title} {count} 次" in head
+        group = text[text.index(f"来自「{title}」的查询"):]
+        group = group[:group.index("\n\n", 1)] if "\n\n" in group[1:] else group
+        assert f"共 {count} 次" in group and re.search(r"\n {2}r0：", group), f"「{title}」没有展开的查询"
+    briefs = re.findall(r"\n {2}- (Q\d+)：(\w+)，5 行", text)
+    assert briefs, "没有列出被省略的查询"
+    assert {t for _, t in briefs} <= {a[4] for a in AGENTS}
+    assert all(f"Q{n}：" in text for n in range(1, 165)), "每次查询都要在清单里（展开的或摘要）"
+    # 结构化结论排在查询前面；抽取失败的卡券核销写明结构化结论为空、不代表没有数据
+    assert text.index("各 Agent 交来的结构化结论") < text.index("来自「")
+    assert "「卡券核销」：结构化结论为空" in text
+    # 写作规则里有缺失断言那一条
+    assert ABSENCE_RULE in next(m.content for m in first if isinstance(m, HumanMessage))
+    [trimmed] = logs(evts, "report_catalog_trimmed")
+    assert trimmed["queries"] == 164 and trimmed["budget"] == DEFAULT_CATALOG_BUDGET
+    assert [n["title"] for n in sorted(trimmed["nodes"], key=lambda n: n["title"])] == \
+        sorted(a[1] for a in AGENTS)
+    assert trimmed["omitted"] == sum(n["omitted"] for n in trimmed["nodes"]) > 0
+
+    # 3. 缺失断言：认出来、交回重写一次；改成「本报告未覆盖」后通过
+    assert len(world.writer) == 2
+    request = str(world.writer[1][-1].content)
+    assert "本报告未覆盖" in request and "没有查到" in request
+    [repair] = logs(evts, "report_repair")
+    assert "数据缺失" in repair["message"] or "没有查到" in repair["message"]
+    [checked] = [e.data for e in evts if e.type == "report.checked"]
+    assert not [v for v in checked["violations"] if v["code"] == "absence_claim"]
+    assert not logs(evts, "report_absence_claim")
+
+
+# --------------------------------------------------------------------------
+# 重写后仍然写缺失断言：只标注、不判失败
+# --------------------------------------------------------------------------
+
+SMALL = {"member": 6, "store": 4, "coupon": 12, "campaign": 8}
+
+
+async def test_an_absence_claim_that_survives_the_rewrite_is_flagged_not_failed(monkeypatch, store_db):
+    monkeypatch.setattr(report_mod, "catalog_budget", lambda window: 4000)
+    world = World(monkeypatch, store_db, counts=SMALL, drafts=[ABSENT], outcome=lambda *a: None)
+    row = await finish(graph(on_violation="fail", max_repairs=3))
+    assert row.status == "succeeded", row.error
+    assert len(world.writer) == 2, "缺失断言只交回重写一次，不按最多重写次数反复重写"
+    evts = await events(row.id)
+    [checked] = [e.data for e in evts if e.type == "report.checked"]
+    assert checked["failed"] is False
+    claims = [v for v in checked["violations"] if v["code"] == "absence_claim"]
+    assert {v["text"] for v in claims} == {"数据缺失", "没有查到"}
+    assert all(v.get("segment") and v.get("context") for v in claims), "带着片段和原句：证据面板能定位"
+    [flag] = logs(evts, "report_absence_claim")
+    assert flag["level"] == "warn" and flag["count"] == 2 and flag["texts"] == ["数据缺失", "没有查到"]
+    assert "本报告未覆盖" in flag["message"] or "依据不足" in flag["message"]
+
+
+async def test_without_omissions_an_absence_statement_is_left_alone(monkeypatch, store_db):
+    """目录全给了：「没有查到」可能就是看了数据说的，不查。"""
+    counts = {"member": 1, "store": 1, "coupon": 1, "campaign": 1}
+    world = World(monkeypatch, store_db, counts=counts, drafts=[ABSENT], outcome=lambda *a: None)
+    row = await finish(graph())
+    assert row.status == "succeeded", row.error
+    assert len(world.writer) == 1
+    evts = await events(row.id)
+    assert not logs(evts, "report_catalog_trimmed") and not logs(evts, "report_absence_claim")
+    prompt = next(m.content for m in world.writer[0] if isinstance(m, HumanMessage))
+    assert ABSENCE_RULE not in prompt and "目录有省略" not in prompt

frontend/src/lib/terms.ts

@@ -203,6 +203,19 @@ export function evidenceTally(cited: number, total: number, other = 0): string {
   return parts.join(' · ')
 }
 
+/**
+ * 缺失断言：证据目录有省略时，报告写了「没有查到」「不存在」「缺失」这类结论（违规 absence_claim，重写后仍有时
+ * 报告节点另发日志 report_absence_claim）。证据面板的违规清单和运行时间线共用这一套说法
+ */
+export const ABSENCE_CLAIM_TEXT = {
+  label: '缺失断言',
+  hint: '写作模型没有看到全部证据：被省略的查询仍在证据台账里，看不到不等于不存在，这类结论应写作「本报告未覆盖」。'
+    + '报告照常产出，这一处只标注、不判失败',
+  title: (n?: number) => `报告写了${n ? ` ${formatNumber(n)} 处` : ''}缺失断言，已标注（证据目录有省略）`,
+  next: '被省略的查询仍在证据台账里，可在证据面板中查看。请在报告撰写节点的指令中要求未展开的方向写作「本报告未覆盖」，'
+    + '或换用上下文窗口更大的写作模型后重新运行',
+}
+
 /** 证据面板和证据条的文案 */
 export const EVIDENCE_TEXT = {
   allCited: (n: number) => `${formatNumber(n)} 个数字都有出处`,

frontend/src/run/EvidencePanel.tsx

@@ -18,7 +18,7 @@ import {
 } from '../lib/evidence'
 import { humanizeError } from '../lib/errors'
 import { formatDateTime, formatNumber, NONE, shortId } from '../lib/format'
-import { EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms'
+import { ABSENCE_CLAIM_TEXT, EVIDENCE_TEXT, JUDGE_TEXT, MERGE_TEXT, NODE_TYPE_LABEL, RUN_SQL_CHECK_TEXT, SQL_CHECK_TEXT, nodeTypeLabel } from '../lib/terms'
 import { useCatalog } from '../store/catalog'
 import { askKey, segmentKey, useEvidence, useExplore, useVerdicts } from '../store/evidence'
 import type {
@@ -1552,6 +1552,7 @@ const CODE_LABEL: Record<string, string> = {
   unresolved_ref: '引用无法解析',
   unknown_entity: '疑似不存在的名称',
   unverified_entity: '无法核实的名称',
+  absence_claim: ABSENCE_CLAIM_TEXT.label,
 }
 
 /** 违规清单：报告撰写节点核对出来的全部问题，画得出线的能定位，画不出的说清在哪 */
@@ -1563,7 +1564,9 @@ function ViolationList({ doc, onLocate }: { doc: EvidenceDocData; onLocate: (seg
       <p className="text-2xs text-faint">{EVIDENCE_TEXT.violationsHint}</p>
       <ul className="space-y-2" data-ev-violations="">
         {list.map((v, i) => {
-          const seg = locatable(doc, v)
+          // 缺失断言说的是一整句话，正文里普通文字不是可聚焦的片段：不给「定位」，原句写在下面
+          const sentence = v.code === 'absence_claim'
+          const seg = sentence ? null : locatable(doc, v)
           return (
             <li key={i} className="rounded border px-2 py-1.5" data-ev-violation={v.code}
                 data-ev-locatable={seg ? 'yes' : 'no'}>
@@ -1576,8 +1579,11 @@ function ViolationList({ doc, onLocate }: { doc: EvidenceDocData; onLocate: (seg
                 )}
               </div>
               {v.message && <p className="mt-1 text-dim">{v.message}</p>}
+              {v.code === 'absence_claim' && (
+                <p className="mt-0.5 text-2xs text-dim" data-ev-absence-hint="">{ABSENCE_CLAIM_TEXT.hint}</p>
+              )}
               {v.context && <p className="mono mt-0.5 text-2xs text-faint [overflow-wrap:anywhere]">「{v.context.trim()}」</p>}
-              {!seg && (
+              {!seg && !sentence && (
                 <p className="mt-0.5 text-2xs" style={{ color: 'var(--st-waiting)' }}>
                   {v.segment ? EVIDENCE_TEXT.structural : EVIDENCE_TEXT.noSegment}
                 </p>

frontend/src/run/decode.ts

@@ -2,7 +2,7 @@ import type { CatalogDriftTable, RunEvent, SqlCheckItem, TeamMember, TeamRound,
 import type { NodeState, RunPhase } from './trace'
 import { formatDuration, formatNumber } from '../lib/format'
 import {
-  CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, COPILOT_CONTEXT_TEXT, JUDGE_TEXT, MERGE_TEXT, RUN_SQL_CHECK_TEXT, SCHEMA_PARTIAL_TEXT, SQL_CHECK_TEXT, TYPE_LABEL,
+  ABSENCE_CLAIM_TEXT, CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, COPILOT_CONTEXT_TEXT, JUDGE_TEXT, MERGE_TEXT, RUN_SQL_CHECK_TEXT, SCHEMA_PARTIAL_TEXT, SQL_CHECK_TEXT, TYPE_LABEL,
   claimTally,
   evidenceTally, issuanceLabel, nodeTypeLabel,
 } from '../lib/terms'
@@ -756,6 +756,16 @@ function explainLog(code: string | undefined, message: string, d: Record<string,
         sub: message.match(/(?:没|未)通过核对（(.+?)）/)?.[1],
       }
     }
+    case 'report_absence_claim': {
+      // 证据目录有省略时报告写了缺失断言，重写后仍在（report.py）：读结构化字段 count、texts，不从原话里认
+      const n = typeof d.count === 'number' && d.count > 0 ? d.count : undefined
+      const texts = strs(d.texts)
+      return {
+        title: ABSENCE_CLAIM_TEXT.title(n),
+        sub: texts.length ? texts.map((t) => `「${t}」`).join('、') : undefined,
+        next: ABSENCE_CLAIM_TEXT.next,
+      }
+    }
     case 'report_catalog_trimmed': {
       // 写作目录超出篇幅（report.py _trimmed_message）。读结构化字段（queries、omitted、nodes），不从原话里认
       const total = typeof d.queries === 'number' ? d.queries : undefined

scripts/check-decode.mjs

@@ -1320,6 +1320,26 @@ await section('报告核对 report.checked、口径卡的台账（可点击证
     .find((s) => s.code === 'report_catalog_trimmed')
   check('……缺字段时标题照样说人话，不出 undefined', trimmedBare?.title === '证据目录超出篇幅，部分查询只列了编号和摘要'
     && trimmedBare?.sub === undefined, `${trimmedBare?.title} / ${trimmedBare?.sub}`)
+
+  // 目录有省略时报告写了缺失断言（「没有查到」「数据缺失」），交回重写一次仍在：只标注、不判失败，时间线单独一行说清
+  check('后端发缺失断言的警告，带着 count、texts', reportPy.includes('code="report_absence_claim"')
+    && reportPy.includes('count=len(absent), texts=texts'), '改了 report.py 的字段就同步改 decode.ts 的 explainLog 和这里的夹具')
+  const ABSENT = '报告在证据目录有省略时写了 2 处缺失断言（「数据缺失」、「没有查到」），已要求模型重写 1 次仍未改掉：'
+    + '被省略的查询仍在证据台账里，这类结论依据不足，应写作本报告未覆盖。报告照常产出，这几处已标注'
+  const absentRun = flatten(mod.decodeRun([E(1, 'node.started', 'w', { node_type: 'report', label: '写报告' }),
+    E(2, 'log', 'w', { level: 'warn', code: 'report_absence_claim', message: ABSENT, count: 2, texts: ['数据缺失', '没有查到'] }),
+    E(3, 'report.checked', 'w', { doc_artifact: 'abc', repairs: 1, ok: false, on_violation: 'fail', failed: false,
+      stats: { numbers: 0, numbers_cited: 0, uncited_numbers: 0, unresolved: 0 },
+      violations: [{ code: 'absence_claim', text: '数据缺失', message: '证据目录有省略时写了缺失断言：写作模型没有看到全部证据，「没有查到」「不存在」「缺失」这类结论依据不足' }] }),
+  ]))
+  const ab = absentRun.find((s) => s.code === 'report_absence_claim')
+  check('report_absence_claim：标题说几处缺失断言、已标注、为什么（目录有省略）', ab?.title === '报告写了 2 处缺失断言，已标注（证据目录有省略）', ab?.title)
+  check('……副标题点名那几处说法', ab?.sub === '「数据缺失」、「没有查到」', ab?.sub)
+  check('……下一步说被省略的仍在台账里、应写「本报告未覆盖」；是提醒（warn），原话留在展开区',
+    !!ab?.next?.includes('本报告未覆盖') && !!ab?.next?.includes('证据台账') && ab?.level === 'warn' && ab?.detail === ABSENT, ab?.next)
+  const rcAbsent = absentRun.find((s) => s.code === 'report_checked')
+  check('……核对那一行照常完成（不判失败），违规清单里写着这条的说明', rcAbsent?.status === 'done'
+    && !!rcAbsent?.detail?.includes('缺失断言') && !!rcAbsent?.detail?.includes('依据不足'), `${rcAbsent?.status} ${rcAbsent?.detail}`)
 })
 
 await section('工具门控 tool.gated（工具信任三档）', async () => {

scripts/check-evidence.mjs

@@ -1094,7 +1094,34 @@ await section('sqlcheck', 'SQL 检查（数据目录阶段 4B）：查询步骤
   await bad.ctx.close()
 })
 
-await section('q-narrow', '查询步骤在 360px：栏内展开、底部抽屉都不横向滚动；减少动效', async () => {
+await section('absence', '缺失断言：证据目录有省略时写了「没有查到」，违规清单写明是什么、为什么依据不足，带着原句', async () => {
+  // 违规的形状：backend/app/engine/evidence.py 的 absence_claims（code、message、text、span、segment、unit、context）
+  const MESSAGE = '证据目录有省略时写了缺失断言：写作模型没有看到全部证据，「没有查到」「不存在」「缺失」这类结论依据不足'
+  const r = await probeQ((id, body) => body, {
+    patchQuery: (f) => {
+      const units = f.doc.blocks.flatMap((b) => b.units ?? [])
+      const unit = units.find((u) => (u.segments ?? []).some((s) => s.kind === 'text' && s.text.trim().length > 2))
+      const seg = unit.segments.find((s) => s.kind === 'text' && s.text.trim().length > 2)
+      f.doc.violations = [...(f.doc.violations ?? []), { code: 'absence_claim', message: MESSAGE, text: '没有查到',
+        span: [seg.span[0], seg.span[0] + 1], segment: seg.id, unit: unit.id, context: '核销数据缺失：本轮没有查到任何结果。' }]
+      return f
+    },
+  })
+  await openQ(r.page, '1,288')
+  await r.page.locator('[data-evidence-panel] [data-ev-open-violations]').click()
+  await r.page.waitForSelector('[data-ev-violations]')
+  const item = r.page.locator('[data-ev-violation="absence_claim"]')
+  const text = await item.innerText().catch(() => '')
+  check('违规清单里有这一条，标「缺失断言」，不是笼统的「文档核对未通过」', text.includes('缺失断言') && !text.includes('文档核对未通过'), text)
+  check('……写明为什么依据不足、该怎么写（文案在 terms.ts）', text.includes(MESSAGE)
+    && (await item.locator('[data-ev-absence-hint]').innerText().catch(() => '')).includes('本报告未覆盖'), text.replace(/\s+/g, ' '))
+  check('……带着原句；说的是一整句，不给「定位」，也不说「正文里没有对应文本」', text.includes('本轮没有查到任何结果')
+    && await item.getByRole('button', { name: '定位' }).count() === 0 && !text.includes('没有对应文本'), text.replace(/\s+/g, ' '))
+  check('没有运行时报错', r.errors.length === 0, r.errors.join(' | '))
+  await r.ctx.close()
+})
+
+await section('q-narrow','查询步骤在 360px：栏内展开、底部抽屉都不横向滚动；减少动效', async () => {
   const overflow = () => page.evaluate((sel) => {
     const el = document.querySelector(sel)
     return { sw: el.scrollWidth, cw: el.clientWidth }