diff --git a/backend/app/engine/deadline.py b/backend/app/engine/deadline.py new file mode 100644 index 00000000..da112e03 --- /dev/null +++ b/backend/app/engine/deadline.py @@ -0,0 +1,167 @@ +"""这一段执行的截止时刻：核查节点据此在运行的时长上限之内收尾，不让整次运行被中止。 + +runner 给每一段执行（两次人工介入之间的那一段）套一道 asyncio.timeout(max_run_seconds)，到点整次运行判失败。核查 +节点一个就能把这一段的时间用光：反方审稿逐句串行反查，默认 5 句、每句 90 秒再加 30 秒收尾宽限，最坏就是 600 秒；双盲 +复算每项 120 秒加宽限，指标多了也一样。被中止时节点里装着反查、盲算的那个 @task 还没做完、没进 checkpoint，「继续 +运行」把它从头再做一遍，多花一份钱，再次超时。 + +所以 runner 进入时长上限时把截止时刻放进这里的 ContextVar（segment），退出时清掉：节点、节点里的 @task、子工作流都 +在这个上下文里执行，都看得到。核查节点每开始一项（一句反查、一项盲算）之前问 allot()：给下游留出余量之后还够不够 +再做一项，这一项的时长上限收紧到多少；不够就不再开始，剩下的照各自「没能核查」的路径记下来，节点正常收尾。决定都 +发生在节点自己的 @task 里，接着运行时取 checkpoint 里的结果，重放确定。 + +截止时刻用事件循环的钟（loop.time()），和 asyncio.timeout 同一个钟。没有上限时（不经 runner 执行的节点：单元测试、 +审批质询接口）为 None，allot() 原样放行，节点的行为和没有这道检查时一样。 + +因此没核成的，节点发日志时带上 cause: RUN_DEADLINE 和这一段的总时长（run_seconds），另起一条，不和别的原因混在一起： +界面据此说「运行时长上限将至」、给对的下一步——调高核查自己的上限只会更快撞线，检查模型、数据源也无济于事。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import math +from collections.abc import AsyncIterator, Callable +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Any + +from app.core.events import EventType +from app.engine.schema import GraphSpec, NodeType + +#: 给下游节点留的余量，按份算。一份给收尾的节点：核查节点后面通常还有成果节点（出具契约复核报告），核查把时间用到 +#: 最后一刻，后面的节点一开始就撞上时长上限，整次运行照样被中止。一份取 RESERVE_S 和这一段总时长的 RESERVE_SHARE 中 +#: 大的：60 秒够成果节点复核、再做一两次普通的模型调用；管理员调大了时长上限的，多半是因为节点本来就慢，余量跟着放大 +RESERVE_S = 60.0 +RESERVE_SHARE = 0.1 +#: 这一段不到 RESERVE_S / SHORT_SHARE（240 秒）时，一份按总时长的 SHORT_SHARE 算：时长上限调得很短（比如 180 秒）时， +#: 固定的 60 秒余量加上收尾宽限，前面的节点用掉一分钟核查就一项都开始不了，比没有这道检查时还差 +SHORT_SHARE = 0.25 +#: 下游这一段里每个还要调模型的节点再各加一份：Copilot 指南里的惯常接法是「口径卡 → 双盲复算 → 写报告 → 反方审稿 → +#: 成果」，双盲复算只留一份的话，写报告那次模型调用一慢，运行就在写报告时被中止。反方审稿也记一份：默认 600 秒的一段 +#: 一份是 60 秒，正好够它开始一句（它自己那份余量之外，还要收尾宽限和最短的查库时长各 30 秒）。估得粗，宁可多留： +#: 核查少做几项记成没核成，好过整次运行被中止 +MODEL_NODES = frozenset({NodeType.LLM, NodeType.AGENT, NodeType.SUPERVISOR, NodeType.SUBGRAPH, NodeType.REPORT, + NodeType.DUAL_CHECK, NodeType.CHALLENGE}) +#: 余量合计最多占这一段总时长的这么多：下游节点再多，核查节点也总还能开始几项 +RESERVE_CAP = 0.5 +#: 日志里的 cause：这几项没核成是因为运行时长上限将至（不再开始、收紧后到了点） +RUN_DEADLINE = "run_deadline" +#: 收紧过的核查到了点：核查 agent 自己发的 verify_limit 改成这一句（relabel） +SQUEEZED_LIMIT = ("核查已达上限（本次运行即将达到 {seconds:g} 秒的时长上限，时长已按剩余时间收紧到 {timeout:g} 秒），" + "已停止查询，按已有结果收尾") + + +@dataclass(frozen=True) +class Segment: + """这一段执行：截止时刻（事件循环时间）和总时长（秒，说明里写、按它算余量）。""" + + deadline: float + seconds: float + + +_SEGMENT: ContextVar[Segment | None] = ContextVar("agentlab_segment", default=None) + + +@contextlib.asynccontextmanager +async def segment(deadline: float | None, seconds: float) -> AsyncIterator[None]: + """runner 用：这一段执行的截止时刻（asyncio.Timeout.when()）和总时长，退出时清掉。写成异步的上下文管理器，好和 + 时长上限写在同一个 async with 里。deadline 为 None（没有上限）时放 None：里面看不到外面的截止时刻。""" + token = _SEGMENT.set(Segment(float(deadline), float(seconds)) if deadline is not None else None) + try: + yield + finally: + _SEGMENT.reset(token) + + +def current() -> Segment | None: + """这一段执行的截止时刻；不在 runner 的执行里时为 None。""" + return _SEGMENT.get() + + +def reserve(seconds: float, after: int = 0) -> float: + """给下游节点留的余量（秒）：一份给收尾的节点，after 是下游这一段里还要调模型的节点数（models_after），各再加一份； + 合计不超过总时长的 RESERVE_CAP。一份多大、为什么，见 RESERVE_S、SHORT_SHARE、MODEL_NODES。""" + share = max(min(RESERVE_S, SHORT_SHARE * seconds), RESERVE_SHARE * seconds) + return min(share * (1 + max(after, 0)), RESERVE_CAP * seconds) + + +def models_after(spec: GraphSpec, node_id: str) -> int: + """这个节点下游、这一段执行里还要调模型的节点（MODEL_NODES）有几个。沿连线往下走，人工审批节点不往后走：审批之后是 + 新的一段，重新计时。分支汇合、环路里的节点只算一次，自己不算；分支上不一定执行的也算上，宁可多留。""" + nodes = spec.node_map() + children: dict[str, list[str]] = {} + for edge in spec.edges: + children.setdefault(edge.source, []).append(edge.target) + seen: set[str] = {node_id} + stack = list(children.get(node_id, [])) + count = 0 + while stack: + cur = stack.pop() + node = nodes.get(cur) + if cur in seen or node is None: + continue + seen.add(cur) + if node.type == NodeType.HUMAN: + continue + if node.type in MODEL_NODES: + count += 1 + stack.extend(children.get(cur, [])) + return count + + +def _now() -> float: + return asyncio.get_running_loop().time() + + +@dataclass(frozen=True) +class Allot: + """allot() 的答复。start：够不够再开始一项。timeout_s：这一项的时长上限，没收紧时就是传进来的那个值（不限时为 + None）。tightened：按剩余时间收紧过（比配置的短，或者配置的是不限）。seconds：这一段的总时长，不再开始、收紧过的 + 时候说明里写；没有上限时为 None。""" + + start: bool + timeout_s: float | None + tightened: bool = False + seconds: float | None = None + + +def allot(timeout_s: float | None, *, grace: float, after: int = 0) -> Allot: + """开始一项核查之前问一句：够不够，这一项的时长上限收紧到多少。timeout_s 是这一项配置的时长上限（None 不限）， + grace 是到点之后收尾那次调用的宽限（verify_agent.SETTLE_GRACE_S），after 是下游这一段里还要调模型的节点数 + （models_after，按它多留余量）。 + + 一项最坏要用「时长上限 + 收尾宽限」。剩下的时间先扣掉给下游的余量、再扣掉收尾宽限，就是这一项查库最多能用的时长： + 比配置的短就收紧到它（取到 0.1 秒，往下取：时间线上写「时长上限 69.7 秒」，不写一串小数）。收紧之后不到 + min(配置的时长上限, 收尾宽限) 就不再开始：收尾宽限是留给一次模型调用的时间，查库的时长比它还短，连一轮都做不完， + 只会白花一次收尾的钱、记成没能核查。配置的时长上限本来就比它短的，按配置的整段算；配置的是不限的，按收尾宽限算。 + 没有上限时原样放行。""" + found = _SEGMENT.get() + if found is None: + return Allot(True, timeout_s) + room = math.floor((found.deadline - _now() - reserve(found.seconds, after) - grace) * 10) / 10 + limit = room if timeout_s is None else min(timeout_s, room) + floor = grace if timeout_s is None else min(timeout_s, grace) + if limit < floor: + return Allot(False, None, seconds=found.seconds) + return Allot(True, limit, tightened=timeout_s is None or room < timeout_s, seconds=found.seconds) + + +def relabel(emit: Callable[..., Any], allot: Allot) -> Callable[..., Any]: + """交给核查 agent 的 emit。这一项的时长收紧过时，它到点自己发的 verify_limit 写的是收紧后的秒数（「时长上限 52.3 + 秒」），和任何设置都对不上，读的人会去找这个数是在哪儿配的、甚至去调大它：改说运行时长上限将至，带上 cause 和这一段 + 的总时长。别的事件、别的上限原样转发；没收紧时原样交回 emit。""" + if not allot.tightened: + return emit + + def wrapped(kind: Any, **data: Any) -> Any: + if str(kind) == EventType.LOG and data.get("code") == "verify_limit" and data.get("limit") == "timeout_s": + data = {**data, "cause": RUN_DEADLINE, "run_seconds": allot.seconds, + "message": SQUEEZED_LIMIT.format(seconds=allot.seconds, timeout=allot.timeout_s)} + return emit(kind, **data) + + return wrapped + + +__all__ = ["MODEL_NODES", "RESERVE_CAP", "RESERVE_S", "RESERVE_SHARE", "RUN_DEADLINE", "SHORT_SHARE", "Allot", + "Segment", "allot", "current", "models_after", "relabel", "reserve", "segment"] diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index 53cfe6e6..b5c555e3 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,11 @@ 可审的结论句、报告文档读不到时也直通（都记一条 challenge_skipped）。 逐句串行反查：并行的话几句的工具调用在时间线上交错，看不出哪条查询在审哪一句；每日余额按先后扣，结果也才确定。 +串行就可能撞上运行的时长上限（默认 5 句、每句 90 秒再加 30 秒收尾宽限，最坏 600 秒，正好是一段执行的上限），整次运行 +被中止。所以每句开始之前看这一段执行还剩多少时间（engine/deadline.py）：给下游留出余量后不够再审一句，这句和后面的 +都不再开始，记为未能反查，节点照常收尾；够的话这句的时长上限按剩余时间收紧，到点照样按查到的结果收尾。这两种在 +时间线上都带 cause: run_deadline（没开始的另起一条 challenge_unreviewed，收紧后到点的是 challenge_limit 里单独的 +一条），界面据此说运行时长上限将至，不叫人调大每句的上限，也不叫人去查模型和数据源。 整个反查连同存审稿记录、发日志都在一个 @task 里：接着跑时取 checkpoint 里的结果，不重复查库、不重复记账，时间线 上也不再发一遍。结果带着输入指纹，接着跑时改了审稿本身的配置就重审；同一次执行里的重试沿用已经审完的（见 _reviewed）。 """ @@ -52,6 +57,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 +95,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（老界面认得），另带 cause +_SQUEEZED = "squeezed" CHALLENGE_SYSTEM = ( "你是反方审稿人：报告已经写完，你站在反方，专门找其中一句结论的反例。你不写报告，也不改报告。\n" @@ -507,6 +519,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 +531,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 +575,16 @@ class _Review: broken: str | None = None # 核查模型本身用不了：后面的句子照这个原因记，不再逐句去建模型 unreviewed: list[str] = [] # 没能反查的句子各自的原因（触顶的由 challenge_limit 说，不在这里） partial: list[list[str]] = [] # 反查了、但该查的源有停用了的句子：各自没查成的源 + late: tuple[str, str] | None = None # 运行时长上限将至：(每句的理由, 时间线上的原因)，后面的句子都照这个记 + overdue = 0 # 因此没开始的句数：时间线另起一条（带 cause），不混进 unreviewed + # 下游这一段里还要调模型的节点：各多留一份余量（engine/deadline.py）。图在运行里不变，算一次 + after = deadline.models_after(ctx.run.spec, ctx.node.id) 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 +592,10 @@ class _Review: if broken is not None: unreached(cand, f"反查未完成：{broken}", broken) continue + if late is not None: + claims.append(_claim(cand, "inconclusive", late[0])) + overdue += 1 + continue try: own = claim_sources(cand.aliases, self.catalog, self.fetch) sources = own or everywhere @@ -592,11 +616,24 @@ 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, after=after) + if not allot.start: + self.run_seconds = allot.seconds + late = (_LATE_CLAIM.format(seconds=allot.seconds), + _LATE_REST.format(seconds=allot.seconds, count=len(self.picked) - at)) + claims.append(_claim(cand, "inconclusive", late[0])) + overdue += 1 + continue + if allot.tightened: + self.run_seconds = allot.seconds + # 收紧过的：核查 agent 到点自己发的 verify_limit 改说运行时长上限将至（deadline.relabel），不写收紧后的秒数 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, - 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)) + output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=allot.timeout_s, + daily_max_usd=self.daily_max_usd, emit=deadline.relabel(ctx.emit, allot), run_id=ctx.run.run_id, + node_id=ctx.node.id, data_versions=run_versions(ctx.run)) if outcome.error and not outcome.model: broken = outcome.error # 连模型都没建成（没配好、接入出错）：换一句也一样 model = model or outcome.model @@ -617,7 +654,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" @@ -625,9 +663,9 @@ class _Review: unreviewed.append(why) counts = {v: sum(1 for c in claims if c["verdict"] == v) for v in VERDICTS} - # 没能反查的句数：没派出去、派出去却没查成的（unreviewed 里各有原因），加上触顶时没能给出结论的。都记为无法判断， - # 可「无法判断 2 句」既可能是查了、数据不够，也可能是根本没查：审批卡要分开说 - missed = len(unreviewed) + sum(skipped for _, skipped in hit.values()) + # 没能反查的句数：没派出去、派出去却没查成的（unreviewed 里各有原因），运行时长上限将至没开始的，加上触顶时没能给出 + # 结论的。都记为无法判断，可「无法判断 2 句」既可能是查了、数据不够，也可能是根本没查：审批卡要分开说 + missed = len(unreviewed) + overdue + sum(skipped for _, skipped in hit.values()) artifact = await self._store({"report": self.report_id, "doc_artifact": self.doc_artifact, "claims": claims, "model": model}) for limit, (touched, skipped) in hit.items(): @@ -635,13 +673,24 @@ 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) + # 收紧过的那一组照记 timeout_s（老界面认得），另带 cause、这一段的总时长和触顶的句数：界面据此说运行时长 + # 上限将至，不叫人调高每句的上限；同一节点里还有句子触了自己的上限时，两条也分得开 + squeezed = {"cause": deadline.RUN_DEADLINE, "run_seconds": self.run_seconds, "claims": touched} \ + if limit == _SQUEEZED else {} + ctx.emit(EventType.LOG, level="warn", code="challenge_limit", + limit="timeout_s" if limit == _SQUEEZED else limit, claims_skipped=skipped, message=message, + **squeezed) if unreviewed: more = "等" if len(set(unreviewed)) > 1 else "" ctx.emit(EventType.LOG, level="warn", code="challenge_unreviewed", claims_skipped=len(unreviewed), reason=unreviewed[0], message=f"反方审稿有 {len(unreviewed)} 句结论未能反查，已记为无法判断：{unreviewed[0]}{more}") + if overdue and late is not None: + # 运行时长上限将至没开始的另起一条，带 cause：不被前面句子的原因盖掉、藏进「等」里，界面也不会为它叫人去查 + # 模型和数据源 + ctx.emit(EventType.LOG, level="warn", code="challenge_unreviewed", cause=deadline.RUN_DEADLINE, + run_seconds=self.run_seconds, claims_skipped=overdue, reason=late[1], + message=f"反方审稿有 {overdue} 句结论未能反查，已记为无法判断：{late[1]}") if partial: names = list(dict.fromkeys(s for gone in partial for s in gone)) ctx.emit(EventType.LOG, level="warn", code="challenge_sources_lost", claims=len(partial), sources=names, diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index f9b82a93..04607ec8 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -32,7 +32,9 @@ SQL，就跟着差异说明进了审批卡、对照工件和目录日志；遮 选了会怎样要在卡上看得出：payload 在约定的几项之外带上实际生效的 on_reject，摘要里每个指标带 catalog——选 a、选 b 按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）；几个指标落到目录 同一项上时每种选择组合合起来会怎样写（preview_joint），放在对照工件的 catalog_joint 里。 -盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。 +盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。运行这一段的 +时长上限将至时不再开始新的盲算、开始的按剩余时间收紧时长（engine/deadline.py），没来得及的照此记，运行不被中止； +这几个另发一条 dual_check_unchecked，带 cause: run_deadline，界面据此说原因、给对的下一步。 **产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial； 有不一致、人裁决过的是 resolved。产出和台账都带 counts（一致、不一致、无法复算各几个）。 @@ -64,7 +66,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 recompute +from app.engine import deadline, recompute from app.engine.approval import read_decision from app.engine.context import NodeContext, NodeError from app.engine.dual_check_facts import preview, preview_joint, record_choices @@ -126,6 +128,12 @@ GAP_BLIND = "指标「{name}」两路结果不一致，审核人选择了盲算 GAP_NEITHER = "指标「{name}」两路结果不一致，审核人判断两边都不对" GAP_UNANSWERED = "指标「{name}」两路结果不一致，审核人没有作出选择" GAP_UNCHECKED = "指标「{name}」无法复算：{reason}" +#: 运行时长上限将至（engine/deadline.py）：没来得及开始的盲算、时长按剩余时间收紧后到了点的盲算，都按无法复算记 +LATE = "本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，未进行盲算" +SQUEEZED = "本次运行即将达到 {seconds:g} 秒的时长上限，盲算时长已按剩余时间收紧，到时未能完成" +#: 这几个指标另起的那条 dual_check_unchecked（带 cause）的原话 +LATE_UNCHECKED = ("本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，双盲复算有 {count} 个指标未能复算：" + "{names}，按无法复算计入缺口") BLIND_SYSTEM = ( "你在做双盲复算：另一条流程已经算出了口径卡上的一个指标，你要在看不到它的 SQL 和结果的情况下，按业务定义独立算出" @@ -431,7 +439,7 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An if agreed and not any(r["status"] == DIFFERENT for r in rows): ctx.emit(EventType.LOG, level="info", code="dual_check_agreed", metrics=agreed, message=f"双盲复算一致：{agreed} 个指标两边的结果相同，已放行") - if unchecked := [r for r in rows if r["status"] == UNCHECKED]: + if unchecked := [r for r in rows if r["status"] == UNCHECKED and not r.get("cause")]: # 结构上没法单独复算的（合并查询、遮罩列）以前一条 dual_check_* 日志都没有，时间线上看不出为什么没核成 parts = "；".join(f"「{r['name']}」（{r.get('reason') or '原因不明'}）" for r in unchecked[:5]) ctx.emit(EventType.LOG, level="warn", code="dual_check_unchecked", @@ -439,6 +447,15 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An for r in unchecked], message=f"双盲复算未能核对 {len(unchecked)} 个指标：{parts}" + ("等" if len(unchecked) > 5 else "") + "，按无法复算计入缺口") + if late := [r for r in rows if r["status"] == UNCHECKED and r.get("cause")]: + # 运行时长上限将至没核成的（_blind 记下 cause）另起一条：界面据此说原因、给下一步，调高每项的上限只会更快撞线 + seconds = late[0].get("run_seconds") + ctx.emit(EventType.LOG, level="warn", code="dual_check_unchecked", cause=late[0]["cause"], run_seconds=seconds, + metrics=[{"id": r["metric_id"], "name": r["name"], "reason": r.get("reason") or "原因不明"} + for r in late], + message=LATE_UNCHECKED.format(seconds=seconds or 0, count=len(late), + names="".join(f"「{r['name']}」" for r in late[:5]) + + ("等" if len(late) > 5 else ""))) return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage} @@ -803,10 +820,12 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op 先单独做第一个：核查模型本身用不了（没配好、接入出错）时，后面的一样用不了，照这一次的原因记，不再一个个去试。 其余的同时进行，最多 BLIND_CONCURRENCY 个：串行时几个慢的输入（每个最多 timeout_s 加收尾的宽限）就能把运行这一段 - 的时长上限耗光，整段盲算没有 checkpoint，接着运行时全部重做、再花一次钱。结果按格子写回，日志按任务的先后发， - 和完成的先后无关。""" + 的时长上限耗光，整段盲算没有 checkpoint，接着运行时全部重做、再花一次钱。指标多了并行也一样耗得光，所以每项开始 + 之前还要看这一段还剩多少时间（engine/deadline.py）：不够就不再开始、按无法复算记，够的话时长上限按剩余时间收紧。 + 结果按格子写回，日志按任务的先后发，和完成的先后无关。""" from app.api.settings import verify_model_spec - from app.engine.verify_agent import run_verify_agent + from app.engine import verify_agent + from app.engine.verify_agent import VerifyOutcome, run_verify_agent from app.tools.datasource import run_versions usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0} @@ -820,13 +839,28 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op if job["source"] not in catalogs: catalogs[job["source"]] = await _catalog_text(sources, job["source"]) outcomes: dict[str, Any] = {} + squeezed: dict[str, float] = {} # 时长按剩余时间收紧过的盲算 → 这一段执行的总时长（到了点时说明里写） + late: dict[str, float] = {} # 运行时长上限将至、没开始的盲算 → 同上 + # 下游这一段里还要调模型的节点（写报告、反方审稿……）：各多留一份余量（engine/deadline.py） + after = deadline.models_after(ctx.run.spec, ctx.node.id) async def one(key: str, job: dict[str, Any]) -> None: + # 给下游留出余量后不够再做一项：不再开始，交回一个没有模型的结果，按无法复算记——第一项就不够时，和核查模型 + # 用不了一样，后面的不再一个个去试。在 @task 里定，重放时不再重新定 + allot = deadline.allot(options.timeout_s, grace=verify_agent.SETTLE_GRACE_S, after=after) + if not allot.start: + late[key] = allot.seconds + outcomes[key] = VerifyOutcome(result=None, cells=[], queries=[], rationale=None, usage={}, + error=LATE.format(seconds=allot.seconds)) + return + if allot.tightened: + squeezed[key] = allot.seconds + # 收紧过的：核查 agent 到点自己发的 verify_limit 改说运行时长上限将至（deadline.relabel），不写收紧后的秒数 outcomes[key] = await run_verify_agent( spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]], - output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=options.timeout_s, - daily_max_usd=options.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id, - data_versions=run_versions(ctx.run)) + output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=allot.timeout_s, + daily_max_usd=options.daily_max_usd, emit=deadline.relabel(ctx.emit, allot), run_id=ctx.run.run_id, + node_id=ctx.node.id, data_versions=run_versions(ctx.run)) keys = list(jobs) await one(keys[0], jobs[keys[0]]) @@ -848,6 +882,14 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op results[key] = dict(results[keys[0]]) continue results[key] = _slim(outcome) + if key in late: + # 没开始的：带上 cause 和这一段的总时长，指标按它记无法复算时时间线另起一条（_check） + results[key].update(cause=deadline.RUN_DEADLINE, run_seconds=late[key]) + elif key in squeezed and results[key]["limit"] == "timeout_s": + # 收紧过的时长到了点：照样按无法复算记（同触了每项时长上限），原因说运行时长上限将至。不留 limit，免得发 + # dual_check_limit——它叫人调高每项的上限，调高了只会更快撞线 + results[key].update(value=None, limit=None, reason=SQUEEZED.format(seconds=squeezed[key]), + cause=deadline.RUN_DEADLINE, run_seconds=squeezed[key]) for k in usage: usage[k] += outcome.usage.get(k) or 0 if outcome.model and model is None: @@ -978,7 +1020,8 @@ def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any row.update(status=UNCHECKED, reason="盲算" + _limit_text(found["limit"], options)) return if found.get("value") is None: - row.update(status=UNCHECKED, reason=found.get("reason") or "盲算没有交回可用的数值") + row.update(status=UNCHECKED, reason=found.get("reason") or "盲算没有交回可用的数值", + **{k: found[k] for k in ("cause", "run_seconds") if k in found}) # 运行时长上限将至（_blind） return values[inp["path"]] = found["value"] outcome = recompute.recompute(metric, recompute.Changes(inputs=values), recompute.Sources( diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 4677078b..2e65f504 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -24,6 +24,7 @@ from app.db.models import Approval, Run, RunEvent, Workflow from app.engine.approval import read_decision from app.engine.compiler import compile_graph, initial_state from app.engine.context import NodeError, RunContext +from app.engine.deadline import segment from app.engine.replay import PROTOCOL, PROTOCOL_KEY, protocol_of from app.engine.schema import GraphSpec, NodeType, loop_steps, topology_of, validate_graph from app.tools.datasource import RUN_VERSIONS_KEY @@ -714,7 +715,8 @@ class RunManager: # 代码执行它——一个挂住的模型或工具能把运行和它占着的并发名额一直拖着。 # 按"这一段执行"计：等人工审批时 _drive 已经结束，那段时间不算。 deadline = asyncio.timeout(settings.max_run_seconds) - async with deadline: + # 截止时刻同时交给节点（engine/deadline.py）：核查节点据此在上限之内收尾，不把整次运行拖到被中止 + async with deadline, segment(deadline.when(), settings.max_run_seconds): async for mode, chunk in app.astream( payload, config=config, stream_mode=["custom", "updates"] ): diff --git a/backend/tests/test_run_deadline.py b/backend/tests/test_run_deadline.py new file mode 100644 index 00000000..6f02da0a --- /dev/null +++ b/backend/tests/test_run_deadline.py @@ -0,0 +1,280 @@ +"""这一段执行的截止时刻（engine/deadline.py）：runner 进入时长上限时放进 ContextVar、退出时清掉，核查节点开始每一项之前 +问 allot() 够不够、时长收紧到多少。 + +- 没有截止时刻（不经 runner 执行）：原样放行，时长上限还是传进来的那个值 +- 有截止时刻：先给下游留出余量（一份取 60 秒或这一段总时长的 10% 中大的，这一段不到 240 秒时按 25%；下游这一段里每个 + 还要调模型的节点再各加一份，合计不超过一半）、再扣收尾宽限；比配置的时长短就收紧，收紧之后不到 min(配置的时长, + 收尾宽限) 就不再开始；配置的是不限时也按剩余时间收紧 +- 收紧过的核查到了点，核查 agent 自己发的 verify_limit 改说运行时长上限将至，带上 cause（relabel） +- runner 每一段执行都放一份：节点、节点里的 @task 都看得到，截止时刻就是 asyncio.timeout 的那一个；审批恢复后的那一段 + 是新的截止时刻；运行结束后清掉 +""" +from __future__ import annotations + +import asyncio + +import pytest +from langgraph.func import task +from sqlalchemy import select + +from app.core.config import settings +from app.core.events import EventType +from app.db.base import SessionLocal +from app.db.models import Approval, Run +from app.engine import compiler, deadline +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, NodeType + + +class Clock: + """假的事件循环钟：allot() 看到的「现在」由测试定。""" + + def __init__(self, monkeypatch, now: float = 1000.0) -> None: + self.now = now + monkeypatch.setattr(deadline, "_now", lambda: self.now) + + +@pytest.fixture +def clock(monkeypatch) -> Clock: + return Clock(monkeypatch) + + +async def test_without_a_deadline_everything_passes_unchanged(clock): + assert deadline.current() is None + for timeout in (90, 0.4, None): + answer = deadline.allot(timeout, grace=30.0) + assert answer == deadline.Allot(True, timeout), "不经 runner 执行：照旧放行，时长不动" + assert answer.timeout_s is timeout, "传进来的就是那个值（int 不变成 float），和没有这道检查时逐字一样" + + +async def test_plenty_of_time_keeps_the_configured_limit(clock): + async with deadline.segment(clock.now + 600, 600): + answer = deadline.allot(90, grace=30.0) + assert answer.start and answer.timeout_s == 90 and isinstance(answer.timeout_s, int) and not answer.tightened + assert answer.seconds == 600 + + +@pytest.mark.parametrize("left, start, limit", [ + (180.0, True, 90), # 60 秒余量 + 90 秒 + 30 秒宽限：刚好够整段 + (150.0, True, 60.0), # 够收紧到 60 秒 + (120.0, True, 30.0), # 收紧到 30 秒：和收尾宽限一样长，还开始 + (119.0, False, None), # 不到 30 秒：连一次模型调用都不够，不再开始 + (10.0, False, None), + (-5.0, False, None), # 截止时刻已经过了 +]) +async def test_the_limit_is_tightened_by_what_is_left(clock, left, start, limit): + async with deadline.segment(clock.now + left, 600): + answer = deadline.allot(90, grace=30.0) + assert answer.start is start + assert answer.timeout_s == limit + assert answer.tightened is (start and limit != 90) + assert answer.seconds == 600, "不再开始、收紧过的说明里要写这一段的总时长" + + +async def test_a_configured_limit_shorter_than_the_grace_runs_whole_or_not_at_all(clock): + """配置的时长上限本来就比收尾宽限短（10 秒）：不收紧，够整段才开始。""" + async with deadline.segment(clock.now + 100, 600): + assert deadline.allot(10, grace=30.0) == deadline.Allot(True, 10, seconds=600) + async with deadline.segment(clock.now + 99, 600): + assert deadline.allot(10, grace=30.0) == deadline.Allot(False, None, seconds=600) + + +async def test_a_tightened_limit_is_cut_to_a_tenth_of_a_second(clock): + """收紧到的时长取到 0.1 秒、往下取：时间线上写「时长上限 69.7 秒」，不写 69.7012；往下取不会超出算好的余量。""" + async with deadline.segment(clock.now + 159.7012, 600): + assert deadline.allot(90, grace=30.0).timeout_s == 69.7 + async with deadline.segment(clock.now + 120.05, 600): + assert deadline.allot(90, grace=30.0).timeout_s == 30.0 + async with deadline.segment(clock.now + 119.99, 600): + assert not deadline.allot(90, grace=30.0).start, "差一点也是不够：往下取之后不到 30 秒" + + +async def test_an_unlimited_item_is_bounded_by_the_deadline(clock): + """配置的是不限（None）：有截止时刻时照样按剩余时间收紧，不然它能一直查到运行被中止。""" + async with deadline.segment(clock.now + 400, 600): + answer = deadline.allot(None, grace=30.0) + assert answer.start and answer.tightened and answer.timeout_s == pytest.approx(310.0) + async with deadline.segment(clock.now + 119, 600): + assert not deadline.allot(None, grace=30.0).start + + +@pytest.mark.parametrize("seconds, kept", [(600, 60.0), (300, 60.0), (1800, 180.0), (3600, 360.0)]) +def test_the_reserve_is_sixty_seconds_or_a_tenth(seconds, kept): + assert deadline.reserve(seconds) == kept + + +@pytest.mark.parametrize("seconds, kept", [(240, 60.0), (200, 50.0), (180, 45.0), (120, 30.0), (60, 15.0)]) +def test_a_short_segment_keeps_a_quarter(seconds, kept): + """这一段不到 240 秒：一份余量按总时长的 25% 算，不再固定 60 秒——不然 180 秒的一段，前面的节点用掉 60 秒之后核查 + 一项都开始不了，比没有这道检查时还差。""" + assert deadline.reserve(seconds) == kept + + +@pytest.mark.parametrize("seconds, after, kept", [ + (600, 1, 120.0), (600, 2, 180.0), (600, 4, 300.0), (600, 9, 300.0), # 最多留一半 + (1800, 2, 540.0), (180, 1, 90.0), (180, 3, 90.0), +]) +def test_each_model_node_downstream_adds_a_share(seconds, after, kept): + """下游这一段里每个还要调模型的节点再各加一份：双盲复算后面接着写报告、反方审稿时，只留 60 秒的话写报告的那次模型 + 调用一慢，运行照样被中止。合计不超过这一段总时长的一半，核查节点总还能开始几项。""" + assert deadline.reserve(seconds, after=after) == kept + + +async def test_the_reserve_follows_the_segment_length(clock): + """时长上限调到 1800 秒：给下游留 180 秒，剩 300 秒时这一项最多 90 秒（300 − 180 − 30）。""" + async with deadline.segment(clock.now + 300, 1800): + assert deadline.allot(120, grace=30.0).timeout_s == pytest.approx(90.0) + + +async def test_downstream_model_nodes_take_their_share(clock): + """下游还有写报告、反方审稿：剩 300 秒时给下游留 180 秒，这一项最多 90 秒（300 − 180 − 30）；剩 239 秒时（不到 + 180 + 30 + 30）不再开始。""" + async with deadline.segment(clock.now + 300, 600): + assert deadline.allot(120, grace=30.0, after=2) == deadline.Allot(True, 90.0, tightened=True, seconds=600) + assert deadline.allot(120, grace=30.0).timeout_s == 120, "下游没有调模型的节点：照旧只留一份" + async with deadline.segment(clock.now + 239, 600): + assert not deadline.allot(120, grace=30.0, after=2).start + + +async def test_a_short_segment_still_starts(clock): + """单次运行时限只有 150 秒、前面用掉 50 秒：留 37.5 秒余量、30 秒收尾宽限，这一项还能查 32.5 秒，照样开始。""" + async with deadline.segment(clock.now + 100, 150): + answer = deadline.allot(90, grace=30.0) + assert answer.start and answer.timeout_s == pytest.approx(32.5) and answer.tightened + + +def spec_of(*nodes, edges) -> GraphSpec: + return GraphSpec.model_validate({"nodes": [node(nid, ntype) for nid, ntype in nodes], + "edges": [{"source": a, "target": b} for a, b in edges]}) + + +def test_model_nodes_after_count_this_segment_only(): + """数下游这一段里还要调模型的节点：写报告、反方审稿、模型、Agent……各算一个；取数、整形、成果这类不算；人工审批之后 + 是新的一段（重新计时），不算；分支汇合到同一个节点只算一次。""" + spec = spec_of(("start", "input"), ("fetch", "tool"), ("card", "metrics"), ("dual", "dual_check"), + ("write", "report"), ("chal", "challenge"), ("shape", "transform"), ("out", "output"), + edges=[("start", "fetch"), ("fetch", "card"), ("card", "dual"), ("dual", "write"), + ("write", "chal"), ("chal", "shape"), ("shape", "out")]) + assert deadline.models_after(spec, "dual") == 2 + assert deadline.models_after(spec, "write") == 1 + assert deadline.models_after(spec, "chal") == 0 + gated = spec_of(("dual", "dual_check"), ("ask", "llm"), ("gate", "human"), ("write", "report"), ("out", "output"), + edges=[("dual", "ask"), ("dual", "gate"), ("gate", "write"), ("write", "out"), ("ask", "out")]) + assert deadline.models_after(gated, "dual") == 1, "人工审批之后的写报告在下一段，不占这一段的时间" + forked = spec_of(("dual", "dual_check"), ("a", "agent"), ("b", "agent"), ("join", "report"), + edges=[("dual", "a"), ("dual", "b"), ("a", "join"), ("b", "join")]) + assert deadline.models_after(forked, "dual") == 3 + looped = spec_of(("chal", "challenge"), ("ask", "llm"), edges=[("chal", "ask"), ("ask", "chal")]) + assert deadline.models_after(looped, "chal") == 1, "环回到自己：自己不算" + assert deadline.models_after(looped, "nowhere") == 0 + + +async def test_a_squeezed_check_says_the_run_deadline_when_it_stops(): + """收紧过的核查到了点：核查 agent 发的 verify_limit 写的是收紧后的秒数（「时长上限 52.3 秒」），和任何设置都对不上。 + relabel 包住 emit，把这一条改说运行时长上限将至、带上 cause 和这一段的总时长；别的事件、别的上限原样转发；没收紧 + 时原样交回 emit。""" + sent: list[tuple[str, dict]] = [] + + def emit(kind, **data): + sent.append((str(kind), data)) + + assert deadline.relabel(emit, deadline.Allot(True, 90)) is emit + assert deadline.relabel(emit, deadline.Allot(True, 90, seconds=600)) is emit, "有截止时刻但没收紧：照旧" + wrapped = deadline.relabel(emit, deadline.Allot(True, 52.3, tightened=True, seconds=600)) + wrapped(EventType.LOG, level="warn", code="verify_limit", limit="timeout_s", + message="核查已达上限（时长上限 52.3 秒），已停止查询，按已有结果收尾") + wrapped(EventType.LOG, level="warn", code="verify_limit", limit="max_queries", + message="核查已达上限（最多查询 6 次），已停止查询，按已有结果收尾") + wrapped(EventType.TOOL_END, tool="db_query__shop", limit="timeout_s") + squeezed, queries, tool = sent + assert squeezed == (EventType.LOG, { + "level": "warn", "code": "verify_limit", "limit": "timeout_s", "cause": deadline.RUN_DEADLINE, + "run_seconds": 600, + "message": "核查已达上限（本次运行即将达到 600 秒的时长上限，时长已按剩余时间收紧到 52.3 秒），已停止查询，" + "按已有结果收尾"}) + assert deadline.RUN_DEADLINE == "run_deadline" + assert queries[1]["message"] == "核查已达上限（最多查询 6 次），已停止查询，按已有结果收尾" and "cause" not in queries[1] + assert tool == (EventType.TOOL_END, {"tool": "db_query__shop", "limit": "timeout_s"}) + + +async def test_the_segment_is_cleared_on_the_way_out(clock): + async with deadline.segment(clock.now + 600, 600): + assert deadline.current() == deadline.Segment(clock.now + 600, 600.0) + async with deadline.segment(None, 600): + assert deadline.current() is None, "没有上限的一段：里面看不到外面的截止时刻" + assert deadline.current() is not None + assert deadline.current() is None + with pytest.raises(RuntimeError): + async with deadline.segment(clock.now + 600, 600): + raise RuntimeError("节点出错") + assert deadline.current() is None, "出错退出也清掉" + + +# -------------------------------------------------------------------------- +# runner 放进来的截止时刻 +# -------------------------------------------------------------------------- + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) -> Run: + for _ in range(400): + 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}：{row.error}") + + +@task +async def _peek_in_task() -> object: + return deadline.current() + + +async def test_the_runner_hands_each_segment_its_deadline(monkeypatch, engine_up): + """节点、节点里的 @task 都看得到这一段的截止时刻：就是 asyncio.timeout 的那一个（开始时刻 + max_run_seconds）。 + 审批恢复后的那一段是新的截止时刻；运行结束后发起它的地方看不到。""" + seen: list[tuple[str, object, object, float]] = [] + + async def peek(state, ctx): + seen.append((ctx.node.id, deadline.current(), await _peek_in_task(), asyncio.get_running_loop().time())) + return {} + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, peek) + monkeypatch.setattr(settings, "max_run_seconds", 321) + peeked = {"mode": "template", "template": "好", "assign_to": "x"} + spec = chain(node("start", "input"), node("before", "transform", **peeked), + node("gate", "human", mode="approve", title="继续吗"), node("after", "transform", **peeked), + node("out", "output", fields=[{"name": "结果", "value": "好"}])) + run = await run_manager.start(graph=spec, input_payload={}) + row = await wait(run.id) + assert row.status == "interrupted", row.error + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where(Approval.run_id == run.id))).scalar_one() + await run_manager.resume(run.id, {"approved": True}, approval_id=approval.id) + row = await wait(run.id, ("succeeded", "failed")) + assert row.status == "succeeded", row.error + + (first_id, first, first_task, first_at), (second_id, second, second_task, second_at) = seen + assert (first_id, second_id) == ("before", "after") + for segment, in_task, at in ((first, first_task, first_at), (second, second_task, second_at)): + assert isinstance(segment, deadline.Segment) and segment.seconds == 321 + assert in_task == segment, "节点里的 @task 看到的是同一个截止时刻" + assert 300 < segment.deadline - at <= 321, "截止时刻是这一段开始时刻 + max_run_seconds，用事件循环的钟" + assert second.deadline > first.deadline, "审批恢复后的那一段从恢复时重新计时" + assert deadline.current() is None, "截止时刻只在 runner 的执行里，发起运行的地方看不到" diff --git a/backend/tests/test_verify_deadline.py b/backend/tests/test_verify_deadline.py new file mode 100644 index 00000000..dbb7bdc5 --- /dev/null +++ b/backend/tests/test_verify_deadline.py @@ -0,0 +1,512 @@ +"""核查节点在运行时长上限内收尾（engine/deadline.py）：反方审稿逐句、双盲复算逐项开始之前，看这一段执行还剩多少时间。 + +- 反方审稿：5 句、截止时刻只够审 2 句 → 审 2 句，其余 3 句记为未能反查（理由、时间线写明运行时长上限将至），运行 + 成功、不被中止，日志和产出的 unreviewed 对得上；剩下的时间不够整段时这句的时长收紧，到点按查到的结果收尾，说的是 + 运行时长上限将至，不说「每句时长上限 90 秒」；接着运行时取 checkpoint 里的审稿，不重审 +- 双盲复算：同理，没来得及开始的盲算按无法复算记（照现有的路径记缺口）；收紧过的时长到点也按无法复算记，不发提醒 + 调高上限的 dual_check_limit；审批恢复后重放，盲算不再执行、哪几项没来得及照旧 +- 因运行时长上限将至没核成的，时间线另起一条、带 cause: run_deadline 和这一段的总时长（run_seconds）：界面据此说清 + 原因、给对的下一步（调高核查自己的上限只会更快撞线，检查模型、数据源也无济于事），不和别的原因混在一条里、被「等」 + 盖掉；收紧过的核查 agent 自己发的 verify_limit 也改说运行时长上限将至 +- 下游还有写报告这类要调模型的节点时，核查给它们多留一份余量：双盲复算后面接着写报告，运行不在写报告时被中止 +- 时间充裕时（默认 600 秒的一段）时长上限照传原值：和没有这道检查时一样 + +两种钟：精确的计数用假的钟（每派出一项就把 allot() 看到的钟往前拨若干秒，真实只过去几毫秒，runner 的 asyncio.timeout +远远不到）；「运行不被中止」用真的钟、真的短时长上限和真的慢 agent，只断言不随机器快慢变的事。模型都是 MockChatModel +剧本，数据源是现造的 SQLite 文件，区域、金额都是虚构的。 +""" +from __future__ import annotations + +import asyncio +from collections.abc import Callable + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.core import artifact_store +from app.core.config import settings +from app.db.base import SessionLocal +from app.db.models import Approval, Run +from app.engine import compiler, deadline, verify_agent +from app.engine.nodes import challenge as challenge_mod +from app.engine.nodes import dual_check as dual_check_mod +from app.engine.runner import run_manager, verify_manifest +from app.engine.schema import NodeType +from app.main import app +from app.providers.mock_model import MockChatModel +from tests.fixtures.dual_check_shop import WEEK, Blind, add_source, build +from tests.fixtures.sources import drop_source +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["cause"] == deadline.RUN_DEADLINE and warn["run_seconds"] == 600, "界面按 cause 说原因、给下一步" + 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 limit["cause"] == deadline.RUN_DEADLINE and limit["run_seconds"] == 600 and limit["claims"] == 1, \ + "界面认 cause：不叫人调高每句的上限" + 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 and warn["cause"] == deadline.RUN_DEADLINE + + +async def test_the_run_deadline_gets_its_own_line(monkeypatch, shop): + """第 1 句反查一条查询都没执行成功（另一个原因），第 3 句起来不及审：时间线两条 challenge_unreviewed，各说各的原因。 + 运行时长上限将至的那条带 cause，不被前一句的原因盖掉、藏进「等」里——界面也就不会为了它叫人去查模型和数据源。""" + plays = dict(FIVE_PLAYS) + plays["居各区域之首"] = Play([], "weakened", "没有查询。") + models = Models(monkeypatch, shop, writer=(FIVE,), plays=plays) + Ticker(monkeypatch, challenge_mod, step=250) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + assert models.reviewed() == ORDER[:2] + out = await output_of(row.id, "chal") + assert out["unreviewed"] == 4 and out["counts"]["inconclusive"] == 4 + + other, late = await logs(row.id, "challenge_unreviewed") + assert other["claims_skipped"] == 1 and other["reason"] == "反查没有执行成功任何查询" and "cause" not in other + assert other["message"] == "反方审稿有 1 句结论未能反查，已记为无法判断：反查没有执行成功任何查询" + assert late["claims_skipped"] == 3 and late["reason"] == LATE_REST + assert late["cause"] == deadline.RUN_DEADLINE and late["run_seconds"] == 600 + assert late["message"] == f"反方审稿有 3 句结论未能反查，已记为无法判断：{LATE_REST}" + + +async def test_its_own_limit_and_the_run_deadline_are_told_apart(monkeypatch, shop): + """第 1 句触了自己的每句时长上限（90 秒，没收紧），第 2 句的时长按剩余时间收紧后也到了点：两条 challenge_limit 都是 + timeout_s，只有收紧的那条带 cause——时间线上不是两行一样的「反方审稿已达时长上限」，叫人调高每句上限的只有第 1 条。""" + Models(monkeypatch, shop, writer=(FIVE,), plays=FIVE_PLAYS) + ticker = Ticker(monkeypatch, challenge_mod, step=440, hit=lambda timeout: True) + row = await finish(graph(shop)) + assert row.status == "succeeded", row.error + first, second = ticker.timeouts + assert first == 90 and second < 90, ticker.timeouts + own, squeezed = await logs(row.id, "challenge_limit") + assert own["limit"] == squeezed["limit"] == "timeout_s" + assert "cause" not in own and "每句时长上限 90 秒" in own["message"] + assert squeezed["cause"] == deadline.RUN_DEADLINE and squeezed["run_seconds"] == 600 and squeezed["claims"] == 1 + assert "每句时长上限" not in squeezed["message"] + + +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"] and warn["cause"] == deadline.RUN_DEADLINE + 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"] and warn["cause"] == deadline.RUN_DEADLINE + # 核查 agent 到点自己也发 verify_limit：时长收紧过的那几句说运行时长上限将至，写的秒数不会是一个和设置对不上的数 + for said in await logs(row.id, "verify_limit"): + if said.get("cause") == deadline.RUN_DEADLINE: + assert "本次运行即将达到 7 秒的时长上限，时长已按剩余时间收紧到" in said["message"] and said["run_seconds"] == 7 + else: + assert "时长上限 1.5 秒" in said["message"] or said["limit"] != "timeout_s", said + + +# -------------------------------------------------------------------------- +# 双盲复算 +# -------------------------------------------------------------------------- + +#: 一条查询算出五个指标（按支付日、排除退款）：销售额 330.0、订单数 3、付款客户数 2、最大单笔 200.0、最小单笔 30.0 +FIVE_SQL = (f"SELECT SUM(amount) AS gmv, COUNT(*) AS orders, COUNT(DISTINCT customer_id) AS buyers, " + f"MAX(amount) AS top, MIN(amount) AS low FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}") +NAMES = {"gmv": "销售额", "orders": "订单数", "buyers": "付款客户数", "top": "最大单笔", "low": "最小单笔"} +AGGS = {"gmv": "SUM(amount)", "orders": "COUNT(*)", "buyers": "COUNT(DISTINCT customer_id)", "top": "MAX(amount)", + "low": "MIN(amount)"} +METRICS = [{"id": mid, "name": name, "expression": f"cell(nodes.fetch, 0, '{mid}')"} for mid, name in NAMES.items()] +#: 盲算各自写一条只算这一个数的 SQL：和原流程的结果一致 +PLANS = {name: (f"SELECT {AGGS[mid]} AS blind_v FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}", "blind_v") + for mid, name in NAMES.items()} +#: 没来得及开始的盲算（默认 600 秒的一段）：指标的原因、出具缺口 +LATE_BLIND = "本次运行即将达到 600 秒的时长上限，为保证运行按时完成，未进行盲算" + + +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 dual_graph(source: str, **dual) -> dict: + """取数 → 口径卡（五个指标）→ 双盲复算 → 成果。成果的叙述里没有数字：没有缺口时是完整出具。""" + nodes = [ + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": FIVE_SQL}), + node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=METRICS), + node("dual", "dual_check", "复核五个指标", **dual), + node("out", "output", fields=[{"name": "结论", "value": "本周指标见口径卡"}], + contract={"metrics_from": "card", "narrative": "本周指标见口径卡"}), + ] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +@pytest.fixture +async def orders(tmp_path): + name, source_id = await add_source(build(tmp_path / "orders.db")) + yield name + await drop_source(source_id) + + +async def dual_run(spec: dict, **kwargs) -> Run: + run = await run_manager.start(graph=spec, input_payload={}, run_class="formal", **kwargs) + return await wait(run.id, ("succeeded", "failed", "interrupted")) + + +def unchecked_reasons(out: dict) -> dict[str, str]: + return {m["metric_id"]: m["reason"] for m in out["metrics"] if m["status"] == "unchecked"} + + +async def test_five_metrics_with_time_for_two_check_two(monkeypatch, orders): + """截止时刻只够盲算 2 项（每项像花了 260 秒）：前 2 项照常盲算、一致，后 3 项不再开始，按无法复算记缺口；运行成功、 + 出具降档。时间线上 dual_check_unchecked 点名这 3 项和原因，不发提醒调高上限的 dual_check_limit。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=260) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.runs == 2 and ticker.timeouts == [120, 120], "时间充裕的两项照传每项时长上限，不收紧" + + out = await output_of(row.id, "dual") + assert out["status"] == "partial" and out["counts"] == {"agreed": 2, "disagreed": 0, "unchecked": 3} + assert [m["status"] for m in out["metrics"]] == ["agreed", "agreed", "unchecked", "unchecked", "unchecked"] + assert unchecked_reasons(out) == {"buyers": LATE_BLIND, "top": LATE_BLIND, "low": LATE_BLIND} + gaps = [f"指标「{NAMES[m]}」无法复算：{LATE_BLIND}" for m in ("buyers", "top", "low")] + assert out["issuance_gaps"] == gaps + assert row.output["_issuance"]["tier"] == "degraded" + assert row.output["_issuance"]["gaps"] == [f"双盲复算「复核五个指标」：{gap}" for gap in gaps] + + [unchecked] = await logs(row.id, "dual_check_unchecked") + assert unchecked["metrics"] == [{"id": m, "name": NAMES[m], "reason": LATE_BLIND} for m in ("buyers", "top", "low")] + assert unchecked["cause"] == deadline.RUN_DEADLINE and unchecked["run_seconds"] == 600, "界面按 cause 说原因、给下一步" + assert unchecked["message"] == ("本次运行即将达到 600 秒的时长上限，为保证运行按时完成，双盲复算有 3 个指标未能复算：" + "「付款客户数」「最大单笔」「最小单笔」，按无法复算计入缺口") + assert await logs(row.id, "dual_check_limit") == [] + assert (await verify_manifest(row.id))["ok"] is True + + +async def test_blind_checks_cut_by_the_deadline_get_their_own_line(monkeypatch, orders): + """「订单数」的盲算没查到这个数（另一个原因），后 3 项来不及开始：时间线两条 dual_check_unchecked。运行时长上限将至的 + 那条带 cause、只列那 3 项，界面据此说原因、给下一步；另一条照旧，不带 cause。""" + Blind(monkeypatch, orders, {name: plan for name, plan in PLANS.items() if name != "订单数"}) + Ticker(monkeypatch, verify_agent, step=260) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + out = await output_of(row.id, "dual") + assert unchecked_reasons(out) == {"orders": "盲算没有查到这个数", "buyers": LATE_BLIND, "top": LATE_BLIND, + "low": LATE_BLIND} + other, late = await logs(row.id, "dual_check_unchecked") + assert other["metrics"] == [{"id": "orders", "name": "订单数", "reason": "盲算没有查到这个数"}] and "cause" not in other + assert other["message"] == "双盲复算未能核对 1 个指标：「订单数」（盲算没有查到这个数），按无法复算计入缺口" + assert late["cause"] == deadline.RUN_DEADLINE and late["run_seconds"] == 600 + assert [m["id"] for m in late["metrics"]] == ["buyers", "top", "low"] + assert all(m["reason"] == LATE_BLIND for m in late["metrics"]) + + +async def test_no_time_for_even_the_first_blind_check(monkeypatch, orders): + """一开始就只剩 40 秒：第一项都不开始（后面的也就不再一个个去试），五项都按无法复算记，节点状态是一个都没核成。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=0, start=560) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.calls == [] and ticker.timeouts == [] + out = await output_of(row.id, "dual") + assert out["status"] == "unchecked" and out["counts"] == {"agreed": 0, "disagreed": 0, "unchecked": 5} + assert set(unchecked_reasons(out).values()) == {LATE_BLIND} + assert len(out["issuance_gaps"]) == 5 + + +async def test_a_blind_check_squeezed_by_the_deadline_says_so(monkeypatch, orders): + """第 2 项开始前只剩约 150 秒：时长收紧到 60 秒以内；它到点时按无法复算记，原因说运行时长上限将至、时长已收紧， + 不说「已达每项时长上限（120 秒）」，也不发叫人调大上限的 dual_check_limit。""" + blind = Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=450, hit=lambda timeout: timeout < 120) + row = await dual_run(dual_graph(orders)) + assert row.status == "succeeded", row.error + assert blind.runs == 2 + first, second = ticker.timeouts + assert first == 120 and 50 < second < 60, ticker.timeouts + out = await output_of(row.id, "dual") + reasons = unchecked_reasons(out) + assert set(reasons) == {"orders", "buyers", "top", "low"} + assert "本次运行即将达到 600 秒的时长上限" in reasons["orders"] and "收紧" in reasons["orders"] + assert "每项时长上限" not in reasons["orders"] + assert {reasons[m] for m in ("buyers", "top", "low")} == {LATE_BLIND} + assert await logs(row.id, "dual_check_limit") == [] + [unchecked] = await logs(row.id, "dual_check_unchecked") + assert unchecked["cause"] == deadline.RUN_DEADLINE and [m["id"] for m in unchecked["metrics"]] == \ + ["orders", "buyers", "top", "low"], "收紧后到点的和没来得及开始的是同一个原因：一条说完" + + +def dual_report_graph(source: str) -> dict: + """取数 → 口径卡 → 双盲复算 → 写报告 → 成果：Copilot 指南里的惯常接法，双盲复算下游还有一次写报告的模型调用。""" + nodes = dual_graph(source)["nodes"] + nodes.insert(4, node("write", "report", "写周报", instructions="写周报")) + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +@pytest.mark.parametrize("build, low, high", [(dual_graph, 100, 110), (dual_report_graph, 40, 50)]) +async def test_a_report_downstream_gets_its_share(monkeypatch, orders, build, low, high): + """第 3 项开始时只剩约 200 秒：下游只有成果节点时留 60 秒余量，这一项还能查 110 秒左右；下游还要写报告时再给它留 + 60 秒，这一项只查 50 秒左右——双盲复算用到最后一刻的话，写报告那次模型调用一慢，整次运行照样被中止。""" + Blind(monkeypatch, orders, PLANS) + ticker = Ticker(monkeypatch, verify_agent, step=200) + row = await dual_run(build(orders)) + assert row.status == "succeeded", row.error + first, second, third = ticker.timeouts + assert first == second == 120 and low < third < high, ticker.timeouts + + +async def test_resuming_after_the_review_replays_the_blind_checks(monkeypatch, orders): + """销售额两路不一致、挂起等审核人裁决；后 3 项没来得及。审批恢复后节点从头重放：盲算取 checkpoint 里的，不再执行， + 没来得及的 3 项照旧——不看恢复后这一段还剩多少时间。""" + plans = dict(PLANS) + plans["销售额"] = (f"SELECT SUM(amount) AS blind_v FROM orders WHERE paid_date {WEEK}", "blind_v") # 没排除退款：380.0 + blind = Blind(monkeypatch, orders, plans) + ticker = Ticker(monkeypatch, verify_agent, step=260) + row = await dual_run(dual_graph(orders)) + assert row.status == "interrupted", row.error + calls, timeouts = list(blind.calls), list(ticker.timeouts) + assert blind.runs == 2 + + ticker.offset = 0 # 恢复后的一段：时间又充裕了，重做的话五项都会盲算 + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where( + Approval.run_id == row.id, Approval.status == "pending"))).scalar_one() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + res = await client.post(f"/api/approvals/{approval.id}/decide", + json={"approved": True, "value": {"choices": {"gmv": "a"}, "note": ""}}) + assert res.status_code == 200, res.text + row = await wait(row.id) + assert row.status == "succeeded", row.error + assert blind.calls == calls and ticker.timeouts == timeouts, "恢复后盲算又执行了：没取 checkpoint" + out = await output_of(row.id, "dual") + assert out["status"] == "resolved" and out["choices"] == {"gmv": "a"} + assert [m["status"] for m in out["metrics"]] == ["disagreed", "agreed", "unchecked", "unchecked", "unchecked"] + assert set(unchecked_reasons(out).values()) == {LATE_BLIND} + assert len(await logs(row.id, "dual_check_unchecked")) == 1, "盲算那一步的日志只发一次" + + +async def test_slow_blind_checks_no_longer_abort_the_run(monkeypatch, orders): + """真的钟：这一段只有 5 秒，五项盲算一项接一项（并行数 1），每项约 1 秒（模型每次调用 0.3 秒，每项 3 次），合起来 + 超过 5 秒——以前整次运行在盲算到一半时被中止。现在时间不够就不再开始、开始的按剩余时间收紧，没来得及的按无法复算 + 记，运行成功。每项时长上限 3 秒，一项自己用不完：无法复算的只会是因为运行时长上限将至。几项来得及随机器快慢变， + 只断言不变的。""" + monkeypatch.setattr(settings, "max_run_seconds", 5) + monkeypatch.setattr(deadline, "RESERVE_S", 1.2) + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 0.4) + monkeypatch.setattr(dual_check_mod, "BLIND_CONCURRENCY", 1) + + async def slow_model(session, spec): + return MockChatModel(model_name="slow-verify", latency=0.3), "slow-verify" + + monkeypatch.setattr(verify_agent, "get_chat_model", slow_model) + Blind(monkeypatch, orders, PLANS) + row = await dual_run(dual_graph(orders), options={"verify": {"dual_check": {"timeout_s": 3}}}) + assert row.status == "succeeded", row.error + assert "时长上限，已中止" not in (row.error or "") + + out = await output_of(row.id, "dual") + reasons = unchecked_reasons(out) + assert out["counts"]["agreed"] >= 1 and reasons, out["metrics"] + assert out["counts"]["agreed"] + len(reasons) == 5 and out["counts"]["disagreed"] == 0 + assert all("本次运行即将达到 5 秒的时长上限" in why for why in reasons.values()), reasons + assert [m["status"] for m in out["metrics"]][:out["counts"]["agreed"]] == ["agreed"] * out["counts"]["agreed"], \ + "一项接一项：来得及的在前" + [unchecked] = await logs(row.id, "dual_check_unchecked") + assert {m["id"] for m in unchecked["metrics"]} == set(reasons) + + +async def test_a_slow_report_after_the_blind_checks_still_finishes(monkeypatch, orders): + """真的钟：双盲复算 → 写报告 → 成果，这一段只有 10 秒。盲算一项接一项，每项都查到时长用完（查不完的查询）；写报告的 + 执行器换成一个要 3 秒的替身（像写作模型慢）。双盲复算给下游的写报告多留了一份余量（4 秒），收尾之后报告写得完，运行 + 照常完成——只留一份（2 秒）的话，盲算用到最后，报告写到一半整次运行就被中止。几项来得及随机器快慢变，只断言不变的。""" + monkeypatch.setattr(settings, "max_run_seconds", 10) + monkeypatch.setattr(deadline, "RESERVE_S", 2.0) + monkeypatch.setattr(verify_agent, "SETTLE_GRACE_S", 0.4) + monkeypatch.setattr(dual_check_mod, "BLIND_CONCURRENCY", 1) + + async def slow_model(session, spec): + return MockChatModel(model_name="slow-verify", latency=0.1), "slow-verify" + + monkeypatch.setattr(verify_agent, "get_chat_model", slow_model) + endless = [f"SELECT COUNT(*) AS blind_v FROM orders WHERE id > {i}" for i in range(300)] + Blind(monkeypatch, orders, {name: (endless, "blind_v") for name in NAMES.values()}) + wrote: list[float] = [] + + async def slow_report(state, ctx): + await asyncio.sleep(3.0) + wrote.append(asyncio.get_running_loop().time()) + return {} + + monkeypatch.setitem(compiler.RUNNERS, NodeType.REPORT, slow_report) + row = await dual_run(dual_report_graph(orders), + options={"verify": {"dual_check": {"timeout_s": 3, "max_queries": None}}}) + assert row.status == "succeeded", row.error + assert "时长上限，已中止" not in (row.error or "") + assert len(wrote) == 1, "报告写完了" + out = await output_of(row.id, "dual") + reasons = unchecked_reasons(out) + assert any("本次运行即将达到 10 秒的时长上限" in why for why in reasons.values()), reasons diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 9ef3d6f2..5c0e6215 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -3846,6 +3846,12 @@ export const DUAL_CHECK_TEXT = { } as Record, limitAny: '上限', limitDailyNext: '今日核查费用已达上限。请在「设置 → 偏好设置 → 核查」中调高每日金额上限，或等到次日再运行', + /** + * 单次运行时限将至（日志带 cause: run_deadline，服务端 engine/deadline.py）：一段执行（两次人工审批之间）有单次运行 + * 时限，核查节点为后续节点预留时间，剩下的时间不够就不再开始新的一项，开始了的按剩余时间缩短时长。双盲复算、反方审稿 + * 这几行共用这句副标题，seconds 是单次运行时限（设置页「运行环境」里的叫法） + */ + deadlineSub: (seconds: number) => `单次运行时限为 ${formatNumber(seconds)} 秒，已为后续节点预留时间`, log: { skippedDisabled: '双盲复算未启用，已跳过', skippedDisabledNext: '如需复算，请在节点的「启用双盲复算」中选择「开启」，或在「设置 → 偏好设置 → 核查」中启用双盲复算', @@ -3887,6 +3893,14 @@ export const DUAL_CHECK_TEXT = { unchecked: (n?: number) => (n ? `双盲复算有 ${formatNumber(n)} 个指标无法复算，已记为缺口` : '双盲复算有指标无法复算，已记为缺口'), uncheckedNext: '无法复算的原因见展开内容。来自合并查询、遮罩列的输入无法单独复算；盲算触顶、出错的，可调高「设置 → 偏好设置 → 核查」中的上限后重新运行', + /** + * dual_check_unchecked 带 cause: run_deadline：单次运行时限将至，没来得及盲算、缩短时长后到时没算完的指标（服务端 + * 另起这一条，原因都一样，副标题只点名）。调高每项的上限只会更快撞线，检查模型、数据源也无济于事：下一步说怎么腾出时间 + */ + deadline: (n?: number) => (n ? `运行时限将至，双盲复算有 ${formatNumber(n)} 个指标未能复算，已记为缺口` + : '运行时限将至，双盲复算有指标未能复算，已记为缺口'), + deadlineNext: '如需全部复算，可在节点的「复算的指标」中减少指标，或在双盲复算前加人工审批（时限按审批前后分段计算）；' + + '确需更长时间，请联系管理员调大单次运行时限（AGENTLAB_MAX_RUN_SECONDS）', }, /** * 出具缺口里双盲复算、反方审稿报上来的那几条（缺口以「双盲复算「节点名」：」「反方审稿：」开头）：时间线、成果卡片、 @@ -4072,6 +4086,19 @@ export const CHALLENGE_TEXT = { unreviewed: (n?: number) => (n ? `反方审稿有 ${formatNumber(n)} 句结论未能反查，已记为无法判断` : '反方审稿有结论未能反查，已记为无法判断'), unreviewedNext: '原因见展开内容。请检查核查模型和数据源是否可用，修正后重新运行；报告未引用数据库查询的结论无法反查', + /** + * 带 cause: run_deadline 的两条（服务端 engine/deadline.py）：单次运行时限将至，反方审稿为后续节点预留时间。 + * challenge_unreviewed 是剩下的句子不再反查；challenge_limit 是开始了的句子按剩余时间缩短了反查时长、到时按查到的结果 + * 收尾（n：这样的句数，skipped：其中未能给出判断的）。调高每句的上限只会更快撞线，检查模型、数据源也无济于事 + */ + deadline: (n?: number) => (n ? `运行时限将至，反方审稿有 ${formatNumber(n)} 句结论未反查，已记为无法判断` + : '运行时限将至，反方审稿有结论未反查，已记为无法判断'), + deadlineLimit: (n?: number, skipped?: number) => { + const head = n ? `运行时限将至，反方审稿缩短了 ${formatNumber(n)} 句结论的反查时长` : '运行时限将至，反方审稿缩短了反查时长' + return skipped ? `${head}，其中 ${formatNumber(skipped)} 句未能给出判断` : `${head}，已按查到的结果给出判断` + }, + deadlineNext: '如需全部审完，可在节点或「设置 → 偏好设置 → 核查」中减少每份报告审的句数，或在反方审稿前加人工审批' + + '（时限按审批前后分段计算）；确需更长时间，请联系管理员调大单次运行时限（AGENTLAB_MAX_RUN_SECONDS）', /** challenge_sources_lost：反查了，但该查的数据源有停用了的，判断不含它们的数据 */ sourcesLost: (n?: number) => (n ? `反方审稿有 ${formatNumber(n)} 句结论只查了部分数据源` : '反方审稿有结论只查了部分数据源'), sourcesLostSub: (names: string[]) => `未能查询：${names.map((x) => `「${x}」`).join('、')}（已不存在或已停用），判断不含其中的数据`, @@ -4135,9 +4162,14 @@ export const VERIFY_SETTING_TEXT = { } as Record, toleranceHint: '两边的值相差不超过这个比例时算一致，例如 0.000001 表示百万分之一，0 表示必须完全相同', unlimited: '不限', - /** 某项设成不限时写明它还受什么约束，不要悄悄生效。by：眼下还设着上限的相关项 */ - unlimitedNote: (kind: 'queries' | 'claims' | 'timeout' | 'daily', by: string[]) => { + /** + * 某项设成不限时写明它还受什么约束，不要悄悄生效。by：眼下还设着上限的相关项。runTimeout 是双盲复算、反方审稿的 + * 时长：它们在运行里执行，一段执行有单次运行时限，时限将至时按剩余时间缩短（服务端 engine/deadline.py），设成不限 + * 也一样；timeout 是审批质询的，不在运行的那一段里执行 + */ + unlimitedNote: (kind: 'queries' | 'claims' | 'timeout' | 'runTimeout' | 'daily', by: string[]) => { if (kind === 'timeout') return '不设上限，时长仅受模型接口自身的超时约束' + if (kind === 'runTimeout') return '不设上限，时长仍受单次运行时限约束，时限将至时按剩余时间缩短' const what = kind === 'queries' ? '查询次数' : kind === 'claims' ? '句数' : '每日核查费用' return by.length ? `不设上限，${what}仅受${by.join('、')}约束` : `不设上限，${what}不受其他上限约束` }, diff --git a/frontend/src/pages/VerifySettings.tsx b/frontend/src/pages/VerifySettings.tsx index e410375c..305ecff2 100644 --- a/frontend/src/pages/VerifySettings.tsx +++ b/frontend/src/pages/VerifySettings.tsx @@ -179,7 +179,8 @@ export function VerifySection({ cur, errors, onEdit, refresh }: { /** 某项设成不限时，它还受哪些眼下设着的上限约束 */ const note = (k: LimitKey): string => { - if (TIMEOUTS.has(k)) return S.unlimitedNote('timeout', []) + // 审批质询不在运行的那一段里执行，只受模型接口的超时约束；双盲复算、反方审稿还受单次运行时限约束 + if (TIMEOUTS.has(k)) return S.unlimitedNote(k === 'inquiry.timeout_s' ? 'timeout' : 'runTimeout', []) if (k === 'challenge.max_claims') return S.unlimitedNote('claims', [S.bound.claims]) if (k === 'daily_max_usd') { const counts = [...COUNTS].some((c) => cur[c] !== null) diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index fd01b6c7..499944a0 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -807,6 +807,14 @@ const cjkText = (v: unknown): string | undefined => (typeof v === 'string' && /[ /** 个数：给的是数就取整，给的是清单就数一下，别的（缺了、写坏了）当不知道 */ const countOf = (v: unknown): number | undefined => (typeof v === 'number' && Number.isFinite(v) && v >= 0 ? Math.round(v) : Array.isArray(v) ? v.length : undefined) +/** + * 核查类日志的 cause：单次运行时限将至（服务端 engine/deadline.py）。核查节点为后续节点预留时间，没开始的、缩短时长后 + * 到时没做完的另起一条带上它。这几行不叫人调高核查的上限（只会更快撞线），也不叫人去查模型和数据源 + */ +const RUN_DEADLINE = 'run_deadline' +/** run_seconds（单次运行时限）→ 那句副标题；不是正数（缺了、写坏了）就不写 */ +const deadlineSub = (v: unknown): string | undefined => + (typeof v === 'number' && Number.isFinite(v) && v > 0 ? DC.deadlineSub(v) : undefined) /** * 几种「看着跑完了、其实没做成」的状况（NI-3/4/5）。后端这几条日志是写给排查的，原样 @@ -1148,17 +1156,20 @@ function explainLog(code: string | undefined, message: string, d: Record { const name = str(m.name) ?? str(m.id) - const why = cjkText(m.reason) + const why = late ? undefined : cjkText(m.reason) return name ? `「${name}」${why ? `：${why}` : ''}` : '' }).filter(Boolean) + const sub = late ? [parts.join('、'), deadlineSub(d.run_seconds)].filter(Boolean).join('；') : parts.join('；') return { - title: DC.log.unchecked(list.length || undefined), - sub: parts.length ? parts.join('；') : undefined, - next: DC.log.uncheckedNext, + title: (late ? DC.log.deadline : DC.log.unchecked)(list.length || undefined), + sub: sub || undefined, + next: late ? DC.log.deadlineNext : DC.log.uncheckedNext, level: 'warn', } } @@ -1201,6 +1212,16 @@ function explainLog(code: string | undefined, message: string, d: Record { + // 一段执行（两次人工审批之间）有单次运行时限。双盲复算、反方审稿为后续节点预留时间：剩下的时间不够就不再开始，开始了的 + // 按剩余时间缩短时长（服务端 engine/deadline.py）。以前这几项照常记成「已达时长上限」「未能反查」「无法复算」，下一步叫人 + // 调高每句、每项的上限（调高了只会更快撞线），或者去查核查模型和数据源（它们本来就没问题）。现在服务端另起一条、带 + // cause: run_deadline 和 run_seconds，界面照它说。先核对后端还是这么发的 + const dlPy = readFileSync(`${root}backend/app/engine/deadline.py`, 'utf8') + const chPy = readFileSync(`${root}backend/app/engine/nodes/challenge.py`, 'utf8') + const dcPy = readFileSync(`${root}backend/app/engine/nodes/dual_check.py`, 'utf8') + check('后端发的还是这几种形状', dlPy.includes('RUN_DEADLINE = "run_deadline"') + && chPy.includes('code="challenge_unreviewed", cause=deadline.RUN_DEADLINE') + && chPy.includes('{"cause": deadline.RUN_DEADLINE, "run_seconds": self.run_seconds, "claims": touched}') + && dcPy.includes('code="dual_check_unchecked", cause=late[0]["cause"], run_seconds=seconds'), + '改了 deadline.py、challenge.py、dual_check.py 里这几条日志的字段，就同步改 decode.ts 的 explainLog 和这里的夹具') + const E = (seq, type, node_id, data = {}) => ({ seq, type, node_id, ts: 1790000000 + seq, data }) + const LATE = '本次运行即将达到 600 秒的时长上限，为保证运行按时完成，剩下 3 句未审' + const BLIND = '本次运行即将达到 600 秒的时长上限，为保证运行按时完成，未进行盲算' + const events = [ + E(1, 'run.started', null, { total: 3 }), + E(2, 'node.started', 'dc', { node_type: 'dual_check', label: '复核五个指标' }), + // 同一节点里别的原因照旧一条，时限将至的另起一条 + E(3, 'log', 'dc', { level: 'warn', code: 'dual_check_unchecked', message: '双盲复算未能核对 1 个指标：「订单数」（盲算没有查到这个数），按无法复算计入缺口', + metrics: [{ id: 'orders', name: '订单数', reason: '盲算没有查到这个数' }] }), + E(4, 'log', 'dc', { level: 'warn', code: 'dual_check_unchecked', cause: 'run_deadline', run_seconds: 600, + message: `本次运行即将达到 600 秒的时长上限，为保证运行按时完成，双盲复算有 3 个指标未能复算：「付款客户数」「最大单笔」「low」，按无法复算计入缺口`, + metrics: [{ id: 'buyers', name: '付款客户数', reason: BLIND }, { id: 'top', name: '最大单笔', reason: BLIND }, { id: 'low', name: '', reason: BLIND }] }), + E(5, 'node.finished', 'dc', { duration_ms: 20 }), + E(6, 'node.started', 'ch', { node_type: 'challenge', label: '审周报' }), + // 第 1 句触了自己的每句时长上限，第 2 句的时长缩短过、到时没给出判断：两条都是 limit: timeout_s + E(7, 'log', 'ch', { level: 'warn', code: 'challenge_limit', limit: 'timeout_s', claims_skipped: 0, + message: '有 1 句反查已达上限（每句时长上限 90 秒），已按查到的结果收尾' }), + E(8, 'log', 'ch', { level: 'warn', code: 'challenge_limit', limit: 'timeout_s', claims_skipped: 1, cause: 'run_deadline', run_seconds: 600, claims: 1, + message: '有 1 句反查已达上限（本次运行即将达到 600 秒的时长上限，反查时长已按剩余时间收紧），已按查到的结果收尾，其中 1 句未能给出结论' }), + E(9, 'log', 'ch', { level: 'warn', code: 'challenge_unreviewed', claims_skipped: 1, reason: '反查没有执行成功任何查询', + message: '反方审稿有 1 句结论未能反查，已记为无法判断：反查没有执行成功任何查询' }), + E(10, 'log', 'ch', { level: 'warn', code: 'challenge_unreviewed', cause: 'run_deadline', run_seconds: 600, claims_skipped: 3, reason: LATE, + message: `反方审稿有 3 句结论未能反查，已记为无法判断：${LATE}` }), + E(11, 'node.finished', 'ch', { duration_ms: 30 }), + E(12, 'run.finished', null, { status: 'succeeded' }), + ] + const all = flatten(mod.decodeRun(events)) + const rows = (code) => all.filter((s) => s.code === code) + const [dcOther, dcLate] = rows('dual_check_unchecked') + const [ownLimit, lateLimit] = rows('challenge_limit') + const [chOther, chLate] = rows('challenge_unreviewed') + const late = [dcLate, lateLimit, chLate] + check('每条一行，挂在自己的节点下', all.find((s) => s.nodeId === 'dc' && s.kind === 'node')?.children?.filter((x) => x.code === 'dual_check_unchecked').length === 2 + && all.find((s) => s.nodeId === 'ch' && s.kind === 'node')?.children?.filter((x) => /^challenge_/.test(x.code ?? '')).length === 4, + all.map((s) => s.code).filter(Boolean).join('、')) + check('dual_check_unchecked：标题说运行时限将至，副标题只点名（没名字的写 id）、带上时限', dcLate?.title === '运行时限将至，双盲复算有 3 个指标未能复算，已记为缺口' + && dcLate.sub === '「付款客户数」、「最大单笔」、「low」；单次运行时限为 600 秒，已为后续节点预留时间' && dcLate.level === 'warn', `${dcLate?.title} / ${dcLate?.sub}`) + check('challenge_limit：缩短过的那条不说「已达时长上限」，说几句缩短了、几句没给出判断', lateLimit?.title === '运行时限将至，反方审稿缩短了 1 句结论的反查时长，其中 1 句未能给出判断' + && lateLimit.sub === '单次运行时限为 600 秒，已为后续节点预留时间' && lateLimit.level === 'warn', `${lateLimit?.title} / ${lateLimit?.sub}`) + check('challenge_unreviewed：标题说几句未反查、记为无法判断，副标题说时限', chLate?.title === '运行时限将至，反方审稿有 3 句结论未反查，已记为无法判断' + && chLate.sub === '单次运行时限为 600 秒，已为后续节点预留时间' && chLate.level === 'warn', `${chLate?.title} / ${chLate?.sub}`) + check('……三行的下一步都说怎么腾出时间：减少句数或指标、在核查前加人工审批、请管理员调大单次运行时限', late.every((s) => !!s?.next?.includes('人工审批') + && s.next.includes('单次运行时限') && s.next.includes('AGENTLAB_MAX_RUN_SECONDS')) + && dcLate?.next?.includes('「复算的指标」') && !!chLate?.next?.includes('句数') && lateLimit?.next === chLate?.next, late.map((s) => s?.next).join(' / ')) + check('……不叫人调高核查的上限，也不叫人去查核查模型和数据源', late.every((s) => !/调高|核查模型|数据源是否/.test(`${s?.title}${s?.sub ?? ''}${s?.next ?? ''}`)), + late.map((s) => s?.next).join(' / ')) + check('……原话留在展开区，标题、副标题、下一步里没有内部代码', late.every((s, i) => s?.detail === events[[3, 7, 9][i]].data.message + && !/run_deadline|cause|run_seconds|timeout_s/.test(`${s.title}${s.sub ?? ''}${s.next ?? ''}`))) + // 别的原因照旧：同一节点里的两条分得开，不是两行一样的「已达时长上限」 + check('别的原因那几行照旧：触了自己的上限仍叫人调高上限，未能反查、无法复算仍按原因说', ownLimit?.title === '反方审稿已达时长上限' + && !!ownLimit.next?.includes('调高') && ownLimit.title !== lateLimit?.title + && chOther?.title === '反方审稿有 1 句结论未能反查，已记为无法判断' && chOther.sub === '反查没有执行成功任何查询' && !!chOther.next?.includes('核查模型') + && dcOther?.title === '双盲复算有 1 个指标无法复算，已记为缺口' && dcOther.sub === '「订单数」：盲算没有查到这个数' && !!dcOther.next?.includes('合并查询'), + [ownLimit, chOther, dcOther].map((s) => `${s?.title}|${s?.sub ?? ''}`).join(' / ')) + + // 都缩短了、都给出了判断；缺字段、类型不对（别的服务端版本）：不崩，不出 undefined、NaN、[object Object]，不瞎填数 + const odd = flatten(mod.decodeRun([ + E(1, 'node.started', 'ch', { node_type: 'challenge' }), + E(2, 'log', 'ch', { level: 'warn', code: 'challenge_limit', limit: 'timeout_s', claims_skipped: 0, cause: 'run_deadline', run_seconds: 1800, claims: 2, message: 'x' }), + E(3, 'log', 'ch', { level: 'warn', code: 'challenge_limit', limit: 'timeout_s', cause: 'run_deadline', run_seconds: 'x', claims: 'y', message: 'x' }), + E(4, 'log', 'ch', { level: 'warn', code: 'challenge_unreviewed', cause: 'run_deadline', claims_skipped: 'x', reason: LATE, message: 'x' }), + E(5, 'log', 'ch', { level: 'warn', code: 'dual_check_unchecked', cause: 'run_deadline', run_seconds: -1, metrics: 'x', message: 'x' }), + E(6, 'node.finished', 'ch', {}), + ])).filter((s) => s.code && s.kind !== 'node') + const [settled, bareLimit, bareUnreviewed, bareUnchecked] = odd + check('都给出了判断：说按查到的结果给出判断，时限照写（大于 1,000 用千分位）', settled?.title === '运行时限将至，反方审稿缩短了 2 句结论的反查时长，已按查到的结果给出判断' + && settled.sub === '单次运行时限为 1,800 秒，已为后续节点预留时间', `${settled?.title} / ${settled?.sub}`) + check('缺字段：标题不带数，没给时限就不写那句副标题（未能反查的退回原因）', bareLimit?.title === '运行时限将至，反方审稿缩短了反查时长，已按查到的结果给出判断' + && bareLimit.sub === undefined && bareUnreviewed?.title === '运行时限将至，反方审稿有结论未反查，已记为无法判断' && bareUnreviewed.sub === LATE + && bareUnchecked?.title === '运行时限将至，双盲复算有指标未能复算，已记为缺口' && bareUnchecked.sub === undefined, + odd.map((s) => `${s.title}|${s.sub ?? ''}`).join(' ; ')) + check('……不出 undefined、NaN、[object Object]', !odd.some((s) => /undefined|NaN|object Object/.test(`${s.title}${s.sub ?? ''}${s.next ?? ''}`))) +}) + console.log(failed ? `\n✗ ${failed} 项未通过` : '\n✓ 解码器全部通过') process.exit(failed ? 1 : 0) diff --git a/scripts/check-verify.mjs b/scripts/check-verify.mjs index 62ecfdfb..d3c6ca5e 100644 --- a/scripts/check-verify.mjs +++ b/scripts/check-verify.mjs @@ -1640,6 +1640,42 @@ await section('设置页「核查」一节：老服务端没有这一组时不 } }) +await section('设置页「核查」一节：双盲复算、反方审稿的时长设成不限时，写明仍受单次运行时限约束；审批质询的照旧', async () => { + // 双盲复算、反方审稿在运行里执行：一段执行有单次运行时限，时限将至时每项、每句的时长按剩余时间缩短（服务端 + // engine/deadline.py），设成不限也一样。以前这里写「仅受模型接口自身的超时约束」，管理员会以为设成不限就能查到底。 + // 审批质询不在运行的那一段里执行，不受单次运行时限约束，照旧 + const off = (part) => ({ ...fx.verify[part], timeout_s: null }) + const { page, errors } = await settingsPage({ verify: { ...fx.verify, dual_check: off('dual_check'), challenge: off('challenge'), inquiry: off('inquiry') } }) + await page.waitForSelector('[data-verify-settings]') + const note = (k) => text(page.locator(`[data-verify-limit="${k}"] [data-verify-unlimited]`)) + const RUN = '不设上限，时长仍受单次运行时限约束，时限将至时按剩余时间缩短' + check('双盲复算、反方审稿：说仍受单次运行时限约束', await note('dual_check.timeout_s') === RUN && await note('challenge.timeout_s') === RUN, + `${await note('dual_check.timeout_s')} / ${await note('challenge.timeout_s')}`) + check('审批质询：照旧只受模型接口自身的超时约束', await note('inquiry.timeout_s') === '不设上限，时长仅受模型接口自身的超时约束', + await note('inquiry.timeout_s')) + check('没有运行时报错', errors.length === 0, errors.join('；')) + // 这句比原来长：390px 下照样折行、不横向溢出，深浅色都读得清 + for (const theme of ['dark', 'light']) { + const narrow = await settingsPage({ width: 390, height: 844, theme, + verify: { ...fx.verify, dual_check: off('dual_check'), challenge: off('challenge') } }) + await narrow.page.waitForSelector('[data-verify-settings]') + await narrow.page.locator('[data-verify-limit="challenge.timeout_s"]').scrollIntoViewIfNeeded() + const w = await narrow.page.evaluate(() => { + const box = document.querySelector('[data-verify-settings]') + const vw = document.documentElement.clientWidth + const right = Math.max(...[box, ...box.querySelectorAll('*')].map((e) => e.getBoundingClientRect().right)) + const hint = document.querySelector('[data-verify-limit="challenge.timeout_s"] [data-verify-unlimited]') + return { box: box.scrollWidth - box.clientWidth, beyond: Math.round(right - vw), color: hint && getComputedStyle(hint).color, + base: getComputedStyle(document.body).backgroundColor } + }) + const ratio = contrast(w.color, w.base, w.base) + check(`390px（${theme}）：这句说明折行、不横向溢出，对比度 ≥ 4.5`, w.box <= 0 && w.beyond <= 0 && ratio >= 4.5, + `${JSON.stringify({ box: w.box, beyond: w.beyond })} · ${ratio.toFixed(2)}`) + await shot(narrow.page, `settings-run-timeout-${theme}`) + check(`390px（${theme}）：没有运行时报错`, narrow.errors.length === 0, narrow.errors.join('；')) + } +}) + await browser.close() console.log(failed ? `\n✗ 双盲复算与反方审稿：${failed} 项未通过` : '\n✓ 双盲复算与反方审稿：全部通过') process.exit(failed ? 1 : 0)