"""主循环的行为：停止判定顺序、写入序列、恢复守卫、取消收尾、并发隔离。 `CLAUDE.md` §3 点名的四类高危产物里有三类落在这个模块上，它们的共同点是**错了不会当场炸**。 所以这里的每一条都尽量断言「日志里留下了什么」而不只是「返回了什么」——返回值对了而写入序列 错了的话，不中断跑完时一切正常，只有崩溃之后才看得出来，而那时已经晚了。 """ import asyncio import dataclasses from collections.abc import Mapping, Sequence import pytest from polyloop._recovery import CorruptLogError from polyloop.ports import ( Action, EventKind, FinalAnswer, InvalidDecision, ModelCall, ParsedReply, RunLog, ToolCall, ) from polyloop.session import ( AgentDefinition, ParameterDriftError, RunIdentityError, RunRequest, resume, run, ) from polyloop.tools import ToolRegistry, ToolSpec from polyloop.types import ( ActionOutcome, ActionStatus, Budget, Context, Injection, Intent, IntentKind, Message, ModelCallResult, ModelReply, ReplayPolicy, Role, RunFinished, RunStarted, StepCompleted, StepRecord, StopReason, SyntheticObservations, TextBlock, ) pytestmark = pytest.mark.unit SYNTHETIC = SyntheticObservations( action_rejected="动作被拒绝了", env_failed="环境坏了", model_call_failed="模型调用失败了", ) TEMPLATE = "观察：{observation}" # --------------------------------------------------------------------------- # 测试替身 # --------------------------------------------------------------------------- class FakeStore: """一个记下所有写入顺序的内存存储。 `writes` 是一份逐条的流水，测试靠它断言写入序列——四次写的顺序是契约的一部分，而顺序错了 只在崩溃之后才看得出来。 """ 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 = 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"} def of_type(self, cls: type) -> list[object]: return [entry for entry in self.writes if isinstance(entry, cls)] class FakeModel: """按脚本回复。脚本里放异常就抛出来。""" def __init__( self, script: Sequence[object], parameters: Mapping[str, str] | None = None ) -> None: self._script = list(script) self.calls: list[ModelCall] = [] self._parameters = dict(parameters or {"model": "fake"}) async def call(self, call: ModelCall) -> ModelReply: self.calls.append(call) item = ( self._script[len(self.calls) - 1] if len(self.calls) <= len(self._script) else self._script[-1] ) if isinstance(item, BaseException): raise item return item # type: ignore[return-value] def parameters(self) -> Mapping[str, str]: return self._parameters class FakeParser: """按模型回复的正文查表。查不到就当成无效决策。""" def __init__(self, table: Mapping[str, object]) -> None: self._table = dict(table) def parse(self, reply: ModelReply) -> ParsedReply: decision = self._table.get(reply.content, InvalidDecision(explanation="看不懂")) return ParsedReply(history_text=reply.content, decision=decision) # type: ignore[arg-type] def parameters(self) -> Mapping[str, str]: return {"parser": "table"} class FakeExecutor: """按脚本返回动作结果。""" def __init__(self, script: Sequence[ActionOutcome]) -> None: self._script = list(script) self.actions: list[Action] = [] async def execute(self, action: Action) -> ActionOutcome: self.actions.append(action) index = min(len(self.actions) - 1, len(self._script) - 1) return self._script[index] def parameters(self) -> Mapping[str, str]: return {"env": "fake"} class FakeSink: def __init__(self) -> None: self.events: list[object] = [] async def emit(self, event: object) -> None: self.events.append(event) def parameters(self) -> Mapping[str, str]: return {"sink": "fake"} def _outcome( status: ActionStatus = ActionStatus.EXECUTED, *, observation: str = "环境说了这些", completed: bool = False, ) -> ActionOutcome: return ActionOutcome( status=status, observation=observation, observation_is_synthetic=False, env_reported_completion=completed, observation_truncated_chars=0, ) def _definition( store: FakeStore, model: FakeModel, parser: FakeParser, sink: object | None = None ) -> AgentDefinition: return AgentDefinition( model_client=model, decision_parser=parser, store=store, event_sink=sink or FakeSink(), # type: ignore[arg-type] synthetic_observations=SYNTHETIC, ) def _request( executor: FakeExecutor, *, run_id: str = "run-1", budget: Budget | None = None, tools: ToolRegistry | None = None, binding: Mapping[str, str] | None = None, ) -> RunRequest: return RunRequest( run_id=run_id, budget=budget or Budget( max_steps=10, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=100000 ), action_executor=executor, tools=tools or ToolRegistry(), context=Context( run_level=(Message(role=Role.USER, content=(TextBlock(text="你是助手"),)),), goal_level=(Message(role=Role.USER, content=(TextBlock(text="任务"),)),), ), injections={}, model_binding=dict(binding or {"item": "a"}), model_replay_policy=ReplayPolicy.NEVER, observation_template=TEMPLATE, cancel_grace_seconds=1.0, ) def _reply(content: str, call_id: str = "c1") -> ModelReply: return ModelReply(call_id=call_id, content=content, thinking="") ACT = Action(text="做点事", tool_call=None) # --------------------------------------------------------------------------- # 写入序列 # --------------------------------------------------------------------------- async def test_one_step_writes_four_records_in_order() -> None: """一步之内四次写，顺序是契约的一部分：模型意图 → 模型结果 → 动作意图 → 步记录。 前两次里第一次是耐久屏障（必须落盘才能发出调用），第三次也是（必须落盘才能执行动作）。 顺序错了不中断跑完时一切正常，只有崩溃之后才看得出来。 """ store = FakeStore() model = FakeModel([_reply("go")]) executor = FakeExecutor([_outcome(completed=True)]) definition = _definition(store, model, FakeParser({"go": ACT})) await run(definition, _request(executor)) kinds = [type(entry).__name__ for entry in store.writes] assert kinds == [ "RunStarted", "Intent", "ModelCallResult", "Intent", "StepCompleted", "RunFinished", ] intents = store.of_type(Intent) assert [entry.kind for entry in intents] == [IntentKind.MODEL_CALL, IntentKind.ACTION] # type: ignore[attr-defined] async def test_result_ids_are_derived_not_random() -> None: """结果 ID 从运行标识与序号推出来，所以重放同一步拿到的是同一个 ID。 随机 ID 还会往轨迹里塞一列每次都不同的值，让两次运行的逐字段比对多一个要排除的字段。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) assert [entry.result_id for entry in store.of_type(Intent)] == [ # type: ignore[attr-defined] "run-1#model#0", "run-1#action#0", ] async def test_the_finished_marker_is_written_before_the_result_comes_back() -> None: """结束标记由库在把结果交给调用方之前写下。 让项目自己落盘的话，「跑完了、库返回了、项目存的时候崩了」这种情况下重启会陷入歧义： 续跑重复执行最后一步的副作用，不续跑就丢掉一次已经花完钱的运行。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) finished = store.of_type(RunFinished) assert len(finished) == 1 assert finished[0].result == result # type: ignore[attr-defined] # --------------------------------------------------------------------------- # 停止判定 # --------------------------------------------------------------------------- async def test_an_env_reported_completion_ends_the_run() -> None: store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) assert result.stop_reason is StopReason.TASK_COMPLETED assert len(result.steps) == 1 async def test_a_completion_marked_tool_ends_the_run_without_any_env_signal() -> None: """提交型完成：环境状态一点没变，靠注册表上的完成标记收尾。 查不到这个标记的话，只有这条完成通路的项目会一路跑到预算耗尽，而它明明在第几步就已经 提交完了。 """ async def _submit(arguments: Mapping[str, object]) -> str: return "已提交" registry = ToolRegistry( [ ToolSpec( name="submit", description="交卷", parameters={}, completes_run=True, handler=_submit, ) ] ) action = Action(text="submit(...)", tool_call=ToolCall(name="submit", arguments={})) store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": action})) request = dataclasses.replace( _request(FakeExecutor([]), tools=registry), action_executor=registry.executor() ) result = await run(definition, request) assert result.stop_reason is StopReason.TASK_COMPLETED assert result.steps[0].env_reported_completion is False async def test_exactly_filling_the_step_budget_still_counts_as_completed() -> None: """「恰好在最后一个允许的步骤上做完了」记成目标达成，不是预算耗尽。 预算结算放在下一次迭代的开头正是为了这个。放在本次结尾的话它会先撞上预算上限，而两者的 轨迹长度一模一样，事后从数据里分不出来。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) budget = Budget( max_steps=1, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=100000 ) result = await run( definition, _request(FakeExecutor([_outcome(completed=True)]), budget=budget) ) assert result.stop_reason is StopReason.TASK_COMPLETED assert len(result.steps) == 1 async def test_running_out_of_steps_gives_the_step_budget_reason() -> None: store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) budget = Budget( max_steps=2, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=100000 ) result = await run(definition, _request(FakeExecutor([_outcome()]), budget=budget)) assert result.stop_reason is StopReason.STEP_BUDGET assert len(result.steps) == 2 async def test_a_final_answer_ends_the_run_without_touching_the_environment() -> None: """最终回答那一支环境根本没被碰过，和动作执行之后的完成是两种不同的结束。""" store = FakeStore() executor = FakeExecutor([_outcome()]) definition = _definition( store, FakeModel([_reply("done")]), FakeParser({"done": FinalAnswer(text="42")}) ) result = await run(definition, _request(executor)) assert result.stop_reason is StopReason.AGENT_FINISHED assert result.final_answer == "42" assert executor.actions == [] assert store.of_type(Intent) == store.of_type(Intent)[:1] # 只有模型调用意图，没有动作意图 async def test_repeated_parse_failures_stop_the_run_and_feed_each_explanation_back() -> None: """无效决策那一支跳过完成判定，说明文本原样进步记录。 那段说明**就是**回喂给模型的观察，不是从一个固定串里取——压成一句会改掉模型收到的纠错 信息，它的纠错行为也就跟着变。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("???")]), FakeParser({})) budget = Budget( max_steps=10, max_actions=10, max_consecutive_parse_failures=2, max_prompt_chars=100000 ) result = await run(definition, _request(FakeExecutor([_outcome()]), budget=budget)) assert result.stop_reason is StopReason.PARSE_FAILED_REPEATEDLY assert len(result.steps) == 2 assert all(step.parse_error == "看不懂" for step in result.steps) assert all(step.observation == "看不懂" for step in result.steps) async def test_a_valid_decision_clears_the_consecutive_parse_failure_count() -> None: """任何一个有效决策把连续失败计数清零，散落的失败不该累加到上限。""" store = FakeStore() model = FakeModel([_reply("???"), _reply("go"), _reply("???"), _reply("go2")]) definition = _definition( store, model, FakeParser({"go": ACT, "go2": Action(text="再来", tool_call=None)}) ) budget = Budget( max_steps=4, max_actions=10, max_consecutive_parse_failures=2, max_prompt_chars=100000 ) result = await run(definition, _request(FakeExecutor([_outcome()]), budget=budget)) assert result.stop_reason is StopReason.STEP_BUDGET async def test_a_failed_model_call_writes_a_result_record_then_stops() -> None: """失败也必须落一条结果记录。 只写步不写结果的话，进程在写完步、还没写运行结束时崩溃，恢复读到「意图有、结果无」会判为 状态未知走重放策略——而这次调用的状态一点都不未知，它明确地失败过。 """ store = FakeStore() definition = _definition(store, FakeModel([RuntimeError("网关连不上")]), FakeParser({})) result = await run(definition, _request(FakeExecutor([_outcome()]))) assert result.stop_reason is StopReason.LLM_ERROR results = store.of_type(ModelCallResult) assert len(results) == 1 assert results[0].reply is None # type: ignore[attr-defined] assert "网关连不上" in results[0].failure # type: ignore[attr-defined,operator] assert result.steps[0].observation == SYNTHETIC.model_call_failed assert result.steps[0].parse_error is None async def test_an_oversized_prompt_stops_without_writing_a_step_or_an_intent() -> None: """规模超限是唯一一种「真的一步都没走」的终止。 命中时模型还没被调用、没花钱、没有调用标识需要对账。伪造一条空步会在轨迹里多出一条永远 连不上账目的记录。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) budget = Budget( max_steps=10, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=1 ) result = await run(definition, _request(FakeExecutor([_outcome()]), budget=budget)) assert result.stop_reason is StopReason.CONTEXT_OVERFLOW assert result.steps == () assert store.of_type(Intent) == [] assert store.of_type(StepCompleted) == [] async def test_an_env_error_ends_the_run() -> None: store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) result = await run(definition, _request(FakeExecutor([_outcome(ActionStatus.ENV_ERROR)]))) assert result.stop_reason is StopReason.ENV_ERROR async def test_a_rejected_action_neither_completes_nor_counts_as_an_executed_action() -> None: """未执行不做完成判定，也不计入已执行动作数，但计入步数。 两个计数混成一个就防不住「模型一直调用不存在的工具」这种不收敛。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) budget = Budget( max_steps=3, max_actions=1, max_consecutive_parse_failures=9, max_prompt_chars=100000 ) result = await run( definition, _request(FakeExecutor([_outcome(ActionStatus.NOT_EXECUTED)]), budget=budget) ) assert result.stop_reason is StopReason.STEP_BUDGET assert len(result.steps) == 3 # --------------------------------------------------------------------------- # 观察投影 # --------------------------------------------------------------------------- @pytest.mark.parametrize( ("status", "expected"), [ (ActionStatus.NOT_EXECUTED, SYNTHETIC.action_rejected), (ActionStatus.ENV_ERROR, SYNTHETIC.env_failed), ], ) async def test_the_library_synthesises_the_observation_when_the_action_did_not_run( status: ActionStatus, expected: str ) -> None: """执行器给的那段在这两档下不进历史。 它是「模型看得见的东西」，而那种东西必须能进参数快照。执行器每次现造一段文本的话，两次 运行之间它可以变而不会有任何地方报错，于是同一份配置跑出来的两次运行在模型看来其实不同。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) budget = Budget( max_steps=1, max_actions=9, max_consecutive_parse_failures=9, max_prompt_chars=100000 ) result = await run( definition, _request(FakeExecutor([_outcome(status, observation="执行器自己写的")]), budget=budget), ) assert result.steps[0].observation == expected assert result.steps[0].observation_is_synthetic is True assert result.steps[0].observation_truncated_chars == 0 async def test_an_executed_action_keeps_the_executor_observation() -> None: store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) result = await run( definition, _request(FakeExecutor([_outcome(observation="真的输出", completed=True)])) ) assert result.steps[0].observation == "真的输出" assert result.steps[0].observation_is_synthetic is False async def test_the_observation_is_fed_back_to_the_next_model_call() -> None: """上一步的观察套上模板进下一次调用的消息序列，不套的话模型分不出哪句是环境说的。""" store = FakeStore() model = FakeModel([_reply("go"), _reply("done")]) definition = _definition(store, model, FakeParser({"go": ACT, "done": FinalAnswer(text="ok")})) await run(definition, _request(FakeExecutor([_outcome(observation="输出A")]))) texts = [block.text for block in model.calls[1].messages[-1].content] assert texts == ["观察：输出A"] # --------------------------------------------------------------------------- # 装配校验 # --------------------------------------------------------------------------- def test_a_request_refuses_an_executor_derived_from_another_registry() -> None: """模型看见的 schema 来自一个注册表、实际分发走另一个，表现是「模型调了一个它看得见的 工具却说不存在」。""" async def _noop(arguments: Mapping[str, object]) -> str: return "" full = ToolRegistry( [ ToolSpec(name="read", description="", parameters={}, handler=_noop), ToolSpec(name="write", description="", parameters={}, handler=_noop), ] ) narrowed = _request(FakeExecutor([]), tools=full.restrict_to(["read"])) with pytest.raises(ValueError, match="派生自另一个注册表"): dataclasses.replace(narrowed, action_executor=full.executor()) def test_a_request_accepts_an_executor_derived_from_the_same_registry() -> None: """内容相同的两个注册表按值相等，所以这条校验不会把两份等价的装配错判成冲突。""" async def _noop(arguments: Mapping[str, object]) -> str: return "" registry = ToolRegistry([ToolSpec(name="read", description="", parameters={}, handler=_noop)]) dataclasses.replace( _request(FakeExecutor([]), tools=registry), action_executor=registry.executor() ) def test_a_request_refuses_a_template_without_the_placeholder() -> None: from polyloop._assembly import AssemblyError with pytest.raises(AssemblyError): dataclasses.replace(_request(FakeExecutor([])), observation_template="没有占位符") def test_a_budget_of_zero_is_refused() -> None: """零或负数会让第一档预算准入立刻命中，产出一次「零步、预算耗尽」的运行。 那和一次真的跑满了上限的运行在停止原因上完全一样，混进统计里分不出来。 """ with pytest.raises(ValueError, match="max_steps"): Budget(max_steps=0, max_actions=1, max_consecutive_parse_failures=1, max_prompt_chars=1) # --------------------------------------------------------------------------- # 参数快照 # --------------------------------------------------------------------------- def test_the_snapshot_carries_every_binding_key() -> None: """崩溃后用同一个运行标识、换一组绑定续跑，后面每一次调用会被记到另一套坐标上， 而两段轨迹在文件里看起来是同一次运行。""" request = _request(FakeExecutor([]), binding={"book": "b1", "task": "t7"}) snapshot = request.parameter_snapshot() assert snapshot["request.binding.book"] == "b1" assert snapshot["request.binding.task"] == "t7" def test_the_snapshot_asks_all_five_seams() -> None: """五个接缝都要上报，动作执行接缝也不例外——它可能是一个已经开好的会话，而会话是不是 有状态会改变跨步语义。""" store = FakeStore() definition = _definition(store, FakeModel([]), FakeParser({})) request = _request(FakeExecutor([])) merged = {**definition.parameter_snapshot(), **request.parameter_snapshot()} assert merged["model_client.model"] == "fake" assert merged["decision_parser.parser"] == "table" assert merged["store.kind"] == "memory" assert merged["event_sink.sink"] == "fake" assert merged["action_executor.env"] == "fake" def test_the_context_never_enters_the_snapshot() -> None: """上下文与注入正文是这次运行的输入数据不是参数，进快照会让快照变成一份数据副本。""" request = dataclasses.replace( _request(FakeExecutor([])), injections={"skill": (Injection(entry_id="e1", content="很长的一段正文"),)}, ) snapshot = request.parameter_snapshot() assert "很长的一段正文" not in "".join(snapshot.values()) assert snapshot["request.injected_entry_ids"] == "e1" # --------------------------------------------------------------------------- # 两条入口的失败方式 # --------------------------------------------------------------------------- async def test_run_refuses_a_run_id_that_already_has_a_log() -> None: """覆盖会毁掉一次已经花完钱的运行的留痕，接着跑是 resume 的事。""" store = FakeStore(RunLog(started=RunStarted(run_id="run-1", parameter_snapshot={}))) definition = _definition(store, FakeModel([]), FakeParser({})) with pytest.raises(RunIdentityError, match="已经有日志"): await run(definition, _request(FakeExecutor([]))) async def test_resume_refuses_a_run_id_with_no_log() -> None: store = FakeStore() definition = _definition(store, FakeModel([]), FakeParser({})) with pytest.raises(RunIdentityError, match="读不到任何日志"): await resume(definition, _request(FakeExecutor([]))) async def test_resume_refuses_when_the_assembly_drifted() -> None: """续跑不能顺便改预算或换模型——那不是限制，是这条守卫的全部意义。""" store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) other_budget = Budget( max_steps=99, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=100000 ) with pytest.raises(ParameterDriftError, match="request.max_steps"): await resume(definition, _request(FakeExecutor([_outcome()]), budget=other_budget)) async def test_resume_hands_back_a_finished_run_without_rerunning_it() -> None: store = FakeStore() model = FakeModel([_reply("go")]) definition = _definition(store, model, FakeParser({"go": ACT})) first = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) again = await resume(definition, _request(FakeExecutor([_outcome()]))) assert again == first assert len(model.calls) == 1 async def test_resume_stops_when_the_state_is_unknown_and_replay_is_forbidden() -> None: """撞上未知状态是一个可预期的正常终态，不是库自身的缺陷。 抛异常会丢掉「跑到第几步、已经花了多少、前面那些步的轨迹」，而那些正是项目决定「重跑还是 人工介入」时要看的。 """ store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) 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, ), ), ) result = await resume(definition, request) assert result.stop_reason is StopReason.RESUME_STATE_UNKNOWN assert result.steps == () async def test_resume_refuses_a_corrupt_log() -> None: """结果有、意图无是结构上说不通的状态，拒绝续跑而不是修好它接着跑。""" store = FakeStore( RunLog( started=RunStarted(run_id="run-1", parameter_snapshot={}), model_results=( ModelCallResult(run_id="run-1", result_id="orphan", reply=None, failure="x"), ), ) ) definition = _definition(store, FakeModel([]), FakeParser({})) with pytest.raises(CorruptLogError): await resume(definition, _request(FakeExecutor([]))) # --------------------------------------------------------------------------- # 取消 # --------------------------------------------------------------------------- async def test_cancellation_propagates_and_still_writes_the_finished_marker() -> None: """取消要能穿过模型调用，`CancelledError` 原样重抛，`run` 不返回结果。 但「这次运行结束了」这个标记必须写下去——不写的话，恢复读到的是一次没有结束标记的运行， 会被当成可以续跑，而它其实是被人主动叫停的。 """ store = FakeStore() class _Hanging(FakeModel): async def call(self, call: ModelCall) -> ModelReply: await asyncio.Event().wait() raise AssertionError("到不了这里") definition = _definition(store, _Hanging([]), FakeParser({})) task = asyncio.ensure_future(run(definition, _request(FakeExecutor([])))) await asyncio.sleep(0) await asyncio.sleep(0) task.cancel() with pytest.raises(asyncio.CancelledError): await task finished = store.of_type(RunFinished) assert len(finished) == 1 assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] # --------------------------------------------------------------------------- # 并发隔离 # --------------------------------------------------------------------------- async def test_two_runs_sharing_a_definition_do_not_see_each_other() -> None: """定义可以并发复用，它自己不持有任何一次运行的状态。 持有了的话，并发跑同一份定义的两次运行会互相写到对方的计数里，而那种错在单线程测试里 永远不出现。 """ store_a = FakeStore() store_b = FakeStore() model = FakeModel([_reply("go")]) parser = FakeParser({"go": ACT}) definition_a = _definition(store_a, model, parser) definition_b = _definition(store_b, FakeModel([_reply("go")]), parser) budget_a = Budget( max_steps=1, max_actions=9, max_consecutive_parse_failures=9, max_prompt_chars=100000 ) budget_b = Budget( max_steps=3, max_actions=9, max_consecutive_parse_failures=9, max_prompt_chars=100000 ) result_a, result_b = await asyncio.gather( run(definition_a, _request(FakeExecutor([_outcome()]), run_id="a", budget=budget_a)), run(definition_b, _request(FakeExecutor([_outcome()]), run_id="b", budget=budget_b)), ) 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="") # --------------------------------------------------------------------------- # 事件出口 # --------------------------------------------------------------------------- class _RaisingSink(FakeSink): """按脚本抛异常的出口。抛完仍然把这条记下来，好断言「库有没有再发一次」。""" def __init__(self, error: BaseException) -> None: super().__init__() self._error = error async def emit(self, event: object) -> None: self.events.append(event) raise self._error async def test_every_step_emits_one_event_carrying_the_whole_record() -> None: """一步一条，带的是整条步记录而不是挑几个字段拼的摘要。 摘要是一次投影，而投影会漂移——步记录加一个字段，带整条的话事件里自动就有。 """ store = FakeStore() sink = FakeSink() model = FakeModel([_reply("go"), _reply("go")]) definition = _definition(store, model, FakeParser({"go": ACT}), sink) result = await run(definition, _request(FakeExecutor([_outcome(), _outcome(completed=True)]))) assert [event.kind for event in sink.events] == [EventKind.STEP_FINISHED] * 2 assert [event.step for event in sink.events] == list(result.steps) async def test_the_event_carries_the_run_id_and_the_project_binding() -> None: """运行标识让共用同一个出口的并发运行分得开；绑定没法从运行标识倒推。""" store = FakeStore() sink = FakeSink() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) await run( definition, _request(FakeExecutor([_outcome(completed=True)]), binding={"item": "a", "task": "t7"}), ) (event,) = sink.events assert event.run_id == "run-1" assert event.model_binding == {"item": "a", "task": "t7"} async def test_the_event_goes_out_after_the_step_landed_not_before() -> None: """先发后写的话，进程崩在两者之间会让观察者看见一步而存储里没有。 事件流的全部安全性建立在「它带的事实在存储里另有一份」上，而这个顺序是那条不变量在 崩溃点上的兑现方式。 """ store = FakeStore() seen_at_emit: list[int] = [] class _OrderSink(FakeSink): async def emit(self, event: object) -> None: seen_at_emit.append(len(store.of_type(StepCompleted))) await super().emit(event) definition = _definition( store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), _OrderSink() ) await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) assert seen_at_emit == [1] async def test_a_failing_sink_does_not_stop_the_run_and_is_counted() -> None: """事件是观察通道不是控制通道：进度回写的数据库连不上，运行照跑完。 计数放在返回值上而不是只记日志，因为日志没人看。 """ store = FakeStore() sink = _RaisingSink(ConnectionError("进度库连不上")) model = FakeModel([_reply("go"), _reply("go")]) definition = _definition(store, model, FakeParser({"go": ACT}), sink) result = await run(definition, _request(FakeExecutor([_outcome(), _outcome(completed=True)]))) assert result.stop_reason is StopReason.TASK_COMPLETED assert len(result.steps) == 2 assert result.event_delivery_failures == 2 async def test_a_delivery_failure_is_not_re_emitted_through_the_same_sink() -> None: """失败不转成一条事件从同一个出口再发一次——那会自我喂食。 一个持续失败的出口会让失败处理路径变成递归，而递归的表现是进程卡住或栈溢出，不是一条 错误日志。所以出口收到的条数必须恰好等于步数。 """ store = FakeStore() sink = _RaisingSink(ConnectionError("一直连不上")) definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) assert len(sink.events) == len(result.steps) == 1 async def test_cancellation_during_delivery_is_not_swallowed() -> None: """接的是 `Exception` 不是 `BaseException`：在这一下吞掉取消，取消会晚一整步才生效。""" store = FakeStore() sink = _RaisingSink(asyncio.CancelledError()) definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) with pytest.raises(asyncio.CancelledError): await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) finished = store.of_type(RunFinished) assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] assert finished[0].result.event_delivery_failures == 0 # type: ignore[attr-defined] async def test_steps_read_back_from_the_log_are_not_re_emitted() -> None: """续跑不给已经完成的步补发事件。 补发等于宣称一件早就发生过的事刚刚发生，而接进度表的那一侧会多出一批重复行。判据是这次 进程里有没有真的执行过，观察者要补全前半段就从存储里读。 """ sink = FakeSink() store = FakeStore() definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) request = _request(FakeExecutor([_outcome(completed=True)])) store._log = _log_with_one_finished_step( # noqa: SLF001 {**definition.parameter_snapshot(), **request.parameter_snapshot()} ) result = await resume(definition, request) # 第 0 步是从日志里读回来的，第 1 步是这次进程里真的走的。只有后者发了事件。 assert [step.step_idx for step in result.steps] == [0, 1] assert [event.step.step_idx for event in sink.events] == [1] def _log_with_one_finished_step(snapshot: Mapping[str, str]) -> RunLog: """一份「第 0 步完整走完、还没写结束记录」的日志。 手工搭而不是先跑一次再续跑：跑出来的那一步要么带完成信号（续跑会当场收尾，走不到第二步）， 要么撞预算上限（续跑在预算准入那一档就停了），两种都验不到「读回来的不发、真跑的发」这条 边界。 """ outcome = _outcome() return RunLog( 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, ), Intent( run_id="run-1", kind=IntentKind.ACTION, call_index=0, result_id="run-1#action#0", replay_policy=ReplayPolicy.NEVER, ), ), model_results=( ModelCallResult( run_id="run-1", result_id="run-1#model#0", reply=_reply("go"), failure=None ), ), steps=( StepCompleted( run_id="run-1", result_id="run-1#action#0", action_outcome=outcome, step=StepRecord( step_idx=0, raw_output="go", content_chars=2, thinking_chars=0, action="做点事", parse_ok=True, parse_error=None, observation=outcome.observation, observation_is_synthetic=False, observation_truncated_chars=0, prompt_chars=10, call_id="c1", step_wall_ms=1, action_status=ActionStatus.EXECUTED, env_reported_completion=False, ), ), ), ) async def test_the_log_keeps_both_the_raw_and_the_repaired_model_output() -> None: """模型原文与解释器改写之后的文本各有位置，两份都不可丢（`design/0013` 决策二）。 GovDoc 有一条硬纪律：agent 的原始输出、修复后的输出、恢复来源全程留痕，禁止静默修复。 承载它的是意图日志而不是事件流——事件可丢，一件只存在于可丢通道里的事实撑不起「禁止 静默修复」。 两份文本天然分开存，是写入序列决定的：模型调用结算时写结果记录，那时还没解释；解释完、 动作走完之后才写步记录，那里面的文本是解释器交回来的。 """ class _RewritingParser(FakeParser): """把第一个代码围栏之后的内容整段丢掉——模型常在代码块后面编造执行结果。""" def parse(self, reply: ModelReply) -> ParsedReply: return ParsedReply(history_text=reply.content.split("|", 1)[0], decision=ACT) store = FakeStore() definition = _definition( store, FakeModel([_reply("真动作|模型编的执行结果")]), _RewritingParser({}) ) result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) (call_result,) = store.of_type(ModelCallResult) assert call_result.reply.content == "真动作|模型编的执行结果" # type: ignore[attr-defined,union-attr] assert result.steps[0].raw_output == "真动作"