iomgaa
fa09e873c5
一步之内写四次（模型意图 → 模型结果 → 动作意图 → 步记录原子写），所以断点有四个， 这个模块的全部工作就是把断点认出来： - 模型意图有、结果无 → 状态未知，按请求声明的模型重放策略决定重调还是停下来报告 - 结果条目在但记的是失败 → 状态一点都不未知，补上那一步以模型调用失败收尾 （判成未知走重放的话，一次已知的失败会被当成可能成功过） - 回复已存、动作意图未写 → 副作用还没发生，重新解释那条回复，不再花一次调用的钱 - 动作意图有、步记录无 → 按意图记录里存下的那个策略决定；重放沿用原来那个预分配 ID， 另分配一个的话原意图永远配不上结果，下次恢复读到的还是「状态未知」 四态表最后一行（结果有、意图无）判为日志损坏拒绝续跑，另加几种撞号与缺号：同一步两条 同种意图（并发写）、步序号不连续（中间某一步的原子写整个丢了）、意图跑到步记录前面、 有动作意图却没有模型调用意图、有记录却没有运行开始记录。 模块 docstring 里写明了一条此前只是隐含的对齐：步序号与模型调用序号是同一个数。没有它， result_id 为空的步记录（解析失败、模型调用失败、最终回答三档）认不出属于哪一步。
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""恢复判定的行为。
|
|
|
|
一步之内写四次，所以断点有四个，每个断点对应一条续跑路径。下面按断点逐个走一遍，外加
|
|
四态表最后一行那几种「结构上说不通」。
|
|
|
|
**这里的每一条都没有失败现场。** 判错了不会当场炸，只会让恢复出来的历史比不中断跑完时
|
|
少一轮或者多一轮，而模型看到的东西一变，后面每一步都跟着偏——事后只表现成「跑出来的结果
|
|
有点不一样」。
|
|
"""
|
|
|
|
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)
|