feat(_recovery): 落成四态判定与四个断点各自的续跑路径
一步之内写四次（模型意图 → 模型结果 → 动作意图 → 步记录原子写），所以断点有四个， 这个模块的全部工作就是把断点认出来： - 模型意图有、结果无 → 状态未知，按请求声明的模型重放策略决定重调还是停下来报告 - 结果条目在但记的是失败 → 状态一点都不未知，补上那一步以模型调用失败收尾 （判成未知走重放的话，一次已知的失败会被当成可能成功过） - 回复已存、动作意图未写 → 副作用还没发生，重新解释那条回复，不再花一次调用的钱 - 动作意图有、步记录无 → 按意图记录里存下的那个策略决定；重放沿用原来那个预分配 ID， 另分配一个的话原意图永远配不上结果，下次恢复读到的还是「状态未知」 四态表最后一行（结果有、意图无）判为日志损坏拒绝续跑，另加几种撞号与缺号：同一步两条 同种意图（并发写）、步序号不连续（中间某一步的原子写整个丢了）、意图跑到步记录前面、 有动作意图却没有模型调用意图、有记录却没有运行开始记录。 模块 docstring 里写明了一条此前只是隐含的对齐：步序号与模型调用序号是同一个数。没有它， result_id 为空的步记录（解析失败、模型调用失败、最终回答三档）认不出属于哪一步。
This commit is contained in:
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user