diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index 53cfe6e6..bbc0e9d7 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -24,9 +24,9 @@ 出具降档（nodes/io.py 的 _upstream_issuance_gaps，所以反方审稿要接在成果节点上游，画布校验会提醒）；fail 判节点 失败；flag 只记录。产出和审批卡里的 on_refuted 是实际生效的那一种（只有 degrade / flag，见 _finish）；degrades 说被推翻的到底会不会让出具降档（按降档处置、又接在带出具契约的成果节点上游），unreviewed 是未能反查的句数（核查 - 模型不可用、触顶没给出结论、数据源不在了、报告没有引用数据库查询……这几句记为无法判断，但反查没有完成，触顶的 - 可能已经查过几次）。每一句都记为无法判断，所以 unreviewed 不会大于 counts 里的 inconclusive。审批卡（human.py 的 - _reviewed_reports）附上审稿概况，据这两项说「出具会降档」「N 句未能反查」 + 模型不可用、触顶没给出结论、数据源不在了、报告没有引用数据库查询、运行时长上限将至没来得及审……这几句记为无法 + 判断，但反查没有完成，触顶的可能已经查过几次）。每一句都记为无法判断，所以 unreviewed 不会大于 counts 里的 + inconclusive。审批卡（human.py 的 _reviewed_reports）附上审稿概况，据这两项说「出具会降档」「N 句未能反查」 上限：句数、每句查询次数、每句时长、全局每日金额（核查和裁判各记各的账，engine/spend.py）。都按发起时快照的核查 设置（run.options["verify"]），节点上写了的盖过它：审批停下期间改了设置，恢复、接着跑仍按发起时的那份。没有快照 @@ -34,6 +34,10 @@ 可审的结论句、报告文档读不到时也直通（都记一条 challenge_skipped）。 逐句串行反查：并行的话几句的工具调用在时间线上交错，看不出哪条查询在审哪一句；每日余额按先后扣，结果也才确定。 +串行就可能撞上运行的时长上限（默认 5 句、每句 90 秒再加 30 秒收尾宽限，最坏 600 秒，正好是一段执行的上限），整次运行 +被中止。所以每句开始之前看这一段执行还剩多少时间（engine/deadline.py）：给下游留出余量后不够再审一句，这句和后面的 +都不再开始，记为未能反查（challenge_unreviewed 写明运行时长上限将至），节点照常收尾；够的话这句的时长上限按剩余时间 +收紧，到点照样按查到的结果收尾。 整个反查连同存审稿记录、发日志都在一个 @task 里：接着跑时取 checkpoint 里的结果，不重复查库、不重复记账，时间线 上也不再发一遍。结果带着输入指纹，接着跑时改了审稿本身的配置就重审；同一次执行里的重试沿用已经审完的（见 _reviewed）。 """ @@ -52,6 +56,7 @@ from app.core.errors import describe_exception from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import DataSource +from app.engine import deadline, verify_agent 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 @@ -89,6 +94,12 @@ _PLACEHOLDER = re.compile(r" ?⟦\?[^⟧]*⟧ ?") _STALE_LIMIT = 16 #: 同一次执行里审完的结果放在 RunContext.extra 的这个键下：{(节点, 第几次执行, 输入指纹): 审稿结果}。重试沿用它 _REVIEWS_KEY = "challenge_reviews" +#: 运行时长上限将至、不再开始的句子（engine/deadline.py）：每句的理由，时间线上的原因（这句和后面的一共几句） +_LATE_CLAIM = "本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，这句未反查" +_LATE_REST = "本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，剩下 {count} 句未审" +#: 时长按剩余时间收紧过的句子触了时长上限：单记一类（_limit_words 说运行时长上限将至，不说「每句时长上限 90 秒」—— +#: 那会叫人调大每句的上限，调大了只会更快撞线），challenge_limit 照记 timeout_s，界面认得 +_SQUEEZED = "squeezed" CHALLENGE_SYSTEM = ( "你是反方审稿人：报告已经写完，你站在反方，专门找其中一句结论的反例。你不写报告，也不改报告。\n" @@ -507,6 +518,8 @@ class _Review: max_claims, max_queries, timeout_s, daily_max_usd self.spec, self.masks, self.writer_model = spec, masks, writer_model self.fetch = _fetcher() + #: 这一段执行的总时长（engine/deadline.py）：有句子的时长按剩余时间收紧过时记下，那几句触顶时说明里写 + self.run_seconds: float | None = None #: 这份报告目录里的查询编号：摘录里指向它们的写法标上「原报告」（mark_original） self.queries = {a for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") == "query"} self.key = content_hash(canonical_json({ @@ -517,6 +530,8 @@ class _Review: def _limit_words(self, limit: str) -> str: if limit == "max_claims": return f"每份报告最多审 {self.max_claims} 句" + if limit == _SQUEEZED: + return f"本次运行即将达到 {self.run_seconds:g} 秒的时长上限，反查时长已按剩余时间收紧" if limit == "max_queries": return f"每句最多查询 {self.max_queries} 次" if limit == "timeout_s": @@ -559,12 +574,13 @@ class _Review: broken: str | None = None # 核查模型本身用不了：后面的句子照这个原因记，不再逐句去建模型 unreviewed: list[str] = [] # 没能反查的句子各自的原因（触顶的由 challenge_limit 说，不在这里） partial: list[list[str]] = [] # 反查了、但该查的源有停用了的句子：各自没查成的源 + late: tuple[str, str] | None = None # 运行时长上限将至：(每句的理由, 时间线上的原因)，后面的句子都照这个记 def unreached(cand: judging.Candidate, rationale: str, why: str) -> None: claims.append(_claim(cand, "inconclusive", rationale)) unreviewed.append(why) - for cand in self.picked: + for at, cand in enumerate(self.picked): if broke: hit["daily_max_usd"][1] += 1 claims.append(_claim(cand, "inconclusive", self._capped())) @@ -572,6 +588,9 @@ class _Review: if broken is not None: unreached(cand, f"反查未完成：{broken}", broken) continue + if late is not None: + unreached(cand, *late) + continue try: own = claim_sources(cand.aliases, self.catalog, self.fetch) sources = own or everywhere @@ -592,9 +611,19 @@ class _Review: why = f"数据源{_quoted(sources)}已不存在或已停用" unreached(cand, f"{where}{why}，没有可以反查的数据源，这句未反查", why) continue + # 派出去之前看这一段执行还剩多少时间：给下游留出余量后不够再审一句，这句和后面的都不再开始、记为未能反查， + # 节点照常收尾，运行不被中止；够的话这句的时长上限按剩余时间收紧。在 @task 里定，接着运行时不再重新定 + allot = deadline.allot(self.timeout_s, grace=verify_agent.SETTLE_GRACE_S) + if not allot.start: + late = (_LATE_CLAIM.format(seconds=allot.seconds), + _LATE_REST.format(seconds=allot.seconds, count=len(self.picked) - at)) + unreached(cand, *late) + continue + if allot.tightened: + self.run_seconds = allot.seconds outcome = await run_verify_agent( spec=self.spec, system=CHALLENGE_SYSTEM, task=task_text, sources=live, - output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=self.timeout_s, + output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=allot.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)) if outcome.error and not outcome.model: @@ -617,7 +646,8 @@ class _Review: claims.append(claim) usage = _add_usage(usage, outcome.usage) if outcome.limit in _CLAIM_LIMITS: - counted = hit.setdefault(outcome.limit, [0, 0]) + squeezed = outcome.limit == "timeout_s" and allot.tightened + counted = hit.setdefault(_SQUEEZED if squeezed else outcome.limit, [0, 0]) counted[0] += 1 counted[1] += why is not None broke = outcome.limit == "daily_max_usd" @@ -635,8 +665,8 @@ class _Review: 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="warn", code="challenge_limit", + limit="timeout_s" if limit == _SQUEEZED else limit, claims_skipped=skipped, message=message) if unreviewed: more = "等" if len(set(unreviewed)) > 1 else "" ctx.emit(EventType.LOG, level="warn", code="challenge_unreviewed", claims_skipped=len(unreviewed), diff --git a/backend/tests/test_verify_deadline.py b/backend/tests/test_verify_deadline.py new file mode 100644 index 00000000..5084aa6b --- /dev/null +++ b/backend/tests/test_verify_deadline.py @@ -0,0 +1,205 @@ +"""核查节点在运行时长上限内收尾（engine/deadline.py）：反方审稿逐句开始之前，看这一段执行还剩多少时间。 + +- 反方审稿：5 句、截止时刻只够审 2 句 → 审 2 句，其余 3 句记为未能反查（理由、时间线写明运行时长上限将至），运行 + 成功、不被中止，日志和产出的 unreviewed 对得上；剩下的时间不够整段时这句的时长收紧，到点按查到的结果收尾，说的是 + 运行时长上限将至，不说「每句时长上限 90 秒」；接着运行时取 checkpoint 里的审稿，不重审 +- 时间充裕时（默认 600 秒的一段）时长上限照传原值：和没有这道检查时一样 + +两种钟：精确的计数用假的钟（每派出一句就把 allot() 看到的钟往前拨若干秒，真实只过去几毫秒，runner 的 asyncio.timeout +远远不到）；「运行不被中止」用真的钟、真的短时长上限和真的慢 agent，只断言不随机器快慢变的事。模型都是 MockChatModel +剧本，数据源是现造的 SQLite 文件，区域、金额都是虚构的。 +""" +from __future__ import annotations + +import asyncio +from collections.abc import Callable + +from app.core import artifact_store +from app.core.config import settings +from app.engine import deadline, verify_agent +from app.engine.nodes import challenge as challenge_mod +from app.engine.runner import run_manager, verify_manifest +from tests.test_challenge_node import ( # noqa: F401 - engine_up、shop 是夹具，按名字引入 + NET, + REFUNDS, + Models, + Play, + engine_up, + finish, + graph, + logs, + output_of, + shop, + wait, +) + +#: 五句都挂了依据的结论，按文档先后反查：华东居首（被削弱）、华南超过华东（预埋的错误，被推翻）、其余三句经受住 +FIVE = ("华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]华南销售额超过华东。[[see:Q1]]" + "华北销售额垫底。[[see:Q1]]华东领先华北。[[see:Q1]]华南高于华北。[[see:Q1]]") +FIVE_PLAYS = { + "居各区域之首": Play([REFUNDS], "weakened", "华东含 [[v:Q1.r0.退款笔数]] 笔退款，扣除后仍居首，但销售额被高估。"), + "华南销售额超过华东": Play([NET], "refuted", + "扣除退款后华南为 [[v:Q1.r1.净销售额]]，仍低于华东的 [[v:Q1.r0.净销售额]]。[[see:Q1]]"), + "华北销售额垫底": Play([NET], "stands", "华北净销售额为 [[v:Q1.r2.净销售额]]，确实最低。"), + "华东领先华北": Play([NET], "stands", "没有找到反例。"), + "华南高于华北": Play([NET], "stands", "没有找到反例。"), +} +ORDER = list(FIVE_PLAYS) +#: 截止时刻将至、不再开始的句子：每句的理由，时间线上的原因（默认 600 秒的一段，剩下 3 句） +LATE_CLAIM = "本次运行即将达到 600 秒的时长上限，为保证运行按时完成，这句未反查" +LATE_REST = "本次运行即将达到 600 秒的时长上限，为保证运行按时完成，剩下 3 句未审" + + +class Ticker: + """假的钟：allot() 看到的「现在」= 真的事件循环时间 + offset。包住节点调用的核查 agent：每派出一项先把钟往前拨 + step 秒（像这一项真花了这么久），再照常执行，并记下这一项拿到的时长上限。hit：拿到的时长上限满足它时，把结果改成 + 触了时长上限（像收紧过的时长真到了点），用来看收紧过的触顶怎么说。""" + + def __init__(self, monkeypatch, module, *, step: float, start: float = 0.0, + hit: Callable[[float], bool] | None = None) -> None: + self.offset = start + self.step = step + self.timeouts: list[float | None] = [] + real = module.run_verify_agent + monkeypatch.setattr(deadline, "_now", lambda: asyncio.get_running_loop().time() + self.offset) + + async def spy(**kwargs): + self.timeouts.append(kwargs.get("timeout_s")) + self.offset += self.step + outcome = await real(**kwargs) + if hit is not None and kwargs.get("timeout_s") is not None and hit(kwargs["timeout_s"]): + outcome.limit = "timeout_s" + return outcome + + monkeypatch.setattr(module, "run_verify_agent", spy) + + +# -------------------------------------------------------------------------- +# 反方审稿 +# -------------------------------------------------------------------------- + + +async def test_five_claims_with_time_for_two_review_two(monkeypatch, shop): + """截止时刻只够审 2 句（每句像花了 250 秒）：审 2 句，其余 3 句不再派 agent、记为未能反查；运行成功，被推翻的 + 那句照样让出具降档；时间线和产出的 unreviewed 都是 3。""" + models = Models(monkeypatch, shop, writer=(FIVE,), plays=FIVE_PLAYS) + ticker = Ticker(monkeypatch, challenge_mod, step=250) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ORDER[:2], "第 3 句开始前只剩约 100 秒：留出 60 秒余量和 30 秒收尾宽限，不够再审一句" + assert ticker.timeouts == [90, 90], "时间充裕的两句照传每句时长上限，不收紧" + + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 3} + assert out["unreviewed"] == 3, "没来得及审的 3 句是未能反查，不是查过了判断不了" + assert row.output["_issuance"]["tier"] == "degraded", "审到的那句被推翻：照样降档" + + record = artifact_store.load(out["challenge_artifact"]) + assert [c["verdict"] for c in record["claims"]] == ["weakened", "refuted", "inconclusive", "inconclusive", + "inconclusive"] + for claim in record["claims"][2:]: + assert claim["rationale"] == LATE_CLAIM and claim["queries"] == [] and claim["rationale_doc"] is None + assert claim["usage"]["calls"] == 0, "没派 agent：一分钱都没花" + + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == 3 and warn["reason"] == LATE_REST + assert warn["message"] == f"反方审稿有 3 句结论未能反查，已记为无法判断：{LATE_REST}" + assert await logs(row.id, "challenge_limit") == [], "没有哪一句触了自己的上限" + [done] = await logs(row.id, "challenge_done") + assert done["counts"] == out["counts"] + assert "其中 3 句未能反查" in done["message"] + assert (await verify_manifest(row.id))["ok"] is True + + +async def test_a_claim_squeezed_by_the_deadline_says_so(monkeypatch, shop): + """第 2 句开始前只剩约 160 秒：留出余量和收尾宽限，这句的时长收紧到 70 秒以内；它到点按查到的结果收尾时，时间线 + 说的是运行时长上限将至、时长已收紧，不说「每句时长上限 90 秒」（那会叫人去调大每句的上限，调大了只会更快撞线）。""" + models = Models(monkeypatch, shop, writer=(FIVE,), plays=FIVE_PLAYS) + ticker = Ticker(monkeypatch, challenge_mod, step=440, hit=lambda timeout: timeout < 90) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ORDER[:2] + first, second = ticker.timeouts + assert first == 90 and 60 < second < 70, ticker.timeouts + + [limit] = await logs(row.id, "challenge_limit") + assert limit["limit"] == "timeout_s" and limit["claims_skipped"] == 0 + assert "本次运行即将达到 600 秒的时长上限" in limit["message"] and "收紧" in limit["message"] + assert "每句时长上限" not in limit["message"] + out = await output_of(row.id, "chal") + assert out["counts"]["refuted"] == 1, "收紧后按查到的结果收尾：反方给的结论照常采信" + assert out["unreviewed"] == 3 + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == 3 + + +async def test_no_time_at_all_reviews_nothing_but_finishes(monkeypatch, shop): + """一开始就只剩 50 秒：一句都不派，5 句都记为未能反查，节点照常收尾、运行成功；没有审到的句子，不降档。""" + models = Models(monkeypatch, shop, writer=(FIVE,), plays=FIVE_PLAYS) + ticker = Ticker(monkeypatch, challenge_mod, step=0, start=550) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.calls == [] and ticker.timeouts == [] + out = await output_of(row.id, "chal") + assert out["counts"] == {"stands": 0, "weakened": 0, "refuted": 0, "inconclusive": 5} and out["unreviewed"] == 5 + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == 5 and "剩下 5 句未审" in warn["reason"] + assert row.output["_issuance"]["tier"] == "formal" + + +async def test_continuing_takes_the_review_from_the_checkpoint(monkeypatch, shop): + """审了 2 句、3 句没来得及之后，节点落盘时出错；接着运行时这一段有了新的截止时刻，审稿照样取 checkpoint 里的： + 不再派 agent，没来得及的 3 句照旧记为未能反查。重放确定，不看接着运行时还剩多少时间。""" + models = Models(monkeypatch, shop, writer=(FIVE,), plays=FIVE_PLAYS) + ticker = Ticker(monkeypatch, challenge_mod, step=250) + 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, timeouts = len(models.calls), list(ticker.timeouts) + assert models.reviewed() == ORDER[:2] + + broken["on"] = False + ticker.offset = 0 # 新的一段：时间又充裕了，重审的话 5 句都会派出去 + await run_manager.continue_failed(run.id) + row = await wait(run.id) + assert row.status == "succeeded", row.error + assert len(models.calls) == calls and ticker.timeouts == timeouts, "接着运行时又派了 agent：审稿没取 checkpoint" + out = await output_of(row.id, "chal") + assert out["unreviewed"] == 3 and out["counts"]["inconclusive"] == 3 + assert len(await logs(row.id, "challenge_done")) == 1 and len(await logs(row.id, "challenge_unreviewed")) == 1 + + +async def test_a_slow_review_no_longer_aborts_the_run(monkeypatch, shop): + """真的钟：这一段只有 7 秒，每句反查都要查满 1.5 秒的时长上限（模型每次调用 0.1 秒，查不完的查询、不限次数），5 句串行 + 要 8 秒多——以前整次运行在审到第 5 句时被中止。现在审到时间不够就停下，剩下的记为未能反查，运行成功。 + 几句来得及审随机器快慢变，只断言不变的：至少审了 1 句、至少 1 句没来得及，没来得及的都写明了原因、计数对得上。""" + monkeypatch.setattr(settings, "max_run_seconds", 7) + monkeypatch.setattr(deadline, "RESERVE_S", 2.0) + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 0.5) + endless = [f"SELECT COUNT(*) AS n FROM sales WHERE id > {i}" for i in range(60)] + plays = {key: Play(endless, "stands", "查到的 [[v:Q1.r0.n]] 行里没有反例。") for key in ORDER} + models = Models(monkeypatch, shop, writer=(FIVE,), plays=plays, latency=0.1) + row = await finish(graph(shop, timeout_s=1.5, max_queries=100)) + assert row.status == "succeeded", row.error + assert "时长上限，已中止" not in (row.error or "") + + out = await output_of(row.id, "chal") + record = artifact_store.load(out["challenge_artifact"]) + late = [c for c in record["claims"] if "即将达到" in c["rationale"] and not c["queries"]] + reviewed = [c for c in record["claims"] if c["queries"]] + assert reviewed and late, [c["rationale"] for c in record["claims"]] + assert len(reviewed) + len(late) == 5 + assert models.reviewed() == ORDER[:len(reviewed)], "按顺序审，停下之后一句都不再派" + assert out["unreviewed"] == len(late) + [warn] = await logs(row.id, "challenge_unreviewed") + assert warn["claims_skipped"] == len(late) and f"剩下 {len(late)} 句未审" in warn["reason"] + assert "本次运行即将达到 7 秒的时长上限" in warn["reason"]