AgentLab/backend/app/engine/nodes/report.py

"""报告撰写节点：模型只写引用标记，数字由系统从口径卡取出来渲染。

以前「给人看的报告」是 llm 节点写的自由文本：数字是模型转写的，出处要到出口再按
数值回头猜——同值的指标一多就猜错，模型顺手写的数也要到出口才被发现。这里把核对
挪到写的那一刻：

1. 建目录：收齐上游口径卡的指标和运行输入，编成 [[m:gmv]]、[[i:week]] 这样的引用。值取自遮罩格的指标（cell() 取的
   遮罩格、取自改了名字的遮罩列、转抄上游口径卡里这类指标的；只按确定取自遮罩列的认，和单元格引用的自查同一个口径）
   在目录里写「已遮罩」：写作目录、存进文档的证据目录都不写原值，引用了的照样交回重写（报告自查记 masked_cell）
2. 流式写作：模型写标记，StreamRenderer 边收边换成真值，流里看到的就是最终的数字
3. 核对：compose_doc 切块、分句、切片段，找出裸数字和解析不了的引用。模型把整篇包在一个 ```markdown 代码块
   里的，先去掉外层围栏（_unfenced）：不然整篇是一个代码块，一句结论都切不出来
4. 修复：有违规就把清单交回去重写，最多 max_repairs 次
5. 收尾：文档落成 report_doc 工件，report.checked 事件进封存；仍然违规时按
   on_violation 失败（fail）或者照常产出、把违规片段标出来（flag）

三期起目录里还有表和字段（这次运行冻结的表结构）、知识库检索的原话。可疑实体（反引号里
写了个哪里都找不到的名字）只标注：不重写、不失败，受管级别的正式出具按缺口降档；表结构快照
不全时核对不了的名字（unverified_entity）只标注。entities: off 整层不管表名、字段名。
结论句策略 claims：off（默认）不管；require_citation 在写作提示里要求结论句挂依据（自动链接的
名字不算依据），没挂的计入缺口、出口降档；judge 另外请一个模型按证据逐句判断（engine/judge.py）：

- 正式运行在节点里判：写完、核对完之后判，判定写进文档的 unit.verdict，随文档一起封存。判定只标注、
  不改写正文；裁判没跑成、触顶了照样产出，没判的记未裁判并留下缺口，出口按 on_unsupported 判档
- 探索运行不在节点里花这笔钱：候选句标「未裁判 · 按需」，点开哪句再判哪句（证据接口）
- rewrite_once（默认关）：证据相矛盾、证据不足的句子连同理由交回写作者只改这几句，再判一次，最多一轮；
  改写稿冒出原稿没有的违规（只标注的可疑实体也算）就不采用
- 裁判调用包在 @task 里，排在节点所有已有调用之后：「接着跑」时直接取 checkpoint 里的判定，
  不重复调用、不重复计费。写作调用不进 checkpoint，接着跑时会重写一稿——所以判定带着输入指纹，
  对不上这一稿的不用

出口契约复核（io.py）用 report_catalog 按同一套参数重建目录、独立重算，不信这里的自查。
"""
from __future__ import annotations

import asyncio
import re
import time
from collections import Counter
from typing import Any

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langgraph.func import task

from app.core.errors import describe_exception, not_configured
from app.core.events import EventType
from app.db.base import SessionLocal
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,
    MARKER_RULES,
    SEE_AND_CODE_RULES,
    StreamRenderer,
    WriterCatalog,
    build_catalog,
    catalog_budget,
    compose_doc,
    describe_violations,
    ledger_enabled,
    withhold_metrics,
    writer_catalog,
)
from app.engine.nodes.llm import _invoke_streaming, _model_spec, _report_call
from app.engine.schema import (
    CLAIMS_JUDGE,
    CLAIMS_VALUES,
    JUDGE_ON_UNSUPPORTED,
    GraphNode,
    GraphSpec,
    NodeType,
    _ancestors,
    claims_problem,
    judge_problems,
)
from app.engine.state import GraphState, message_text, thinking_text
from app.engine.toolcalls import leaked_markup, markup_warning
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")
#: 实体层：link（默认）核对表名、字段名并自动链接；off 整层不管（目录里不收表和字段）
ENTITIES = ("link", "off")
#: 结论句策略，和画布校验同一份（schema.CLAIMS_VALUES）：off / require_citation / judge
CLAIMS = CLAIMS_VALUES
#: 只标注、不让写作者重写也不判失败的违规：可疑实体（受管的正式出具按缺口降档）、表结构快照不全时
#: 核对不了的名字（出口只标注）
SOFT = frozenset({"unknown_entity", "unverified_entity"})
#: 交回写作者重写一次、重写不好也不判失败的违规：句中的 [[see:]]（渲染后留下断句，数字和出处都没问题）；
#: 写作目录有省略时的缺失断言（「没有查到」「数据缺失」：写作者没看全证据，结论依据不足，但数字和出处都核对过）。
#: 只剩这几类时不按 max_repairs 反复重写：重写一次没改掉，再写也多半一样，留在核对结果里标注
REWRITE_ONLY = frozenset({"inline_see", "absence_claim"})
#: 重写一次就是多一次整篇的模型调用，写不对的模型多给几次也大多写不对
MAX_REPAIRS = 3

ROLE = ("你是报告撰写人。报告里的每个数字都由系统从证据里取出来、按口径卡的格式渲染；"
        "你负责挑重点、组织结构、说清变化和原因。")

MARKUP_NUDGE = ("你上一条回复把工具调用写成了文字。报告撰写不调用任何工具：能用的数据都在上面的"
                "证据目录里，直接用引用标记写报告正文。")
#: 节点失败原因（给人看）。给模型的那句是上面的 MARKUP_NUDGE，「证据目录」是模型那边的说法
MARKUP_ERROR = ("模型输出了工具调用的原始标记，而报告撰写节点不调用工具，数据已由系统提供。"
                "请换一个模型重试")
#: 接着跑时 checkpoint 里缓存的判定最多翻几条去找这一稿的（每条都是先前某一稿的，对不上就往后翻）
_STALE_LIMIT = 16


# --------------------------------------------------------------------------
# 目录与放行名单：出口复核要用同一套参数重建，所以不依赖 NodeContext
# --------------------------------------------------------------------------


def _setting(spec: GraphSpec, node: GraphNode, key: str, default: Any = None) -> Any:
    """和 NodeContext.cfg 同一个规则：节点上没写就取图级 defaults。"""
    value = node.config.get(key)
    if value in (None, ""):
        return spec.defaults.get(key, default)
    return value


def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[str, Any]:
    """报告节点的证据目录。

    出口契约复核必须用这一个函数重建：参数（metrics_from、evidence_from、祖先范围）
    差一点，alias 集合就不一样——同一个指标 id 在两张卡里时，只收一张写 m:gmv，
    全收就得写 m:east.gmv——复核会把一份好好的报告判成引用解析不了。

    值取自遮罩格、写成「已遮罩」的指标（evidence.withhold_metrics）按这个报告节点产出里记下的认（masked_metrics）：报告
    节点写作之前按证据面板的判断认好（api/evidence.py 的 withheld_metrics，要查库、读快照，这里做不了），随产出记下；
    出口契约复核拿同一份重建，正文里的「已遮罩」才对得上，不算文档被改过。报告节点写作时交进来的状态里没有它自己的产出
    （run_report），拿到的是原样的目录。
    """
    ledger = [e for e in state.get("evidence") or [] if isinstance(e, dict)]
    sources = _setting(spec, node, "evidence_from", "ancestors")
    if isinstance(sources, str) and sources != "ancestors":
        sources = [sources]
    if isinstance(sources, list):
        # evidence_from 只管查询、检索这类证据（连同查询当时的表结构）从哪几个节点来；指标从哪来由 metrics_from 管
        keep = {str(s) for s in sources}
        ledger = [e for e in ledger
                  if e.get("kind") not in ("query", "retrieval", "schema") or e.get("node_id") in keep]
    catalog = build_catalog(
        nodes=state.get("nodes") or {},
        ledger=ledger,
        inputs=state.get("input") or {},
        metrics_from=_setting(spec, node, "metrics_from") or None,
        allowed=_ancestors(spec, node.id),
    )
    if report_entities(spec, node) == "off":
        # 目录里没有表和字段，写作目录也就不列；compose / verify 另传 entities=False，解析不了的原因照实说
        catalog = {alias: e for alias, e in catalog.items() if e.get("kind") not in ENTITY_KINDS}
    recorded = (state.get("nodes") or {}).get(node.id)
    withheld = recorded.get("masked_metrics") if isinstance(recorded, dict) else None
    if isinstance(withheld, list) and withheld:
        catalog = withhold_metrics(catalog, [str(a) for a in withheld], _masked_word())
    return catalog


def _masked_word() -> str:
    """「已遮罩」：证据面板、推演同一个说法。api.evidence 导入了引擎里的模块，用到时再导入。"""
    from app.api.evidence import MASKED

    return MASKED


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:
    """写作目录有省略时的那条警告（给人看）：省略的是什么、各来源省略了几次，写作模型看不到哪些明细。

    放不下不一定省略了查询：查询全都列出、只是有的只列前几行，或者只省略了表和字段的说明。以前一律按「省略查询」写，
    全部列出时成了「0 次只列编号和摘要（）。写作模型看不到这些查询的明细」，说的不是实情。"""
    head = f"证据目录超出篇幅（约 {writer.budget} 字，按写作模型的上下文窗口确定）："
    if writer.omitted:
        per = "、".join(f"{r['title']}省略 {r['omitted']} 次" for r in writer.nodes if r["omitted"])
        shown = writer.queries - len(writer.omitted)
        return (f"{head}{writer.queries} 次查询中 {shown} 次列出了数据，{len(writer.omitted)} 次只列编号和摘要（{per}）。"
                "写作模型看不到这些查询的明细，报告不得据此断言没有数据")
    if writer.compact:
        per = "、".join(f"{r['title']} {r['compact']} 次" for r in writer.nodes if r.get("compact"))
        return (f"{head}{writer.queries} 次查询都列出了数据，其中 {writer.compact} 次只列前 {writer.compact_rows} 行"
                f"（{per}）。写作模型看不到这些查询其余的行，报告不得据此断言没有数据")
    return f"{head}{writer.queries} 次查询都列出了数据，表和字段、知识库检索的说明有省略"


def report_entities(spec: GraphSpec, node: GraphNode) -> str:
    """报告节点的实体层：link（默认）/ off。出口复核调 verify_doc 时传 entities=(这个值 == "link")。

    写错的值执行时节点会报错，这里只把 off 当 off，别的一律按默认。
    """
    return "off" if _setting(spec, node, "entities", "link") == "off" else "link"


def report_claims(spec: GraphSpec, node: GraphNode) -> str:
    """报告节点的结论句策略：off / require_citation / judge。出口判档按它（和节点执行时同一个取值规则）。

    写错的值画布校验就拦了，执行时节点也会报错，这里只把没写当成 off。
    """
    value = _setting(spec, node, "claims", "off")
    return value if isinstance(value, str) and value in CLAIMS else "off"


def report_judge(spec: GraphSpec, node: GraphNode) -> dict[str, Any]:
    """claims: judge 的子配置：节点上的 judge（没写取图级 defaults），写了的键原样保留——上限写 null 就是
    不限，没写的键运行时取设置里的默认（judge.report_budget）。rewrite_once、on_unsupported 补上缺省
    （false、degrade）：出口判档读 on_unsupported。写错的值画布校验就拦了，执行时节点也会报错。
    """
    raw = _setting(spec, node, "judge")
    raw = raw if isinstance(raw, dict) else {}
    return {**raw, "rewrite_once": raw.get("rewrite_once") is True,
            "on_unsupported": raw.get("on_unsupported") if raw.get("on_unsupported") in JUDGE_ON_UNSUPPORTED
            else "degrade"}


def report_allowance(spec: GraphSpec, node: GraphNode) -> list[Any]:
    """报告自查放行的字面量：出口契约 report_from 指向它时，契约的 allow_numbers。

    契约放行、报告却不放行的话，写作者会为一个最终根本不算违规的数白白重写一次。
    反过来不成立：放行名单只由契约定，报告节点没有自己的名单——出口复核是独立的裁判。
    """
    allowed: list[Any] = []
    for other in spec.nodes:
        contract = other.config.get("contract") if other.type == NodeType.OUTPUT else None
        if isinstance(contract, dict) and contract.get("report_from") == node.id:
            extra = contract.get("allow_numbers") or []
            allowed.extend(extra if isinstance(extra, list) else [extra])
    return allowed


def report_cells_allowed(spec: GraphSpec, node: GraphNode, *, governed: bool) -> bool:
    """报告能不能直接引用查询单元格：出口契约的 report_from 指着它、而那份契约不允许的，写作时就不给。

    出口复核按契约判（io.py），这里提前照同一条规则建目录：不然写作者照着提示写了单元格、自查
    通过，到出口才整段判成解析不了——受管出具直接不予出具。
    """
    from app.engine.governance import cells_allowed

    contracts = [other.config.get("contract") for other in spec.nodes if other.type == NodeType.OUTPUT]
    return all(cells_allowed(c, governed=governed) for c in contracts
               if isinstance(c, dict) and c.get("report_from") == node.id)


#: 目录里能被报告引用的几种来源（沙箱代码节点的登记只为说清为什么不能引用，不算）
_CITABLE = {"metric": "口径卡指标", "query": "查询结果", "retrieval": "知识库检索", "input": "运行输入"}


def _no_evidence(catalog: dict[str, Any], cells: bool) -> str | None:
    """目录里没有可引用的来源时给一句话，说清缺的是哪种；有就返回 None。"""
    kinds = {entry.get("kind") for entry in catalog.values()}
    if not kinds & set(_CITABLE):
        return ("报告撰写节点的上游没有可引用的来源（" + "、".join(_CITABLE.values()) + "都没有），"
                f"报告中的数字都会被判为没有出处。请在它前面接入取数节点或口径卡，或检查「{field_label('metrics_from')}」")
    if not cells and not kinds & {"metric", "input"}:
        # 只有查询和检索：单元格引用不了，检索片段本来就只能当依据，一个数都没处引
        return (f"报告撰写节点的上游只有查询结果，没有口径卡指标，而{CELLS_REASON}，当前无法引用单元格，"
                "报告中的数字都会被判为没有出处。请在出具契约中开启「单元格引用」，或把要写的数字登记到口径卡")
    return None


# --------------------------------------------------------------------------
# 执行器
# --------------------------------------------------------------------------


class _RenderedStream:
    """借 _invoke_streaming 的流式与回退逻辑，只在出口把 llm.token 换成渲染后的字。

    模型写的是 [[m:gmv]]，前端要看到的是「45,678.5元」。标记会被切在两个 chunk
    中间，StreamRenderer 攒到没有未闭合的 [[ 再放；调用结束时 flush 掉剩下的。
    """

    def __init__(self, ctx: NodeContext, catalog: dict[str, Any], cells: bool = True) -> None:
        self._ctx = ctx
        self._catalog = catalog
        self._cells = cells
        self._renderer = StreamRenderer(catalog, cells_allowed=cells)

    def __getattr__(self, name: str) -> Any:
        return getattr(self._ctx, name)

    def emit(self, event_type: EventType | str, **data: Any) -> None:
        if event_type == EventType.LLM_TOKEN:
            if text := self._renderer.feed(str(data.get("delta") or "")):
                self._ctx.emit(EventType.LLM_TOKEN, delta=text)
            return
        if event_type == EventType.LOG and data.get("code") == "stream_fallback":
            # 流断在半路、退回非流式重来：攒着的半截不再放出去，免得和重来的那份接在一起
            self._renderer = StreamRenderer(self._catalog, cells_allowed=self._cells)
        self._ctx.emit(event_type, **data)

    def flush(self) -> None:
        if tail := self._renderer.flush():
            self._ctx.emit(EventType.LLM_TOKEN, delta=tail)


class _MaskedCells:
    """报告自查：写作者引用了取自遮罩列的格子（[[v:Q2.r0.tel]]），记一条违规 masked_cell，和没有出处的数字一样交回写作者
    重写（evidence.masked_cells）。写作目录里的查询结果本身不遮（写作者要看得到数据），只拦引用；口径卡指标不一样，值取自
    遮罩格的在目录里就写成了「已遮罩」（文档的证据目录收着全部指标，原值写着就随文档出去了），引用它的组装文档时就记了
    违规（evidence.masked_metrics），不在这里。

    哪几列算按 api/evidence.py 的 _cited_masks 认：列名就是遮罩列的，SQL 里给遮罩列起了别名、套了函数、经子查询或 WITH
    转一道的（_mask_sets 的 sure，和证据面板同一个函数）；合并查询按逐格来历和合并 SQL 认。遮罩按列名在整个数据源上
    生效，和证据面板一样：哪张表（含视图）里叫这个名字的列都算，设了 name 时门店名和客户姓名都算，违规说明点名取自哪张表
    的哪一列。证据面板为了展示宁可多遮（SQL 认不准时整份的列都遮），自查不跟着多算：多算一列就是一次判失败，写作者也
    改不掉。

    只查引用，不按值在正文里找遮罩值：两个字的姓名会撞上「销售高峰」这类常用词，正式运行因此判失败（复核第 3 轮去掉了
    按值认）；正文里照抄的号码本来就会被「数字没有出处」拦下。

    组装好一稿之后才认：只读正文引用到的那几份快照（读盘复验哈希放进线程），每份只读一次，重写的稿子不再读；没引用格子的
    一份都不读。"""

    def __init__(self, catalog: dict[str, Any]) -> None:
        from app.api.evidence import _Masks

        self._catalog = catalog
        self._masks = _Masks()
        self._memo: dict[str, dict[str, list[str]]] = {}

    async def __call__(self, doc: dict[str, Any]) -> dict[str, Any]:
        """把这一稿引用了遮罩列的格子记进 doc 的违规（接在最后，和 compose_doc 给了 masked 时一样），交回同一份文档。"""
        from app.engine.evidence import iter_segments, masked_cells

        wanted: dict[str, Any] = {}
        for seg in iter_segments(doc):
            cite = seg.get("cite") or {}
            entry = self._catalog.get(cite.get("alias")) or {}
            artifact = entry.get("artifact")
            if cite.get("kind") == "cell" and cite.get("status") == "resolved" and isinstance(artifact, str) \
                    and artifact and artifact not in self._memo:
                wanted.setdefault(artifact, entry.get("source"))
        for artifact, source in wanted.items():
            await self._read(artifact, source)
        cited = masked_cells(doc, self._catalog, {a: m for a, m in self._memo.items() if m})
        doc["violations"] = [*doc["violations"], *cited]
        return doc

    async def _read(self, artifact: str, source: Any) -> None:
        """读一份查询快照，引用了就算违规的列（_cited_masks：{结果列: 它取自的遮罩列}）记进 _memo。读不出来的记空：引用它的
        格子本来就解析不了。"""
        from app.api.evidence import _cited_masks
        from app.core.artifact_store import load

        try:
            snap = await asyncio.to_thread(load, artifact)
        except ValueError:
            snap = None
        self._memo[artifact] = await _cited_masks(snap, source, self._masks, artifact) if isinstance(snap, dict) else {}


def _masked_said(hits: list[dict[str, Any]]) -> str:
    """报告引用了遮罩列的那几处在缺口、日志里怎么说：「引用了 2 处取自遮罩列的格子（「v:Q2.r0.tel」、…）」，引用了值取自
    遮罩格的指标的另说「指标」（违规上记着 metric）。只点名引用，不写值。"""
    metrics = sum(1 for v in hits if v.get("metric"))
    parts = [f"{n} 处取自遮罩列的{what}" for n, what in ((len(hits) - metrics, "格子"), (metrics, "指标")) if n]
    return f"引用了 {'、'.join(parts)}（{_named(hits)}）"


def _masked_targets(hits: list[dict[str, Any]]) -> str:
    """去改写作指令时要去掉的是什么：引用了格子说「这些列」，引用了指标说「这些指标」，都有的两样都说。"""
    metrics = any(v.get("metric") for v in hits)
    cells = any(not v.get("metric") for v in hits)
    return "这些列、指标" if metrics and cells else "这些指标" if metrics else "这些列"


async def _run_class(run_id: str) -> str | None:
    """按运行记录上的 run_class 取：子图、审批之后续跑都是另起的执行上下文，只有运行记录靠得住。"""
    from app.db.models import Run

    async with SessionLocal() as session:
        run = await session.get(Run, run_id)
    return run.run_class if run is not None else None


async def _default_on_violation(run_id: str) -> str:
    """探索运行默认标注、照常产出；正式运行默认失败——封存的发布版不能带着裸数字出具。"""
    return "fail" if await _run_class(run_id) == "formal" else "flag"


def _options(ctx: NodeContext) -> tuple[str, int]:
    numbers = str(ctx.cfg("numbers", "strict") or "strict")
    if numbers not in NUMBERS:
        raise NodeError(ctx.node.id, _only("numbers", NUMBERS, numbers))
    # 画布校验也拦这两项；执行时再守一道，报的是同一句话，免得写错的值被悄悄当成默认
    if problem := claims_problem(ctx.cfg("claims")):
        raise NodeError(ctx.node.id, problem)
    if ctx.cfg("claims") == CLAIMS_JUDGE and (problems := judge_problems(ctx.cfg("judge"))):
        raise NodeError(ctx.node.id, "；".join(message for _, message in problems))
    entities = ctx.cfg("entities", "link") or "link"
    if entities not in ENTITIES:
        raise NodeError(ctx.node.id, _only("entities", ENTITIES, entities))
    raw = ctx.cfg("max_repairs", 1)
    try:
        repairs = int(raw)
    except (TypeError, ValueError):
        raise NodeError(ctx.node.id, _repairs_problem(raw)) from None
    if isinstance(raw, bool) or not 0 <= repairs <= MAX_REPAIRS:
        raise NodeError(ctx.node.id, _repairs_problem(raw))
    return numbers, repairs


def _only(key: str, allowed: tuple[str, ...], got: Any) -> str:
    """「「表名、字段名」只能是「核对」或「不核对」，当前为「x」」：写界面上的叫法，不写键名和枚举值。"""
    names = [f"「{option_label(key, v)}」" for v in allowed]
    listed = "或".join(names) if len(names) <= 2 else "、".join(names[:-1]) + "或" + names[-1]
    return f"「{field_label(key, 'report')}」只能是{listed}，当前为「{got}」"


def _repairs_problem(raw: Any) -> str:
    return f"「{field_label('max_repairs', 'report')}」需要填写 0 到 {MAX_REPAIRS} 的整数，当前为「{raw}」"


def _blocking(violations: list[dict[str, Any]], numbers: str) -> list[dict[str, Any]]:
    """要让写作者改的违规。numbers=off 时裸数字只记录不拦（文档和出口复核照样看得到）。

    可疑实体只标注（SOFT）：一个反引号里的业务词也可能被当成名字，为它整篇重写、判失败都不值；
    它留在文档和出口复核里，受管的正式出具按缺口降档。表结构快照不全时核对不了的名字同样只标注。
    """
    return [v for v in violations if v.get("code") not in SOFT
            and not (numbers == "off" and v.get("code") == "uncited_number")]


def _fatal(blocking: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """要判失败的违规：重写之后还在的、不属于 REWRITE_ONLY 的。"""
    return [v for v in blocking if v.get("code") not in REWRITE_ONLY]


def _fail_advice(fatal: list[dict[str, Any]]) -> str:
    """判失败时给人看的下一步。引用了遮罩列的格子多半是写作指令点名要的（「联系电话 [[v:Q2.r0.tel]]」），换模型没用；
    改成「标注」照常产出，遮罩列的值就随报告出去了——这两条都不建议，只说去改写作指令。别的违规照旧建议换模型、先标注。"""
    hits = [v for v in fatal if v.get("code") == "masked_cell"]
    if hits and len(hits) == len(fatal):
        return f"数据源设置了遮罩的列不应写进报告，请在报告撰写节点的指令中去掉对{_masked_targets(hits)}的引用"
    if hits:
        return (f"引用了遮罩列的那几处，请在报告撰写节点的指令中去掉对{_masked_targets(hits)}的引用；"
                "其余几处可换用指令遵循能力更强的模型")
    return (f"请换用指令遵循能力更强的模型，或先将「{field_label('on_violation')}」设为"
            f"「{option_label('on_violation', 'flag')}」，查看标注后再调整")


def _violation_key(violation: dict[str, Any]) -> tuple[str, str]:
    """认一处违规：(code, 点名的字)。点名的字按 text、ref、message 的顺序取第一个有的。"""
    named = violation.get("text") or violation.get("ref") or violation.get("message") or ""
    return str(violation.get("code") or ""), str(named)


def _new_violations(before: list[dict[str, Any]], after: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """after 里 before 没有的违规，按 (code, 点名的字) 逐个抵消（同一个裸数字多写一处也算新的）。

    全部违规都比，包括只标注、不让重写的那几类（SOFT）：改写稿冒出一个编造的表名，受管的正式出具就会
    因此降档；去掉一处、又冒出另一处，总数没变，冒出来的也还是新的。
    """
    left = Counter(_violation_key(v) for v in before)
    fresh: list[dict[str, Any]] = []
    for violation in after:
        key = _violation_key(violation)
        if left[key] > 0:
            left[key] -= 1
        else:
            fresh.append(violation)
    return fresh


def _named(violations: list[dict[str, Any]], limit: int = 3) -> str:
    """违规里点名的那几个字：「45678」「m:nope」。"""
    names = [str(v.get("text") or v.get("ref") or "") for v in violations]
    names = [n for n in dict.fromkeys(names) if n][:limit]
    return "、".join(f"「{n}」" for n in names)


def _repair_request(violations: list[dict[str, Any]]) -> str:
    return (
        "你写的报告没有通过系统核对，问题如下：\n"
        f"{describe_violations(violations)}\n\n"
        "请重写整篇报告：所有数字都用引用标记写，只引用证据目录里有的指标和输入；"
        "不要解释改了什么，只输出新的报告正文。"
    )


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 "根据下面的证据写一份简洁的报告，先总后分。"
    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 "")
    return [
        SystemMessage(content="\n\n".join(p for p in (system, ROLE) if p)),
        HumanMessage(content=(
            f"{instructions}\n\n{writer.text}\n\n{rules}\n\n"
            "只输出报告正文（Markdown），不要写撰写说明。"
        )),
    ]


#: 整篇报告外面包的那层代码块（_unfenced）。第一行：围栏（三个以上反引号或波浪号）加信息串
_OPEN_FENCE = re.compile(r"(`{3,}|~{3,})[ \t]*([^\s`]*)[ \t]*")
#: 最后一行：行首最多缩进三格（同 CommonMark），缩进四格的是一行缩进代码，不算收尾
_CLOSE_FENCE = re.compile(r" {0,3}(`{3,}|~{3,})")
#: 正文里以围栏开头的行（行首空白不讲究）：和外层同种符号、比外层长的，分不清哪一行收尾
_FENCE_RUN = re.compile(r"\s*(`{3,}|~{3,})")
#: 其中 compose_doc 的切块（evidence._FENCE）和前端 Markdown.tsx 认作围栏的：信息串只能是紧跟着的一个词（字母、数字、
#: 下划线）。去掉外层之后正文按它们切，``` sql、```sql title 这样的行它们不认，是一行正文
_INNER_FENCE = re.compile(r"\s*(`{3,}|~{3,})([A-Za-z0-9_]*)\s*")
#: 信息串只认这几个（大小写不敏感）：写成 sql、json 的是模型真要贴一段代码
_PROSE_INFO = frozenset({"", "markdown", "md"})


def _unfenced(raw: str) -> str:
    """模型把整篇报告包成一个代码块（```markdown … ```）时去掉外层围栏，返回里面的正文；别的一律原样返回。

    不去的话整篇在 compose_doc 里是一个代码块：裸数字照样标出，可一句结论都切不出来，裁判、反方审稿没有句子可审，
    界面上整篇显示成代码。只在拿得准的时候去：去掉首尾空白后，第一行是围栏、信息串为空或 markdown / md，最后一行
    是同一道围栏（同种符号、同样长、不带信息串）。sql、json 这类信息串、围栏外还有前言结语的，都不是整篇报告被包了
    起来，原样交给 compose_doc。

    正文自己的代码块可以套在里面，中间的围栏行这样看（只看和外层同种符号的）：
    - 比外层短：收不掉外层，外层收在哪一行没有疑问，跳过
    - 比外层长：分不清哪一行收尾，不去
    - 和外层一样长：模型最常见的写法是 ```markdown 里再套 ```sql … ```。带信息串的一行开一个代码块，不带的一行收掉它，
      两两配对；去掉外层后，compose_doc 和前端每见到一行 ``` 开合一次，配好的这一对正好切成一个代码块。配不成对的
      不去：没有代码块可收的 ``` 说明外层在这里就收了，后面是围栏外的字（好几个代码块）；开着的代码块里又来一个
      ```sql，compose_doc 和前端会拿它当收尾；扫到最后还开着的，后半篇会被吞进代码块

    换一种符号的（外层反引号、里面 ~~~）也跳过，照常去掉外层。跳过的这些行去掉外层后怎么切，和模型不包外层、直接
    这么写时一样：compose_doc 和前端只认 ``` 围栏，里面 ~~~、四个以上反引号的代码块按正文切，当一段话核对——比整篇
    一个代码块强，真正的结论句至少有得审。
    """
    text = (raw or "").replace("\r\n", "\n").replace("\r", "\n").strip()
    lines = text.split("\n")
    head = _OPEN_FENCE.fullmatch(lines[0])
    tail = _CLOSE_FENCE.fullmatch(lines[-1]) if len(lines) > 1 else None
    if head is None or tail is None or head.group(2).lower() not in _PROSE_INFO or tail.group(1) != head.group(1):
        return raw
    fence = head.group(1)
    nested = False      # 正文自己的、和外层一样长的代码块开着
    for line in lines[1:-1]:
        run = _FENCE_RUN.match(line)
        if run is None or run.group(1)[0] != fence[0] or len(run.group(1)) < len(fence):
            continue
        if len(run.group(1)) > len(fence):
            return raw
        if (inner := _INNER_FENCE.fullmatch(line)) is None:
            continue    # ``` sql 这样的行 compose_doc 和前端不认，去掉外层后是一行正文
        # 一样长的：带信息串的只能开（已经开着就配不成对），不带的只能收（没有开着的就是外层在这里收了）
        if bool(inner.group(2)) == nested:
            return raw
        nested = not nested
    return raw if nested else "\n".join(lines[1:-1])


async def _store(doc: dict[str, Any], ctx: NodeContext) -> str | None:
    from app.core.artifact_store import put_json

    try:
        return await put_json(doc, kind="report_doc", run_id=ctx.run.run_id, node_id=ctx.node.id,
                              meta={"stats": doc.get("stats") or {}})
    except Exception as e:  # noqa: BLE001 - 工件写失败不毁掉运行，只是这份报告点不开证据
        ctx.emit(EventType.LOG, level="warn", code="evidence_store_failed",
                 message=f"报告文档保存失败（{describe_exception(e)}），报告中的数字将无法查看证据")
        return None


async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
    numbers, max_repairs = _options(ctx)
    on_violation = str(ctx.cfg("on_violation", "") or "") or await _default_on_violation(ctx.run.run_id)
    if on_violation not in ON_VIOLATION:
        raise NodeError(ctx.node.id, _only("on_violation", ("flag", "fail"), on_violation))

    async with SessionLocal() as session:
        try:
            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
    # 目录按上游的产出建：这个节点自己上一次的产出（循环里重跑时）不算，它记下的 masked_metrics 这一次重新认
    upstream = {**state, "nodes": {k: v for k, v in (state.get("nodes") or {}).items() if k != ctx.node.id}}
    catalog = report_catalog(upstream, spec, ctx.node)  # type: ignore[arg-type]
    # 值取自遮罩格的指标（cell() 取的遮罩格、取自改了名字的遮罩列、转抄上游口径卡里这类指标的）：写作目录、存进文档的
    # 证据目录里都写「已遮罩」。只按确定取自遮罩列的认，和单元格引用的自查同一个口径（api/evidence.py 的 withheld_metrics）：
    # SQL 认不准时没确定取自遮罩列的普通金额照给，不然引用它的正式运行判失败、写作者也改不掉。记进产出，出口契约复核按
    # 同一份重建目录
    from app.api.evidence import withheld_metrics

    withheld = await withheld_metrics(catalog, nodes=state.get("nodes") or {}, ledger=state.get("evidence") or [])
    catalog = withhold_metrics(catalog, withheld, _masked_word())
    # 报告引用了取自遮罩列的格子：自查记一条违规，和没有出处的数字一样交回写作者重写（_MaskedCells）；引用了上面那几个
    # 指标的，组装文档时就记了（evidence.masked_metrics）
    masked = _MaskedCells(catalog)
    allow = report_allowance(spec, ctx.node)
    cells = True
    # 结论句策略只对升级后发起的运行生效：升级前的运行（图里也不会有 claims）事件、产出一个字不加
    evidence_on = ledger_enabled(ctx.run)
    claims = report_claims(spec, ctx.node) if evidence_on else "off"
    if evidence_on:
        # 升级前发起的运行没有查询条目，单元格无从谈起，也就不多查这一次
        from app.engine.governance import governed_formal

        cells = report_cells_allowed(spec, ctx.node, governed=await governed_formal(ctx.run.run_id))
    if warning := _no_evidence(catalog, cells):
        ctx.emit(EventType.LOG, level="warn", code="report_no_evidence", message=warning)

    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),
                 compact=writer.compact, compact_rows=writer.compact_rows,
                 nodes=[{k: r[k] for k in ("node_id", "title", "total", "omitted", "compact")} for r in writer.nodes])
    messages = _messages(ctx, state, writer, claims)
    started = time.perf_counter()
    spent: list[dict[str, Any]] = []
    nudged = False

    async def draft(payload: list[BaseMessage]) -> tuple[BaseMessage, str]:
        nonlocal nudged
        stream = _RenderedStream(ctx, catalog, cells)
        began = time.perf_counter()
        response = await _invoke_streaming(model, payload, stream, model_id)  # type: ignore[arg-type]
        stream.flush()
        spent.append(_report_call(ctx, response, model_id, began))
        text = message_text(response)
        if snippet := leaked_markup(text):
            # 写成了一段工具调用标记：纠正一次，还这样就判失败，别把标记当报告往下游送
            if nudged:
                raise NodeError(ctx.node.id, MARKUP_ERROR)
            nudged = True
            ctx.emit(EventType.LOG, level="warn", code="tool_markup_leak",
                     message=f"{markup_warning(snippet)}，已要求模型重试一次")
            return await draft([*payload, response, HumanMessage(content=MARKUP_NUDGE)])
        if not text and thinking_text(response):
            ctx.emit(EventType.LOG, level="warn", code="empty_completion",
                     message="模型只输出了思考内容，报告正文为空。请调大「最大输出 token」，或关闭「思考模式」。")
        return response, text

    entities = report_entities(spec, ctx.node) == "link"

    async def compose(raw: str) -> dict[str, Any]:
        # 第一稿、重写稿、裁判退回的改写稿都从这里过：整篇包在代码块里的先去掉外层围栏。文档的 source 记的是去掉
        # 围栏之后的正文——markdown 由它渲染，出口契约复核取回的也是这份文档。引用了遮罩列的格子组装好之后再认
        # （_MaskedCells）：只读正文里引用到的那几份快照
        return await masked(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))

    response, raw = await draft(messages)
    doc = await compose(raw)
    repairs = 0
    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)}），"
                         f"已要求模型重写（第 {repairs} 次）")
        messages = [*messages, response, HumanMessage(content=_repair_request(blocking))]
        response, raw = await draft(messages)
        doc = await compose(raw)

    judgement: dict[str, Any] | None = None
    if claims == CLAIMS_JUDGE:
        from app.engine.nodes.human import reviewed_by_human

        judge_cfg = report_judge(spec, ctx.node)
        if await _run_class(ctx.run.run_id) != "formal" and not reviewed_by_human(spec, ctx.node.id):
            # 探索运行默认按需裁判：点开哪句才判哪句，节点里不花这笔钱。要交给人工审批的报告例外：审批人据此
            # 拍板，批准前就该知道哪句结论站不住；而按需裁判要等运行封存之后才能判，封存前放开会动证据清单的
            # 完整性约束，所以这种报告探索运行也在节点里当场判（和正式运行同一条路）
            judgement = judging.on_demand(doc, on_unsupported=judge_cfg["on_unsupported"],
                                          rewrite_once=judge_cfg["rewrite_once"], writer_model=model_id)
        elif not (_fatal(blocking) and on_violation == "fail"):
            # 反正要判失败的报告不值得再花钱判（节点紧接着就报错）
            doc, response, blocking, judgement, usages = await _judge_inline(
                ctx, doc, catalog, judge_cfg, model_id=model_id, messages=[*messages, response],
                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}："
                         "被省略的查询仍在证据台账里，这类结论依据不足，应写作本报告未覆盖。报告照常产出，这几处已标注")
    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["ref"]) for v in hits if v.get("ref")))
        gaps.append(f"报告{_masked_said(hits)}，{tried}：数据源设置了遮罩的列不应写进报告")
        ctx.emit(EventType.LOG, level="warn", code="report_masked_cell", count=len(hits), refs=refs,
                 columns=list(dict.fromkeys(str(v["column"]) for v in hits if v.get("column"))),
                 message=f"{gaps[0]}。报告照常产出，这几处已标注，已记为缺口。"
                         f"请在报告撰写节点的指令中去掉对{_masked_targets(hits)}的引用")
    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,
             on_violation=on_violation, failed=bool(_fatal(blocking)) and on_violation == "fail",
             **({"claims": claims} if evidence_on else {}),
             **({"judge": judgement} if judgement is not None else {}))
    if (fatal := _fatal(blocking)) and on_violation == "fail":
        blocking = fatal
        head = "；".join(v["message"] for v in blocking[:5])
        more = f"；…另有 {len(blocking) - 5} 处" if len(blocking) > 5 else ""
        raise NodeError(
            ctx.node.id,
            f"报告有 {len(blocking)} 处未通过核对（已要求重写 {repairs} 次）：{head}{more}。{_fail_advice(blocking)}",
        )

    usage = {key: sum(one[key] for one in spent) for key in spent[0]}
    usage["cost_usd"] = round(usage["cost_usd"], 6)
    text = doc["markdown"]
    result = {
        "text": text,
        "doc_artifact": doc_artifact,
        "stats": doc["stats"],
        "violations": doc["violations"][:20],
        "repairs": repairs,
        "model": model_id,
        "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 {}),
        # 写成「已遮罩」的指标：出口契约复核按它重建目录（report_catalog）。同上，只在有的时候带
        **({"masked_metrics": withheld} if withheld else {}),
    }
    updates: dict[str, Any] = {"nodes": {ctx.node.id: result}, "usage": usage}
    if ctx.cfg("emit_message", True):
        updates["messages"] = [AIMessage(content=text or "（空）")]
    var_name = ctx.cfg("assign_to", "")
    if var_name:
        updates["vars"] = {var_name: text}
    return updates


# --------------------------------------------------------------------------
# claims: judge——正式运行在节点里裁判
# --------------------------------------------------------------------------


def _rewrite_request(sentences: list[tuple[str, str]]) -> str:
    listed = "\n".join(f"{i}.「{text}」——{why or '证据不支持这句话'}" for i, (text, why) in enumerate(sentences, 1))
    return (
        "另一个模型按证据逐句核对了你的报告，下面这几句它认为证据不支持（和证据相矛盾，或者证据不足）：\n"
        f"{listed}\n\n"
        "请只改这几句：补上证据目录里真有的依据 [[see:…]]，或者把说法收窄到证据能支持的程度，或者删掉；"
        "其余部分一字不改。数字仍然只能用引用标记写。不要解释改了什么，只输出改好的整篇报告正文。"
    )


def _rewrite_reason(verdict: dict[str, Any]) -> str:
    """交回写作者的理由：裁判的理由，证据不足的再写上缺的是什么。"""
    why = str(verdict.get("rationale") or "")
    if verdict.get("status") == "insufficient" and verdict.get("missing"):
        why = f"{why}（缺少：{verdict['missing']}）" if why else f"证据不足，缺少：{verdict['missing']}"
    return why


def _merge(first: dict[str, Any], second: dict[str, Any], budget: judging.Budget) -> dict[str, Any]:
    """改写前后两轮裁判合成一份：判定、句数、缺口按改写后这一稿，花费、调用、用时两轮相加。"""
    merged = dict(second)
    for key in ("cost_usd", "estimated_usd"):
        merged[key] = round(first[key] + second[key], 6)
    for key in ("calls", "duration_ms"):
        merged[key] = first[key] + second[key]
    merged["notes"] = list(dict.fromkeys([*first["notes"], *second["notes"]]))
    merged["budget"] = budget.as_dict()
    merged["model"] = second.get("model") or first.get("model")
    merged["priced"] = first.get("priced")
    return merged


async def _judge_inline(ctx: NodeContext, doc: dict[str, Any], catalog: dict[str, Any], judge_cfg: dict[str, Any], *,
                        model_id: str, messages: list[BaseMessage], draft: Any, compose: Any, numbers: str,
                        blocking: list[dict[str, Any]]
                        ) -> tuple[dict[str, Any], BaseMessage, list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]]:
    """写完、核对完之后请裁判判一遍，判定写进文档。返回 (文档, 写作者最后一稿, 要拦的违规, 裁判摘要, 用量)。

    messages 是写作的对话、最后一条是写作者这一稿。rewrite_once 时把证据不支持的句子交回去只改这几句，
    再判一次（没改的句子直接用第一次的判定），最多一轮；改写稿冒出原稿没有的违规就不采用，保留原稿。

    摘要里的 rewrite = {units, sentences, applied, reason, changed}：
    - sentences：交回去的句子（原稿里的原文）
    - units：这几句在封存文档里的编号。没采用时封存的是原稿，就是原稿的编号；采用了封存的是改写稿，
      编号按改写稿重排，按原文认：原句原样留在改写稿里的给新编号，改掉了的是 null
    - changed：采用时改写稿里新出来的句子（原稿里没有这段原文）在封存文档里的编号；没采用是 []
    """
    from app.engine.replay import _in_graph

    async with SessionLocal() as session:
        settings = await judging.judge_settings(session)
        judge_spec = await judging.judge_model_spec(session, judge_cfg, settings=settings)
    budget = judging.report_budget(settings, judge_cfg)
    masks = await judging.source_masks(catalog)

    @task
    async def judge_step(request: judging.JudgeRequest, key: str, limits: judging.Budget, known: dict[str, Any],
                         first: bool) -> dict[str, Any]:
        # 裁判进 checkpoint：「接着跑」时直接取这里的判定，不再调用、不再计费（每日计数也记在这里面）。
        # 事件都在这里面发：取缓存时不会再发一遍。第二轮 limits 是剩下的钱和时间，说给人看的仍是配置的上限
        outcome = await judging.run_request(request, spec=judge_spec, budget=limits, known=known, emit=ctx.emit,
                                            shown=budget)
        if first and outcome.get("model") and outcome["model"] == model_id:
            outcome["same_model"] = True
            ctx.emit(EventType.LOG, level="warn", code="judge_same_model",
                     message=f"裁判模型与写作模型相同（「{model_id}」），难以发现写作模型自身的错误。"
                             "请在「设置 → 偏好设置 → 证据裁判」中选择另一个模型，"
                             f"或在节点的「{field_label('judge')}」中指定裁判模型")
        return {**outcome, "key": key}

    async def judge_round(round_doc: dict[str, Any], limits: judging.Budget, known: dict[str, Any],
                          first: bool) -> dict[str, Any]:
        # 摘录里每份查询快照逐份遮，改了名字的遮罩列（c.phone AS tel）也不给，和按需裁判同一个函数（_judge_loader）
        from app.api.evidence import _judge_loader

        request = judging.prepare(round_doc, catalog, masked=masks, loader=await _judge_loader(round_doc, catalog))
        key = judging.request_key(request, judge_spec, limits, known)
        if _in_graph():
            # 写作不进 checkpoint，接着跑时写作者会重写一稿：缓存里先前那一稿的判定对不上指纹，往后翻
            for _ in range(_STALE_LIMIT):
                outcome = await judge_step(request, key, limits, known, first)
                if outcome.get("key") == key:
                    return outcome
        return await judging.run_request(request, spec=judge_spec, budget=limits, known=known, emit=ctx.emit,
                                         shown=budget)

    outcome = await judge_round(doc, budget, {}, True)
    same_model = bool(outcome.get("same_model"))
    usages = [outcome["usage"]]
    response = messages[-1]
    rewrite: dict[str, Any] | None = None
    # 交回改写的：证据相矛盾的（含旧取值 unsupported）和证据不足的——两样原来都判作「证据不支持」
    bad = [uid for uid, v in outcome["verdicts"].items() if v["status"] in (*judging.CONTRADICTED, "insufficient")]
    if judge_cfg["rewrite_once"] and bad:
        texts = judging.unit_texts(doc)
        sentences = [(texts.get(uid, ""), _rewrite_reason(outcome["verdicts"][uid])) for uid in bad]
        ctx.emit(EventType.LOG, level="warn", code="report_rewrite",
                 message=f"裁判认为 {len(bad)} 句结论证据不支持（{_named([{'text': t} for t, _ in sentences])}），"
                         "已退回写作模型仅修改这几句")
        again, raw = await draft([*messages, HumanMessage(content=_rewrite_request(sentences))])
        rewritten = await compose(raw)
        worse = _blocking(rewritten["violations"], numbers)
        fresh = _new_violations(doc["violations"], rewritten["violations"])
        rewrite = {"units": bad, "sentences": [t for t, _ in sentences], "applied": False, "reason": None,
                   "changed": []}
        if fresh:
            rewrite["reason"] = (f"改写稿新增了 {len(fresh)} 处原稿没有的问题（{_named(fresh)}），未予采用，"
                                 "已保留原稿及原有判定")
            ctx.emit(EventType.LOG, level="warn", code="report_rewrite_rejected", message=rewrite["reason"])
        else:
            known = {outcome["keys"][u]: v for u, v in outcome["verdicts"].items() if v["status"] in judging.JUDGED}
            # 第二轮用第一轮剩下的钱和时间：每份报告的上限管的是整份报告
            left = judging.Budget(
                max_claims=budget.max_claims,
                max_cost_usd=None if budget.max_cost_usd is None else max(0.0, budget.max_cost_usd - outcome["cost_usd"]),
                timeout_s=None if budget.timeout_s is None else max(0.0, budget.timeout_s - outcome["duration_ms"] / 1000),
                daily_max_usd=budget.daily_max_usd)
            second = await judge_round(rewritten, left, known, False)
            usages.append(second["usage"])
            outcome = _merge(outcome, second, budget)
            doc, response, blocking = rewritten, again, worse
            # 封存的是改写稿，编号按它重排：按原文认回去，改掉了的句子在改写稿里没有编号
            now = judging.unit_texts(rewritten)
            at: dict[str, str] = {}
            for uid, text in now.items():
                at.setdefault(text, uid)
            before = set(texts.values())
            rewrite.update(applied=True, units=[at.get(texts.get(uid, "")) for uid in bad],
                           changed=[uid for uid, text in now.items() if text not in before])
    summary = judging.summarize(outcome, mode="inline", on_unsupported=judge_cfg["on_unsupported"],
                                rewrite_once=judge_cfg["rewrite_once"], same_model=same_model, rewrite=rewrite,
                                writer_model=model_id)
    judging.apply_judgement(doc, outcome, summary)
    return doc, response, blocking, summary, usages