diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index a2348aad..f1902e78 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -1,21 +1,480 @@ -"""反方审稿节点：让另一个模型站在反方，专挑上游报告的毛病，逐条提出反驳意见。 +"""反方审稿节点：报告写完后，挑几句关键结论，各派一个只读 agent 站在反方主动找反例。 -目前只是占位：图里放了这个节点也能运行，执行器原样跳过、记一条日志，不改任何状态。 -真正的审稿由后续工作包补上；没能回应的反驳通过节点产出里的 issuance_gaps 交给成果节点的 -出具契约降档（见 nodes/io.py 的 _upstream_issuance_gaps）。 +结论句裁判（engine/judge.py）只看写作者挂的证据，判「说法有没有被证据支持」；可证据本身就可能挑过——只查了含 +退款的销售额，只看了一个月。反方审稿换个方向：不看证据够不够，自己写查询去证伪。 + +1. 选句：报告文档里的结论句按裁判的预筛排好（judge.candidates / claim_priority：挂了依据的、带比较或因果的在前）， + 取前「最多审几句」句；裁判判成「不是结论」的不审。同一份文档永远选出同样的句子、同样的顺序 +2. 反查：每句一个只读 agent（verify_agent.run_verify_agent）。工具只给这句引用的查询所在的数据源（指标按它的输入 + 取自的查询算，合并查询按它合并的那几个源算）；这句没挂依据，就在这份报告引用的数据源里查。任务里给这句结论 + 和它挂的证据摘录（judge.prepare，数据源遮罩的列照旧不给），要它先想清楚「如果这句不成立，数据里会看到什么」， + 写只读查询去找，最后交结论（经受住 / 被削弱 / 被推翻 / 无法判断）和理由。理由里的数字只能引用它自己查询的 + 格子，裸数字由 compose_doc 标出来 +3. 记录：每句的结论、理由（渲染后的正文和组装好的文档）、查询快照、用量写进审稿记录（challenge_artifact），随运行 + 封存。查询照常发 tool.end（role: verify，带查询快照，可下钻）；台账只记 kind challenge，绝不记 query——核查的 + 查询进了写作目录，下游的写作者会把它们当成可以引用的证据 +4. 处置：被推翻的按 on_refuted 处置。degrade（默认）在产出里写 issuance_gaps，成果节点的出具契约照单收进缺口、 + 出具降档（nodes/io.py 的 _upstream_issuance_gaps）；fail 判节点失败；flag 只记录。审批卡（human.py 的 + _reviewed_reports）附上审稿概况 + +上限：句数、每句查询次数、每句时长、全局每日金额（核查和裁判各记各的账，engine/spend.py）。都按发起时快照的核查 +设置（run.options["verify"]），节点上写了的盖过它：审批停下期间改了设置，恢复、接着跑仍按发起时的那份。没有快照 +的运行按设置的当前值。全局关闭且节点没写「启用」时直通；报告没有可审的结论句也直通。 + +逐句串行反查：并行的话几句的工具调用在时间线上交错，看不出哪条查询在审哪一句；每日余额按先后扣，结果也才确定。 +整个反查连同存审稿记录、发日志都在一个 @task 里：接着跑时取 checkpoint 里的结果，不重复查库、不重复记账，时间线 +上也不再发一遍。 """ from __future__ import annotations +from collections.abc import Callable, Iterable from typing import Any -from app.core.events import EventType -from app.engine.context import NodeContext -from app.engine.state import GraphState +from langgraph.func import task -#: 占位期间的跳过原因：节点产出和日志里都写这一句 -PENDING_REASON = "反方审稿功能开发中，本节点暂未生效，已跳过" +from app.core import artifact_store +from app.core.errors import describe_exception +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.evidence import ENTITY_KINDS, iter_units, ledger_enabled, next_exec +from app.engine.judge import _usd +from app.engine.merge_query import MERGE_SOURCE +from app.engine.replay import _in_graph +from app.engine.schema import CHALLENGE_LABEL, CHALLENGE_ON_REFUTED_LABEL, GraphSpec, NodeType, challenge_problems, \ + challenge_report +from app.engine.state import GraphState +from app.engine.verify_agent import RATIONALE, VerifyOutcome, run_verify_agent +from app.tools.datasource import run_versions + +#: 四种结论：经受住 / 被削弱 / 被推翻 / 无法判断。counts 的键照这个顺序 +VERDICTS = ("stands", "weakened", "refuted", "inconclusive") +#: 节点产出、审批卡里最多列几句（审稿记录里一句不少） +ITEMS_MAX = 10 +#: 出具缺口、报错里最多点几句的原文，每句最多几个字 +_NAMED = 3 +_QUOTE_CHARS = 40 +#: 审稿记录里每次查询记的几项 +_QUERY_KEYS = ("alias", "artifact", "sql", "row_count", "truncated", "source") +#: 每句反查都会遇到的上限：触顶时发 challenge_limit。步数上限（verify_agent 的 max_steps）由核查自己的 +#: verify_limit 说明，不在这里另记 +_CLAIM_LIMITS = ("max_queries", "timeout_s", "daily_max_usd") +#: 列「审批卡上的几句」时谁先留：被推翻、被削弱、没能判断的比经受住的更要让审批人看到 +_RANK = {"refuted": 0, "weakened": 1, "inconclusive": 2, "stands": 3} + +CHALLENGE_SYSTEM = ( + "你是反方审稿人：报告已经写完，你站在反方，专门找其中一句结论的反例。你不写报告，也不改报告。\n" + "- 先想清楚：如果这句结论不成立，数据里会看到什么；再写只读查询去找这样的反例\n" + "- 换一个角度核对：按别的维度拆开、换时间范围、核对口径和过滤条件（退款、作废、重复记录）、找被忽略的数据，" + "不要只是重复写作者的查询\n" + "- verdict 四选一：stands 认真找过、没有找到反例；weakened 大体成立，但有重要的例外或前提，或者把范围、程度、" + "原因说过了头；refuted 查到的数据和这句结论相矛盾；inconclusive 查不到判断所需的数据\n" + "- 没有找到反例不等于结论被推翻：只有数据和结论冲突时才判 refuted" +) +CHALLENGE_TASK_PROMPT = ( + "要审的结论（报告「{report}」中的一句{section}）：\n「{claim}」\n\n" + "写作者给这句挂的证据摘录（编号属于原报告；你的查询另起编号，理由里只能引用你自己的查询）：\n{evidence}\n\n" + "可以查询的数据源：{sources}\n\n" + "按上面的做法反查，最后交结论（verdict）和理由（rationale）：理由说清查了什么、看到了什么，里面的数字一律写成" + "对你自己查询结果的引用，不要直接写数字。" +) +#: 交给核查 agent 的结果结构：结论一项；理由由 verify_agent 自动加上（rationale） +CHALLENGE_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": {"verdict": { + "type": "string", "enum": list(VERDICTS), + "description": "stands 经受住、weakened 被削弱、refuted 被推翻、inconclusive 无法判断", + }}, + "required": ["verdict"], +} + + +# -------------------------------------------------------------------------- +# 执行器 +# -------------------------------------------------------------------------- async def run_challenge(state: GraphState, ctx: NodeContext) -> dict[str, Any]: - ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON) - return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}} + spec, node = ctx.run.spec, ctx.node + # 节点自己的配置直接读 node.config，不走 ctx.cfg：图级 defaults 里的 provider / model 是写作模型的， + # 拿来当反方模型就成了自己审自己 + config = node.config + # 画布校验也拦这几项；执行时再守一道，报的是同一句话（跳过校验执行的旧图、改坏了的配置） + if problems := challenge_problems(config): + raise NodeError(node.id, "；".join(message for _, message in problems)) + report_id, problem = challenge_report(spec, node) + if report_id is None: + raise NodeError(node.id, problem or "反方审稿找不到要审的报告") + title = spec.node_map()[report_id].title + + settings = await _verify_settings(ctx) + part = settings.get("challenge") if isinstance(settings.get("challenge"), dict) else {} + enabled = config.get("enabled") + if not (part.get("enabled", True) if enabled is None else enabled): + from app.api.settings import VERIFY_SETTING_LABEL + + return _skip(ctx, "disabled", f"反方审稿已关闭（设置中的「{VERIFY_SETTING_LABEL['challenge.enabled']}」，" + f"或节点的「{CHALLENGE_LABEL['enabled']}」），本节点已跳过") + written = (state.get("nodes") or {}).get(report_id) + written = written if isinstance(written, dict) else {} + doc_artifact = written.get("doc_artifact") if isinstance(written.get("doc_artifact"), str) else None + doc = _load_doc(doc_artifact) + if doc is None or doc_artifact is None: + return _skip(ctx, "no_report", f"报告「{title}」的报告文档无法读取（没有写出，或保存失败），本节点已跳过") + claims = pick_claims(doc) + if not claims: + return _skip(ctx, "no_claims", f"报告「{title}」中没有可审的结论句，本节点已跳过") + + max_claims = _limit(config, part, "max_claims") + max_queries = _limit(config, part, "max_queries") + timeout_s = _limit(config, part, "timeout_s") + catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} + from app.api.settings import verify_model_spec # 在这里导入：api.settings 导入了 engine 里的模块 + + async with SessionLocal() as session: + model_spec = await verify_model_spec(session, config, settings=settings) + review = _Review( + ctx, report_id=report_id, title=title, doc_artifact=doc_artifact, doc=doc, catalog=catalog, claims=claims, + max_claims=None if max_claims is None else int(max_claims), + max_queries=None if max_queries is None else int(max_queries), + timeout_s=None if timeout_s is None else float(timeout_s), daily_max_usd=settings.get("daily_max_usd"), + spec=model_spec, masks=await judging.source_masks(catalog), + writer_model=written.get("model") if isinstance(written.get("model"), str) else None, + ) + if _in_graph(): + @task + async def challenge_step() -> dict[str, Any]: + # 反查进 checkpoint：接着跑时直接取这里的结果，不再查库、不再记账（每日账也记在这里面）； + # 事件都在这里面发，取缓存时不会再发一遍 + return await review.run() + + outcome = await challenge_step() + else: + outcome = await review.run() + return _finish(ctx, state, outcome, on_refuted=str(config.get("on_refuted") or "degrade")) + + +def _skip(ctx: NodeContext, reason: str, message: str) -> dict[str, Any]: + """直通：不改任何状态，记一条 challenge_skipped。审批卡认不出审稿结果（没有 counts），不会附上概况。""" + ctx.emit(EventType.LOG, level="info", code="challenge_skipped", reason=reason, message=message) + return {"nodes": {ctx.node.id: {"skipped": True, "reason": reason}}} + + +async def _verify_settings(ctx: NodeContext) -> dict[str, Any]: + """核查设置：发起时快照进运行选项的那份（run.options["verify"]），补齐默认、去掉坏值；没有快照的运行按设置的 + 当前值。""" + from app.api.settings import verify_settings, verify_values + from app.engine.runner import RUN_OPTIONS_KEY + + options = ctx.run.extra.get(RUN_OPTIONS_KEY) + snapshot = options.get("verify") if isinstance(options, dict) else None + if isinstance(snapshot, dict): + return verify_values(snapshot) + async with SessionLocal() as session: + return await verify_settings(session) + + +def _limit(config: dict[str, Any], part: dict[str, Any], key: str) -> Any: + """节点上写了的上限盖过设置；留空（不写、写 null）沿用设置，设置里的 null 才是不限。""" + value = config.get(key) + return part.get(key) if value in (None, "") else value + + +def _load_doc(artifact: str | None) -> dict[str, Any] | None: + if not artifact: + return None + try: + doc = artifact_store.load(artifact) + except ValueError: # 和哈希对不上：被改过的文档不审 + return None + return doc if isinstance(doc, dict) else None + + +def pick_claims(doc: dict[str, Any]) -> list[judging.Candidate]: + """要审的结论句：裁判预筛排好的候选（judge.candidates），去掉裁判已经判成「不是结论」的。""" + units = {u.get("id"): u for _, u in iter_units(doc)} + out: list[judging.Candidate] = [] + for cand in judging.candidates(doc): + verdict = (units.get(cand.unit) or {}).get("verdict") + if isinstance(verdict, dict) and verdict.get("status") == "not_a_claim": + continue + out.append(cand) + return out + + +# -------------------------------------------------------------------------- +# 数据源：这句引用的证据背后是哪几个库 +# -------------------------------------------------------------------------- + + +def _fetcher() -> Callable[[Any], Any]: + """按工件 id 取工件（复验哈希），同一份只读一次；取不到、被改过的当没有。""" + memo: dict[str, Any] = {} + + def fetch(artifact: Any) -> Any: + key = str(artifact or "") + if key not in memo: + try: + memo[key] = artifact_store.load(key) if key else None + except Exception: # noqa: BLE001 - 取不回来就当没有：少一个源，不让整个审稿失败 + memo[key] = None + return memo[key] + + return fetch + + +def _snapshot_sources(snapshot: Any) -> list[str]: + """一份查询快照背后的数据源：合并查询是它合并的那几个源，其余是快照记的源。""" + if not isinstance(snapshot, dict): + return [] + if snapshot.get("source") == MERGE_SOURCE: + return [str(s) for s in snapshot.get("sources") or [] if s] + return [str(snapshot["source"])] if snapshot.get("source") else [] + + +def claim_sources(aliases: Iterable[str], catalog: dict[str, Any], fetch: Callable[[Any], Any]) -> list[str]: + """这几件证据背后的数据源（去重保序）：查询的、口径卡指标的输入取自的查询的、表和字段的。知识库检索、运行输入 + 背后没有数据源。认法同裁判找遮罩（judge.source_masks）。""" + out: list[str] = [] + for alias in aliases: + entry = catalog.get(alias) + if not isinstance(entry, dict): + continue + kind = entry.get("kind") + found: list[str] = [] + if kind == "query": + source = entry.get("source") + found = [str(source)] if source and source != MERGE_SOURCE else _snapshot_sources(fetch(entry.get("artifact"))) + elif kind == "metric": + card = fetch(entry.get("artifact")) + wanted = (entry.get("locator") or {}).get("metric") + for metric in (card.get("metrics") if isinstance(card, dict) else None) or []: + if isinstance(metric, dict) and metric.get("id") == wanted: + for item in metric.get("inputs") or []: + if isinstance(item, dict) and item.get("artifact"): + found += _snapshot_sources(fetch(item["artifact"])) + elif kind in ENTITY_KINDS: + source = entry.get("source") + found = [str(source)] if source else next( + (_snapshot_sources(fetch(o.get("artifact"))) for o in entry.get("sources") or [] + if isinstance(o, dict) and o.get("kind") == "schema"), []) + out.extend(s for s in found if s not in out) + return out + + +# -------------------------------------------------------------------------- +# 逐句反查 +# -------------------------------------------------------------------------- + + +def _zero_usage() -> dict[str, Any]: + return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0, "unpriced_calls": 0} + + +def _add_usage(total: dict[str, Any], more: dict[str, Any]) -> dict[str, Any]: + out = dict(total) + for key, value in (more or {}).items(): + if isinstance(value, (int, float)) and not isinstance(value, bool) and key in out: + out[key] += value + out["cost_usd"] = round(out["cost_usd"], 6) + return out + + +def _claim(cand: judging.Candidate, verdict: str, rationale: str, *, doc: dict[str, Any] | None = None, + queries: Iterable[dict[str, Any]] = (), usage: dict[str, Any] | None = None) -> dict[str, Any]: + """审稿记录里的一句。rationale 是渲染后的正文；系统替它说的（没有可查的源、没能反查）rationale_doc 为 None。""" + return {"unit": cand.unit, "text": cand.text, "verdict": verdict, "rationale": rationale, "rationale_doc": doc, + "queries": [{k: q.get(k) for k in _QUERY_KEYS} for q in queries], "usage": usage or _zero_usage()} + + +def _claim_of(cand: judging.Candidate, outcome: VerifyOutcome) -> dict[str, Any]: + if outcome.result is None: + # 没交成结果（模型不可用、调用失败、触顶时连收尾都没赶上）：查过的照记，结论只能是无法判断 + return _claim(cand, "inconclusive", f"反查未完成：{outcome.error or '原因不明'}", queries=outcome.queries, + usage=outcome.usage) + verdict = str(outcome.result.get("verdict") or "").strip().lower() + return _claim(cand, verdict if verdict in VERDICTS else "inconclusive", str(outcome.result.get(RATIONALE) or ""), + doc=outcome.rationale, queries=outcome.queries, usage=outcome.usage) + + +def _named(claims: list[dict[str, Any]]) -> str: + """「「…」、「…」」：点名最多 _NAMED 句，每句截到 _QUOTE_CHARS 个字。""" + names = "、".join(f"「{judging.clip_text(c['text'], _QUOTE_CHARS)}」" for c in claims[:_NAMED]) + return names + ("等" if len(claims) > _NAMED else "") + + +def _refuted_words(claims: list[dict[str, Any]]) -> str: + """出具缺口、时间线、报错里的那句：「反方审稿推翻了 N 句结论：「…」」。不带节点名：出具契约统一加前缀。""" + return f"反方审稿推翻了 {len(claims)} 句结论：{_named(claims)}" + + +def pick_items(claims: list[dict[str, Any]]) -> list[dict[str, Any]]: + """节点产出、审批卡里列的几句 {unit, text, verdict}：最多 ITEMS_MAX 句，照原来的顺序。多出来的时候先留被推翻、 + 被削弱、没能判断的，再按先后——被推翻的那句不能因为排得靠后就不给审批人看。""" + keep = range(len(claims)) if len(claims) <= ITEMS_MAX else \ + sorted(sorted(range(len(claims)), key=lambda i: (_RANK.get(claims[i]["verdict"], len(_RANK)), i))[:ITEMS_MAX]) + return [{k: claims[i][k] for k in ("unit", "text", "verdict")} for i in keep] + + +class _Review: + """一次审稿：逐句派核查 agent，写审稿记录，发日志。run() 的返回值进 checkpoint，只放节点产出用得到的。""" + + def __init__(self, ctx: NodeContext, *, report_id: str, title: str, doc_artifact: str, doc: dict[str, Any], + catalog: dict[str, Any], claims: list[judging.Candidate], max_claims: int | None, + max_queries: int | None, timeout_s: float | None, daily_max_usd: float | None, spec: Any, + masks: dict[str, Any], writer_model: str | None) -> None: + self.ctx = ctx + self.report_id, self.title, self.doc_artifact, self.doc, self.catalog = \ + report_id, title, doc_artifact, doc, catalog + self.total = len(claims) + self.picked = claims if max_claims is None else claims[:max_claims] + self.max_claims, self.max_queries, self.timeout_s, self.daily_max_usd = \ + max_claims, max_queries, timeout_s, daily_max_usd + self.spec, self.masks, self.writer_model = spec, masks, writer_model + self.fetch = _fetcher() + + def _limit_words(self, limit: str) -> str: + if limit == "max_claims": + return f"每份报告最多审 {self.max_claims} 句" + if limit == "max_queries": + return f"每句最多查询 {self.max_queries} 次" + if limit == "timeout_s": + return f"每句时长上限 {self.timeout_s:g} 秒" + return f"全局每日核查金额上限 {_usd(self.daily_max_usd)}" + + def _task(self, cand: judging.Candidate, sources: list[str]) -> str: + """交给核查 agent 的任务：这句结论、它挂的证据摘录（遮罩照旧）、可以查的数据源。摘录开头的编号改成 + 「原报告 Qn」：agent 自己的查询也编 Q1、Q2，不分开的话理由里引用的会是另一份快照的格子。""" + request = judging.prepare(self.doc, self.catalog, units=[cand.unit], loader=self.fetch, masked=self.masks) + aliases = dict.fromkeys(a for c in request.cands for a in c.aliases if a in request.excerpts) + evidence = "\n\n".join(request.excerpts[a].replace(f"【{a}】", f"【原报告 {a}】", 1) for a in aliases) + return CHALLENGE_TASK_PROMPT.format( + report=self.title, section=f"，所在小节「{cand.section}」" if cand.section else "", claim=cand.text, + evidence=evidence or "（这句没有挂依据）", sources="、".join(sources)) + + async def run(self) -> dict[str, Any]: + ctx = self.ctx + if len(self.picked) < self.total: + skipped = self.total - len(self.picked) + ctx.emit(EventType.LOG, level="warn", code="challenge_limit", limit="max_claims", claims_skipped=skipped, + message=f"报告「{self.title}」有 {self.total} 句可审的结论，已达上限（{self._limit_words('max_claims')}），" + f"{skipped} 句未审") + # 这句没挂依据时在整份报告引用的数据源里查：写作者看得到的就是这几个库 + everywhere = claim_sources(self.catalog, self.catalog, self.fetch) + claims: list[dict[str, Any]] = [] + usage = _zero_usage() + model: str | None = None + warned = False + hit: dict[str, list[int]] = {} # 上限 → [触顶的句数, 因此没能给出结论的句数] + broke = False # 全局每日金额用完了：后面的句子不再派 agent + for cand in self.picked: + if broke: + hit["daily_max_usd"][1] += 1 + claims.append(_claim(cand, "inconclusive", f"已达上限（{self._limit_words('daily_max_usd')}），这句未反查")) + continue + try: + sources = claim_sources(cand.aliases, self.catalog, self.fetch) or everywhere + task_text = self._task(cand, sources) if sources else "" + except Exception as e: # noqa: BLE001 - 一句的摘录出了意外（坏形状的工件），只让这一句没审成 + claims.append(_claim(cand, "inconclusive", f"反查未完成：{describe_exception(e)}")) + continue + if not sources: + claims.append(_claim(cand, "inconclusive", + "这份报告引用的证据中没有数据库查询，没有可以反查的数据源，这句未反查")) + continue + outcome = await run_verify_agent( + spec=self.spec, system=CHALLENGE_SYSTEM, task=task_text, sources=sources, + output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=self.timeout_s, + daily_max_usd=self.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id, + data_versions=run_versions(ctx.run)) + model = model or outcome.model + if not warned and outcome.model and outcome.model == self.writer_model: + warned = True + ctx.emit(EventType.LOG, level="warn", code="challenge_same_model", model=outcome.model, + message=f"反方模型与报告的写作模型相同（「{outcome.model}」），难以发现写作模型自身的错误。" + f"请在设置的「核查」一节选择另一个模型，或在节点的「{CHALLENGE_LABEL['model']}」中指定") + claims.append(_claim_of(cand, outcome)) + usage = _add_usage(usage, outcome.usage) + if outcome.limit in _CLAIM_LIMITS: + counted = hit.setdefault(outcome.limit, [0, 0]) + counted[0] += 1 + counted[1] += outcome.result is None + broke = outcome.limit == "daily_max_usd" + + counts = {v: sum(1 for c in claims if c["verdict"] == v) for v in VERDICTS} + artifact = await self._store({"report": self.report_id, "doc_artifact": self.doc_artifact, "claims": claims, + "model": model}) + for limit, (touched, skipped) in hit.items(): + words = self._limit_words(limit) + message = f"已达上限（{words}），{skipped} 句未反查" if limit == "daily_max_usd" else \ + f"有 {touched} 句反查已达上限（{words}），已按查到的结果收尾" + \ + (f"，其中 {skipped} 句未能给出结论" if skipped else "") + ctx.emit(EventType.LOG, level="warn", code="challenge_limit", limit=limit, claims_skipped=skipped, + message=message) + ctx.emit(EventType.LOG, level="info", code="challenge_done", counts=counts, artifact=artifact, + message=f"反方审稿完成：审了 {len(claims)} 句结论，经受住 {counts['stands']} 句、被削弱 " + f"{counts['weakened']} 句、被推翻 {counts['refuted']} 句、无法判断 {counts['inconclusive']} 句") + if refuted := [c for c in claims if c["verdict"] == "refuted"]: + ctx.emit(EventType.LOG, level="warn", code="challenge_refuted", + items=[{"unit": c["unit"], "text": c["text"]} for c in refuted[:ITEMS_MAX]], + message=_refuted_words(refuted)) + return {"artifact": artifact, "counts": counts, "usage": usage, + "claims": [{k: c[k] for k in ("unit", "text", "verdict")} for c in claims]} + + async def _store(self, record: dict[str, Any]) -> str | None: + try: + return await artifact_store.put_json(record, kind="challenge", run_id=self.ctx.run.run_id, + node_id=self.ctx.node.id, meta={"report": self.report_id}) + except Exception as e: # noqa: BLE001 - 记录写失败不毁掉运行：结论照常处置，只是理由和查询点不开 + self.ctx.emit(EventType.LOG, level="warn", code="evidence_store_failed", + message=f"反方审稿记录保存失败（{describe_exception(e)}），审稿的理由和查询将无法查看") + return None + + +def _finish(ctx: NodeContext, state: GraphState, outcome: dict[str, Any], *, on_refuted: str) -> dict[str, Any]: + """反查的结果 → 节点产出；被推翻的按 on_refuted 处置。重放时 outcome 取自 checkpoint，算出来的一样。""" + claims = outcome.get("claims") or [] + refuted = [c for c in claims if c.get("verdict") == "refuted"] + if refuted and on_refuted == "fail": + raise NodeError(ctx.node.id, f"{_refuted_words(refuted)}。请核对这几句结论；如只需记录，可将" + f"「{CHALLENGE_LABEL['on_refuted']}」改为「{CHALLENGE_ON_REFUTED_LABEL['flag']}」") + result: dict[str, Any] = {"challenge_artifact": outcome.get("artifact"), "counts": outcome.get("counts") or {}, + "items": pick_items(claims), "on_refuted": on_refuted} + if refuted and on_refuted == "degrade": + result["issuance_gaps"] = [_refuted_words(refuted)] + updates: dict[str, Any] = {"nodes": {ctx.node.id: result}} + spent = outcome.get("usage") or {} + if spent.get("calls"): + updates["usage"] = {k: spent.get(k, 0) for k in ("input_tokens", "output_tokens", "total_tokens", "cost_usd", + "calls")} + if outcome.get("artifact") and ledger_enabled(ctx.run): + # 台账只记审稿记录本身（kind challenge）：它的查询不进写作目录，报告按台账编 Q1… 时看不到它们 + updates["evidence"] = [{"kind": "challenge", "node_id": ctx.node.id, "exec": next_exec(state, ctx.node.id), + "artifact": outcome["artifact"]}] + return updates + + +# -------------------------------------------------------------------------- +# 审批卡 +# -------------------------------------------------------------------------- + + +def challenge_summary(spec: GraphSpec, outputs: dict[str, Any], report_id: str) -> dict[str, Any] | None: + """审批卡上这份报告的反方审稿概况 {counts, items, artifact, on_refuted}：审它的、已经审完的反方审稿节点给的。 + 没有（没加反方审稿、还没审到、直通了）返回 None，审批卡不加这个键。几个反方审稿节点审同一份报告时取画布上 + 靠后的那个。只看图和状态：重放时算出来的一样。""" + found: dict[str, Any] | None = None + for node in spec.nodes: + result = outputs.get(node.id) + if node.type != NodeType.CHALLENGE or not isinstance(result, dict) or not isinstance(result.get("counts"), dict): + continue + if challenge_report(spec, node)[0] != report_id: + continue + found = {"counts": dict(result["counts"]), "items": list(result.get("items") or [])[:ITEMS_MAX], + "artifact": result.get("challenge_artifact"), "on_refuted": result.get("on_refuted") or "degrade"} + return found + + +__all__ = ["CHALLENGE_SYSTEM", "ITEMS_MAX", "VERDICTS", "challenge_summary", "claim_sources", "pick_claims", + "pick_items", "run_challenge"] diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index f314011b..8d55ea36 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -112,7 +112,9 @@ def _reviewed_reports(state: GraphState, ctx: NodeContext) -> list[dict[str, Any 真实运行里报告有一句结论和证据相矛盾，审批人批准时完全不知道：审批卡只有正文。这里带上报告节点产出里的 stats（同 report.checked）、结论句策略、裁判摘要和报告文档的工件 id，审批卡据此写核对概况、给打开证据的入口。 - 只认已经跑完、带着核对统计的报告；重放时 state 一样，算出来的也一样。""" + 审这份报告的反方审稿已经审完的，再带上它的概况（challenge：几句被推翻、被削弱，打开审稿记录的入口）；没有就 + 不加这个键。只认已经跑完、带着核对统计的报告；重放时 state 一样，算出来的也一样。""" + from app.engine.nodes.challenge import challenge_summary from app.engine.schema import NodeType spec = ctx.run.spec @@ -134,6 +136,8 @@ def _reviewed_reports(state: GraphState, ctx: NodeContext) -> list[dict[str, Any item["doc_artifact"] = result["doc_artifact"] if isinstance(result.get("judge"), dict): item["judge"] = {k: result["judge"][k] for k in _JUDGE_KEYS if k in result["judge"]} + if (challenge := challenge_summary(spec, outputs, rid)) is not None: + item["challenge"] = challenge out.append(item) return out diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index d3ae6325..58d8a337 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -613,6 +613,111 @@ def _check_report_from(node: GraphNode, report_from: Any, spec: GraphSpec, node_id=node.id, field=field, code=code) +# -------------------------------------------------------------------------- +# 反方审稿节点（engine/nodes/challenge.py）：审哪份报告、各项配置 +# -------------------------------------------------------------------------- + +#: 反方审稿节点各项配置在界面上的叫法：报错写叫法，不写键名 +CHALLENGE_LABEL = { + "report": "要审的报告", "max_claims": "最多审几句", "max_queries": "每句最多查询次数", + "timeout_s": "每句时长上限（秒）", "provider": "反方模型 · 接入", "model": "反方模型", + "on_refuted": "结论被推翻时", "enabled": "启用", +} +#: 结论被推翻时怎么办：degrade 计入出具缺口、出具降档（默认）/ flag 只记录 / fail 判节点失败 +CHALLENGE_ON_REFUTED = ("degrade", "flag", "fail") +CHALLENGE_ON_REFUTED_LABEL = {"degrade": "出具降档", "flag": "仅记录", "fail": "判为失败"} + + +def _challenge_field(key: str) -> str: + return f"「{type_label(NodeType.CHALLENGE)}」的「{CHALLENGE_LABEL[key]}」" + + +def challenge_problems(config: dict[str, Any]) -> list[tuple[str, str]]: + """反方审稿各项配置有什么问题（要审的报告另见 challenge_report）：[(字段, 人话)]，没问题返回 []。 + 校验和节点执行时报的是同一句话。 + + 三个上限留空（不写或写 null）就是沿用设置里「核查」一节的值，和「启用」留空一个意思。节点上没有「不限」的写法： + 要放开上限去设置里改，免得一个节点悄悄把核查的花费放开了。 + """ + out: list[tuple[str, str]] = [] + for key, words in (("max_claims", "每份报告最多反查的句数"), ("max_queries", "每句最多执行的查询次数")): + value = config.get(key) + if value not in (None, "") and not _positive(value, integer=True): + out.append((key, f"{_challenge_field(key)}需要填写正整数（{words}），留空表示沿用设置，当前为「{value}」")) + value = config.get("timeout_s") + if value not in (None, "") and not _positive(value): + out.append(("timeout_s", f"{_challenge_field('timeout_s')}需要填写大于 0 的秒数，留空表示沿用设置，" + f"当前为「{value}」")) + for key, what in (("provider", "模型接入的名称"), ("model", "模型 ID")): + value = config.get(key) + if value is not None and not isinstance(value, str): + out.append((key, f"{_challenge_field(key)}需要填写{what}，留空表示使用设置中的核查模型，当前为「{value}」")) + value = config.get("on_refuted") + if value not in (None, "") and value not in CHALLENGE_ON_REFUTED: + names = [f"「{CHALLENGE_ON_REFUTED_LABEL[v]}」" for v in CHALLENGE_ON_REFUTED] + out.append(("on_refuted", f"{_challenge_field('on_refuted')}只能选{'、'.join(names[:-1])}或{names[-1]}，" + f"当前为「{value}」")) + value = config.get("enabled") + if value is not None and not isinstance(value, bool): + out.append(("enabled", f"{_challenge_field('enabled')}只能是开启、关闭或留空（沿用设置），当前为「{value}」")) + return out + + +def challenge_report(spec: GraphSpec, node: GraphNode) -> tuple[str | None, str | None]: + """反方审稿审的是哪份报告：(报告撰写节点 id, None)；定不下来时 (None, 为什么)。校验、执行器、审批卡都按它认。 + + 写了「要审的报告」就按写的，必须是它上游的报告撰写节点。没写就从上游找：直接连着的报告撰写节点只有一个， + 就是它（「初稿 → 终稿 → 反方审稿」审终稿）；否则整个上游里只有一个报告撰写节点才取它。两份以上说不准审哪一份， + 宁可让人选，也不能悄悄审错一份。 + """ + who, label = f"「{type_label(NodeType.CHALLENGE)}」", CHALLENGE_LABEL["report"] + nodes = spec.node_map() + upstream = _ancestors(spec, node.id) + chosen = node.config.get("report") + if chosen in (None, ""): + direct = list(dict.fromkeys(e.source for e in spec.incoming(node.id) + if (n := nodes.get(e.source)) is not None and n.type == NodeType.REPORT)) + if len(direct) == 1: + return direct[0], None + found = [n for n in spec.nodes if n.id in upstream and n.type == NodeType.REPORT] + if len(found) == 1: + return found[0].id, None + if not found: + return None, (f"{who}的上游没有「报告撰写」节点：反方审稿只审报告撰写节点写出的报告。" + f"请把它接在报告撰写节点之后，或在「{label}」中选择一个") + names = "、".join(f"「{n.title}」" for n in found[:5]) + ("等" if len(found) > 5 else "") + return None, f"{who}的上游有多个「报告撰写」节点（{names}），请在「{label}」中选择要审的那一份" + if not isinstance(chosen, str): + return None, f"{who}的「{label}」需要选择一个「报告撰写」节点，当前为「{chosen}」" + target = nodes.get(chosen) + if target is None: + return None, f"{who}的「{label}」选择了「{chosen}」，但工作流中不存在这个节点" + if target.type != NodeType.REPORT: + return None, (f"{who}的「{label}」选择了「{target.title}」，它是「{type_label(target.type)}」节点，" + "不是「报告撰写」节点：反方审稿只审报告撰写节点写出的报告") + if chosen not in upstream: + return None, (f"{who}的「{label}」选择的「{target.title}」不在它的上游：审稿时这份报告还没有写出。" + "请把报告撰写节点连到反方审稿之前") + return chosen, None + + +def _check_challenge(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None: + """反方审稿的配置写错了挡住运行：执行时节点照样会报这一句，与其取数、写完报告才死在审稿这一步，不如在画图时说破。 + 反方模型和要审的报告的写作模型写成同一个的给警告：等于自己审自己。""" + for field, message in challenge_problems(node.config): + result.add(message, node_id=node.id, field=field, code="challenge.config_invalid") + report, problem = challenge_report(spec, node) + if problem or report is None: + result.add(problem or "", node_id=node.id, field="report", code="challenge.report_invalid") + return + writer = spec.node_map()[report] + model = node.config.get("model") + if isinstance(model, str) and model and model == (writer.config.get("model") or spec.defaults.get("model")): + result.add(f"反方模型与报告「{writer.title}」的写作模型相同（「{model}」），等于自己审自己，难以发现写作模型自身" + f"的错误。请在「{CHALLENGE_LABEL['model']}」中选择另一个模型", level="warning", node_id=node.id, + field="model", code="challenge.same_model") + + def validate_graph(spec: GraphSpec) -> ValidationResult: """编译前的静态检查。错误会挡住运行，警告只在画布上提示。""" result = ValidationResult() @@ -679,6 +784,8 @@ def validate_graph(spec: GraphSpec) -> ValidationResult: _check_report_claims(node, spec, result) elif node.type == NodeType.MERGE: _check_merge(node, spec, result) + elif node.type == NodeType.CHALLENGE: + _check_challenge(node, spec, result) elif node.type == NodeType.OUTPUT: contract = cfg.get("contract") if contract and not isinstance(contract, dict): diff --git a/backend/tests/test_challenge_node.py b/backend/tests/test_challenge_node.py new file mode 100644 index 00000000..7483cf00 --- /dev/null +++ b/backend/tests/test_challenge_node.py @@ -0,0 +1,807 @@ +"""反方审稿节点（engine/nodes/challenge.py）：报告写完后挑几句关键结论，各派一个只读 agent 站在反方找反例。 + +- 选句照结论句裁判的预筛（judge.candidates / claim_priority）：挂了依据的、带比较或因果的排前面，取前「最多审几句」句 +- 每句一个只读 agent（engine/verify_agent.py）：工具只给这句引用的查询所在的数据源，任务里给这句结论和它挂的证据摘录 + （judge.prepare，遮罩照旧）；理由里的数字只能引用它自己的查询格子，裸数字被标出来 +- 结论四选一：经受住 / 被削弱 / 被推翻 / 无法判断；完整记录写进 challenge_artifact，随运行封存；台账只记 kind challenge， + 它的查询不进任何报告的写作目录 +- 被推翻的：on_refuted 为 degrade（默认）时写 issuance_gaps，出具降档；fail 判节点失败；flag 只记录 +- 上限：句数、每句查询数、每句时长、全局每日金额；全局关闭、报告没有可审的结论句时直通 +- 整个反查放在一个 @task 里：接着跑时不重复执行、不重复记账 +- 人工审批审这份报告时，审批卡附上反方审稿的概况 +- 画布校验：要审的报告必须是上游的报告撰写节点 + +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、电话都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 +import uuid +from dataclasses import dataclass, field + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from sqlalchemy import delete, select + +from app.core import artifact_store +from app.data import introspect +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import Approval, DataSource, Run, RunEvent, Setting +from app.engine import spend, verify_agent +from app.engine.context import NodeContext, NodeError, RunContext +from app.engine.evidence import iter_units +from app.engine.judge import SPEND_KEY as JUDGE_SPEND_KEY +from app.engine.nodes import challenge as challenge_mod +from app.engine.nodes import report as report_mod +from app.engine.runner import run_manager, verify_manifest +from app.engine.schema import GraphSpec, validate_graph +from app.providers.factory import ProviderNotConfigured +from app.providers.mock_model import MockChatModel + +PRICED = "claude-sonnet-5" +#: 各区域销售额（含退款）：华东 1500、华南 900、华北 400 +QUERY = "SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales GROUP BY region ORDER BY 销售额 DESC" +#: 一句有依据的结论、一句预埋的错误结论（华南其实低于华东） +TEXT = "华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]华南销售额超过华东。[[see:Q1]]" +EAST, SOUTH = "华东销售额 1,500，居各区域之首。", "华南销售额超过华东。" +NET = "SELECT region AS 区域, SUM(amount) AS 净销售额 FROM sales WHERE refunded = 0 GROUP BY region ORDER BY 净销售额 DESC" +REFUNDS = "SELECT COUNT(*) AS 退款笔数 FROM sales WHERE refunded = 1" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_( + [spend.SPEND_KEY, JUDGE_SPEND_KEY, "verify", "judge", "copilot"]))) + await session.commit() + yield + await run_manager.shutdown() + + +async def _source(tmp_path, *, options=None) -> tuple[str, int]: + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript( + "CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, amount INTEGER, refunded INTEGER);" + "INSERT INTO sales VALUES (1, '华东', '13900000001', 1200, 0), (2, '华东', '13900000002', 300, 1)," + " (3, '华南', '13900000003', 900, 0), (4, '华北', '13900000004', 400, 0);") + db.commit() + db.close() + name = f"shop_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=options or {}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + return name, row.id + + +@pytest.fixture +async def shop(tmp_path): + """只读的手工源：四笔订单，华东两笔（其中一笔 300 已退款）、华南一笔 900、华北一笔 400。返回源名。""" + name, source_id = await _source(tmp_path) + yield name + await engines.invalidate(source_id) + + +@pytest.fixture +async def masked_shop(tmp_path): + """同样的库，买家电话按数据源的设置遮罩。""" + name, source_id = await _source(tmp_path, options={"mask_columns": ["buyer_phone"]}) + yield name + await engines.invalidate(source_id) + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def chain(*nodes, extra_edges=()): + edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])] + return {"nodes": list(nodes), "edges": [*edges, *extra_edges]} + + +def graph(source: str, *, sql: str = QUERY, contract: bool = True, **challenge) -> dict: + """取数 → 报告 → 反方审稿 → 成果。成果按报告文档复核（引用模式的出具契约），数字都有出处：不加反方审稿是完整出具。""" + challenge.setdefault("report", "write") + return chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿", **challenge), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + **({"contract": {"report_from": "write"}} if contract else {})), + ) + + +@dataclass +class Play: + """反查 agent 对某一句的剧本：依次执行的查询，最后交的结论和理由。""" + + queries: list[str] = field(default_factory=list) + verdict: str = "stands" + rationale: str = "没有找到反例。" + + +#: 默认剧本：错误结论被推翻（扣除退款后华南仍低于华东），首句被削弱（华东含退款） +PLAYS = { + "华南销售额超过华东": Play([NET], "refuted", "扣除退款后华南为 [[v:Q1.r1.净销售额]]，仍低于华东的 [[v:Q1.r0.净销售额]]。[[see:Q1]]"), + "居各区域之首": Play([REFUNDS], "weakened", "华东含 [[v:Q1.r0.退款笔数]] 笔退款，扣除后仍居首，但销售额被高估。"), +} + + +class Models: + """写作者和反查 agent 两份剧本。写作者依次回 writer 里的几段；反查 agent 按任务里的结论认出是哪一句，照 plays 查询、 + 交结果。calls 记下反查 agent 的每次调用：(step / final, 认出的句子片段, 收到的消息)。""" + + def __init__(self, monkeypatch, source: str, *, writer=(TEXT,), plays=None, model_id=PRICED, latency=0.0, + broken: Exception | None = None): + self.calls: list[tuple[str, str | None, list]] = [] + self.writes: list[list] = [] + #: 反查 agent 每句拿到的工具：{句子片段: 工具名集合} + self.tools: dict[str | None, set[str]] = {} + plays = PLAYS if plays is None else plays + + def which(messages) -> str | None: + task = str(messages[1].content) if len(messages) > 1 else "" + return next((k for k in plays if k in task), None) + + def decide(model, messages): + system = str(messages[0].content) + if challenge_mod.CHALLENGE_SYSTEM in system: + key = which(messages) + play = plays.get(key) or Play() + if model.response_format: + self.calls.append(("final", key, list(messages))) + return AIMessage(content=json.dumps({"verdict": play.verdict, "rationale": play.rationale}, + ensure_ascii=False)) + self.calls.append(("step", key, list(messages))) + self.tools.setdefault(key, set()).update(t["name"] for t in model.tools) + done = sum(1 for m in messages if isinstance(m, ToolMessage)) + if done < len(play.queries): + call = {"name": f"db_query__{source}", "args": {"sql": play.queries[done]}, + "id": f"{(key or 'x')[:4]}-{done}"} + return AIMessage(content="查一下。", tool_calls=[call]) + return AIMessage(content="查完了。") + if report_mod.ROLE in system: + self.writes.append(list(messages)) + return AIMessage(content=writer[min(len(self.writes), len(writer)) - 1]) + raise AssertionError(f"没想到的模型调用：{system[:60]}") + + monkeypatch.setattr(MockChatModel, "_decide", decide) + + async def verify_model(session, spec): + if broken is not None: + raise broken + return MockChatModel(model_name=model_id, latency=latency), model_id + + monkeypatch.setattr(verify_agent, "get_chat_model", verify_model) + + def tasks(self) -> dict[str, str]: + """反查 agent 收到的任务：{句子片段: 任务原文}。""" + return {key: str(messages[1].content) for kind, key, messages in self.calls if kind == "step" and key} + + def reviewed(self) -> list[str | None]: + return list(dict.fromkeys(key for _, key, _ in self.calls)) + + +async def finish(spec: dict, *, run_class="formal", options=None, statuses=("succeeded", "failed")) -> Run: + run = await run_manager.start(graph=spec, input_payload={}, run_class=run_class, options=options) + return await wait(run.id, statuses) + + +async def wait(run_id, statuses=("succeeded", "failed")) -> Run: + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return row + raise AssertionError(f"运行 {run_id} 没跑完：{row.status}") + + +async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +async def logs(run_id: str, code: str) -> list[dict]: + return [e.data for e in await events(run_id, "log") if e.data.get("code") == code] + + +async def output_of(run_id: str, node_id: str) -> dict: + finished = await events(run_id, "node.finished", node_id) + return artifact_store.load(finished[-1].data["artifact"]) + + +async def report_doc(run_id: str) -> dict: + [checked] = await events(run_id, "report.checked", "write") + return artifact_store.load(checked.data["doc_artifact"]) + + +def unit_of(doc: dict, text: str) -> str: + md = doc["markdown"] + return next(u["id"] for _, u in iter_units(doc) if md[u["span"][0]:u["span"][1]].strip() == text) + + +COUNT_KEYS = ("stands", "weakened", "refuted", "inconclusive") + + +# -------------------------------------------------------------------------- +# 被推翻 → 出具降档 +# -------------------------------------------------------------------------- + + +async def test_a_refuted_conclusion_degrades_the_issuance(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"], "按优先级、再按文档先后逐句反查" + + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded" + assert issuance["gaps"] == [f"反方审稿「审稿」：反方审稿推翻了 1 句结论：「{SOUTH}」"], "缺口不带节点名，前缀由出具契约加" + + doc = await report_doc(row.id) + east, south = unit_of(doc, EAST), unit_of(doc, SOUTH) + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0} + assert set(out["counts"]) == set(COUNT_KEYS) + assert out["items"] == [{"unit": east, "text": EAST, "verdict": "weakened"}, + {"unit": south, "text": SOUTH, "verdict": "refuted"}] + assert out["on_refuted"] == "degrade" + assert out["issuance_gaps"] == [f"反方审稿推翻了 1 句结论：「{SOUTH}」"] + + record = artifact_store.load(out["challenge_artifact"]) + assert set(record) == {"report", "doc_artifact", "claims", "model"} + assert record["report"] == "write" and record["model"] == PRICED + [checked] = await events(row.id, "report.checked", "write") + assert record["doc_artifact"] == checked.data["doc_artifact"] + first, second = record["claims"] + assert set(second) == {"unit", "text", "verdict", "rationale", "rationale_doc", "queries", "usage"} + assert (second["unit"], second["text"], second["verdict"]) == (south, SOUTH, "refuted") + assert second["rationale"] == "扣除退款后华南为 900，仍低于华东的 1,200。" + assert second["rationale_doc"]["violations"] == [] + [query] = second["queries"] + assert set(query) == {"alias", "artifact", "sql", "row_count", "truncated", "source"} + assert query["alias"] == "Q1" and query["source"] == shop and "refunded = 0" in query["sql"] + assert artifact_store.load(query["artifact"])["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]] + assert query["row_count"] == 3 and query["truncated"] is False + assert second["usage"]["calls"] >= 2 and second["usage"]["input_tokens"] > 0 + assert first["verdict"] == "weakened" and first["rationale"] == "华东含 1 笔退款，扣除后仍居首，但销售额被高估。" + + # 时间线：完成、推翻各一条，按约定带结构化字段 + [done] = await logs(row.id, "challenge_done") + assert done["counts"] == out["counts"] and done["artifact"] == out["challenge_artifact"] + [refuted] = await logs(row.id, "challenge_refuted") + assert refuted["items"] == [{"unit": south, "text": SOUTH}] + # 反查的查询发 tool.end（role: verify，带查询快照），可下钻、随运行封存 + ends = [e.data for e in await events(row.id, "tool.end", "chal")] + assert len(ends) == 2 and all(d["role"] == "verify" and d.get("query_artifact") for d in ends) + assert {d["query_artifact"] for d in ends} == {first["queries"][0]["artifact"], query["artifact"]} + llm_ends = [e.data for e in await events(row.id, "llm.end", "chal")] + assert llm_ends and all(d["role"] == "verify" for d in llm_ends) + # 台账只记 kind challenge，绝不记 query：核查的查询不能被当成报告可以引用的证据 + [finished] = await events(row.id, "node.finished", "chal") + assert finished.data["evidence"] == [{"kind": "challenge", "node_id": "chal", "exec": 1, + "artifact": out["challenge_artifact"]}] + # 节点用量记进运行用量 + assert row.usage["calls"] >= sum(c["usage"]["calls"] for c in record["claims"]) + verdict = await verify_manifest(row.id) + assert verdict["ok"] is True + assert finished.seq <= row.manifest_seq + + +async def test_flag_only_records_and_fail_stops_the_run(monkeypatch, shop): + Models(monkeypatch, shop) + row = await finish(graph(shop, on_refuted="flag")) + assert row.status == "succeeded", row.error + assert row.output["_issuance"]["tier"] == "formal" and row.output["_issuance"]["gaps"] == [] + out = await output_of(row.id, "chal") + assert out["on_refuted"] == "flag" and "issuance_gaps" not in out and out["counts"]["refuted"] == 1 + assert len(await logs(row.id, "challenge_refuted")) == 1, "flag 也照实记一条" + + Models(monkeypatch, shop) + failed = await finish(graph(shop, on_refuted="fail")) + assert failed.status == "failed" + assert failed.error_node_id == "chal" + assert "反方审稿推翻了 1 句结论" in failed.error and SOUTH in failed.error + # 判失败之前审稿记录已经存下、时间线照常有完成那一条 + [done] = await logs(failed.id, "challenge_done") + record = artifact_store.load(done["artifact"]) + assert [c["verdict"] for c in record["claims"]] == ["weakened", "refuted"] + + +async def test_nothing_refuted_means_no_gap(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET], "inconclusive", "查不到判断所需的数据。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert row.output["_issuance"]["tier"] == "formal" + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 1, "weakened": 0, "refuted": 0, "inconclusive": 1} + assert "issuance_gaps" not in out + assert await logs(row.id, "challenge_refuted") == [] + + +async def test_the_verify_queries_never_reach_a_writer_downstream(monkeypatch, shop): + """反方审稿之后再写一份报告：它的证据目录里只有原来的查询（Q1），没有反查 agent 的查询。""" + Models(monkeypatch, shop, writer=(TEXT, "华东销售额 [[v:Q1.r0.销售额]]。[[see:Q1]]")) + spec = chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿", report="write"), + node("again", "report", "复写", instructions="再写一遍"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.again.text }}"}]), + ) + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + [checked] = await events(row.id, "report.checked", "again") + doc = artifact_store.load(checked.data["doc_artifact"]) + queries = [a for a, e in doc["catalog"].items() if e.get("kind") == "query"] + assert queries == ["Q1"] and doc["catalog"]["Q1"]["node_id"] == "fetch" + + +# -------------------------------------------------------------------------- +# 理由里的裸数字 +# -------------------------------------------------------------------------- + + +async def test_bare_numbers_in_the_rationale_are_flagged(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET], "refuted", "华南只有 900，低于华东的 [[v:Q1.r0.净销售额]]。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert south["rationale"] == "华南只有 900，低于华东的 1,200。" + bare = [v for v in south["rationale_doc"]["violations"] if v["code"] == "uncited_number"] + assert [v["text"] for v in bare] == ["900"], "自己写的数字不采信，标成没有出处" + assert south["rationale_doc"]["stats"]["uncited_numbers"] == 1 + + +# -------------------------------------------------------------------------- +# 上限 +# -------------------------------------------------------------------------- + + +async def test_the_claim_limit_reviews_the_first_claims_only(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop, max_claims=1)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + out = await output_of(row.id, "chal") + assert [i["text"] for i in out["items"]] == [EAST] and sum(out["counts"].values()) == 1 + assert row.output["_issuance"]["tier"] == "formal", "没审到那句错误结论：不能凭空降档" + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "max_claims" and limit["claims_skipped"] == 1 + assert "1 句" in limit["message"] + + +async def test_the_settings_snapshot_sets_the_limits_and_the_node_overrides_it(monkeypatch, shop): + """上限按发起时快照的核查设置（run.options["verify"]）；节点上写了的盖过它。""" + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"challenge": {"max_claims": 1}}}) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + + models = Models(monkeypatch, shop) + row = await finish(graph(shop, max_claims=2), options={"verify": {"challenge": {"max_claims": 1}}}) + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"] + assert await logs(row.id, "challenge_limit") == [] + + +async def test_without_a_snapshot_the_current_settings_apply(monkeypatch, shop): + async with SessionLocal() as session: + session.add(Setting(key="verify", value={"challenge": {"max_claims": 1}})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首"] + + +async def test_the_query_limit_applies_to_each_claim(monkeypatch, shop): + plays = {"华南销售额超过华东": Play([NET, REFUNDS], "refuted", "华南为 [[v:Q1.r1.净销售额]]。"), + "居各区域之首": Play([REFUNDS], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays) + row = await finish(graph(shop, max_queries=1), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert [q["alias"] for q in south["queries"]] == ["Q1"], "第二条查询超过每句的查询次数上限，没有执行" + assert south["verdict"] == "refuted" and south["rationale"] == "华南为 900。", "触顶照样按查到的结果收尾" + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "max_queries" and limit["claims_skipped"] == 0 + + +async def test_the_time_limit_applies_to_each_claim_and_still_settles(monkeypatch, shop): + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 2.0) + endless = [f"SELECT COUNT(*) AS n FROM sales WHERE id > {i}" for i in range(40)] + plays = {"华南销售额超过华东": Play(endless, "inconclusive", "时间到了，没能查完。"), + "居各区域之首": Play([], "stands", "没有找到反例。")} + Models(monkeypatch, shop, plays=plays, latency=0.1) + row = await finish(graph(shop, timeout_s=0.4), run_class="exploratory") + assert row.status == "succeeded", row.error + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + south = record["claims"][1] + assert south["verdict"] == "inconclusive" and 1 <= len(south["queries"]) < 20 + limits = await logs(row.id, "challenge_limit") + assert [(lm["limit"], lm["claims_skipped"]) for lm in limits] == [("timeout_s", 0)] + + +async def test_the_daily_cap_stops_the_remaining_claims(monkeypatch, shop): + """当天的核查金额已经用完：一句都不反查，照实记成无法判断，不降档。""" + async with SessionLocal() as session: + session.add(Setting(key=spend.SPEND_KEY, value={"date": spend._today(), "usd": 1.9999, "calls": 3})) + await session.commit() + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"daily_max_usd": 2.0}}) + assert row.status == "succeeded", row.error + assert [kind for kind, _, _ in models.calls] == [], "没钱就不调用模型" + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 2} + record = artifact_store.load(out["challenge_artifact"]) + assert all("$2.00" in c["rationale"] for c in record["claims"]) + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "daily_max_usd" and limit["claims_skipped"] == 2 + assert row.output["_issuance"]["tier"] == "formal" + + +# -------------------------------------------------------------------------- +# 审批卡 +# -------------------------------------------------------------------------- + + +def gated(source: str, **challenge) -> dict: + challenge.setdefault("report", "write") + nodes = [ + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": QUERY}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿", **challenge), + node("gate", "human", "主管审批", mode="approve", title="审核周报", message="{{ nodes.write.text }}"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ] + spec = chain(*nodes[:-1]) + spec["nodes"].append(nodes[-1]) + spec["edges"].append({"source": "gate", "target": "out", "sourceHandle": "approved"}) + return spec + + +async def paused(spec: dict, *, options=None) -> tuple[Run, Approval]: + row = await finish(spec, run_class="exploratory", options=options, statuses=("interrupted", "failed", "succeeded")) + assert row.status == "interrupted", row.error + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where(Approval.run_id == row.id))).scalar_one() + return row, approval + + +async def test_the_approval_card_carries_the_challenge_summary(monkeypatch, shop): + Models(monkeypatch, shop) + row, approval = await paused(gated(shop)) + [report] = approval.payload["reports"] + out = await output_of(row.id, "chal") + assert report["node_id"] == "write" + assert report["challenge"] == {"counts": out["counts"], "items": out["items"], + "artifact": out["challenge_artifact"], "on_refuted": "degrade"} + assert report["challenge"]["counts"]["refuted"] == 1 + # human.requested 事件里也带着：回放时界面一样读得到 + [requested] = await events(row.id, "human.requested", "gate") + assert requested.data["reports"] == approval.payload["reports"] + + +async def test_the_approval_card_shows_what_flag_means(monkeypatch, shop): + Models(monkeypatch, shop) + _, approval = await paused(gated(shop, on_refuted="flag")) + assert approval.payload["reports"][0]["challenge"]["on_refuted"] == "flag" + + +async def test_a_skipped_challenge_adds_nothing_to_the_card(monkeypatch, shop): + Models(monkeypatch, shop) + _, approval = await paused(gated(shop, enabled=False)) + [report] = approval.payload["reports"] + assert "challenge" not in report, "没审就不加这个键：老行为逐字不变" + + +def test_items_keep_the_refuted_ones_when_there_are_many(): + claims = [{"unit": f"u{i}", "text": f"第 {i} 句", "verdict": "stands"} for i in range(14)] + claims[12]["verdict"] = "refuted" + claims[13]["verdict"] = "weakened" + items = challenge_mod.pick_items(claims) + assert len(items) == 10 + assert [i["unit"] for i in items][-2:] == ["u12", "u13"], "最多 10 条：被推翻、被削弱的不能被挤掉，顺序照原来的" + assert [i["unit"] for i in items][:8] == [f"u{i}" for i in range(8)] + + +# -------------------------------------------------------------------------- +# 重放 +# -------------------------------------------------------------------------- + + +async def test_continuing_after_a_failure_does_not_run_the_agents_again(monkeypatch, shop): + models = Models(monkeypatch, shop) + real = challenge_mod._finish + broken = {"on": True} + + def flaky(*args, **kwargs): + if broken["on"]: + raise RuntimeError("落盘时断了") + return real(*args, **kwargs) + + monkeypatch.setattr(challenge_mod, "_finish", flaky) + run = await run_manager.start(graph=graph(shop), input_payload={}, run_class="formal") + row = await wait(run.id) + assert row.status == "failed" and "落盘时断了" in (row.error or ""), row.error + calls = len(models.calls) + assert calls > 0 + before = await spend.daily_spend() + verify_ends = len(await events(run.id, "llm.end", "chal")) + tool_ends = len(await events(run.id, "tool.end", "chal")) + + broken["on"] = False + await run_manager.continue_failed(run.id) + row = await wait(run.id) + assert row.status == "succeeded", row.error + assert len(models.calls) == calls, "接着跑时反查 agent 又执行了一遍：反查没进 checkpoint" + assert await spend.daily_spend() == before, "接着跑时核查的每日账又记了一次" + assert len(await events(run.id, "llm.end", "chal")) == verify_ends + assert len(await events(run.id, "tool.end", "chal")) == tool_ends + assert len(await logs(row.id, "challenge_done")) == 1 + assert row.output["_issuance"]["tier"] == "degraded" + + +# -------------------------------------------------------------------------- +# 直通 +# -------------------------------------------------------------------------- + + +async def test_switched_off_globally_it_passes_through(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop), options={"verify": {"challenge": {"enabled": False}}}) + assert row.status == "succeeded", row.error + assert models.calls == [] + [skipped] = await logs(row.id, "challenge_skipped") + assert skipped["reason"] == "disabled" and skipped["level"] == "info" + out = await output_of(row.id, "chal") + assert out == {"skipped": True, "reason": "disabled"} + assert row.output["_issuance"]["tier"] == "formal" + assert await events(row.id, "llm.end", "chal") == [] + + +async def test_the_node_can_turn_it_on_or_off(monkeypatch, shop): + models = Models(monkeypatch, shop) + row = await finish(graph(shop, enabled=True), options={"verify": {"challenge": {"enabled": False}}}) + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"], "节点写了 enabled: true，全局关闭也照审" + assert row.output["_issuance"]["tier"] == "degraded" + + models = Models(monkeypatch, shop) + row = await finish(graph(shop, enabled=False)) + assert models.calls == [] and (await logs(row.id, "challenge_skipped"))[0]["reason"] == "disabled" + + +async def test_a_report_without_conclusions_passes_through(monkeypatch, shop): + models = Models(monkeypatch, shop, writer=("下面看细节。",)) + row = await finish(graph(shop, contract=False), run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.calls == [] + [skipped] = await logs(row.id, "challenge_skipped") + assert skipped["reason"] == "no_claims" + assert await output_of(row.id, "chal") == {"skipped": True, "reason": "no_claims"} + + +async def test_claims_without_any_query_behind_them_are_not_sent_out(monkeypatch): + """报告的证据里没有一次数据库查询（数字来自整形节点）：没有数据源可查，不派 agent，照实记无法判断。""" + metrics = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}] + models = Models(monkeypatch, "unused", writer=("本周销售额 [[m:gmv]]，比上周有所增长。[[see:m:gmv]]",)) + spec = chain( + node("start", "input"), + node("fetch", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), + node("caliber", "metrics", caliber="周报口径", caliber_version="v1", metrics=metrics), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ) + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.calls == [] + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 1} + [claim] = artifact_store.load(out["challenge_artifact"])["claims"] + assert claim["verdict"] == "inconclusive" and "数据库查询" in claim["rationale"] + assert claim["queries"] == [] and claim["rationale_doc"] is None + + +async def test_an_unavailable_model_leaves_the_claims_inconclusive(monkeypatch, shop): + Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置可用的模型接入")) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + out = await output_of(row.id, "chal") + assert out["counts"]["inconclusive"] == 2 + record = artifact_store.load(out["challenge_artifact"]) + assert all(c["rationale"].startswith("反查未完成：核查模型不可用") for c in record["claims"]) + assert row.output["_issuance"]["tier"] == "formal" + + +# -------------------------------------------------------------------------- +# 同模型提示、默认的报告、摘录遮罩 +# -------------------------------------------------------------------------- + + +async def test_reviewing_with_the_writers_own_model_is_called_out(monkeypatch, shop): + Models(monkeypatch, shop, model_id="mock-fast") + row = await finish(graph(shop), run_class="exploratory") + assert row.status == "succeeded", row.error + [warn] = await logs(row.id, "challenge_same_model") + assert warn["model"] == "mock-fast" and warn["level"] == "warn" + assert "mock-fast" in warn["message"] + + Models(monkeypatch, shop) + other = await finish(graph(shop), run_class="exploratory") + assert await logs(other.id, "challenge_same_model") == [] + + +async def test_the_upstream_report_is_taken_when_none_is_chosen(monkeypatch, shop): + models = Models(monkeypatch, shop) + spec = graph(shop) + spec["nodes"][3]["data"]["config"].pop("report") + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.reviewed() == ["居各区域之首", "华南销售额超过华东"] + assert artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"])["report"] == "write" + + +async def test_the_task_gives_the_claim_its_evidence_and_only_its_sources(monkeypatch, masked_shop): + """任务里给这句结论和它挂的证据摘录：遮罩的列照旧不给；工具只给这句引用的查询所在的数据源。""" + sql = "SELECT region AS 区域, buyer_phone, amount AS 金额 FROM sales ORDER BY id" + plays = {"华东有一笔": Play(["SELECT COUNT(*) AS 笔数 FROM sales WHERE region = '华东'"], "stands", "没有找到反例。")} + models = Models(monkeypatch, masked_shop, writer=("华东有一笔 [[v:Q1.r0.金额]] 的订单。[[see:Q1]]",), plays=plays) + row = await finish(graph(masked_shop, sql=sql), run_class="exploratory") + assert row.status == "succeeded", row.error + task = models.tasks()["华东有一笔"] + assert "华东有一笔 1,200 的订单。" in task + assert "原报告" in task and sql.split(" FROM")[0][:20] in task, "摘录里有这句挂的查询" + assert "13900000001" not in task and "buyer_phone" not in task.split("列：")[1].split("\n")[0] + assert "遮罩" in task + assert models.tools["华东有一笔"] == {f"db_schema__{masked_shop}", f"db_query__{masked_shop}"} + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + assert [q["source"] for q in record["claims"][0]["queries"]] == [masked_shop] + + +async def test_each_claim_only_gets_the_sources_it_cites(monkeypatch, tmp_path, shop): + """两个数据源各查一次：挂了 Q1 的那句只拿得到 Q1 那个源的工具，挂了 Q2 的只拿得到另一个；没挂依据的结论在 + 这份报告用到的数据源里反查。""" + other, other_id = await _source(tmp_path) + try: + text = ("华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]退款只有 [[v:Q2.r0.退款笔数]] 笔。[[see:Q2]]" + "增长主要来自华东。") + plays = {"居各区域之首": Play([NET]), "退款只有": Play([REFUNDS]), "增长主要来自": Play([REFUNDS])} + models = Models(monkeypatch, shop, writer=(text,), plays=plays) + spec = chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("fetch2", "tool", tool=f"db_query__{other}", args={"sql": REFUNDS}), + node("write", "report", "写周报", instructions="写周报"), + node("chal", "challenge", "审稿"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ) + row = await finish(spec, run_class="exploratory") + assert row.status == "succeeded", row.error + assert models.tools["居各区域之首"] == {f"db_schema__{shop}", f"db_query__{shop}"} + assert models.tools["退款只有"] == {f"db_schema__{other}", f"db_query__{other}"} + assert models.tools["增长主要来自"] == {f"db_schema__{s}" for s in (shop, other)} \ + | {f"db_query__{s}" for s in (shop, other)} + assert models.reviewed() == ["居各区域之首", "退款只有", "增长主要来自"], "挂了依据的排在前面" + finally: + await engines.invalidate(other_id) + + +# -------------------------------------------------------------------------- +# 画布校验 +# -------------------------------------------------------------------------- + + +def spec_of(*nodes, edges=None) -> GraphSpec: + if edges is None: + edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])] + return GraphSpec.model_validate({"nodes": list(nodes), "edges": edges}) + + +def problems(spec: GraphSpec, nid: str = "chal") -> list[tuple[str | None, str, str | None]]: + return [(i.field, i.message, i.code) for i in validate_graph(spec).issues if i.level == "error" and i.node_id == nid] + + +BASE = [node("start", "input"), node("fetch", "tool", tool="db_query__shop", args={"sql": "SELECT 1"}), + node("write", "report", "写周报", instructions="写周报")] + + +def test_the_report_must_be_an_upstream_report_writer(): + ok = spec_of(*BASE, node("chal", "challenge", report="write")) + assert problems(ok) == [] + assert problems(spec_of(*BASE, node("chal", "challenge"))) == [], "不写就取上游唯一的报告" + + [(fld, message, code)] = problems(spec_of(*BASE, node("chal", "challenge", report="fetch"))) + assert fld == "report" and code == "challenge.report_invalid" + assert "「fetch」" in message and "调用工具" in message and "报告撰写" in message + + [(_, message, _)] = problems(spec_of(*BASE, node("chal", "challenge", report="ghost"))) + assert "「ghost」" in message and "不存在" in message + + later = spec_of(*BASE, node("chal", "challenge", report="late"), node("late", "report", "后写的报告")) + [(_, message, _)] = problems(later) + assert "后写的报告" in message and "上游" in message + + [(fld, message, _)] = problems(spec_of(node("start", "input"), node("chal", "challenge"))) + assert fld == "report" and "没有" in message and "报告撰写" in message + + two = spec_of(node("start", "input"), node("a", "report", "周报甲"), node("b", "report", "周报乙"), + node("chal", "challenge"), + edges=[{"source": "start", "target": "a"}, {"source": "start", "target": "b"}, + {"source": "a", "target": "chal"}, {"source": "b", "target": "chal"}]) + [(fld, message, _)] = problems(two) + assert fld == "report" and "周报甲" in message and "周报乙" in message and "选择" in message + + # 两份报告前后相接：直接连着的那一份就是要审的 + stacked = spec_of(node("start", "input"), node("a", "report", "初稿"), node("b", "report", "终稿"), + node("chal", "challenge")) + assert problems(stacked) == [] + from app.engine.schema import challenge_report + + assert challenge_report(stacked, stacked.node_map()["chal"]) == ("b", None) + + +@pytest.mark.parametrize("config, field_name", [ + ({"max_claims": 0}, "max_claims"), ({"max_claims": 1.5}, "max_claims"), ({"max_claims": True}, "max_claims"), + ({"max_queries": -1}, "max_queries"), ({"timeout_s": 0}, "timeout_s"), ({"timeout_s": "30"}, "timeout_s"), + ({"on_refuted": "withhold"}, "on_refuted"), ({"enabled": "yes"}, "enabled"), ({"model": 3}, "model"), + ({"provider": ["a"]}, "provider"), +]) +def test_bad_settings_are_named_in_chinese(config, field_name): + [(fld, message, code)] = problems(spec_of(*BASE, node("chal", "challenge", report="write", **config))) + assert fld == field_name and code == "challenge.config_invalid" + assert field_name not in message, "写界面上的叫法，不写键名" + assert message.startswith("「反方审稿」的「") + + +def test_good_settings_pass(): + spec = spec_of(*BASE, node("chal", "challenge", report="write", max_claims=3, max_queries=2, timeout_s=45.5, + on_refuted="flag", enabled=None, provider=None, model="claude-sonnet-5")) + assert problems(spec) == [] + + +def test_reviewing_with_the_writer_model_is_a_warning(): + spec = spec_of(node("start", "input"), node("write", "report", "写周报", model="claude-sonnet-5"), + node("chal", "challenge", model="claude-sonnet-5")) + warnings = [i for i in validate_graph(spec).issues if i.level == "warning" and i.node_id == "chal"] + assert [(i.field, i.code) for i in warnings] == [("model", "challenge.same_model")] + assert "自己审自己" in warnings[0].message or "难以发现" in warnings[0].message + + +async def test_the_node_refuses_the_same_bad_settings_at_run_time(): + """跳过画布校验直接执行（旧版本发布的图、改坏了的配置）：节点自己报同一句话。""" + spec = spec_of(*BASE, node("chal", "challenge", report="write", on_refuted="withhold")) + [(_, message, _)] = problems(spec) + ctx = NodeContext(node=spec.node_map()["chal"], run=RunContext(run_id="r", thread_id="r", spec=spec)) + with pytest.raises(NodeError) as caught: + await challenge_mod.run_challenge({"nodes": {}}, ctx) + assert str(caught.value) == message + + wrong = spec_of(*BASE, node("chal", "challenge", report="fetch")) + [(_, message, _)] = problems(wrong) + ctx = NodeContext(node=wrong.node_map()["chal"], run=RunContext(run_id="r", thread_id="r", spec=wrong)) + with pytest.raises(NodeError) as caught: + await challenge_mod.run_challenge({"nodes": {}}, ctx) + assert str(caught.value) == message diff --git a/backend/tests/test_issuance_upstream_gaps.py b/backend/tests/test_issuance_upstream_gaps.py index 3427336d..2cf2fce6 100644 --- a/backend/tests/test_issuance_upstream_gaps.py +++ b/backend/tests/test_issuance_upstream_gaps.py @@ -86,8 +86,8 @@ async def test_upstream_gap_degrades_issuance_and_names_the_node(): async def test_stub_gate_nodes_pass_through_without_touching_issuance(): - """双盲复算、反方审稿目前是占位：放在图里照样能运行，出具不受影响""" - for kind in ("dual_check", "challenge"): + """双盲复算目前是占位：放在图里照样能运行，出具不受影响。反方审稿已有真实现，见 test_challenge_node.py""" + for kind in ("dual_check",): row = await finish(graph(node("gate", kind))) issuance = row.output["_issuance"] assert issuance["tier"] == "formal", (kind, issuance)