写作者的出处清单按写作模型的窗口定篇幅，放不下时分优先级、按来源均衡，省略了什么照实说

一次真实运行里证据台账有上百次查询（4 个 Agent 查询次数悬殊），清单写死 12000 字，十万字以上的目录被截到一万两千字以内，只有前二十来次查询进了写作者的视野；查询按编号排，前面的 Agent 占满了预算，有的 Agent 一次都没露面。

- 预算：写作模型上下文窗口的 40%（字数按 1 字约 1 token 保守算），窗口先看模型接入里给这个模型登记的 context，再看模型目录；拿不到按 128K（和 Agent 护栏同一个保守值），即 51200 字；下限 12000、上限 160000
- 第一优先：各 Agent 经按出处核对交来的结构化字段（换成报告目录里的全局编号，值按单元格引用的渲染规则写，给出可照抄的引用标记）、口径卡指标和运行输入，一律列出；结构化结论为空的来源写明不代表没有数据
- 查询按产出它的节点分组、轮流展开：字段引用到的查询先展开，再给每次查询列前 3 行，有余量再补到 20 行；表和字段、知识库检索最多占预算的四分之一
- 没展开的查询列出编号和一行摘要（表、行数）；约定只能在句末写 [[see:Qn]] 作依据，不猜行列、不凭摘要写数，数字照样逐字核对
- 清单开头明说目录有省略、各来源各省略了几次；报告节点发 report_catalog_trimmed 警告（带 queries、omitted、nodes），时间线给它专门的标题和下一步，运行复核算作缺口
- 放得下时清单和以前一字不差（多出的只有结构化字段那一节）

backend/app/engine/evidence.py

@@ -47,7 +47,9 @@ import json
 import math
 import re
 import unicodedata
+from collections import Counter
 from collections.abc import Callable
+from dataclasses import dataclass, field
 from decimal import ROUND_HALF_UP, Context, Decimal
 from typing import Any, Literal, TypedDict
 
@@ -57,6 +59,7 @@ from app.data.names import FILLERS, name_key
 from app.data.provenance_types import DOC_PROVENANCE
 from app.data.tabular import UNSHAPED_NOTE
 from app.engine.expressions import CellError, cell_value, column_kind, locate_cell
+from app.engine.guards import FALLBACK_CONTEXT
 from app.engine.issuance import _tolerance, extract_numbers, number_allowance
 from app.engine.labels import option_label
 
@@ -3094,10 +3097,12 @@ def _sql_check_note(artifact: Any, snaps: _Snapshots) -> str | None:
             "引用时须如实说明，不要当作确定的结论）")
 
 
-def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> list[str]:
+def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool,
+                  shown_rows: int = _PROMPT_ROWS) -> list[str]:
     """一次查询在写作目录里的样子：编号、来源、列，以及能引用的行（行号 + 渲染后的值）。
 
-    值按 [[v:]] 的渲染规则写出来：写作者在这里看到的，就是引用之后报告里显示的字。
+    值按 [[v:]] 的渲染规则写出来：写作者在这里看到的，就是引用之后报告里显示的字。shown_rows 是列出几行：
+    目录放不下时先给每次查询列前几行（_PROMPT_COMPACT_ROWS），再补到 _PROMPT_ROWS。
     """
     alias = entry["alias"]
     cols = entry.get("columns") or []
@@ -3115,7 +3120,7 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool)
     columns = [str(c) for c in snapshot.get("columns") or []][:_PROMPT_COLS]
     rows = snapshot["rows"]
     example = None
-    for r, record in enumerate(rows[:_PROMPT_ROWS]):
+    for r, record in enumerate(rows[:shown_rows]):
         cells = []
         for c in columns:
             try:
@@ -3130,22 +3135,78 @@ def _query_prompt(entry: dict[str, Any], snaps: _Snapshots, cells_allowed: bool)
         lines.append(f"  r{r}：" + "，".join(cells))
     if not rows:
         lines.append("  （0 行，没有可以引用的格）")
-    elif len(rows) > _PROMPT_ROWS:
-        lines.append(f"  …共 {len(rows)} 行，这里只列出前 {_PROMPT_ROWS} 行；后面的行也能按行号引用")
+    elif len(rows) > shown_rows:
+        lines.append(f"  …共 {len(rows)} 行，这里只列出前 {shown_rows} 行；后面的行也能按行号引用")
     if rows and columns:
         lines.append(f"  例：[[v:{alias}.r0.{example or columns[0]}]]")
     return lines
 
 
-def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowed: bool = True,
-                   loader: Loader | None = None) -> str:
-    """证据目录的文字版，放进写作者的 prompt。超过预算就截断，并照实说截断了。
+#: 写作目录的字数预算（catalog_budget）按写作模型的上下文窗口定，不再写死 12000：
+#: - 占窗口的 40%。其余留给写作规则和指令（约 4 千字）、重写轮次（每轮把上一稿和违规清单追加进对话，最多 3 轮）、
+#:   推理模型的思考和报告正文（输出额度）。字数按 1 字 ≈ 1 token 算：中文一个字约 0.6–1 个 token，目录里的数字、
+#:   列名、标点每个字更少，按 1:1 算偏保守，实际占用低于 40%
+#: - 拿不到窗口时按 128K（和 Agent 护栏的 FALLBACK_CONTEXT 同一个保守值），即 51200 字。原来的 12000 字只放得下
+#:   20 多次查询：2026-10 一次上百次查询的运行，十万字以上的目录被截到一万两千字以内，写作者只看到前二十来次
+#: - 下限 12000：窗口很小（32K 以下）的本地模型维持原来的篇幅；上限 160000：百万窗口的模型每写一稿、每重写一次
+#:   都要重发整份目录，花费随之上涨，而 160000 字已经放得下那次运行的查询全部展开
+CATALOG_WINDOW_SHARE = 0.4
+CATALOG_MIN_CHARS = 12_000
+CATALOG_MAX_CHARS = 160_000
+#: 目录放不下时，先给每次查询列几行
+_PROMPT_COMPACT_ROWS = 3
+#: 放不下时表和字段、知识库检索两节合起来最多占预算的几分之一（剩下的给查询）
+_PROMPT_TAIL_SHARE = 4
+_PROMPT_CUT = "\n…（目录太长，后面的省略了）"
 
-    查询结果会列出能引用的行（要读快照）；cells_allowed=False 时只列编号，说明只能当依据。
-    沙箱代码节点不列：它的产出不能引用。
+
+def catalog_budget(window: int | None) -> int:
+    """写作目录的字数预算：写作模型上下文窗口（token）的 40%，夹在上下限之间；拿不到窗口按 FALLBACK_CONTEXT。"""
+    size = window if isinstance(window, int) and not isinstance(window, bool) and window > 0 else FALLBACK_CONTEXT
+    return max(CATALOG_MIN_CHARS, min(CATALOG_MAX_CHARS, int(size * CATALOG_WINDOW_SHARE)))
+
+
+DEFAULT_CATALOG_BUDGET = catalog_budget(None)
+
+#: 目录有省略时，开头跟在省略情况后面的约定（给模型看）。和数字核对一致：没展开的查询只能当依据，数照样逐字核对
+OMITTED_RULES = (
+    "没展开的查询照样在证据台账里，只是这里没有列出它的数据：\n"
+    "- 不要据此断言「没有查到」「不存在」「缺失」「没有数据」，只能写「本报告未覆盖」，并说明依据不足；\n"
+    "- 要用没展开的查询，只能在句末写 [[see:Qn]] 作依据（Qn 换成它的编号）；不要写 [[v:]]、[[table:]] 去猜它的行和列，"
+    "也不要凭摘要写任何数字——数字照样逐字核对，没有出处的数会被打回重写。结构化结论里写好的引用标记可以照抄。"
+)
+#: 结构化结论为空的来源后面跟的那句（给模型看）
+EMPTY_SOURCE_PROMPT = ("结构化结论为空（字段都记为空值）。这不代表没有数据：不得据此断言没有查到、不存在或缺失，"
+                       "只能写本报告未覆盖")
+
+
+@dataclass
+class WriterCatalog:
+    """给写作者的出处清单（writer_catalog）：文字，以及省略了什么。
+
+    - trimmed：放不下、有省略；nodes 是各来源的查询数 [{node_id, title, total, full, compact, omitted}]，
+      full 列出前 20 行、compact 只列前几行、omitted 只列编号和摘要；omitted 是只列摘要的查询编号
+    - empty_sources：结构化结论为空的来源（Agent 的名字）
+    清单有省略、或有来源的结构化结论为空时，报告节点加写作规则（ABSENCE_RULE）；有省略时组装文档另查缺失断言
     """
-    snaps = _Snapshots(loader)
-    lines = ["可引用的证据（写数字只能用下面的引用标记，系统会换成真实数值）："]
+
+    text: str
+    budget: int
+    queries: int = 0
+    trimmed: bool = False
+    nodes: list[dict[str, Any]] = field(default_factory=list)
+    omitted: list[str] = field(default_factory=list)
+    empty_sources: list[str] = field(default_factory=list)
+
+    @property
+    def guarded(self) -> bool:
+        """写作者看不全证据：缺失断言没有依据。"""
+        return self.trimmed or bool(self.empty_sources)
+
+
+def _metric_prompt(catalog: dict[str, Any]) -> list[str]:
+    """口径卡指标和运行输入在写作目录里的样子。"""
+    lines: list[str] = []
     groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
     for entry in catalog.values():
         if entry.get("kind") == "metric":
@@ -3172,23 +3233,249 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe
     if inputs:
         lines.append("\n运行输入：")
         lines.extend(f"- [[{e['alias']}]] = {e.get('rendered')}" for e in inputs)
+    return lines
+
+
+def _dig_path(data: Any, path: str) -> Any:
+    for key in path.split("."):
+        data = data.get(key) if isinstance(data, dict) else None
+    return data
+
+
+def _field_prompt(path: str, value: Any, ev: dict[str, Any], aliases: dict[str, str], snaps: _Snapshots,
+                  cells_allowed: bool) -> str:
+    """一个结构化字段在写作目录里的那一行：值（按 [[v:]] 的渲染规则）和引用它的标记。"""
+    status = ev.get("status")
+    if status == "missing":
+        return f"- {path}：抽取时没有找到这个值（记为空，不代表库里没有）"
+    if status not in ("verified", "mismatch"):
+        return f"- {path}：出处核对不了（记为空）"
+    locator = ev.get("locator") if isinstance(ev.get("locator"), dict) else {}
+    alias = aliases.get(str(ev.get("artifact") or ""))
+    if alias is None:
+        return f"- {path}：所依据的查询不在这份目录里，不能引用"
+    if not cells_allowed:
+        return f"- {path}：出自 {alias}，只能在句末写 [[see:{alias}]] 作依据"
+    if isinstance(locator.get("rows"), list) and len(locator["rows"]) == 2:
+        first, last = locator["rows"]
+        named = locator.get("columns") if isinstance(locator.get("columns"), dict) else {"": locator.get("column")}
+        cols = ",".join(dict.fromkeys(str(c) for c in named.values() if c))
+        many = f"（共 {last - first + 1} 行，整表最多 {TABLE_MAX_ROWS} 行，要分成几张表）" \
+            if last - first + 1 > TABLE_MAX_ROWS else ""
+        return (f"- {path}：数组，出自 {alias} 第 {first}–{last} 行，可以单独一行写 "
+                f"[[table:{alias} cols={cols} rows={first}-{last}]]{many}")
+    row, column = locator.get("row"), locator.get("column")
+    shown = MISSING
+    snapshot, why = snaps.get(alias, ev.get("artifact"))
+    try:
+        if why:
+            raise CellError(why)
+        raw, column = locate_cell(snapshot, row, column)
+        shown = render_cell(raw, kind=column_kind(snapshot, column))
+    except (CellError, RenderError):
+        try:
+            shown = render_cell(value)
+        except RenderError:
+            pass
+    return f"- {path} = {shown}：[[v:{alias}.r{row}.{column}]]"
+
+
+def _fields_prompt(sources: list[dict[str, Any]], catalog: dict[str, Any], snaps: _Snapshots,
+                   cells_allowed: bool) -> tuple[list[str], list[str], list[str]]:
+    """各 Agent 经按出处核对交来的结构化字段：(写作目录里的几行, 结构化结论为空的来源, 字段引用到的查询编号)。
+
+    字段的出处记的是 Agent 自己的编号（它的第几次查询），这里换成报告目录里的全局编号：按快照工件认，同一个节点的
+    优先（工件内容寻址，两次一模一样的查询共用一份快照，指哪一次都是同一格）。
+    """
+    mine: dict[tuple[str, str], str] = {}
+    anyone: dict[str, str] = {}
+    for e in catalog.values():
+        if e.get("kind") == "query":
+            mine.setdefault((str(e.get("node_id") or ""), str(e.get("artifact") or "")), e["alias"])
+            anyone.setdefault(str(e.get("artifact") or ""), e["alias"])
+    lines: list[str] = []
+    empty: list[str] = []
+    cited: list[str] = []
+    for source in sources:
+        evidence = source.get("evidence")
+        if not isinstance(evidence, dict) or not evidence:
+            continue
+        node_id = str(source.get("node_id") or "")
+        title = str(source.get("title") or node_id)
+        data = source.get("data") if isinstance(source.get("data"), dict) else {}
+        if all(_dig_path(data, path) is None for path in evidence):
+            empty.append(title)
+            lines.append(f"「{title}」：{EMPTY_SOURCE_PROMPT}")
+            continue
+        aliases = {artifact: mine.get((node_id, artifact)) or alias for artifact, alias in anyone.items()}
+        lines.append(f"「{title}」：")
+        for path, ev in evidence.items():
+            ev = ev if isinstance(ev, dict) else {}
+            lines.append(_field_prompt(path, _dig_path(data, path), ev, aliases, snaps, cells_allowed))
+            if ev.get("status") in ("verified", "mismatch") and (alias := aliases.get(str(ev.get("artifact") or ""))):
+                cited.append(alias)
+    if not lines:
+        return [], [], []
+    return (["\n各 Agent 交来的结构化结论（系统已逐格核对到查询快照；要写这些数，用后面的引用标记）：", *lines],
+            empty, list(dict.fromkeys(cited)))
+
+
+def _query_brief(entry: dict[str, Any], tables: list[str]) -> str:
+    """没展开的查询在目录里的一行摘要：编号、查的表（没有表结构时写工具名）、行数。"""
+    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 "查询")
+    size = f"{rows} 行" if isinstance(rows, int) else "行数不明"
+    return f"  - {entry['alias']}：{what}，{size}" + ("（结果已截断）" if entry.get("truncated") else "")
+
+
+def _cut_lines(lines: list[str], limit: int, note: str) -> list[str]:
+    """从头取到放得下 limit 个字为止，后面换成一句 note。"""
+    out: list[str] = []
+    used = len(note) + 1
+    for line in lines:
+        if used + len(line) + 1 > limit:
+            return [*out, note]
+        out.append(line)
+        used += len(line) + 1
+    return out
+
+
+def writer_catalog(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUDGET, cells_allowed: bool = True,
+                   loader: Loader | None = None, fields: list[dict[str, Any]] | None = None,
+                   titles: dict[str, str] | None = None) -> WriterCatalog:
+    """证据目录的文字版，放进写作者的 prompt，连同省略了什么（WriterCatalog）。
+
+    查询结果会列出能引用的行（要读快照）；cells_allowed=False 时只列编号，说明只能当依据。
+    沙箱代码节点不列：它的产出不能引用。fields 是上游 Agent 经按出处核对交来的结构化字段
+    [{node_id, title, data, evidence}]，titles 是节点在画布上的名字（分组标题用）。
+
+    放得下就整份给（没有结构化字段时和以前一字不差）。放不下时按优先级、按来源均衡（_trimmed_catalog）：
+    口径卡指标、运行输入、结构化字段一律列出；查询按产出它的节点分组、轮流展开；没展开的列编号和摘要；
+    开头照实说目录有省略、各来源省略了几次。
+    """
+    snaps = _Snapshots(loader)
+    head = ["可引用的证据（写数字只能用下面的引用标记，系统会换成真实数值）："]
+    field_lines, empty, cited = _fields_prompt(fields or [], catalog, snaps, cells_allowed)
+    first = [*_metric_prompt(catalog), *field_lines]
     queries = [e for e in catalog.values() if e.get("kind") == "query"]
-    if queries:
-        lines.append(f"\n{CELL_RULES}" if cells_allowed else
-                     "\n查询结果（这次出具不允许直接引用单元格，只能写在 [[see:Q1]] 这样的依据里）：")
-        for e in queries:
-            lines.extend(_query_prompt(e, snaps, cells_allowed))
+    rules = ([f"\n{CELL_RULES}" if cells_allowed else
+              "\n查询结果（这次出具不允许直接引用单元格，只能写在 [[see:Q1]] 这样的依据里）："] if queries else [])
+    tail: list[str] = []
     if (index := snaps.entities(catalog)).active:
-        lines.extend(_entity_prompt(catalog, partial=index.partial, snaps=snaps))
+        tail.extend(_entity_prompt(catalog, partial=index.partial, snaps=snaps))
     retrievals = [e for e in catalog.values() if e.get("kind") == "retrieval"]
     if retrievals:
-        lines.append(f"\n{QUOTE_RULES}")
+        tail.append(f"\n{QUOTE_RULES}")
         for e in retrievals:
-            lines.extend(_retrieval_prompt(e, snaps))
-    text = "\n".join(lines)
+            tail.extend(_retrieval_prompt(e, snaps))
+    full = {e["alias"]: _query_prompt(e, snaps, cells_allowed) for e in queries}
+    text = "\n".join([*head, *first, *rules, *(line for e in queries for line in full[e["alias"]]), *tail])
+    if len(text) <= budget:
+        return WriterCatalog(text=text, budget=budget, queries=len(queries), empty_sources=empty)
+    return _trimmed_catalog(catalog, head, first, rules, queries, full, tail, budget=budget, snaps=snaps,
+                    cells_allowed=cells_allowed, titles=titles or {}, empty=empty, pinned=cited)
+
+
+def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], rules: list[str],
+                     queries: list[dict[str, Any]], full: dict[str, list[str]], tail: list[str], *, budget: int,
+                     snaps: _Snapshots, cells_allowed: bool, titles: dict[str, str], empty: list[str],
+                     pinned: list[str]) -> WriterCatalog:
+    """放不下时的目录。以前从头截断：查询按编号排，前面的节点占满预算，后面的节点一次都不露面。
+
+    1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）
+    2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE
+    3. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开：
+       第一轮每次列前 _PROMPT_COMPACT_ROWS 行，第二轮补到 _PROMPT_ROWS 行，放不下为止
+    4. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES）
+    最后整份量一次：超了就把最后展开的退回去；指标和字段本身就超（极少见）才从尾部截断。
+    """
+    if tail and len("\n".join(tail)) > budget // _PROMPT_TAIL_SHARE:
+        tail = _cut_lines(tail, budget // _PROMPT_TAIL_SHARE, "…（表和字段、知识库检索太长，后面的省略了）")
+    tables: dict[str, list[str]] = {}
+    for e in catalog.values():
+        if e.get("kind") == "table":
+            for alias in e.get("queries") or []:
+                tables.setdefault(alias, []).append(str(e.get("name")))
+    groups: dict[str, list[dict[str, Any]]] = {}
+    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}
+    shapes = {"brief": brief, "compact": compact, "full": full}
+    tier = {e["alias"]: "brief" for e in queries}
+
+    def rows() -> list[dict[str, Any]]:
+        out = []
+        for nid, entries in groups.items():
+            got = Counter(tier[e["alias"]] for e in entries)
+            out.append({"node_id": nid, "title": titles.get(nid) or nid or "其他", "total": len(entries),
+                        "full": got["full"], "compact": got["compact"], "omitted": got["brief"]})
+        return out
+
+    def assemble() -> str:
+        stats = rows()
+        per = "、".join(f"{r['title']} {r['total']} 次（展开 {r['full'] + r['compact']} 次，省略 {r['omitted']} 次）"
+                        for r in stats)
+        n = {k: sum(r[k] for r in stats) for k in ("full", "compact", "omitted")}
+        notice = (f"注意：证据太多，这份目录有省略（篇幅上限约 {budget} 字）。共 {len(queries)} 次查询："
+                  f"{n['full']} 次列出前 {_PROMPT_ROWS} 行，{n['compact']} 次只列前 {_PROMPT_COMPACT_ROWS} 行，"
+                  f"{n['omitted']} 次只列编号和摘要。各来源：{per}。")
+        body = [*head, notice, OMITTED_RULES, *first, *rules]
+        for r, entries in zip(stats, groups.values()):
+            body.append(f"\n来自「{r['title']}」的查询（共 {r['total']} 次，展开 {r['full'] + r['compact']} 次）：")
+            for e in entries:
+                if tier[e["alias"]] != "brief":
+                    body.extend(shapes[tier[e["alias"]]][e["alias"]])
+            if r["omitted"]:
+                body.append(f"  没展开（{r['omitted']} 次，只列编号和摘要）：")
+                body.extend(brief[e["alias"]][0] for e in entries if tier[e["alias"]] == "brief")
+        return "\n".join([*body, *tail])
+
+    def size(lines: list[str]) -> int:
+        return sum(len(line) + 1 for line in lines)
+
+    # 省略情况那句里的数字会变长几位：留一点余量，最后再整份量一次
+    remaining = budget - len(assemble()) - 8 * (len(groups) + 3)
+    upgraded: list[tuple[str, str]] = []
+    for alias in pinned:
+        if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining:
+            tier[alias] = "compact"
+            remaining -= delta
+            upgraded.append((alias, "brief"))
+    for target, source in (("compact", "brief"), ("full", "compact")):
+        moved = True
+        while moved:
+            moved = False
+            for entries in groups.values():
+                for e in entries:
+                    alias = e["alias"]
+                    if tier[alias] != source:
+                        continue
+                    delta = size(shapes[target][alias]) - size(shapes[source][alias])
+                    if delta <= remaining:
+                        tier[alias] = target
+                        remaining -= delta
+                        upgraded.append((alias, source))
+                        moved = True
+                        break
+    text = assemble()
+    while len(text) > budget and upgraded:
+        alias, source = upgraded.pop()
+        tier[alias] = source
+        text = assemble()
     if len(text) > budget:
-        text = text[:budget].rsplit("\n", 1)[0] + "\n…（目录太长，后面的省略了）"
-    return text
+        text = text[:budget - len(_PROMPT_CUT)].rsplit("\n", 1)[0] + _PROMPT_CUT
+    return WriterCatalog(text=text, budget=budget, queries=len(queries), trimmed=True, nodes=rows(),
+                         omitted=[e["alias"] for e in queries if tier[e["alias"]] == "brief"], empty_sources=empty)
+
+
+def catalog_prompt(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUDGET, cells_allowed: bool = True,
+                   loader: Loader | None = None, fields: list[dict[str, Any]] | None = None,
+                   titles: dict[str, str] | None = None) -> str:
+    """证据目录的文字版（writer_catalog 的文字）。超过预算时分优先级、按来源均衡，并照实说省略了什么。"""
+    return writer_catalog(catalog, budget=budget, cells_allowed=cells_allowed, loader=loader, fields=fields,
+                          titles=titles).text
 
 
 def unshaped_table(entry: dict[str, Any], fetch: Callable[[Any], Any]) -> bool:

backend/app/engine/nodes/report.py

@@ -50,11 +50,13 @@ from app.engine.evidence import (
     MARKER_RULES,
     SEE_AND_CODE_RULES,
     StreamRenderer,
+    WriterCatalog,
     build_catalog,
-    catalog_prompt,
+    catalog_budget,
     compose_doc,
     describe_violations,
     ledger_enabled,
+    writer_catalog,
 )
 from app.engine.nodes.llm import _invoke_streaming, _model_spec, _report_call
 from app.engine.schema import (
@@ -70,7 +72,8 @@ from app.engine.schema import (
 )
 from app.engine.state import GraphState, message_text, thinking_text
 from app.engine.toolcalls import leaked_markup, markup_warning
-from app.providers.factory import ProviderNotConfigured, get_chat_model
+from app.providers import catalog as model_catalog
+from app.providers.factory import ModelSpec, ProviderNotConfigured, get_chat_model, resolve_provider
 
 ON_VIOLATION = ("fail", "flag")
 NUMBERS = ("strict", "off")
@@ -140,6 +143,57 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[
     return catalog
 
 
+def report_fields(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[dict[str, Any]]:
+    """上游 Agent 经按出处核对交来的结构化字段：[{node_id, title, data, evidence}]，按画布上的顺序。
+
+    范围和 report_catalog 收查询的范围一致：祖先节点，写了 evidence_from 就只收那几个。只进写作目录（第一优先，
+    放不下也列出），不进目录本身：出口复核按目录重建，字段的引用落到的是查询单元格，和以前一样核对。
+    """
+    sources = _setting(spec, node, "evidence_from", "ancestors")
+    if isinstance(sources, str) and sources != "ancestors":
+        sources = [sources]
+    keep = {str(s) for s in sources} if isinstance(sources, list) else None
+    allowed = _ancestors(spec, node.id)
+    outputs = state.get("nodes") or {}
+    found: list[dict[str, Any]] = []
+    for other in spec.nodes:
+        result = outputs.get(other.id)
+        if other.id not in allowed or (keep is not None and other.id not in keep) or not isinstance(result, dict):
+            continue
+        if isinstance(result.get("data_evidence"), dict) and result["data_evidence"]:
+            found.append({"node_id": other.id, "title": other.title or other.id, "data": result.get("data"),
+                          "evidence": result["data_evidence"]})
+    return found
+
+
+async def model_window(session: Any, spec: ModelSpec, model_id: str) -> int | None:
+    """写作模型的上下文窗口（token）：模型接入里给这个模型登记的 context 优先，其次模型目录
+    （providers/catalog.py）；都没有返回 None，由 catalog_budget 取保守值。"""
+    try:
+        provider = await resolve_provider(session, spec.provider, spec.model or model_id)
+    except ProviderNotConfigured:
+        provider = None
+    for entry in (provider.models or []) if provider is not None else []:
+        if isinstance(entry, dict) and entry.get("id") == model_id:
+            raw = entry.get("context")
+            try:
+                value = int(raw) if not isinstance(raw, bool) and raw not in (None, "") else 0
+            except (TypeError, ValueError):
+                value = 0
+            if value > 0:
+                return value
+    return model_catalog.find_context(model_id)
+
+
+def _trimmed_message(writer: WriterCatalog) -> str:
+    """写作目录有省略时的那条警告（给人看）：各来源省略了几次，写作模型看不到哪些明细。"""
+    per = "、".join(f"{r['title']}省略 {r['omitted']} 次" for r in writer.nodes if r["omitted"])
+    shown = writer.queries - len(writer.omitted)
+    return (f"证据目录超出篇幅（约 {writer.budget} 字，按写作模型的上下文窗口确定）：{writer.queries} 次查询中 "
+            f"{shown} 次列出了数据，{len(writer.omitted)} 次只列编号和摘要（{per}）。写作模型看不到这些查询的明细，"
+            "报告不得据此断言没有数据")
+
+
 def report_entities(spec: GraphSpec, node: GraphNode) -> str:
     """报告节点的实体层：link（默认）/ off。出口复核调 verify_doc 时传 entities=(这个值 == "link")。
 
@@ -351,8 +405,7 @@ def _repair_request(violations: list[dict[str, Any]]) -> str:
     )
 
 
-def _messages(ctx: NodeContext, state: GraphState, catalog: dict[str, Any], cells: bool = True,
-              claims: str = "off") -> list[BaseMessage]:
+def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off") -> list[BaseMessage]:
     system = ctx.render_str(ctx.cfg("system", ""), state)
     instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \
         or "根据下面的证据写一份简洁的报告，先总后分。"
@@ -363,7 +416,7 @@ def _messages(ctx: NodeContext, state: GraphState, catalog: dict[str, Any], cell
     return [
         SystemMessage(content="\n\n".join(p for p in (system, ROLE) if p)),
         HumanMessage(content=(
-            f"{instructions}\n\n{catalog_prompt(catalog, cells_allowed=cells)}\n\n{rules}\n\n"
+            f"{instructions}\n\n{writer.text}\n\n{rules}\n\n"
             "只输出报告正文（Markdown），不要写撰写说明。"
         )),
     ]
@@ -392,6 +445,8 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
             model, model_id = await get_chat_model(session, _model_spec(ctx))
         except ProviderNotConfigured as e:
             raise NodeError(ctx.node.id, not_configured(e)) from e
+        # 写作目录的篇幅按这个模型的上下文窗口定（evidence.catalog_budget）
+        window = await model_window(session, _model_spec(ctx), model_id)
 
     spec = ctx.run.spec
     catalog = report_catalog(state, spec, ctx.node)
@@ -408,7 +463,14 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     if warning := _no_evidence(catalog, cells):
         ctx.emit(EventType.LOG, level="warn", code="report_no_evidence", message=warning)
 
-    messages = _messages(ctx, state, catalog, cells, claims)
+    writer = writer_catalog(catalog, budget=catalog_budget(window), cells_allowed=cells,
+                            fields=report_fields(state, spec, ctx.node),
+                            titles={n.id: n.title or n.id for n in spec.nodes})
+    if writer.trimmed:
+        ctx.emit(EventType.LOG, level="warn", code="report_catalog_trimmed", message=_trimmed_message(writer),
+                 budget=writer.budget, window=window, queries=writer.queries, omitted=len(writer.omitted),
+                 nodes=[{k: r[k] for k in ("node_id", "title", "total", "omitted")} for r in writer.nodes])
+    messages = _messages(ctx, state, writer, claims)
     started = time.perf_counter()
     spent: list[dict[str, Any]] = []
     nudged = False

backend/app/engine/review.py

@@ -89,6 +89,8 @@ WARN_CODES: dict[str, tuple[str, str]] = {
     "report_repair": ("report_repair", DEGRADED),
     # 报告撰写节点的上游没有可引用的口径卡指标：报告里的数都没有出处
     "report_no_evidence": ("report_no_evidence", DEGRADED),
+    # 证据目录超出篇幅，有的查询只列了编号和摘要：写作模型没看全证据，报告的覆盖面有缺口
+    "report_catalog_trimmed": ("report_catalog_trimmed", DEGRADED),
     # 口径卡的指标集、报告文档没能落进工件库：答案照常，只是点不开证据
     "evidence_store_failed": ("evidence_store_failed", DEGRADED),
     # agent 的结构化字段：模型报的值和查询快照对不上（已按快照取值），或者出处核对不了

backend/tests/fixtures/evidence/store_run.py

@@ -0,0 +1,106 @@
+"""合成的「4 个专项 Agent 并行查库 → 报告」场景：查询多、分布不均，每个 Agent 都交了按出处核对的字段。
+
+形状仿一次真实运行（4 个 Agent 并行、查询次数悬殊，合计上百次），次数和内容全部虚构：门店、会员、卡券、活动。
+不连库、不调模型：台账条目、查询快照、表结构快照、Agent 的 data / data_evidence 都在内存里拼好，快照经 loader 取。
+
+台账按节点跑完的先后排（state["evidence"] 是各节点结束时追加的），所以全局编号 Q1… 前 26 个都是会员专项的。
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from app.engine.evidence import cell_eid, make_eid
+
+#: (节点 id, 画布上的名字, 查询次数, 查的表)
+AGENTS: list[tuple[str, str, int, str]] = [
+    ("agent_member", "会员专项", 26, "member_daily"),
+    ("agent_store", "门店专项", 17, "store_daily"),
+    ("agent_coupon", "卡券核销", 79, "coupon_redemption"),
+    ("agent_campaign", "活动内容", 42, "campaign_view"),
+]
+COLUMNS = ["store_name", "biz_time", "cnt", "amount"]
+TYPES = {"store_name": "text", "biz_time": "datetime", "cnt": "number", "amount": "number"}
+ROWS = 30
+#: 结构化抽取失败（两次都被截断）的那个 Agent：字段全是空值
+EMPTY_AGENT = "agent_coupon"
+FIELDS = ["total_cnt", "top_store", "latest_time", "top_stores", "churn_rate"]
+
+
+def artifact_of(node_id: str, i: int) -> str:
+    return f"{node_id}-q{i:03d}".ljust(64, "0")
+
+
+def snapshot(node_id: str, i: int, rows: int = ROWS) -> dict[str, Any]:
+    """第 i 次查询的快照：每行一个门店，时间是 ISO 写法（快照里存的就是这样）。"""
+    seed = sum(map(ord, node_id)) + i * 7
+    return {
+        "columns": list(COLUMNS),
+        "column_types": dict(TYPES),
+        "rows": [[f"门店{(seed + r) % 90 + 1:02d}", f"2026-06-{r % 28 + 1:02d}T08:{(seed + r) % 60:02d}:39",
+                  (seed * 13 + r * 17) % 900 + 10, f"{(seed * 31 + r * 7) % 9000 + 100}.50"] for r in range(rows)],
+        "truncated": False,
+    }
+
+
+def schema_snapshot(table: str) -> dict[str, Any]:
+    return {"source": "shop", "tables": {table: {"columns": [{"name": c, "type": TYPES[c]} for c in COLUMNS]}}}
+
+
+def build() -> tuple[list[dict[str, Any]], dict[str, Any], dict[str, Any], dict[str, str]]:
+    """(台账, 快照 {工件 id: 内容}, state["nodes"], {节点 id: 名字})。"""
+    ledger: list[dict[str, Any]] = []
+    store: dict[str, Any] = {}
+    nodes: dict[str, Any] = {}
+    titles: dict[str, str] = {}
+    for node_id, title, count, table in AGENTS:
+        titles[node_id] = title
+        schema_id = f"schema-{table}".ljust(64, "0")
+        store[schema_id] = schema_snapshot(table)
+        ledger.append({"kind": "schema", "node_id": node_id, "exec": 1, "artifact": schema_id, "source": "shop",
+                       "tables": {table: list(COLUMNS)}})
+        for i in range(1, count + 1):
+            aid = artifact_of(node_id, i)
+            store[aid] = snapshot(node_id, i)
+            ledger.append({"kind": "query", "node_id": node_id, "exec": 1, "artifact": aid,
+                           "via": f"tool-{aid}"[:64], "call_id": f"call_{i}", "tool": "db_query__shop", "source": "shop",
+                           "columns": list(COLUMNS), "rows": ROWS, "truncated": False,
+                           "schema_artifact": schema_id, "tables": [table]})
+        nodes[node_id] = agent_output(node_id, store)
+    return ledger, store, nodes, titles
+
+
+def agent_output(node_id: str, store: dict[str, Any]) -> dict[str, Any]:
+    """Agent 交出的结构化字段：本节点的 Q3 那一格、Q5 的前 5 行；EMPTY_AGENT 抽取失败，字段全空。"""
+    if node_id == EMPTY_AGENT:
+        reason = "结构化抽取没有完成：模型的输出额度在推理阶段就用完了（最后一次的额度为 32768 token），结构化结论没能写出来"
+        return {"text": "查到了大量核销和兑换记录。", "data": {f: None for f in FIELDS},
+                "data_evidence": {f: {"status": "unresolved", "reason": reason} for f in FIELDS}}
+    q3, q5 = artifact_of(node_id, 3), artifact_of(node_id, 5)
+    first = store[q3]["rows"][0]
+
+    def cell(column: str, index: int) -> dict[str, Any]:
+        return {"ref": f"Q3.r0.{column}", "call": "Q3", "artifact": q3, "via": f"tool-{q3}"[:64],
+                "locator": {"row": 0, "column": column}, "eid": cell_eid(q3, 0, column), "status": "verified"}
+
+    rows_locator = {"rows": [0, 4], "column": "store_name"}
+    return {
+        "text": "结论见字段。",
+        "data": {"total_cnt": first[2], "top_store": first[0], "latest_time": first[1],
+                 "top_stores": [r[0] for r in store[q5]["rows"][:5]], "churn_rate": None},
+        "data_evidence": {
+            "total_cnt": cell("cnt", 2), "top_store": cell("store_name", 0), "latest_time": cell("biz_time", 1),
+            "top_stores": {"ref": "Q5.r0-4.store_name", "call": "Q5", "artifact": q5, "via": f"tool-{q5}"[:64],
+                           "locator": rows_locator, "eid": make_eid("rows", q5, rows_locator), "status": "verified"},
+            "churn_rate": {"ref": None, "status": "missing"},
+        },
+    }
+
+
+def global_alias(node_id: str, local: int) -> str:
+    """本节点第 local 次查询在报告目录里的全局编号（台账按节点先后排）。"""
+    offset = 0
+    for nid, _, count, _ in AGENTS:
+        if nid == node_id:
+            return f"Q{offset + local}"
+        offset += count
+    raise KeyError(node_id)

backend/tests/test_report_catalog_budget.py

@@ -0,0 +1,209 @@
+"""写作者的出处清单：预算按写作模型的上下文窗口定；超出预算时分优先级、按来源均衡，省略了什么照实说。
+
+2026-10 一次运行：证据台账里上百次查询（4 个 Agent 查询次数悬殊），出处清单写死 12000 字，十万字以上的目录
+被截到一万两千字以内，只有前二十来次查询进了写作者的视野，后面只有一句「目录太长，后面的省略了」。查询按编号排，前面的
+Agent 占满了预算，有的 Agent 一次都没露面；写作者于是写出「本轮没有查到任何核销结果」，可那个 Agent 查到了上百条。
+
+- 预算：写作模型上下文窗口的 40%（字数按 1 字 ≈ 1 token 保守算），拿不到窗口按 128K；下限 12000、上限 160000
+- 第一优先：各 Agent 经按出处核对交来的结构化字段、口径卡指标，一律列出
+- 查询按产出它的节点分组、轮流取：先给每次查询前几行，再补到 20 行；每个节点都有代表
+- 没展开的查询列出编号和一行摘要（表、行数）；约定只能写 [[see:Qn]] 作依据，数字照样逐字核对
+- 清单开头明说「目录有省略」，各节点各省略了多少条
+
+夹具全部合成（tests/fixtures/evidence/store_run.py）。
+"""
+from __future__ import annotations
+
+import re
+
+import pytest
+
+import app.db.models  # noqa: F401 - 建表（conftest 的 create_all）前要先登记模型
+from app.engine.evidence import (
+    CATALOG_MAX_CHARS,
+    CATALOG_MIN_CHARS,
+    DEFAULT_CATALOG_BUDGET,
+    build_catalog,
+    catalog_budget,
+    catalog_prompt,
+    writer_catalog,
+)
+from tests.fixtures.evidence import store_run
+
+LEDGER, STORE, NODES, TITLES = store_run.build()
+
+
+def load(artifact):
+    return STORE.get(artifact)
+
+
+@pytest.fixture
+def catalog():
+    return build_catalog(nodes=NODES, ledger=LEDGER)
+
+
+def sources():
+    return [{"node_id": nid, "title": TITLES[nid], "data": NODES[nid]["data"],
+             "evidence": NODES[nid]["data_evidence"]} for nid, *_ in store_run.AGENTS]
+
+
+def written(catalog, budget=DEFAULT_CATALOG_BUDGET, **kw):
+    return writer_catalog(catalog, budget=budget, loader=load, fields=sources(), titles=TITLES, **kw)
+
+
+def group_of(text: str, title: str) -> str:
+    """清单里「来自「<名字>」的查询」那一段（到下一个来源或下一节为止）。"""
+    start = text.index(f"来自「{title}」的查询")
+    rest = text[start + 1:]
+    ends = [m.start() for m in re.finditer(r"\n(?:来自「|\n)", rest)]
+    return text[start:start + 1 + (ends[0] if ends else len(rest))]
+
+
+# --------------------------------------------------------------------------
+# 预算
+# --------------------------------------------------------------------------
+
+
+def test_budget_follows_the_model_window():
+    assert DEFAULT_CATALOG_BUDGET == catalog_budget(None) == 51_200, "拿不到窗口按 128K 的 40%"
+    assert catalog_budget(128_000) == 51_200
+    assert catalog_budget(200_000) == 80_000
+    assert catalog_budget(32_000) == 12_800
+    assert catalog_budget(8_000) == CATALOG_MIN_CHARS == 12_000, "很小的窗口维持原来的篇幅"
+    assert catalog_budget(1_000_000) == CATALOG_MAX_CHARS == 160_000, "百万窗口也有上限：每写一稿都要重发整份目录"
+    assert catalog_budget(0) == catalog_budget(-5) == DEFAULT_CATALOG_BUDGET
+
+
+def test_the_default_is_far_above_the_old_cut():
+    assert DEFAULT_CATALOG_BUDGET >= 4 * 12_000
+
+
+# --------------------------------------------------------------------------
+# 放得下：照旧
+# --------------------------------------------------------------------------
+
+
+def test_a_catalog_that_fits_is_not_trimmed():
+    small = build_catalog(nodes={}, ledger=[e for e in LEDGER if e.get("node_id") == "agent_store"])
+    got = writer_catalog(small, loader=load)
+    assert not got.trimmed and not got.omitted
+    assert "目录有省略" not in got.text
+    assert got.text == catalog_prompt(small, loader=load), "没有结构化字段时和 catalog_prompt 一字不差"
+    assert "Q17：" in got.text and "  r19：" in got.text
+
+
+# --------------------------------------------------------------------------
+# 放不下：分优先级、按来源均衡（那次运行同样形状的场景）
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("budget", [DEFAULT_CATALOG_BUDGET, 12_000])
+def test_every_agent_stays_in_view_within_the_budget(catalog, budget):
+    got = written(catalog, budget)
+    text = got.text
+    assert got.trimmed and len(text) <= budget, len(text)
+    for nid, title, count, table in store_run.AGENTS:
+        group = group_of(text, title)
+        assert f"共 {count} 次" in group
+        assert re.search(r"\n {2}r0：", group), f"「{title}」一次查询都没展开"
+    # 每次查询都在清单里：展开的有行，没展开的有编号和摘要
+    total = sum(c for *_, c, _ in store_run.AGENTS)
+    for n in range(1, total + 1):
+        assert re.search(rf"(?:^|\n)\s*- Q{n}：", text), f"Q{n} 不见了"
+
+
+def test_omitted_queries_are_listed_with_a_summary(catalog):
+    got = written(catalog)
+    assert got.omitted, "这个场景放不下全部展开"
+    for alias in got.omitted:
+        node = catalog[alias]["node_id"]
+        table = next(t for nid, _, _, t in store_run.AGENTS if nid == node)
+        assert re.search(rf"\n {{2}}- {alias}：{table}，30 行", got.text), alias
+    # 没展开的查询没有行
+    first = got.omitted[0]
+    block = got.text[got.text.index(f"- {first}："):].split("\n", 2)
+    assert not block[1].lstrip().startswith("r0：")
+
+
+def test_the_head_says_what_was_left_out_per_source(catalog):
+    got = written(catalog)
+    head = got.text[:800]
+    assert "目录有省略" in head
+    for row in got.nodes:
+        assert f"{row['title']} {row['total']} 次" in head and f"省略 {row['omitted']} 次" in head, row
+    assert sum(r["omitted"] for r in got.nodes) == len(got.omitted)
+    assert [r["node_id"] for r in got.nodes] == [nid for nid, *_ in store_run.AGENTS]
+
+
+def test_how_to_use_an_omitted_query_is_spelled_out(catalog):
+    text = written(catalog).text
+    assert "[[see:Qn]]" in text and "本报告未覆盖" in text
+    assert "不要凭摘要写" in text and "数字照样逐字核对" in text
+    for phrase in ("没有查到", "不存在", "缺失", "没有数据"):
+        assert phrase in text
+
+
+def test_structured_fields_come_first_and_point_at_global_cells(catalog):
+    got = written(catalog, 12_000)
+    text = got.text
+    fields_at = text.index("各 Agent 交来的结构化结论")
+    assert fields_at < text.index("来自「"), "结构化结论排在查询前面"
+    member_q3 = store_run.global_alias("agent_member", 3)
+    store_q3 = store_run.global_alias("agent_store", 3)
+    campaign_q5 = store_run.global_alias("agent_campaign", 5)
+    assert f"[[v:{member_q3}.r0.cnt]]" in text and f"[[v:{store_q3}.r0.store_name]]" in text
+    assert f"[[table:{campaign_q5} cols=store_name rows=0-4]]" in text
+    assert "churn_rate：抽取时没有找到这个值" in text
+    # 抽取失败的那个 Agent：结构化结论为空，不能据此断言没有数据
+    block = text[text.index("「卡券核销」：", fields_at):]
+    assert block.startswith("「卡券核销」：结构化结论为空") and "本报告未覆盖" in block.split("\n", 1)[0]
+    assert got.empty_sources == ["卡券核销"]
+
+
+def test_queries_the_fields_cite_are_expanded_first(catalog):
+    """结构化字段引用到的那几次查询先展开：写作者照抄字段的引用标记时，看得到那几行。"""
+    got = written(catalog, 12_000)
+    for nid in ("agent_member", "agent_store", "agent_campaign"):
+        for local in (3, 5):
+            alias = store_run.global_alias(nid, local)
+            assert alias not in got.omitted, alias
+            assert re.search(rf"\n- {alias}：.*\n {{2}}r0：", got.text), alias
+
+
+def test_a_field_value_reads_like_the_cell_it_cites(catalog):
+    """字段后面写的值和引用之后报告里显示的一样（[[v:]] 的渲染规则）。"""
+    text = written(catalog).text
+    q3 = STORE[store_run.artifact_of("agent_member", 3)]
+    assert f"total_cnt = {q3['rows'][0][2]:,}：[[v:Q3.r0.cnt]]" in text
+
+
+# --------------------------------------------------------------------------
+# 报告节点：窗口从哪来
+# --------------------------------------------------------------------------
+
+
+async def test_the_window_comes_from_the_provider_entry_then_the_catalog():
+    """模型接入里给这个模型登记的 context 优先，其次模型目录；都没有返回 None（预算按保守值）。"""
+    from sqlalchemy import delete
+
+    from app.db.base import SessionLocal
+    from app.db.models import Provider
+    from app.engine.nodes.report import model_window
+    from app.providers.factory import ModelSpec
+
+    async with SessionLocal() as session:
+        row = Provider(name="fa-window-test", kind="mock", enabled=True, default_model="mock-wide",
+                       models=[{"id": "mock-wide", "context": 64_000}, {"id": "mock-text", "context": "96000"},
+                               {"id": "mock-bare"}])
+        session.add(row)
+        await session.commit()
+        try:
+            assert await model_window(session, ModelSpec(model="mock-wide"), "mock-wide") == 64_000
+            assert await model_window(session, ModelSpec(model="mock-text"), "mock-text") == 96_000
+            # 接入里没登记窗口：按模型目录（deepseek-chat 登记了 128K）；目录里也没有就是 None
+            assert await model_window(session, ModelSpec(model="deepseek-chat"), "deepseek-chat") == 128_000
+            assert await model_window(session, ModelSpec(model="mock-bare"), "mock-bare") is None
+            assert catalog_budget(await model_window(session, ModelSpec(model="mock-wide"), "mock-wide")) == 25_600
+        finally:
+            await session.execute(delete(Provider).where(Provider.id == row.id))
+            await session.commit()

frontend/src/run/decode.ts

@@ -756,6 +756,22 @@ function explainLog(code: string | undefined, message: string, d: Record<string,
         sub: message.match(/(?:没|未)通过核对（(.+?)）/)?.[1],
       }
     }
+    case 'report_catalog_trimmed': {
+      // 写作目录超出篇幅（report.py _trimmed_message）。读结构化字段（queries、omitted、nodes），不从原话里认
+      const total = typeof d.queries === 'number' ? d.queries : undefined
+      const omitted = typeof d.omitted === 'number' ? d.omitted : undefined
+      const rows: any[] = Array.isArray(d.nodes) ? d.nodes : []
+      const per = rows.filter((r) => typeof r?.omitted === 'number' && r.omitted > 0 && str(r?.title))
+        .map((r) => `${r.title} 省略 ${formatNumber(r.omitted)}/${formatNumber(r.total)} 次`)
+      return {
+        title: omitted != null
+          ? `证据目录超出篇幅，${total != null ? `${formatNumber(total)} 次查询中 ` : ''}${formatNumber(omitted)} 次只列了编号和摘要`
+          : '证据目录超出篇幅，部分查询只列了编号和摘要',
+        sub: per.length ? per.join('、') : undefined,
+        next: '写作模型按来源均衡地看到了一部分查询的明细，被省略的查询仍在证据台账里，报告不得据此断言没有数据。'
+          + '需要完整明细时，请换用上下文窗口更大的写作模型，或让 Agent 在 SQL 中先汇总再交出结果',
+      }
+    }
     case 'agent_field_mismatch': {
       // 「有 1 个字段和查询快照对不上，已按快照取值：order_cnt 模型报 1240，快照是 1234」（llm.py _field_warnings）
       const n = message.match(/有\s*(\d+)\s*个字段/)?.[1]

scripts/check-copy.mjs

@@ -230,12 +230,13 @@ const MODEL_ONLY = {
     'NODE_REFERENCE', 'ROLE', 'MARKUP_NUDGE', 'JUDGE_RULES', 'JUDGE_ROLE', 'JUDGE_RULE', 'REVIEW_SYSTEM', 'MARKER_RULES',
     'GATE_SYSTEM', 'TOOL_MARKUP_NUDGE', '_UNKNOWN_ENTITY_FOR_MODEL', 'NODE_FIELD_FIX', '_FIX_TAIL',
     '_STREAM_PROTOCOL', '_RULES_HEAD', '_NO_LOWERING', '_ASK_HUMAN', '_UPGRADE_TASK', 'CELL_RULES', 'ENTITY_RULES',
-    'EXTRACT_SYSTEM', 'GRAPH_SCHEMA', 'REVIEW_SCHEMA', '_CN_N',
+    'EXTRACT_SYSTEM', 'GRAPH_SCHEMA', 'REVIEW_SCHEMA', '_CN_N', 'OMITTED_RULES', 'ABSENCE_RULE',
   ]),
   // 拼提示词的函数：生成 / 修正 / 升级请求、抽取消息、交给抽取模型的输出结构；
   // 助手的数据源上下文（api/copilot_context.py 的各段渲染、挑表请求，copilot.py 里挑表用的需求）
   funcs: new Set(['_user_message', '_repair_request', '_publish_fix_request', '_upgrade_assist_request',
-    '_extract_messages', '_from_rows', 'catalog_prompt',
+    '_extract_messages', '_from_rows', 'catalog_prompt', 'writer_catalog', '_trimmed_catalog', '_metric_prompt',
+    '_fields_prompt', '_field_prompt', '_query_brief',
     '_pick_messages', '_pick_need', '_publish_fix_need', '_upgrade_need',
     '_detail_block', '_selected_block', '_join_lines', '_route_line', '_source_head']),
   // 第二个参数是交回模型改写的指令（第一个是给人看的那句）：engine/evidence.py 的 _Reason(text, fix)、fail(text, fix)

scripts/check-decode.mjs

@@ -1300,6 +1300,26 @@ await section('报告核对 report.checked、口径卡的台账（可点击证
     `${flagged?.status} ${flagged?.level}`)
   const bare = flatten(mod.decodeRun([E(1, 'report.checked', null, {})])).find((s) => s.code === 'report_checked')
   check('载荷缺字段也不崩，只说核对了报告', bare?.title === '核对报告' && !bare.artifact, bare?.title)
+
+  // 写作目录超出篇幅（2026-10 一次运行：上百次查询只有前二十来次进了写作者的视野，界面上一个字都没说）：
+  // 现在按来源均衡地展开，没展开的列编号和摘要，并发一条警告。界面读结构化字段（queries、omitted、nodes）
+  check('后端发目录有省略的警告，带着结构化字段', reportPy.includes('code="report_catalog_trimmed"')
+    && reportPy.includes('queries=writer.queries, omitted=len(writer.omitted)') && reportPy.includes('"node_id", "title", "total", "omitted"'),
+    '改了 report.py 的字段就同步改 decode.ts 的 explainLog 和这里的夹具')
+  const TRIMMED = '证据目录超出篇幅（约 51200 字，按写作模型的上下文窗口确定）：164 次查询中 90 次列出了数据，74 次只列编号和摘要'
+    + '（卡券核销省略 52 次、活动内容省略 22 次）。写作模型看不到这些查询的明细，报告不得据此断言没有数据'
+  const trimmed = flatten(mod.decodeRun([E(1, 'node.started', 'w', { node_type: 'report', label: '写报告' }),
+    E(2, 'log', 'w', { level: 'warn', code: 'report_catalog_trimmed', message: TRIMMED, budget: 51200, window: null, queries: 164,
+      omitted: 74, nodes: [{ node_id: 'a', title: '会员专项', total: 26, omitted: 0 }, { node_id: 'b', title: '卡券核销', total: 79, omitted: 52 },
+        { node_id: 'c', title: '活动内容', total: 42, omitted: 22 }] })])).find((s) => s.code === 'report_catalog_trimmed')
+  check('report_catalog_trimmed：标题说几次查询只列了编号和摘要', trimmed?.title === '证据目录超出篇幅，164 次查询中 74 次只列了编号和摘要', trimmed?.title)
+  check('……副标题按来源列省略了几次（没省略的来源不列）', trimmed?.sub === '卡券核销 省略 52/79 次、活动内容 省略 22/42 次', trimmed?.sub)
+  check('……下一步说被省略的仍在台账里、不得据此断言没有数据，以及怎么看全', !!trimmed?.next?.includes('不得据此断言没有数据')
+    && !!trimmed?.next?.includes('上下文窗口更大') && trimmed?.level === 'warn' && trimmed?.detail === TRIMMED, trimmed?.next)
+  const trimmedBare = flatten(mod.decodeRun([E(1, 'log', 'w', { level: 'warn', code: 'report_catalog_trimmed', message: '目录有省略' })]))
+    .find((s) => s.code === 'report_catalog_trimmed')
+  check('……缺字段时标题照样说人话，不出 undefined', trimmedBare?.title === '证据目录超出篇幅，部分查询只列了编号和摘要'
+    && trimmedBare?.sub === undefined, `${trimmedBare?.title} / ${trimmedBare?.sub}`)
 })
 
 await section('工具门控 tool.gated（工具信任三档）', async () => {