反方审稿在运行时长上限内收尾：每句派出去之前看这一段执行还剩多少时间，不够再审一句就停下，这句和后面的记为未能反查，节点照常收尾、运行不被中止；够的话这句的时长上限按剩余时间收紧

为什么：逐句串行反查（为了时间线看得出哪条查询在审哪一句、每日余额扣得确定，这个设计保留）默认 5 句、每句 90 秒再加 30 秒收尾宽限，最坏正好是一段执行的 600 秒，加上取数、写报告必然被中止；中止时反查的 @task 没做完，「继续运行」从头再审、再次超时。

- 判断用 deadline.allot（给下游留余量、扣收尾宽限），只在要派 agent 的句子上做：没有可查数据源这类不花时间的句子照记原来的原因。一旦停下，后面的句子照同一个原因记（同每日金额用完、核查模型用不了的处理），理由写「本次运行即将达到 600 秒的时长上限，为保证运行按时完成，这句未反查」，challenge_unreviewed 的原因写「……剩下 N 句未审」，产出的 unreviewed 照算进去
- 时长收紧过的句子到点按查到的结果收尾，challenge_limit 照记 timeout_s（界面认得），说法另起：说运行时长上限将至、反查时长已按剩余时间收紧，不说「每句时长上限 90 秒」——那会叫人调大每句的上限，调大了只会更快撞线
- 决定都在原有的 @task 里做，输入指纹不含截止时刻：接着运行时取 checkpoint 里的审稿，不重审，没来得及的照旧
- 时间充裕时时长上限照传原值，和以前逐字一样（现有用例全过）

测试：假的钟（每派出一句拨钟若干秒）钉住 5 句只够审 2 句、收紧的那句触顶怎么说、一句都来不及、接着运行不重审；真的钟用 7 秒的时长上限和查不完的慢 agent，以前被中止，现在运行成功、计数对得上。去掉判断后这几条都变红。

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),

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"]