From 76495d9e393a66bee157ec8e5d5358d1a5b0efe0 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 03:12:42 -0400 Subject: [PATCH] =?UTF-8?q?fix(session,=5Frecovery,types):=20=E4=BF=AE=20C?= =?UTF-8?q?odex=20=E5=AF=B9=E6=8A=97=E5=AE=A1=E6=9F=A5=E6=8A=A5=E7=9A=84?= =?UTF-8?q?=E4=BA=94=E6=9D=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 逐条核实全部成立。最重的是第一条：停止判定的结果只存在于结束记录里，而步记录与结束记录是 两次写。崩在两者之间那次判定就丢了，恢复照常回到预算准入——一次「恰好用满预算完成」被改写 成「预算耗尽」（两条轨迹长度一模一样），一次已经达成目标的运行接着往下跑，一次该以连续解析 失败收尾的运行再花一次模型调用。修法是续跑时把那次判定重演一遍：判定要的东西全在日志里 （动作结果在步记录那条原子写里、工具名在步记录上、模型回复在模型调用结果里）。最终回答那 一档更麻烦，答案文本只存在于结束记录里，靠重新解释那条已存下来的回复找回来——那时副作用还 没发生，重新解释是安全的。五种收尾各一条测试。 其余四条： - 模型调用失败步在续跑时补写，prompt_chars 填了 0。重新装配出来的和被打断时是同一份， 照它算。填 0 是把轨迹里那一列改写成假值。 - 取消正好落在写运行开始记录那一下时不留取消标记，留下一份只有开始记录的日志：run 因标识 已存在而拒绝、resume 当成可以从第 0 步续跑。用 shield 让那条记录一定落地，取消结束记录 才有地方挂（反过来先写结束记录会拼出结构上说不通的日志）。 - ModelCallResult 能同时带 reply 和 failure，而恢复只看 reply is None，于是把一次失败的 调用当成成功、接着解释那段回复执行动作。加构造期不变量。 - call_id 空串一路能进持久化记录。docstring 里那句「绝不能是空串」原本没有任何东西守着。 顺带把主循环里重复的一次上下文装配去掉，并把只在续跑第一次迭代成立的那个分支挪出循环。 --- src/polyloop/_recovery/__init__.py | 40 ++++- src/polyloop/session/__init__.py | 168 +++++++++++++++----- src/polyloop/types/__init__.py | 21 +++ tests/unit/test_session.py | 237 +++++++++++++++++++++++++++-- 4 files changed, 407 insertions(+), 59 deletions(-) diff --git a/src/polyloop/_recovery/__init__.py b/src/polyloop/_recovery/__init__.py index 991ffc4..94fed63 100644 --- a/src/polyloop/_recovery/__init__.py +++ b/src/polyloop/_recovery/__init__.py @@ -45,6 +45,7 @@ from polyloop.types import ( ModelReply, ReplayPolicy, RunResult, + StepCompleted, StepRecord, ) @@ -125,6 +126,15 @@ class ResumePlan: pending_action_result_id: str | None = None #: 结束标记里存着的那份结果。 finished_result: RunResult | None = None + #: 最后一个完整落地的步，连同它的动作结果。 + #: + #: **续跑要拿它把那一步之后的停止判定重演一遍。** 那个判定的结果只存在于结束记录里，崩在 + #: 「步记录已落盘、结束记录没落盘」之间就丢了——而恢复照常回到预算准入，于是一次 + #: 「恰好用满预算完成」会被改写成「预算耗尽」，一次已经达成目标的运行会接着往下跑。 + last_step_completed: StepCompleted | None = None + #: 最后一步那次模型调用的回复。最终回答那一档要靠重新解释它把答案找回来——答案文本只存在 + #: 于结束记录里，步记录存的是回填进历史的那段。 + last_reply: ModelReply | None = None def _indexed_intents(intents: tuple[Intent, ...], kind: IntentKind) -> dict[int, Intent]: @@ -178,6 +188,25 @@ def _trailing_parse_failures(steps: tuple[StepRecord, ...]) -> int: return count +def _reply_for_step( + log: RunLog, model_intents: dict[int, Intent], step: StepCompleted | None +) -> ModelReply | None: + """找出某一步那次模型调用拿到的回复。 + + **靠意图记录的结果 ID 配对，不靠 ID 的拼法。** 结果 ID 长什么样是装配层的事，这个模块 + 不该知道——知道了的话，改一次 ID 拼法就要同时改这里，而漏改不会报错、只会让配对静默失效。 + """ + if step is None: + return None + intent = model_intents.get(step.step.step_idx) + if intent is None: + return None + for result in log.model_results: + if result.result_id == intent.result_id: + return result.reply + return None + + def _check_structural_invariants(log: RunLog) -> None: """把四态表最后一行（结果有、意图无）以及几种撞号在这里一次查完。""" if log.started is None and (log.intents or log.model_results or log.steps or log.finished): @@ -228,18 +257,21 @@ def plan_resume(log: RunLog, model_replay_policy: ReplayPolicy) -> ResumePlan: for entry in log.steps if entry.action_outcome is not None and entry.action_outcome.status is ActionStatus.EXECUTED ) + model_intents = _indexed_intents(log.intents, IntentKind.MODEL_CALL) + action_intents = _indexed_intents(log.intents, IntentKind.ACTION) + interrupted = len(steps) + + last_completed = max(log.steps, key=lambda entry: entry.step.step_idx, default=None) base = { "steps": steps, "steps_appended": len(steps), "actions_executed": executed_actions, "consecutive_parse_failures": _trailing_parse_failures(steps), "next_call_index": len(steps), + "last_step_completed": last_completed, + "last_reply": _reply_for_step(log, model_intents, last_completed), } - model_intents = _indexed_intents(log.intents, IntentKind.MODEL_CALL) - action_intents = _indexed_intents(log.intents, IntentKind.ACTION) - interrupted = len(steps) - ahead = sorted( index for index in set(model_intents) | set(action_intents) if index > interrupted ) diff --git a/src/polyloop/session/__init__.py b/src/polyloop/session/__init__.py index 16715c8..29ffd77 100644 --- a/src/polyloop/session/__init__.py +++ b/src/polyloop/session/__init__.py @@ -306,6 +306,103 @@ class _Driver: ) return result + async def _drain_within_grace(self, writing: "asyncio.Future[None]") -> None: + """等一个已经发出去的写入在宽限期内落完，等不到就放弃。 + + 取消进来时那次写入可能还在半路。不等它就写结束记录的话，两条记录可能以相反的顺序落地， + 而一份「先有结束记录、后有开始记录」的日志在恢复那边是结构上说不通的状态。 + """ + try: + await asyncio.wait_for( + asyncio.shield(writing), timeout=self._request.cancel_grace_seconds + ) + except TimeoutError: + writing.cancel() + logger.error( + "取消宽限期内没写完在途记录，运行 %s 的日志可能不完整", self._request.run_id + ) + except Exception: + logger.exception("取消时在途记录写失败，运行 %s 的日志可能不完整", self._request.run_id) + + async def start(self, snapshot: Mapping[str, str]) -> None: + """写运行开始记录。取消正好落在这一下时，屏蔽它让它写完再补结束记录。 + + 不屏蔽的话会留下一份**只有开始记录**的日志：`run` 因为标识已存在而拒绝，`resume` 把它 + 当成可以从第 0 步续跑——而它其实是被人主动叫停的。屏蔽让这条记录一定落地，取消结束 + 记录才有地方挂：先写结束记录再写开始记录会拼出一份结构上说不通的日志。 + """ + writing: asyncio.Future[None] = asyncio.ensure_future( + self._definition.store.write_run_started( + RunStarted(run_id=self._request.run_id, parameter_snapshot=snapshot) + ) + ) + try: + await asyncio.shield(writing) + except asyncio.CancelledError: + await self._drain_within_grace(writing) + await self._finish_cancelled() + raise + + def _assemble(self) -> tuple[Message, ...]: + return assemble( + context=self._request.context, + injections=self._request.injections, + steps=self._steps, + observation_template=self._request.observation_template, + ) + + def seed(self, plan: ResumePlan) -> None: + """把从日志读回来的计数与步序列装进来。""" + self._counters = RunCounters( + steps_appended=plan.steps_appended, + actions_executed=plan.actions_executed, + consecutive_parse_failures=plan.consecutive_parse_failures, + ) + self._steps = list(plan.steps) + + async def settle_interrupted_tail(self, plan: ResumePlan) -> RunResult | None: + """把最后一步之后的那次停止判定重演一遍。返回 `None` 表示那一步之后确实该接着跑。 + + **这一步不是可选的优化，是正确性。** 停止判定的结果只存在于结束记录里，而步记录与结束 + 记录是两次写；崩在两者之间，那次判定就丢了。恢复照常回到预算准入的话：一次「恰好用满 + 预算完成」会被改写成「预算耗尽」（两者的轨迹长度一模一样，事后分不出来），一次已经 + 达成目标的运行会接着往下跑，一次该以连续解析失败收尾的运行会再花一次模型调用。 + + 判定所需的东西全都在日志里：动作结果在步记录那条原子写里，工具名在步记录上，模型回复 + 在模型调用结果里。这里只是把它们重新读一遍。 + """ + entry = plan.last_step_completed + if entry is None: + return None + step = entry.step + + if not step.parse_ok and step.parse_error is None: + # 模型调用失败那一步。它一写完就该以模型调用失败收尾。 + return await self._finish(StopReason.LLM_ERROR) + + if not step.parse_ok: + stop = parse_failure_admission(self._counters, self._request.budget) + return None if stop is None else await self._finish(stop) + + if entry.action_outcome is not None: + spec = None if step.tool_name is None else self._request.tools.spec_for(step.tool_name) + stop = completion_verdict( + entry.action_outcome, bool(spec is not None and spec.completes_run) + ) + return None if stop is None else await self._finish(stop) + + # 有效决策却没有动作结果，只剩最终回答那一档。答案文本只存在于结束记录里，所以重新 + # 解释那条已经存下来的回复把它找回来——那时副作用还没发生，重新解释是安全的。 + if plan.last_reply is None: + raise CorruptLogError("最后一步是最终回答，却找不到它那次模型调用的回复") + decision = self._definition.decision_parser.parse(plan.last_reply).decision + if not isinstance(decision, FinalAnswer): + raise CorruptLogError( + "最后一步记成了最终回答，重新解释同一条回复却得到别的分支——" + "解释器在两次运行之间被换过，或者它不是确定性的" + ) + return await self._finish(StopReason.AGENT_FINISHED, final_answer=decision.text) + async def _finish_cancelled(self) -> None: """取消进来时尽力写下结束标记，宽限期用完就放弃。 @@ -322,22 +419,13 @@ class _Driver: final_answer=None, steps=tuple(self._steps), ) - writing = asyncio.ensure_future( - self._definition.store.write_run_finished( - RunFinished(run_id=self._request.run_id, result=result) + await self._drain_within_grace( + asyncio.ensure_future( + self._definition.store.write_run_finished( + RunFinished(run_id=self._request.run_id, result=result) + ) ) ) - try: - await asyncio.wait_for( - asyncio.shield(writing), timeout=self._request.cancel_grace_seconds - ) - except TimeoutError: - writing.cancel() - logger.error( - "取消宽限期内没写完结束记录，运行 %s 的日志缺结束标记", self._request.run_id - ) - except Exception: - logger.exception("取消时写结束记录失败，运行 %s 的日志缺结束标记", self._request.run_id) # -- 主循环 ------------------------------------------------------------- @@ -372,26 +460,23 @@ class _Driver: skip_model_intent = plan is not None and plan.action is ResumeAction.REDO_MODEL_CALL skip_action_intent = plan is not None and plan.action is ResumeAction.REPLAY_LAST_ACTION if plan is not None: - self._counters = RunCounters( - steps_appended=plan.steps_appended, - actions_executed=plan.actions_executed, - consecutive_parse_failures=plan.consecutive_parse_failures, - ) - self._steps = list(plan.steps) + self.seed(plan) pending_reply = plan.pending_reply pending_failure = plan.pending_failure + # 模型调用失败那一步的步记录还没写完就断了：补上它，然后以模型调用失败收尾。只可能 + # 发生在续跑的第一次迭代上，所以判在循环外面。 + # **规模照现在重新装配出来的算**，那和被打断时装配出来的是同一份——历史、上下文、注入 + # 内容都没变，而续跑守卫已经比对过它们了。填 0 会把那一列改写成一个假值。 + if pending_failure is not None: + started = time.monotonic() + chars = prompt_chars(self._assemble()) + await self._write_step(self._failed_call_step(len(self._steps), chars, started), None) + return await self._finish(StopReason.LLM_ERROR) + while True: call_index = len(self._steps) - # 模型调用失败那一步的步记录还没写完就断了：补上它，然后以模型调用失败收尾。 - if pending_failure is not None: - step = self._failed_call_step( - call_index, prompt_chars_used=0, started=time.monotonic() - ) - await self._write_step(step, None) - return await self._finish(StopReason.LLM_ERROR) - # A 预算准入。放在开头而不是上一次迭代的结尾：一次「恰好用满预算完成」的运行走的 # 是完成判定，放结尾它会先撞上预算上限，而两者的轨迹长度一模一样。 stop = budget_admission(self._counters, self._request.budget) @@ -399,12 +484,7 @@ class _Driver: return await self._finish(stop) started = time.monotonic() - messages = assemble( - context=self._request.context, - injections=self._request.injections, - steps=self._steps, - observation_template=self._request.observation_template, - ) + messages = self._assemble() chars = prompt_chars(messages) # B 规模判定。命中时不产生步记录、也不写任何意图——模型还没被调用、没花钱、没有 @@ -696,10 +776,9 @@ async def run(definition: AgentDefinition, request: RunRequest) -> RunResult: f"运行标识 {request.run_id!r} 已经有日志了。要接着跑用 resume；" "这里不覆盖，因为那会毁掉一次已经花完钱的运行的留痕" ) - await definition.store.write_run_started( - RunStarted(run_id=request.run_id, parameter_snapshot=_merged_snapshot(definition, request)) - ) - return await _Driver(definition, request).drive() + driver = _Driver(definition, request) + await driver.start(_merged_snapshot(definition, request)) + return await driver.drive() async def resume(definition: AgentDefinition, request: RunRequest) -> RunResult: @@ -736,10 +815,19 @@ async def resume(definition: AgentDefinition, request: RunRequest) -> RunResult: if plan.finished_result is None: raise CorruptLogError("有结束标记却读不出结果，这条记录坏了") return plan.finished_result + driver = _Driver(definition, request) if plan.action is ResumeAction.STOP_UNKNOWN: - return await _Driver(definition, request).finish_resume_unknown(plan) + return await driver.finish_resume_unknown(plan) - return await _Driver(definition, request).drive(plan) + if plan.action is ResumeAction.CONTINUE_AT_NEXT_STEP: + # 上一步是完整的，但它之后那次停止判定的结果只存在于结束记录里，而那条记录没写下来。 + # 先把它重演一遍——不重演的话，一次已经达成目标的运行会被改写成预算耗尽或者接着往下跑。 + driver.seed(plan) + settled = await driver.settle_interrupted_tail(plan) + if settled is not None: + return settled + + return await driver.drive(plan) __all__ = [ diff --git a/src/polyloop/types/__init__.py b/src/polyloop/types/__init__.py index 3ecb225..53d3e65 100644 --- a/src/polyloop/types/__init__.py +++ b/src/polyloop/types/__init__.py @@ -112,6 +112,15 @@ class ModelReply: #: 模型的推理段。同上。 thinking: str + def __post_init__(self) -> None: + """空串的连接键在构造期就拦住。 + + docstring 里那句「绝不能是空串」原本没有任何东西守着，于是一个被写成空串的键能一路 + 进到持久化记录里。它看起来合法、连表时静默匹配不上，而 `None` 至少能被显式筛出来。 + """ + if self.call_id == "": + raise ValueError("call_id 不能是空串：空串是个看起来合法的键，连表时静默匹配不上") + # --------------------------------------------------------------------------- # 动作结果 @@ -368,6 +377,18 @@ class ModelCallResult: #: 只需要知道「失败过」以及失败的大致形态。 failure: str | None + def __post_init__(self) -> None: + """`reply` 与 `failure` 恰好一个有值，这条在构造期就守住。 + + 两个都有值的记录同时声称这次调用成功过和失败过，而恢复只看 `reply is None`——于是它 + 会把一次失败的调用当成成功，接着去解释那段回复、执行动作。两个都为空则是一条什么都 + 没说的结果，恢复会把它当成成功但拿不到回复。 + """ + if (self.reply is None) == (self.failure is None): + raise ValueError( + f"reply 与 failure 必须恰好一个有值：reply={self.reply!r}, failure={self.failure!r}" + ) + @dataclass(frozen=True, slots=True, kw_only=True) class StepCompleted: diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 1d35410..66962cf 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -73,34 +73,35 @@ class FakeStore: 只在崩溃之后才看得出来。 """ - def __init__(self, log: RunLog | None = None) -> None: + def __init__(self, log: RunLog | None = None, *, drop_finished: bool = False) -> None: self.writes: list[object] = [] self._log = log or RunLog() + #: 模拟「步记录已落盘、结束记录没落盘」那个崩溃点：结束记录照常返回成功，但不进日志。 + self._drop_finished = drop_finished async def write_run_started(self, record: RunStarted) -> None: self.writes.append(record) - self._log = RunLog( - started=record, - intents=self._log.intents, - model_results=self._log.model_results, - steps=self._log.steps, - finished=self._log.finished, - ) + self._log = dataclasses.replace(self._log, started=record) async def write_intent(self, record: Intent) -> None: self.writes.append(record) + self._log = dataclasses.replace(self._log, intents=(*self._log.intents, record)) async def write_model_call_result(self, record: ModelCallResult) -> None: self.writes.append(record) + self._log = dataclasses.replace(self._log, model_results=(*self._log.model_results, record)) async def write_step_completed(self, record: StepCompleted) -> None: self.writes.append(record) + self._log = dataclasses.replace(self._log, steps=(*self._log.steps, record)) async def read_log(self, run_id: str) -> RunLog: return self._log async def write_run_finished(self, record: RunFinished) -> None: self.writes.append(record) + if not self._drop_finished: + self._log = dataclasses.replace(self._log, finished=record) def parameters(self) -> Mapping[str, str]: return {"kind": "memory"} @@ -699,13 +700,6 @@ async def test_resume_hands_back_a_finished_run_without_rerunning_it() -> None: model = FakeModel([_reply("go")]) definition = _definition(store, model, FakeParser({"go": ACT})) first = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) - store._log = RunLog( # noqa: SLF001 — 把刚写下的那些记录拼回一份可读的日志 - started=store.of_type(RunStarted)[0], # type: ignore[arg-type] - intents=tuple(store.of_type(Intent)), # type: ignore[arg-type] - model_results=tuple(store.of_type(ModelCallResult)), # type: ignore[arg-type] - steps=tuple(store.of_type(StepCompleted)), # type: ignore[arg-type] - finished=store.of_type(RunFinished)[0], # type: ignore[arg-type] - ) again = await resume(definition, _request(FakeExecutor([_outcome()]))) @@ -823,3 +817,216 @@ async def test_two_runs_sharing_a_definition_do_not_see_each_other() -> None: assert len(result_a.steps) == 1 assert len(result_b.steps) == 3 assert {step.step_idx for step in result_b.steps} == {0, 1, 2} + + +# --------------------------------------------------------------------------- +# 崩在「步记录已落盘、结束记录没落盘」之间 +# --------------------------------------------------------------------------- +# +# 停止判定的结果只存在于结束记录里，而步记录与结束记录是两次写。崩在两者之间，那次判定就 +# 丢了——恢复照常回到预算准入的话，结果会和不中断跑完时不一样，而两条轨迹长度完全相同。 +# 下面每一条都先跑一次「结束记录丢了」的运行，再续跑，断言续跑得出的是原本那个结论。 + + +async def _run_losing_the_finished_record( + model_script: Sequence[object], + parser_table: Mapping[str, object], + outcomes: Sequence[ActionOutcome], + *, + budget: Budget | None = None, +) -> tuple[FakeStore, AgentDefinition, RunRequest]: + store = FakeStore(drop_finished=True) + definition = _definition(store, FakeModel(model_script), FakeParser(parser_table)) + request = _request(FakeExecutor(list(outcomes)), budget=budget) + await run(definition, request) + return store, definition, request + + +async def test_resume_replays_the_completion_verdict_that_was_never_persisted() -> None: + """第 0 步达成目标、结束记录没落盘 → 续跑必须还是「目标达成」，不是「预算耗尽」。 + + 这是最典型的一种：`max_steps=1` 时恢复会走到预算准入，已追加步数正好达到上限，于是报 + 预算耗尽。两者的轨迹长度一模一样，事后从数据里分不出来。 + """ + budget = Budget( + max_steps=1, max_actions=9, max_consecutive_parse_failures=9, max_prompt_chars=100000 + ) + store, definition, request = await _run_losing_the_finished_record( + [_reply("go")], {"go": ACT}, [_outcome(completed=True)], budget=budget + ) + calls_before = len(store.of_type(Intent)) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.TASK_COMPLETED + assert len(result.steps) == 1 + assert len(store.of_type(Intent)) == calls_before # 没有再调一次模型 + + +async def test_resume_does_not_keep_running_a_run_that_already_reached_its_goal() -> None: + """预算还够的时候，同一个洞的表现是「已经做完了还接着跑」。""" + store, definition, request = await _run_losing_the_finished_record( + [_reply("go")], {"go": ACT}, [_outcome(completed=True)] + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.TASK_COMPLETED + assert len(result.steps) == 1 + + +async def test_resume_replays_the_env_error_verdict() -> None: + store, definition, request = await _run_losing_the_finished_record( + [_reply("go")], {"go": ACT}, [_outcome(ActionStatus.ENV_ERROR)] + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.ENV_ERROR + + +async def test_resume_replays_the_repeated_parse_failure_verdict() -> None: + """该以连续解析失败收尾的运行，续跑不能再花一次模型调用。""" + budget = Budget( + max_steps=9, max_actions=9, max_consecutive_parse_failures=2, max_prompt_chars=100000 + ) + store, definition, request = await _run_losing_the_finished_record( + [_reply("???")], {}, [_outcome()], budget=budget + ) + intents_before = len(store.of_type(Intent)) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.PARSE_FAILED_REPEATEDLY + assert len(store.of_type(Intent)) == intents_before + + +async def test_resume_recovers_the_final_answer_by_reparsing_the_stored_reply() -> None: + """最终回答那一档更麻烦：答案文本只存在于结束记录里，步记录存的是回填进历史的那段。 + + 靠重新解释那条已经存下来的模型回复把它找回来——那时副作用还没发生，重新解释是安全的。 + """ + store, definition, request = await _run_losing_the_finished_record( + [_reply("done")], {"done": FinalAnswer(text="42")}, [_outcome()] + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.AGENT_FINISHED + assert result.final_answer == "42" + + +async def test_resume_replays_the_model_call_failure_verdict() -> None: + store, definition, request = await _run_losing_the_finished_record( + [RuntimeError("网关连不上")], {}, [_outcome()] + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.LLM_ERROR + + +async def test_resume_keeps_going_when_the_last_step_really_did_not_stop_the_run() -> None: + """重演出来是「不停」的时候要接着跑，不能把每次续跑都变成立刻收尾。""" + store, definition, request = await _run_losing_the_finished_record( + [_reply("go"), _reply("done")], + {"go": ACT, "done": FinalAnswer(text="ok")}, + [_outcome()], + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.AGENT_FINISHED + assert len(result.steps) == 2 + + +async def test_a_recovered_failed_call_step_keeps_its_real_prompt_size() -> None: + """模型调用结果已落盘、失败步没落盘时补写的那一步，规模不能填 0。 + + 重新装配出来的和被打断时装配出来的是同一份——历史、上下文、注入内容都没变，而续跑守卫 + 已经比对过它们了。填 0 会把轨迹里那一列改写成一个假值。 + """ + store = FakeStore() + definition = _definition(store, FakeModel([RuntimeError("断了")]), FakeParser({})) + request = _request(FakeExecutor([_outcome()])) + snapshot = {**definition.parameter_snapshot(), **request.parameter_snapshot()} + store._log = RunLog( # noqa: SLF001 — 造一个正好断在第二次写之后的日志 + started=RunStarted(run_id="run-1", parameter_snapshot=snapshot), + intents=( + Intent( + run_id="run-1", + kind=IntentKind.MODEL_CALL, + call_index=0, + result_id="run-1#model#0", + replay_policy=ReplayPolicy.NEVER, + ), + ), + model_results=( + ModelCallResult(run_id="run-1", result_id="run-1#model#0", reply=None, failure="断了"), + ), + ) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.LLM_ERROR + assert result.steps[0].prompt_chars > 0 + + +async def test_cancelling_during_the_very_first_write_still_leaves_a_cancelled_marker() -> None: + """取消正好落在写运行开始记录那一下时，也要留下取消标记。 + + 不屏蔽的话会留下一份只有开始记录的日志：`run` 因为标识已存在而拒绝，`resume` 把它当成 + 可以从第 0 步续跑——而它其实是被人主动叫停的，两条路都走不通。 + """ + + class _GatedStore(FakeStore): + def __init__(self) -> None: + super().__init__() + self.gate = asyncio.Event() + + async def write_run_started(self, record: RunStarted) -> None: + await self.gate.wait() + await super().write_run_started(record) + + store = _GatedStore() + definition = _definition(store, FakeModel([]), FakeParser({})) + task = asyncio.ensure_future(run(definition, _request(FakeExecutor([])))) + await asyncio.sleep(0) + await asyncio.sleep(0) + task.cancel() + store.gate.set() + + with pytest.raises(asyncio.CancelledError): + await task + + assert len(store.of_type(RunStarted)) == 1 + finished = store.of_type(RunFinished) + assert len(finished) == 1 + assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# 记录自身的不变量 +# --------------------------------------------------------------------------- + + +def test_a_model_call_result_cannot_claim_both_success_and_failure() -> None: + """两个都有值的记录同时声称成功过和失败过，而恢复只看 `reply is None`。 + + 于是它会把一次失败的调用当成成功，接着去解释那段回复、执行动作。 + """ + with pytest.raises(ValueError, match="恰好一个有值"): + ModelCallResult( + run_id="run-1", result_id="r", reply=_reply("go"), failure="TimeoutError: x" + ) + + +def test_a_model_call_result_must_say_something() -> None: + with pytest.raises(ValueError, match="恰好一个有值"): + ModelCallResult(run_id="run-1", result_id="r", reply=None, failure=None) + + +def test_an_empty_call_id_is_refused_at_construction() -> None: + """空串是个看起来合法的键，连表时静默匹配不上，而 None 至少能被显式筛出来。""" + with pytest.raises(ValueError, match="call_id"): + ModelReply(call_id="", content="hi", thinking="")