合入期 5 前置修复 K4:运行把这一段执行的截止时刻交给节点(engine/deadline.py),反方审稿与双盲复算开始每一项前看剩余时间,按剩余时间收紧每项时长、给下游留出余量,来不及的记为未能反查 / 无法复算并写明「运行时限将至」,整次运行不再被时长上限中止;继续运行取 checkpoint 里的结果不重复反查

This commit is contained in:
ThearchyHelios
2026-10-06 16:26:45 +08:00
11 changed files with 1272 additions and 33 deletions
+167
View File
@@ -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"]
+62 -13
View File
@@ -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,
+53 -10
View File
@@ -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(
+3 -1
View File
@@ -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"]
):
+280
View File
@@ -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 的执行里,发起运行的地方看不到"
+512
View File
@@ -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
+34 -2
View File
@@ -3846,6 +3846,12 @@ export const DUAL_CHECK_TEXT = {
} as Record<string, string>,
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<string, string>,
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}不受其他上限约束`
},
+2 -1
View File
@@ -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)
+36 -6
View File
@@ -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<string,
}
}
case 'dual_check_unchecked': {
// metrics 是 [{id, name, reason}]:副标题点名、说原因(没起名字的写 id)
// metrics 是 [{id, name, reason}]:副标题点名、说原因(没起名字的写 id)。带 cause: run_deadline 的是单次运行时限
// 将至没核成的那几个(服务端另起这一条,原因都一样):副标题只点名、带上时限,下一步说怎么腾出时间
const list = Array.isArray(d.metrics) ? d.metrics.filter(isRecord) : []
const late = d.cause === RUN_DEADLINE
const parts = list.map((m) => {
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<string,
case 'challenge_limit': {
const which = str(d.limit)
const skipped = countOf(d.claims_skipped)
if (d.cause === RUN_DEADLINE) {
// 时长按剩余时间缩短过的那几句到了时(limit 照旧是 timeout_s,服务端单起这一条):不说「已达时长上限」,不叫人
// 调高每句的上限。claims:这样的句数
return {
title: CL.deadlineLimit(countOf(d.claims) || undefined, skipped || undefined),
sub: deadlineSub(d.run_seconds),
next: CL.deadlineNext,
level: 'warn',
}
}
return {
title: CL.limit((which && DC.limitLabel[which]) || DC.limitAny, skipped || undefined),
next: which === 'daily_max_usd' ? DC.limitDailyNext : CL.limitNext,
@@ -1208,7 +1229,16 @@ function explainLog(code: string | undefined, message: string, d: Record<string,
}
}
case 'challenge_unreviewed': {
// claims_skipped:未能反查的句数;reason:第一句未能反查的原因(一句中文)
// claims_skipped:未能反查的句数;reason:第一句未能反查的原因(一句中文)。带 cause: run_deadline 的是单次运行
// 时限将至、剩下没开始的句子(服务端另起这一条):副标题说时限,下一步说怎么腾出时间,不叫人去查模型和数据源
if (d.cause === RUN_DEADLINE) {
return {
title: CL.deadline(countOf(d.claims_skipped) || undefined),
sub: deadlineSub(d.run_seconds) ?? cjkText(d.reason),
next: CL.deadlineNext,
level: 'warn',
}
}
return {
title: CL.unreviewed(countOf(d.claims_skipped) || undefined),
sub: cjkText(d.reason),
+87
View File
@@ -2493,5 +2493,92 @@ await section('审批质询 approval.inquiry(期 4):一行「审批人质
check('……有回答、没写成没成:只说已回答,不说完成', oddRows[4]?.sub === '已回答' && oddRows[4].level === 'info', `${oddRows[4]?.sub} / ${oddRows[4]?.level}`)
})
await section('单次运行时限将至(cause: run_deadline):核查没做完的那几行说时限,下一步不叫人调高上限、不叫人查模型和数据源', async () => {
// 一段执行(两次人工审批之间)有单次运行时限。双盲复算、反方审稿为后续节点预留时间:剩下的时间不够就不再开始,开始了的
// 按剩余时间缩短时长(服务端 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)
+36
View File
@@ -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)