diff --git a/src/polyloop/_recovery/__init__.py b/src/polyloop/_recovery/__init__.py index 3a99afc..104e6d9 100644 --- a/src/polyloop/_recovery/__init__.py +++ b/src/polyloop/_recovery/__init__.py @@ -1,7 +1,306 @@ -"""恢复状态判定与运行身份校验。内部模块。 +"""恢复判定：一份读回来的日志说明上次跑到哪儿了，接下来该从哪儿续。 -读到结构上说不通的状态就失败，不修复也不带着它继续——一个被猜着修好的日志，会让后面 -每一个基于它的判断都建立在猜测上，而且不会有任何地方提示这件事发生过。 +**内部模块**（下划线开头，不进 `polyloop/__init__.py`）。纯逻辑，无 I/O、无事件循环——读日志 +那一下由 `polyloop.session` 调存储接缝完成，这里只吃已经读回来的那份记录。 -**纯逻辑，约束同 `_assembly`。** +**不 import 其余四个逻辑层模块**，五者互不 import。所以这里输出的是三个裸计数而不是 +`_stopping.RunCounters`，由 `session` 拼起来。 + +## 一次执行的四态 + +按「意图有没有 / 结果有没有」判（`research-wiki/design/0002-step-level-resume.md` 决策二）： + +| 意图 | 结果 | 含义 | 做法 | +|---|---|---|---| +| 无 | 无 | 还没开始 | 重跑这一步 | +| 有 | 有 | 执行完了 | 跳过 | +| 有 | 无 | **状态未知** | 按声明的重放策略决定 | +| 无 | 有 | 结构上说不通 | 判为日志损坏，拒绝续跑 | + +最后一行是刻意的：读到说不通的状态就失败，不修复也不带着它继续。一个被猜着修好的日志会让 +后面每一个基于它的判断都建立在猜测上，而且不会有任何地方提示这件事发生过。 + +## 一步之内写四次，所以断点有四个 + +`0003` 决策四那张表：模型调用意图 → 模型调用结果 → 动作意图 → 动作结果与步记录（原子）。 +断在哪两次之间，恢复要做的事就不同——这个模块的全部工作就是把断点认出来。 + +## 步序号与模型调用序号是同一个数 + +一步之内恰好一次模型调用（`0004` 决策三 C 档），所以第 k 步的两条意图都带 `call_index == k`， +而那一步的步记录带 `step_idx == k`。**没有这条对齐，带 `result_id` 为空的步记录就认不出它属于 +哪一步**——那种步记录（解析失败、模型调用失败、最终回答三档）没有动作意图可以配对，只剩序号 +这一条线索。规模超限那一档不产生步记录也不写意图，两边一起不加，对齐不受影响。 """ + +from collections import Counter +from dataclasses import dataclass +from enum import StrEnum + +from polyloop.ports import RunLog +from polyloop.types import ( + ActionStatus, + Intent, + IntentKind, + ModelReply, + ReplayPolicy, + RunResult, + StepRecord, +) + + +class CorruptLogError(Exception): + """日志处在结构上说不通的状态，拒绝续跑。 + + **不继承 `ValueError`**：它不是「调用方传错了参数」，是「存下来的那份数据坏了」，两者 + 要人做的事完全不同。 + """ + + +class ExecutionState(StrEnum): + """一次执行（一次模型调用，或一次动作执行）在日志里处于哪一态。""" + + NOT_STARTED = "not_started" + COMPLETED = "completed" + #: 意图写了、结果没写。执行到底发生没发生，日志答不出来。 + UNKNOWN = "unknown" + #: 结果有、意图没有。结构上说不通。 + CORRUPT = "corrupt" + + +def classify_execution(has_intent: bool, has_result: bool) -> ExecutionState: + """四态表本身。 + + **两种意图（模型调用与动作）共用这一个函数**，因为判定完全相同。分开写两遍的话，改的 + 时候必然有一处漏掉——这正是两种意图合成一个记录类型的同一条理由。 + """ + if has_intent and has_result: + return ExecutionState.COMPLETED + if has_intent: + return ExecutionState.UNKNOWN + if has_result: + return ExecutionState.CORRUPT + return ExecutionState.NOT_STARTED + + +class ResumeAction(StrEnum): + """续跑时下一步该做什么。""" + + #: 日志是空的，这是一次全新的运行。 + START_FRESH = "start_fresh" + #: 有结束标记，这次运行早就跑完了，把存下来的结果原样交回去。 + ALREADY_FINISHED = "already_finished" + #: 上一步是完整的，从下一步的开头（预算准入）接着跑。 + CONTINUE_AT_NEXT_STEP = "continue_at_next_step" + #: 模型调用状态未知且声明可重放：重新调一次模型。 + REDO_MODEL_CALL = "redo_model_call" + #: 模型回复已经存下来了，动作意图还没写——副作用还没发生，重新解释那条回复。 + REPARSE_LAST_REPLY = "reparse_last_reply" + #: 动作状态未知且声明可重放：重新解释那条回复，再执行一次动作。 + REPLAY_LAST_ACTION = "replay_last_action" + #: 状态未知且声明绝不重放：停下来报告，不替谁做决定。 + STOP_UNKNOWN = "stop_unknown" + #: 模型调用明确失败过，步记录还没写：补上那一步，然后以模型调用失败收尾。 + FINISH_MODEL_CALL_FAILED = "finish_model_call_failed" + + +@dataclass(frozen=True, slots=True) +class ResumePlan: + """从一份日志读出来的续跑计划。""" + + action: ResumeAction + #: 已经完成的步，按序号排好。续跑要拿它重建模型看见的那段历史。 + steps: tuple[StepRecord, ...] = () + #: 三个裸计数，由 `session` 拼成 `_stopping.RunCounters`。 + steps_appended: int = 0 + actions_executed: int = 0 + consecutive_parse_failures: int = 0 + #: 下一步的序号，同时也是下一次模型调用的序号。 + next_call_index: int = 0 + #: 被打断的那一步已经拿到的模型回复。重新解释与重放动作两档要它。 + pending_reply: ModelReply | None = None + #: 被打断的那次模型调用记下来的失败说明。 + pending_failure: str | None = None + #: 被打断的那次动作预分配的结果 ID。重放时沿用它，不另分配一个。 + pending_action_result_id: str | None = None + #: 结束标记里存着的那份结果。 + finished_result: RunResult | None = None + + +def _indexed_intents(intents: tuple[Intent, ...], kind: IntentKind) -> dict[int, Intent]: + """把某一种意图按调用序号排开，撞号就是损坏。 + + 同一步写了两条同种意图，说明有两个进程在往同一个运行标识里写——那时哪一条对应哪次执行 + 没有答案，而随便挑一条会让后面每一步都建立在猜测上。 + """ + by_index: dict[int, Intent] = {} + for intent in intents: + if intent.kind is not kind: + continue + if intent.call_index in by_index: + raise CorruptLogError( + f"第 {intent.call_index} 步有两条 {kind.value} 意图，日志被并发写过" + ) + by_index[intent.call_index] = intent + return by_index + + +def _ordered_steps(log: RunLog) -> tuple[StepRecord, ...]: + """按序号排好已完成的步，并检查序号是不是从 0 起连续。 + + 缺号意味着中间某一步的原子写整个丢了。带着缺口续跑的话，重建出来的历史比不中断跑完时 + 少一轮，模型看到的东西不一样，后面每一步都跟着偏——而这件事在轨迹里看不出来。 + """ + steps = sorted((entry.step for entry in log.steps), key=lambda step: step.step_idx) + for expected, step in enumerate(steps): + if step.step_idx != expected: + raise CorruptLogError( + f"步记录的序号不连续：第 {expected} 位上是 step_idx={step.step_idx}" + ) + return tuple(steps) + + +def _trailing_parse_failures(steps: tuple[StepRecord, ...]) -> int: + """末尾连着几步没解释出有效决策。 + + 只数末尾那一串：任何一个有效决策把计数清零，所以中间那些散落的失败不算数。 + """ + count = 0 + for step in reversed(steps): + if step.parse_ok: + break + count += 1 + return count + + +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): + raise CorruptLogError("日志里有记录却没有运行开始记录，读不出这份日志是按哪份配置跑的") + + model_intent_ids = { + intent.result_id for intent in log.intents if intent.kind is IntentKind.MODEL_CALL + } + action_intent_ids = { + intent.result_id for intent in log.intents if intent.kind is IntentKind.ACTION + } + + for result_id, times in Counter(entry.result_id for entry in log.model_results).items(): + if times > 1: + raise CorruptLogError(f"结果 ID {result_id!r} 有 {times} 条模型调用结果") + for result in log.model_results: + if result.result_id not in model_intent_ids: + raise CorruptLogError( + f"模型调用结果 {result.result_id!r} 对不上任何一条意图" + "（四态表最后一行：结果有、意图无）" + ) + + for entry in log.steps: + if entry.result_id is not None and entry.result_id not in action_intent_ids: + raise CorruptLogError( + f"步记录带的动作结果 ID {entry.result_id!r} 对不上任何一条动作意图" + "（四态表最后一行：结果有、意图无）" + ) + + +def plan_resume(log: RunLog, model_replay_policy: ReplayPolicy) -> ResumePlan: + """读一份日志，给出续跑计划。日志说不通就抛 `CorruptLogError`。 + + `model_replay_policy` 来自本次运行请求，必填无默认——模型调用的重放策略不像工具那样能 + 从注册表查到，只有调用方知道这次调用能不能重来。 + """ + _check_structural_invariants(log) + + if log.finished is not None: + return ResumePlan(action=ResumeAction.ALREADY_FINISHED, finished_result=log.finished.result) + + if log.started is None: + return ResumePlan(action=ResumeAction.START_FRESH) + + steps = _ordered_steps(log) + executed_actions = sum( + 1 + for entry in log.steps + if entry.action_outcome is not None and entry.action_outcome.status is ActionStatus.EXECUTED + ) + base = { + "steps": steps, + "steps_appended": len(steps), + "actions_executed": executed_actions, + "consecutive_parse_failures": _trailing_parse_failures(steps), + "next_call_index": len(steps), + } + + 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 + ) + if ahead: + raise CorruptLogError( + f"第 {ahead[0]} 步有意图，而第 {interrupted} 步的步记录还没写——中间少了一步" + ) + + model_intent = model_intents.get(interrupted) + if model_intent is None: + if interrupted in action_intents: + raise CorruptLogError( + f"第 {interrupted} 步有动作意图却没有模型调用意图，动作是从哪儿来的" + ) + return ResumePlan(action=ResumeAction.CONTINUE_AT_NEXT_STEP, **base) # type: ignore[arg-type] + + model_result = next( + (entry for entry in log.model_results if entry.result_id == model_intent.result_id), None + ) + model_state = classify_execution(has_intent=True, has_result=model_result is not None) + if model_state is not ExecutionState.COMPLETED or model_result is None: + # 意图在场，所以四态里只剩「状态未知」这一种可能：结果条目缺了。调用可能已经发出去、 + # 也可能没有；发出去了就已经花了钱、已经在网关那边记了账。 + if model_replay_policy is ReplayPolicy.SAFE: + return ResumePlan(action=ResumeAction.REDO_MODEL_CALL, **base) # type: ignore[arg-type] + return ResumePlan(action=ResumeAction.STOP_UNKNOWN, **base) # type: ignore[arg-type] + + if model_result.reply is None: + # 这次调用的状态一点都不未知：它明确地失败过，失败这件事就记在这条结果里。判成状态 + # 未知走重放策略的话，一次已知的失败会被当成可能成功过。 + return ResumePlan( + action=ResumeAction.FINISH_MODEL_CALL_FAILED, + pending_failure=model_result.failure, + **base, # type: ignore[arg-type] + ) + + action_intent = action_intents.get(interrupted) + if action_intent is None: + # 断在模型调用结果与动作意图之间：副作用还没发生，重新解释那条回复是安全的。 + # 不把上次解释的结果填回去——库根本没存过它，存的是模型原文。 + return ResumePlan( + action=ResumeAction.REPARSE_LAST_REPLY, + pending_reply=model_result.reply, + **base, # type: ignore[arg-type] + ) + + # 动作意图写了、步记录没写：动作到底执行没执行，日志答不出来。 + if action_intent.replay_policy is ReplayPolicy.SAFE: + return ResumePlan( + action=ResumeAction.REPLAY_LAST_ACTION, + pending_reply=model_result.reply, + pending_action_result_id=action_intent.result_id, + **base, # type: ignore[arg-type] + ) + return ResumePlan( + action=ResumeAction.STOP_UNKNOWN, + pending_reply=model_result.reply, + **base, # type: ignore[arg-type] + ) + + +__all__ = [ + "CorruptLogError", + "ExecutionState", + "ResumeAction", + "ResumePlan", + "classify_execution", + "plan_resume", +] diff --git a/tests/unit/test_recovery.py b/tests/unit/test_recovery.py new file mode 100644 index 0000000..77ae16d --- /dev/null +++ b/tests/unit/test_recovery.py @@ -0,0 +1,399 @@ +"""恢复判定的行为。 + +一步之内写四次，所以断点有四个，每个断点对应一条续跑路径。下面按断点逐个走一遍，外加 +四态表最后一行那几种「结构上说不通」。 + +**这里的每一条都没有失败现场。** 判错了不会当场炸，只会让恢复出来的历史比不中断跑完时 +少一轮或者多一轮，而模型看到的东西一变，后面每一步都跟着偏——事后只表现成「跑出来的结果 +有点不一样」。 +""" + +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)