"""恢复判定的行为。 一步之内写四次，所以断点有四个，每个断点对应一条续跑路径。下面按断点逐个走一遍，外加 四态表最后一行那几种「结构上说不通」。 **这里的每一条都没有失败现场。** 判错了不会当场炸，只会让恢复出来的历史比不中断跑完时 少一轮或者多一轮，而模型看到的东西一变，后面每一步都跟着偏——事后只表现成「跑出来的结果 有点不一样」。 """ import pytest from polyloop._recovery import ( CorruptLogError, ExecutionState, ResumeAction, classify_execution, plan_resume, ) from polyloop.ports import RunLog from polyloop.types import ( ActionOutcome, ActionStatus, Intent, IntentKind, ModelCallResult, ModelReply, ReplayPolicy, RunFinished, RunResult, RunStarted, StepCompleted, StepRecord, StopReason, ) pytestmark = pytest.mark.unit RUN_ID = "run-1" STARTED = RunStarted(run_id=RUN_ID, parameter_snapshot={"model.name": "x"}) def _step_record(idx: int, *, parse_ok: bool = True) -> StepRecord: return StepRecord( step_idx=idx, raw_output="", content_chars=0, thinking_chars=0, action="print(1)" if parse_ok else None, parse_ok=parse_ok, parse_error=None if parse_ok else "没有代码块", observation="", observation_is_synthetic=False, observation_truncated_chars=0, prompt_chars=0, call_id=None, step_wall_ms=0, ) def _model_intent(idx: int, policy: ReplayPolicy = ReplayPolicy.NEVER) -> Intent: return Intent( run_id=RUN_ID, kind=IntentKind.MODEL_CALL, call_index=idx, result_id=f"model-{idx}", replay_policy=policy, ) def _action_intent(idx: int, policy: ReplayPolicy = ReplayPolicy.NEVER) -> Intent: return Intent( run_id=RUN_ID, kind=IntentKind.ACTION, call_index=idx, result_id=f"action-{idx}", replay_policy=policy, ) def _model_result(idx: int, *, failed: bool = False) -> ModelCallResult: return ModelCallResult( run_id=RUN_ID, result_id=f"model-{idx}", reply=None if failed else ModelReply(call_id=f"call-{idx}", content="hi", thinking=""), failure="连不上" if failed else None, ) def _completed_step(idx: int, *, executed: bool = True, parse_ok: bool = True) -> StepCompleted: """一条完整的步记录。`executed=False` 表示那一步没有动作（解析失败那种）。""" if not executed: return StepCompleted( run_id=RUN_ID, result_id=None, action_outcome=None, step=_step_record(idx, parse_ok=parse_ok), ) return StepCompleted( run_id=RUN_ID, result_id=f"action-{idx}", action_outcome=ActionOutcome( status=ActionStatus.EXECUTED, observation="", observation_is_synthetic=False, env_reported_completion=False, observation_truncated_chars=0, ), step=_step_record(idx, parse_ok=parse_ok), ) def _log(*, intents=(), model_results=(), steps=(), started=STARTED, finished=None) -> RunLog: return RunLog( started=started, intents=tuple(intents), model_results=tuple(model_results), steps=tuple(steps), finished=finished, ) # --------------------------------------------------------------------------- # 四态表本身 # --------------------------------------------------------------------------- @pytest.mark.parametrize( ("has_intent", "has_result", "expected"), [ (False, False, ExecutionState.NOT_STARTED), (True, True, ExecutionState.COMPLETED), (True, False, ExecutionState.UNKNOWN), (False, True, ExecutionState.CORRUPT), ], ) def test_the_four_states(has_intent: bool, has_result: bool, expected: ExecutionState) -> None: """整张表就四格，四格全在这里。 最后一格是刻意的：读到说不通的状态就失败，不修复也不带着它继续。 """ assert classify_execution(has_intent, has_result) is expected # --------------------------------------------------------------------------- # 两端：空日志与跑完了的日志 # --------------------------------------------------------------------------- def test_an_empty_log_means_a_brand_new_run() -> None: """读一个从没写过的运行标识返回空日志，那就是一次全新的运行。""" plan = plan_resume(RunLog(), ReplayPolicy.NEVER) assert plan.action is ResumeAction.START_FRESH assert plan.next_call_index == 0 def test_a_finished_log_hands_back_the_stored_result() -> None: """有结束标记就是跑完了，不重跑最后一步。 结束标记由库在把结果交给调用方之前写下，正是为了消掉「跑完了但项目没存住」那个歧义。 """ result = RunResult( run_id=RUN_ID, stop_reason=StopReason.TASK_COMPLETED, final_answer="42", steps=() ) plan = plan_resume( _log( intents=[_model_intent(0), _action_intent(0)], model_results=[_model_result(0)], steps=[_completed_step(0)], finished=RunFinished(run_id=RUN_ID, result=result), ), ReplayPolicy.NEVER, ) assert plan.action is ResumeAction.ALREADY_FINISHED assert plan.finished_result == result # --------------------------------------------------------------------------- # 四个断点 # --------------------------------------------------------------------------- def test_a_log_that_ends_on_a_step_boundary_continues_at_the_next_step() -> None: """两步都写完了，什么都没断——从第三步的开头接着跑。""" plan = plan_resume( _log( intents=[_model_intent(0), _action_intent(0), _model_intent(1), _action_intent(1)], model_results=[_model_result(0), _model_result(1)], steps=[_completed_step(0), _completed_step(1)], ), ReplayPolicy.NEVER, ) assert plan.action is ResumeAction.CONTINUE_AT_NEXT_STEP assert plan.next_call_index == 2 assert plan.steps_appended == 2 assert plan.actions_executed == 2 @pytest.mark.parametrize( ("policy", "expected"), [ (ReplayPolicy.SAFE, ResumeAction.REDO_MODEL_CALL), (ReplayPolicy.NEVER, ResumeAction.STOP_UNKNOWN), ], ) def test_a_model_call_with_no_result_follows_the_declared_replay_policy( policy: ReplayPolicy, expected: ResumeAction ) -> None: """断点一：模型调用意图写了、结果没写。 调用可能已经发出去、也可能没有。库不猜，按调用方声明的策略办——声明绝不重放就停下来 报告，而不是替谁决定要不要再花一次钱。 """ plan = plan_resume(_log(intents=[_model_intent(0)]), policy) assert plan.action is expected def test_a_model_call_that_failed_finishes_as_a_model_call_failure() -> None: """断点二的一支：结果条目在，但它记的是一次失败。 这次调用的状态一点都不未知——它明确地失败过。判成状态未知走重放策略的话，一次已知的 失败会被当成可能成功过。 """ plan = plan_resume( _log(intents=[_model_intent(0)], model_results=[_model_result(0, failed=True)]), ReplayPolicy.NEVER, ) assert plan.action is ResumeAction.FINISH_MODEL_CALL_FAILED assert plan.pending_failure == "连不上" def test_a_reply_with_no_action_intent_is_reparsed_not_recalled() -> None: """断点二：模型回复存下来了，动作意图还没写。 副作用还没发生，重新解释那条回复是安全的，也不必再花一次模型调用的钱。**重新解释而不是 把上次的结果填回去**——库根本没存过解释结果，存的是模型原文。 """ plan = plan_resume( _log(intents=[_model_intent(0)], model_results=[_model_result(0)]), ReplayPolicy.NEVER, ) assert plan.action is ResumeAction.REPARSE_LAST_REPLY assert plan.pending_reply is not None assert plan.pending_reply.content == "hi" @pytest.mark.parametrize( ("policy", "expected"), [ (ReplayPolicy.SAFE, ResumeAction.REPLAY_LAST_ACTION), (ReplayPolicy.NEVER, ResumeAction.STOP_UNKNOWN), ], ) def test_an_action_with_no_step_record_follows_the_tool_declared_policy( policy: ReplayPolicy, expected: ResumeAction ) -> None: """断点三：动作意图写了、步记录没写。动作到底执行没执行，日志答不出来。 这一档读的是**意图记录里存下来的那个策略**，不是现在去注册表里再查一遍——工具的声明可能 在两次运行之间被改过，而那时该按哪一份办，只有当时写下的那一份说了算。 """ plan = plan_resume( _log( intents=[_model_intent(0), _action_intent(0, policy)], model_results=[_model_result(0)], ), ReplayPolicy.NEVER, ) assert plan.action is expected def test_a_replayed_action_reuses_the_result_id_it_was_given() -> None: """重放沿用原来那个预分配的 ID，不另分配一个。 另分配一个的话，原来那条意图永远配不上任何结果，下一次恢复读到的还是「状态未知」—— 一次成功的重放会在日志里留下一个永远好不了的洞。 """ plan = plan_resume( _log( intents=[_model_intent(0), _action_intent(0, ReplayPolicy.SAFE)], model_results=[_model_result(0)], ), ReplayPolicy.NEVER, ) assert plan.pending_action_result_id == "action-0" # --------------------------------------------------------------------------- # 计数重建 # --------------------------------------------------------------------------- def test_only_executed_actions_count_towards_the_action_budget() -> None: """没有动作的步（解析失败那种）计入步数，不计入已执行动作数。""" plan = plan_resume( _log( intents=[_model_intent(0), _action_intent(0), _model_intent(1)], model_results=[_model_result(0), _model_result(1)], steps=[_completed_step(0), _completed_step(1, executed=False, parse_ok=False)], ), ReplayPolicy.NEVER, ) assert plan.steps_appended == 2 assert plan.actions_executed == 1 def test_only_the_trailing_run_of_parse_failures_counts() -> None: """任何一个有效决策把连续失败计数清零，所以中间那些散落的失败不算数。""" plan = plan_resume( _log( intents=[_model_intent(idx) for idx in range(4)] + [_action_intent(0), _action_intent(1)], model_results=[_model_result(idx) for idx in range(4)], steps=[ _completed_step(0, parse_ok=False), _completed_step(1), _completed_step(2, executed=False, parse_ok=False), _completed_step(3, executed=False, parse_ok=False), ], ), ReplayPolicy.NEVER, ) assert plan.consecutive_parse_failures == 2 # --------------------------------------------------------------------------- # 日志损坏：拒绝续跑，不修复 # --------------------------------------------------------------------------- def test_a_result_with_no_intent_is_corrupt() -> None: """四态表最后一行。结构上说不通的状态就失败，不猜。""" with pytest.raises(CorruptLogError, match="对不上任何一条意图"): plan_resume(_log(model_results=[_model_result(0)]), ReplayPolicy.NEVER) def test_a_step_whose_action_result_has_no_intent_is_corrupt() -> None: with pytest.raises(CorruptLogError, match="对不上任何一条动作意图"): plan_resume( _log( intents=[_model_intent(0)], model_results=[_model_result(0)], steps=[_completed_step(0)], ), ReplayPolicy.NEVER, ) def test_two_intents_for_the_same_step_are_corrupt() -> None: """同一步两条同种意图，说明有两个进程在往同一个运行标识里写。""" with pytest.raises(CorruptLogError, match="并发写过"): plan_resume(_log(intents=[_model_intent(0), _model_intent(0)]), ReplayPolicy.NEVER) def test_a_gap_in_the_step_numbers_is_corrupt() -> None: """缺号意味着中间某一步的原子写整个丢了。 带着缺口续跑的话，重建出来的历史比不中断跑完时少一轮，而这件事在轨迹里看不出来。 """ with pytest.raises(CorruptLogError, match="不连续"): plan_resume( _log( intents=[_model_intent(0), _action_intent(0), _model_intent(2), _action_intent(2)], model_results=[_model_result(0), _model_result(2)], steps=[_completed_step(0), _completed_step(2)], ), ReplayPolicy.NEVER, ) def test_an_intent_that_runs_ahead_of_the_steps_is_corrupt() -> None: """第 1 步有意图而第 0 步的步记录还没写：中间少了一步。""" with pytest.raises(CorruptLogError, match="中间少了一步"): plan_resume( _log(intents=[_model_intent(0), _model_intent(1)], model_results=[_model_result(0)]), ReplayPolicy.NEVER, ) def test_an_action_intent_without_a_model_call_intent_is_corrupt() -> None: """一步之内先写模型调用意图再写动作意图，反过来说明动作是凭空来的。""" with pytest.raises(CorruptLogError, match="动作是从哪儿来的"): plan_resume(_log(intents=[_action_intent(0)]), ReplayPolicy.NEVER) def test_records_without_a_run_started_record_are_corrupt() -> None: """没有运行开始记录就读不出这份日志是按哪份配置跑的，续跑守卫无从比对。""" with pytest.raises(CorruptLogError, match="运行开始记录"): plan_resume(_log(intents=[_model_intent(0)], started=None), ReplayPolicy.NEVER)