iomgaa
3492a2994a
四件事，都来自第一个下游的 issue。 **RunRequest 新增 fingerprints。** 上下文与注入内容刻意不进快照（它们是数据不是参数）， 而这条规则把生成它们的提示词模板也一起挡在外面了。具体的失败场景：换一份模板续跑不报错， 前几步用 A 模板、后几步用 B，那次运行的数据已经废了却没有任何提示。键形状 request.fingerprint.<name>，和 model_binding 那个坐标分开——一个是「这次运行属于哪一格」， 一个是「用的配方是哪一版」，混在一个字段里事后分不开。默认空映射时一个键都不写，所以已有 配置算出来的快照逐字节不变。 **注入的通道维度不再拍平。** 一通道一键，于是「声明了通道但一条都没选中」和「压根没有这个 通道」分得开：前者是值为空串的键，后者是键不存在。有一档实验要比较的正是这两种情形。 **ActionExecutor 的 docstring 写定抛异常时会怎样**：环境故障走 ENV_ERROR 返回值，实现方真 抛了库不接管、异常原样穿出。理由在 design 0016；简言之库替它编一个结算结果就是在编造，而 副作用状态在那一刻是未知的。 **三个映射字段在构造期冻成只读。** 对抗审查发现 frozen=True 不禁止改字段里那个 dict，于是 构造期那道「快照取值必须是字符串」的校验能被绕过去：构造完往 fingerprints 里塞一个整数， 它一路进日志，要到续跑读日志时才炸——而那时这次运行已经完整跑过一遍。复用 tools 里已有的 _frozen，没另写一套。
1554 lines
61 KiB
Python
1554 lines
61 KiB
Python
"""主循环的行为：停止判定顺序、写入序列、恢复守卫、取消收尾、并发隔离。
|
|
|
|
`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.skill"] == "e1"
|
|
|
|
|
|
def test_a_declared_but_empty_channel_still_gets_a_key() -> None:
|
|
"""声明了通道却一条都没选中，是一个值为空串的键。
|
|
|
|
有下游要比较「声明了通道而选中零条」和「连通道都不声明」这两组运行，拍平之后两者在记录
|
|
里长得一样，那一档就测不了。
|
|
"""
|
|
request = dataclasses.replace(_request(FakeExecutor([])), injections={"skill": ()})
|
|
|
|
assert request.parameter_snapshot()["request.injected_entry_ids.skill"] == ""
|
|
|
|
|
|
def test_a_channel_that_was_never_declared_has_no_key_at_all() -> None:
|
|
request = dataclasses.replace(_request(FakeExecutor([])), injections={})
|
|
|
|
snapshot = request.parameter_snapshot()
|
|
|
|
assert not [key for key in snapshot if key.startswith("request.injected_entry_ids")]
|
|
|
|
|
|
def test_two_channels_do_not_get_merged_into_one_key() -> None:
|
|
"""通道名在拍平的写法里只用来定顺序，排完就没了，「哪几条来自哪个通道」事后查不到。"""
|
|
request = dataclasses.replace(
|
|
_request(FakeExecutor([])),
|
|
injections={
|
|
"skill": (
|
|
Injection(entry_id="s1", content="甲"),
|
|
Injection(entry_id="s2", content="乙"),
|
|
),
|
|
"memo": (Injection(entry_id="m1", content="丙"),),
|
|
},
|
|
)
|
|
|
|
snapshot = request.parameter_snapshot()
|
|
|
|
assert snapshot["request.injected_entry_ids.skill"] == "s1,s2"
|
|
assert snapshot["request.injected_entry_ids.memo"] == "m1"
|
|
|
|
|
|
def test_no_fingerprints_means_no_fingerprint_keys() -> None:
|
|
"""默认空映射时一个键都不写，不是写一个值为空串的键。
|
|
|
|
今天已经在跑的配置算出来的快照因此逐字节不变，只有真的传了指纹的运行才多出那几项。
|
|
"""
|
|
snapshot = _request(FakeExecutor([])).parameter_snapshot()
|
|
|
|
assert not [key for key in snapshot if key.startswith("request.fingerprint.")]
|
|
|
|
|
|
def test_every_fingerprint_enters_the_snapshot() -> None:
|
|
"""提示词模板不是数据是参数：它是一份跨运行复用的配方，每次运行拿它渲染出这一次的上下文。
|
|
|
|
不记的话，「这次用的是哪一版提示词」就只剩下调用方的 git 提交这一个粒度，而同一个提交下
|
|
完全可以试好几份不同的模板。
|
|
"""
|
|
request = dataclasses.replace(
|
|
_request(FakeExecutor([])),
|
|
fingerprints={"prompt_template": "sha256:abc", "skill_pack": "sha256:def"},
|
|
)
|
|
|
|
snapshot = request.parameter_snapshot()
|
|
|
|
assert snapshot["request.fingerprint.prompt_template"] == "sha256:abc"
|
|
assert snapshot["request.fingerprint.skill_pack"] == "sha256:def"
|
|
|
|
|
|
def test_a_non_string_binding_value_is_refused_at_construction() -> None:
|
|
"""快照的取值类型已经是持久化契约的一部分，只靠反序列化那一侧的话，这次运行会照常跑完、
|
|
钱花光，才在续跑读日志时发现有一项读不回来。"""
|
|
with pytest.raises(ValueError, match="RunRequest.model_binding"):
|
|
dataclasses.replace(_request(FakeExecutor([])), model_binding={"round": 3})
|
|
|
|
|
|
def test_a_non_string_fingerprint_is_refused_at_construction() -> None:
|
|
"""`model_binding` 与 `fingerprints` 语义同族、形状相同，两个字段的行为必须一样——不一样
|
|
比两个都不校验更难查，查的人会先怀疑自己传错了字段。"""
|
|
with pytest.raises(ValueError, match="RunRequest.fingerprints"):
|
|
dataclasses.replace(_request(FakeExecutor([])), fingerprints={"template": 7})
|
|
|
|
with pytest.raises(ValueError, match="RunRequest.fingerprints"):
|
|
dataclasses.replace(_request(FakeExecutor([])), fingerprints={7: "sha256:abc"})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field_name", "added_value"),
|
|
[("model_binding", "a"), ("fingerprints", "sha256:abc"), ("injections", ())],
|
|
)
|
|
def test_a_mapping_field_cannot_be_changed_after_construction(
|
|
field_name: str, added_value: object
|
|
) -> None:
|
|
"""构造期那道校验拦不住构造之后原地改，所以三个映射字段构造时就被冻成只读的。
|
|
|
|
`frozen=True` 只挡住「把字段重新绑到另一个对象上」。往 `fingerprints` 里塞一个整数，它会
|
|
一路进到运行开始记录的参数快照，要到续跑读日志反序列化那一关才炸——而那时这次运行已经完整
|
|
跑过一遍、钱也花完了。`injections` 那一份还会同时改掉装配出来的消息序列。
|
|
|
|
`fingerprints` 这一档同时守住默认值：不传它的请求手上是一个 `default_factory` 造的空
|
|
dict，不冻的话那个 dict 改得动，而往里塞什么都不经过校验。
|
|
"""
|
|
request = _request(FakeExecutor([]))
|
|
|
|
with pytest.raises(TypeError):
|
|
getattr(request, field_name)["新加的"] = added_value
|
|
|
|
|
|
def test_changing_the_dict_that_was_passed_in_does_not_change_the_request() -> None:
|
|
"""调用方传进来的那个 dict 之后再被改，请求看见的仍是构造那一刻的形状。
|
|
|
|
另一个方向：不拷一份的话，一个被复用的绑定 dict 改一个键，就悄悄改掉了一个已经在跑的运行
|
|
的参数快照。
|
|
"""
|
|
binding = {"item": "a"}
|
|
request = dataclasses.replace(_request(FakeExecutor([])), model_binding=binding)
|
|
|
|
binding["item"] = "b"
|
|
|
|
assert request.parameter_snapshot()["request.binding.item"] == "a"
|
|
|
|
|
|
def test_a_request_derived_with_replace_still_constructs() -> None:
|
|
"""`dataclasses.replace` 是推荐的派生范式，它会把已经冻过的那个只读映射再传进一次构造。
|
|
|
|
冻结那一步不能因此炸——炸的话，派生一个请求就得先把三个映射字段各拆回普通 dict，而那正是
|
|
大多数调用方不会想到要做的一步。派生出来的那个照样是冻的。
|
|
"""
|
|
request = dataclasses.replace(
|
|
_request(FakeExecutor([])), fingerprints={"prompt_template": "sha256:abc"}
|
|
)
|
|
|
|
derived = dataclasses.replace(request, run_id="run-2")
|
|
|
|
assert derived.run_id == "run-2"
|
|
assert derived.parameter_snapshot()["request.fingerprint.prompt_template"] == "sha256:abc"
|
|
with pytest.raises(TypeError):
|
|
derived.fingerprints["prompt_template"] = "sha256:def"
|
|
|
|
|
|
async def test_the_frozen_binding_still_reaches_the_model_call_and_the_event() -> None:
|
|
"""绑定冻成只读映射之后，模型调用与事件上那两份照样读得出来、内容一致。
|
|
|
|
只读是这三个字段的全部改动：透传路径没有变，下游拿它查键、比对、序列化都和以前一样，
|
|
只是拿它反过来改这次运行的配置不再行得通。
|
|
"""
|
|
store = FakeStore()
|
|
sink = FakeSink()
|
|
model = FakeModel([_reply("go")])
|
|
definition = _definition(store, model, FakeParser({"go": ACT}), sink)
|
|
|
|
await run(definition, _request(FakeExecutor([_outcome(completed=True)]), binding={"item": "a"}))
|
|
|
|
(event,) = sink.events
|
|
assert model.calls[0].binding == {"item": "a"}
|
|
assert event.model_binding == {"item": "a"}
|
|
with pytest.raises(TypeError):
|
|
model.calls[0].binding["item"] = "b" # type: ignore[index]
|
|
|
|
|
|
class _BadParameters:
|
|
"""把一个接缝原样代理出去，只把 `parameters()` 换成一份取值不是字符串的映射。"""
|
|
|
|
def __init__(self, seam: object, parameters: Mapping[object, object]) -> None:
|
|
self._seam = seam
|
|
self._parameters = parameters
|
|
|
|
def __getattr__(self, name: str) -> object:
|
|
return getattr(self._seam, name)
|
|
|
|
def parameters(self) -> Mapping[object, object]:
|
|
return self._parameters
|
|
|
|
|
|
@pytest.mark.parametrize("seam", ["model_client", "decision_parser", "store", "event_sink"])
|
|
def test_a_definition_seam_reporting_a_non_string_is_refused(seam: str) -> None:
|
|
"""接缝实现由下游写，它返回什么算外部输入。五个接缝里任何一个返回一个整数，症状都一样：
|
|
这次运行照常跑完，续跑时才炸。
|
|
|
|
报错要指得出是哪个接缝——只说「快照必须是字符串」的话，七个来源里找不出是谁。
|
|
"""
|
|
definition = _definition(FakeStore(), FakeModel([]), FakeParser({}))
|
|
broken = dataclasses.replace(
|
|
definition, **{seam: _BadParameters(getattr(definition, seam), {"oops": 1})}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match=rf"{seam}\.parameters"):
|
|
broken.parameter_snapshot()
|
|
|
|
|
|
def test_the_action_executor_reporting_a_non_string_is_refused() -> None:
|
|
"""第五个接缝挂在请求上，聚合发生在另一处，所以它要单独验一遍。"""
|
|
request = dataclasses.replace(
|
|
_request(FakeExecutor([])),
|
|
action_executor=_BadParameters(FakeExecutor([]), {"session": 42}), # type: ignore[arg-type]
|
|
)
|
|
|
|
with pytest.raises(ValueError, match=r"action_executor\.parameters"):
|
|
request.parameter_snapshot()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 两条入口的失败方式
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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_refuses_when_only_a_fingerprint_changed() -> None:
|
|
"""换一份模板、用同一个运行标识续跑，前几步用 A、后几步用 B，全程零报错，那次运行的数据
|
|
已经废了却没有任何东西提示。守住这个失败场景是 `fingerprints` 存在的全部意义。
|
|
|
|
别的东西一个字都没变：预算、绑定、接缝、注入都一样，只有指纹换了。
|
|
"""
|
|
store = FakeStore()
|
|
definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}))
|
|
first = dataclasses.replace(
|
|
_request(FakeExecutor([_outcome(completed=True)])),
|
|
fingerprints={"prompt_template": "sha256:aaa"},
|
|
)
|
|
await run(definition, first)
|
|
|
|
second = dataclasses.replace(first, fingerprints={"prompt_template": "sha256:bbb"})
|
|
with pytest.raises(ParameterDriftError, match="request.fingerprint.prompt_template"):
|
|
await resume(definition, second)
|
|
|
|
|
|
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]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 动作执行接缝抛异常
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _RaisingExecutor(FakeExecutor):
|
|
"""执行动作时抛出，而不是返回一个填好状态的结果。"""
|
|
|
|
def __init__(self, error: BaseException) -> None:
|
|
super().__init__([])
|
|
self._error = error
|
|
|
|
async def execute(self, action: Action) -> ActionOutcome:
|
|
self.actions.append(action)
|
|
raise self._error
|
|
|
|
|
|
async def test_an_executor_exception_comes_straight_out_of_run() -> None:
|
|
"""库不接管，异常原样穿出去，不被转成任何停止原因。
|
|
|
|
接住就得给这一步编一个动作结果——状态、观察、完成信号、截断字符数四个字段全是库现造的，
|
|
没有一个来自执行器。而一条带着动作结果的完整步记录会被恢复读成「上一步走完了」，那个未知
|
|
状态就被抹掉了。转成 `ENV_ERROR` 还会把实现方的 bug 伪装成环境故障送进下游的统计。
|
|
"""
|
|
store = FakeStore()
|
|
definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}))
|
|
executor = _RaisingExecutor(AttributeError("包装类自己的 bug"))
|
|
|
|
with pytest.raises(AttributeError, match="包装类自己的 bug"):
|
|
await run(definition, _request(executor))
|
|
|
|
assert executor.actions != []
|
|
# 日志停在「动作意图有、步记录无」，也没有结束记录：不写恰好是照实。
|
|
assert [entry.kind for entry in store.of_type(Intent)] == [ # type: ignore[attr-defined]
|
|
IntentKind.MODEL_CALL,
|
|
IntentKind.ACTION,
|
|
]
|
|
assert store.of_type(StepCompleted) == []
|
|
assert store.of_type(RunFinished) == []
|
|
|
|
|
|
async def test_resume_after_an_executor_exception_reads_the_state_as_unknown() -> None:
|
|
"""副作用发生了没有，库无从知道，日志里也没有任何一处记着。
|
|
|
|
这和进程崩在动作执行中途是同一种状态，恢复照四态表把它读成未知；那条动作意图记着「绝不
|
|
重放」，于是这次运行以「续跑时状态未知」收尾，而不是重新执行一次动作。
|
|
"""
|
|
store = FakeStore()
|
|
definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}))
|
|
request = _request(_RaisingExecutor(AttributeError("包装类自己的 bug")))
|
|
with pytest.raises(AttributeError):
|
|
await run(definition, request)
|
|
|
|
result = await resume(definition, request)
|
|
|
|
assert result.stop_reason is StopReason.RESUME_STATE_UNKNOWN
|
|
assert result.steps == ()
|
|
|
|
|
|
async def test_cancellation_from_the_executor_still_writes_the_finished_marker() -> None:
|
|
"""取消要能穿过动作执行，`CancelledError` 原样重抛，但结束标记照常写下去。
|
|
|
|
不写的话，恢复读到的是一次没有结束标记的运行，会被当成可以续跑，而它其实是被人主动叫停的。
|
|
"""
|
|
store = FakeStore()
|
|
definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}))
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await run(definition, _request(_RaisingExecutor(asyncio.CancelledError())))
|
|
|
|
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 == "真动作"
|