From 0132887cb96e23aebeb830fa979493cd0ba20949 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 02:59:34 -0400 Subject: [PATCH] =?UTF-8?q?feat(session):=20=E8=90=BD=E6=88=90=E4=B8=BB?= =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E3=80=81=E4=B8=A4=E4=B8=AA=E8=A3=85=E9=85=8D?= =?UTF-8?q?=E5=AF=B9=E8=B1=A1=E4=B8=8E=E4=B8=A4=E6=9D=A1=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 到 H 八档由这里唯一执行，每档的判定住在 _stopping。三处容易写错的地方都有测试钉着： 恰好用满预算完成记成目标达成而不是预算耗尽（结算在下一次迭代开头）；规模超限不写步也不写 意图（唯一一种真的一步都没走的终止）；未执行与环境故障两档的观察取库合成的那段，执行器 给的不进历史。 写入序列断言成一条流水：模型意图 → 模型结果 → 动作意图 → 步记录，模型调用失败也落一条 结果记录（不落的话恢复会把一次已知的失败判成状态未知走重放）。结果 ID 从运行标识与序号 推出来而不是随机数——重放同一步拿到同一个 ID，轨迹里也少一列每次都不同的值。 写这块时修掉的两个自己的坑： - 重放时会重复写意图，而同一步两条同种意图被 _recovery 判成「日志被并发写过」，一次成功的 重放反倒把日志弄坏。加了两个一次性开关跳过已经落过盘的那条。 - _recovery 数连续解析失败时把模型调用失败那一步也算进去了。它压根没走到解释器，判据改成 「解释器给了一段回喂文本」——解析失败必定带着那段说明，模型调用失败没有。 取消：CancelledError 原样重抛、run 不返回结果，但结束标记要在宽限期内尽力写下去，写不完 只记日志不再抛（再抛会把取消这件事本身盖掉）。并发隔离：全部可变状态住在每次运行一个的 _Driver 里，定义与请求都是 frozen 的，有一条并发跑两次的测试。 Budget 加了构造期校验（四项都必须为正）：零或负数会产出一次「零步、预算耗尽」的运行， 那和一次真的跑满上限的运行在停止原因上完全一样，混进统计里分不出来。 事件出口收下了但没有调用点——Event 还没有字段，发一条内容为空的事件既没用又会变成一份 要兼容的形状。 --- src/polyloop/_recovery/__init__.py | 7 +- src/polyloop/session/__init__.py | 753 +++++++++++++++++++++++++- src/polyloop/types/__init__.py | 17 + tests/unit/test_session.py | 825 +++++++++++++++++++++++++++++ 4 files changed, 1597 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_session.py diff --git a/src/polyloop/_recovery/__init__.py b/src/polyloop/_recovery/__init__.py index 3f34dad..991ffc4 100644 --- a/src/polyloop/_recovery/__init__.py +++ b/src/polyloop/_recovery/__init__.py @@ -164,10 +164,15 @@ def _trailing_parse_failures(steps: tuple[StepRecord, ...]) -> int: """末尾连着几步没解释出有效决策。 只数末尾那一串：任何一个有效决策把计数清零，所以中间那些散落的失败不算数。 + + **判据是「解释器给了一段回喂文本」，不是单看 `parse_ok`。** 模型调用失败那一步的 + `parse_ok` 也是假——那一步压根没走到解释器，把它算进连续解析失败，一次恢复就可能在 + 「连续解析失败」上收尾，而那件事没发生过。解释失败那一步必定带着回喂给模型的说明 + （`0004` 决策三 D 档），模型调用失败那一步没有，两者就靠这个分开。 """ count = 0 for step in reversed(steps): - if step.parse_ok: + if step.parse_ok or step.parse_error is None: break count += 1 return count diff --git a/src/polyloop/session/__init__.py b/src/polyloop/session/__init__.py index 7e11d34..16715c8 100644 --- a/src/polyloop/session/__init__.py +++ b/src/polyloop/session/__init__.py @@ -1,7 +1,752 @@ -"""定义、请求，以及 `run` 与 `resume` 两个动词。 +"""装配层：两个入口协程，以及它们收的两个装配对象。 -**唯一的驱动入口**：不导出低阶循环，也不导出单步。 +这个模块把五个接缝和四个纯逻辑模块编织成一次运行。停止判定的顺序在 +`research-wiki/design/0004-stopping-and-step-record.md` 决策三（A 到 H 八档），本模块的主循环 +是它的唯一执行者；每一档的判定本身住在 `polyloop._stopping`。 -逻辑层那五个模块之间的编织只能发生在这里——它们互不 import，这个模块是它们唯一的会合处。 -代价是这个模块会长，这是接受了的。 +**模块名叫 `session`，但治理单位不叫 session。** 这个名字说的是「这个模块把各部分装配到 +一起」，与一次运行怎么称呼是两件事——`session` 在业界普遍指一个长期存在、可以来回对话的 +东西，而这里的治理单位是有界的、一次性的（`0006` 决策一）。 + +**事件出口现在不发任何事件。** `Event` 还没有字段，事件集与具名回调清单要独立成一份 design +doc；在那之前发一条内容为空的事件既没用又会变成一份要兼容的形状。所以 `event_sink` 这个字段 +收下了但没有调用点，`RunResult.event_delivery_failures` 恒为 0。 """ + +import asyncio +import json +import logging +import time +from collections.abc import Mapping +from dataclasses import dataclass + +from polyloop._assembly import ( + assemble, + check_observation_template, + injected_entry_ids, + prompt_chars, +) +from polyloop._recovery import CorruptLogError, ResumeAction, ResumePlan, plan_resume +from polyloop._stopping import ( + RunCounters, + budget_admission, + completion_verdict, + parse_failure_admission, + prompt_size_admission, +) +from polyloop.ports import ( + Action, + ActionExecutor, + DecisionParser, + EventSink, + FinalAnswer, + InvalidDecision, + ModelCall, + ModelClient, + RunLog, + RunStore, +) +from polyloop.tools import RegistryExecutor, ToolRegistry +from polyloop.types import ( + ActionOutcome, + ActionStatus, + Budget, + Context, + Injection, + Intent, + IntentKind, + Message, + ModelCallResult, + ModelReply, + ReplayPolicy, + RunFinished, + RunResult, + RunStarted, + StepCompleted, + StepRecord, + StopReason, + SyntheticObservations, +) + +logger = logging.getLogger(__name__) + + +class RunIdentityError(Exception): + """运行标识和日志对不上：`run` 撞上已有日志，或者 `resume` 读不到日志。 + + **两条入口的失败方式不同，所以不合并成一个带开关的函数**（`0006` 决策二）：`run` 撞上 + 已有日志要报错，`resume` 读不到日志要报错。合并之后调用方看不出自己走的是哪条，而两种 + 错的处置完全不同。 + """ + + +class ParameterDriftError(Exception): + """续跑时现算的参数快照与日志里那份对不上。 + + **这意味着续跑不能顺便改预算或换模型**，那不是限制而是这条守卫的全部意义：用同一个运行 + 标识换一份定义续跑，前几步与后几步会来自两个不同的配置而全程零报错。 + """ + + +@dataclass(frozen=True, slots=True, kw_only=True) +class AgentDefinition: + """跨运行不变的那一半装配，可以并发复用。 + + 四个接缝挂在这里，因为它们随装配变而不随运行变。**它自己不持有任何一次运行的状态**—— + 持有了的话，并发跑同一份定义的两次运行会互相写到对方的计数里。 + """ + + model_client: ModelClient + decision_parser: DecisionParser + store: RunStore + event_sink: EventSink + #: 库自己合成、回填给模型看的那几段观察。动作被拒绝与环境故障两档不取执行器给的那段， + #: 取这里的（`0007` 决策二）——执行器每次现造一段文本的话，那段文本既不在参数快照里、 + #: 也不受任何约束，两次运行之间它可以变而不会有任何地方报错。 + synthetic_observations: SyntheticObservations + + def parameter_snapshot(self) -> Mapping[str, str]: + """向四个接缝各问一次参数再聚合。 + + **是方法不是字段。** 写成字段就要在构造定义之前先问一遍，而那时定义还不存在；写成 + 方法则每次现问，快照永远是从真实对象上读出来的**事实**而不是一份**声明**。 + + 键带接缝名前缀，免得「哪一侧报的这个键」要靠约定记住。 + """ + snapshot: dict[str, str] = {} + for prefix, seam in ( + ("model_client", self.model_client), + ("decision_parser", self.decision_parser), + ("store", self.store), + ("event_sink", self.event_sink), + ): + for key, value in seam.parameters().items(): + snapshot[f"{prefix}.{key}"] = value + return snapshot + + +@dataclass(frozen=True, slots=True, kw_only=True) +class RunRequest: + """一次运行独有的那一半装配。构造廉价：无 I/O、无网络校验、无哈希计算。""" + + #: 不透明字符串，库不解析。它同时是日志的主键。 + run_id: str + budget: Budget + #: 动作执行接缝。挂在请求上，因为它每次运行都不同。 + action_executor: ActionExecutor + #: 本次可见的那个（子）注册表。**与 `action_executor` 同时存在不是重复**：不注册工具的 + #: 项目传一个空注册表加一个环境句柄，注册了工具的项目传 `tools.executor()`。 + tools: ToolRegistry + context: Context + #: 本次要贴进上下文的条目，按通道分组。 + injections: Mapping[str, tuple[Injection, ...]] + #: 项目自己的标识，库不解释，原样透传给每次模型调用。它的全部键值都进参数快照。 + model_binding: Mapping[str, str] + #: 必填无默认。工具的重放策略能从注册表查到，模型调用的查不到——只有调用方知道这次调用 + #: 能不能重来。 + model_replay_policy: ReplayPolicy + #: 观察回填历史时套的格式，必须含 `{observation}` 占位符。 + observation_template: str + #: 取消进来之后，库留给自己写结束记录的秒数。不写的话，恢复读到的是一次没有结束标记的 + #: 运行，会被当成可以续跑，而它其实是被人主动叫停的。 + cancel_grace_seconds: float + + def __post_init__(self) -> None: + check_observation_template(self.observation_template) + if self.cancel_grace_seconds < 0: + raise ValueError(f"取消宽限期不能为负：{self.cancel_grace_seconds}") + # 模型看见的 schema 来自一个注册表、实际分发走另一个，表现是「模型调了一个它看得见的 + # 工具却说不存在」。执行器不是注册表派生的就一概放行——那是项目自己写执行器的情形， + # 库无从判断也不该判断（`0006` 决策三）。 + if ( + isinstance(self.action_executor, RegistryExecutor) + and self.action_executor.registry != self.tools + ): + raise ValueError( + "action_executor 派生自另一个注册表：模型看见的 schema 与实际分发会来自两份" + f"不同的工具集（执行器持有 {self.action_executor.registry!r}，本次可见 {self.tools!r}）" + ) + + def parameter_snapshot(self) -> Mapping[str, str]: + """请求这一侧的快照，加上向动作执行接缝问的那一次。 + + **上下文与注入内容不进快照。** 它们是这次运行的输入数据不是参数，进快照会让快照变成 + 一份数据副本，而它们可能很大。注入的**条目标识**另行进轨迹，所以「这次贴了哪几条」 + 事后查得到，查不到的只是正文。 + """ + snapshot: dict[str, str] = { + "request.max_steps": str(self.budget.max_steps), + "request.max_actions": str(self.budget.max_actions), + "request.max_consecutive_parse_failures": str( + self.budget.max_consecutive_parse_failures + ), + "request.max_prompt_chars": str(self.budget.max_prompt_chars), + "request.model_replay_policy": self.model_replay_policy.value, + "request.observation_template": self.observation_template, + "request.cancel_grace_seconds": str(self.cancel_grace_seconds), + "request.tools": ",".join(self.tools.names()), + "request.injected_entry_ids": ",".join(injected_entry_ids(self.injections)), + } + # 模型绑定必须进快照，否则给它选字符串映射的那条理由就落空了。失败场景很具体：崩溃后 + # 用同一个运行标识、换一组绑定续跑，后面每一次调用被记到另一套坐标上，而两段轨迹在 + # 文件里看起来是同一次运行。 + for key, value in self.model_binding.items(): + snapshot[f"request.binding.{key}"] = value + for key, value in self.action_executor.parameters().items(): + snapshot[f"action_executor.{key}"] = value + return snapshot + + +def _merged_snapshot(definition: AgentDefinition, request: RunRequest) -> dict[str, str]: + return {**definition.parameter_snapshot(), **request.parameter_snapshot()} + + +def _model_result_id(run_id: str, call_index: int) -> str: + """预分配的模型调用结果 ID。 + + **从运行标识与序号推出来，不用随机数。** 恢复时要精确地问「这个 ID 的结果条目在不在」， + 推得出来就意味着重放同一步时拿到的是同一个 ID——重放沿用原 ID 这件事因此不需要额外传递。 + 随机 ID 还会往轨迹里塞一列每次都不同的值，让两次运行的逐字段比对多一个要排除的字段。 + """ + return f"{run_id}#model#{call_index}" + + +def _action_result_id(run_id: str, call_index: int) -> str: + return f"{run_id}#action#{call_index}" + + +def _serialise_arguments(arguments: Mapping[str, object]) -> str: + """工具参数落进轨迹的那一列。 + + `sort_keys` 是为了同样的参数每次落出同样的字符串——不排的话，字典顺序的差异会让两次运行 + 的逐字段比对报出一堆假差异。`default=str` 兜住不可序列化的取值，因为这一列是给人看的 + 留痕，不该因为某个参数装了个对象就让整次运行失败。 + """ + return json.dumps(arguments, ensure_ascii=False, sort_keys=True, default=str) + + +def _project_observation( + outcome: ActionOutcome, synthetic: SyntheticObservations +) -> tuple[str, bool, int]: + """决定这一步回填进历史的观察是哪一段（`0007` 决策二）。 + + 未执行与环境故障两档取库合成的那段，执行器返回的 `observation` 不进历史——它是「模型看 + 得见的东西」，而那种东西必须能进参数快照。执行器每次现造一段文本的话，两次运行之间它可以 + 变而不会有任何地方报错，于是「同一份配置跑出来的两次运行」在模型看来其实不同。 + + 代价是执行器知道的细节丢了（「哪个参数不合法」只有它知道）。接受它，因为另一头的代价更 + 大；要补的话将来靠事件流把执行器原文送出去做审计——**进历史的东西必须可复现，进审计的 + 不必**。 + """ + if outcome.status is ActionStatus.NOT_EXECUTED: + return synthetic.action_rejected, True, 0 + if outcome.status is ActionStatus.ENV_ERROR: + return synthetic.env_failed, True, 0 + return ( + outcome.observation, + outcome.observation_is_synthetic, + outcome.observation_truncated_chars, + ) + + +class _Driver: + """一次运行的全部可变状态。**每次运行一个实例，绝不跨运行复用。** + + 并发跑同一份定义的多次运行各持一个，所以计数、步序列、时钟互不可见。装配对象那两个是 + frozen 的，共享它们没有问题。 + """ + + __slots__ = ("_counters", "_definition", "_request", "_steps") + + def __init__(self, definition: AgentDefinition, request: RunRequest) -> None: + self._definition = definition + self._request = request + self._counters = RunCounters() + self._steps: list[StepRecord] = [] + + # -- 写入 --------------------------------------------------------------- + + async def _write_step(self, step: StepRecord, outcome: ActionOutcome | None) -> None: + """动作结果与步记录一次原子落地。 + + 存储实现要么两者都可见、要么都不可见。分开写的话，崩在两者之间会让恢复把那一步读成 + 「执行完了，跳过」，那一步的历史文本就永远丢了——恢复出来的消息序列比不中断跑完时少 + 一轮，后面每一步都跟着偏。 + """ + result_id = ( + None if outcome is None else _action_result_id(self._request.run_id, step.step_idx) + ) + await self._definition.store.write_step_completed( + StepCompleted( + run_id=self._request.run_id, + result_id=result_id, + action_outcome=outcome, + step=step, + ) + ) + self._steps.append(step) + self._counters = self._counters.with_step_appended() + + async def _finish(self, stop_reason: StopReason, final_answer: str | None = None) -> RunResult: + """写结束标记，然后返回结果。 + + **标记在把结果交给调用方之前写下。** 让项目自己落盘的话，「跑完了、库返回了、项目存 + 的时候崩了」这种情况下，重启后日志显示最后一步有结果、没有结束标记，而项目那边什么都 + 没有——续跑会重复执行最后一步的副作用，不续跑就丢掉一次已经花完钱的运行。歧义来自 + 结果跨了两个存储。 + """ + result = RunResult( + run_id=self._request.run_id, + stop_reason=stop_reason, + final_answer=final_answer, + steps=tuple(self._steps), + ) + await self._definition.store.write_run_finished( + RunFinished(run_id=self._request.run_id, result=result) + ) + return result + + async def _finish_cancelled(self) -> None: + """取消进来时尽力写下结束标记，宽限期用完就放弃。 + + 不写的话，恢复读到的是一次没有结束标记的运行，会被当成可以续跑，而它其实是被人主动 + 叫停的。**宽限期用完还没写完就放弃，不无限等待**——取消的语义是尽快停下，为了留痕而 + 卡住违背它。 + + 写失败只记日志不再抛：这里已经在 `CancelledError` 的处置路径上，再抛一个异常会把取消 + 这件事本身盖掉，而调用方的结构化并发正等着那个 `CancelledError`。 + """ + result = RunResult( + run_id=self._request.run_id, + stop_reason=StopReason.CANCELLED, + final_answer=None, + steps=tuple(self._steps), + ) + writing = asyncio.ensure_future( + self._definition.store.write_run_finished( + RunFinished(run_id=self._request.run_id, result=result) + ) + ) + try: + await asyncio.wait_for( + asyncio.shield(writing), timeout=self._request.cancel_grace_seconds + ) + except TimeoutError: + writing.cancel() + logger.error( + "取消宽限期内没写完结束记录，运行 %s 的日志缺结束标记", self._request.run_id + ) + except Exception: + logger.exception("取消时写结束记录失败，运行 %s 的日志缺结束标记", self._request.run_id) + + # -- 主循环 ------------------------------------------------------------- + + async def drive(self, plan: ResumePlan | None = None) -> RunResult: + """走 `0004` 决策三那八档，直到某一档给出停止原因。 + + **`CancelledError` 不捕获吞没**：接住只为写一条结束标记，写完原样重抛。取消要能穿过 + 模型调用与动作执行，而 `run` 在取消时**不返回结果**，为的是不破坏调用方的结构化并发 + 语义（`0004` 决策二）。 + """ + try: + return await self._loop(plan) + except asyncio.CancelledError: + await self._finish_cancelled() + raise + + async def finish_resume_unknown(self, plan: ResumePlan) -> RunResult: + """续跑撞上「状态未知且声明绝不重放」时的收尾。 + + 返回一个正常结果而不是抛异常：撞上未知状态是一个可预期的正常终态，不是库自身的缺陷。 + 抛异常会丢掉「跑到第几步、已经花了多少、前面那些步的轨迹」，而那些信息正是项目决定 + 「重跑还是人工介入」时要看的。 + """ + self._steps = list(plan.steps) + return await self._finish(StopReason.RESUME_STATE_UNKNOWN) + + async def _loop(self, plan: ResumePlan | None) -> RunResult: + pending_reply: ModelReply | None = None + pending_failure: str | None = None + # 被打断的那一步的意图已经落过盘了，重放时不能再写一条——同一步两条同种意图会被恢复 + # 判定认成「日志被并发写过」，于是下一次续跑直接拒绝，一次成功的重放反倒把日志弄坏了。 + skip_model_intent = plan is not None and plan.action is ResumeAction.REDO_MODEL_CALL + skip_action_intent = plan is not None and plan.action is ResumeAction.REPLAY_LAST_ACTION + if plan is not None: + self._counters = RunCounters( + steps_appended=plan.steps_appended, + actions_executed=plan.actions_executed, + consecutive_parse_failures=plan.consecutive_parse_failures, + ) + self._steps = list(plan.steps) + pending_reply = plan.pending_reply + pending_failure = plan.pending_failure + + while True: + call_index = len(self._steps) + + # 模型调用失败那一步的步记录还没写完就断了：补上它，然后以模型调用失败收尾。 + if pending_failure is not None: + step = self._failed_call_step( + call_index, prompt_chars_used=0, started=time.monotonic() + ) + await self._write_step(step, None) + return await self._finish(StopReason.LLM_ERROR) + + # A 预算准入。放在开头而不是上一次迭代的结尾：一次「恰好用满预算完成」的运行走的 + # 是完成判定，放结尾它会先撞上预算上限，而两者的轨迹长度一模一样。 + stop = budget_admission(self._counters, self._request.budget) + if stop is not None: + return await self._finish(stop) + + started = time.monotonic() + messages = assemble( + context=self._request.context, + injections=self._request.injections, + steps=self._steps, + observation_template=self._request.observation_template, + ) + chars = prompt_chars(messages) + + # B 规模判定。命中时不产生步记录、也不写任何意图——模型还没被调用、没花钱、没有 + # 调用标识需要对账。这是唯一一种「真的一步都没走」的终止。 + stop = prompt_size_admission(chars, self._request.budget) + if stop is not None: + return await self._finish(stop) + + # C 写模型调用意图，调模型。 + if pending_reply is not None: + reply, pending_reply = pending_reply, None + else: + reply_or_none = await self._call_model( + call_index, messages, write_intent=not skip_model_intent + ) + skip_model_intent = False + if reply_or_none is None: + step = self._failed_call_step(call_index, chars, started) + await self._write_step(step, None) + return await self._finish(StopReason.LLM_ERROR) + reply = reply_or_none + + # D / E 解释决策。 + parsed = self._definition.decision_parser.parse(reply) + decision = parsed.decision + + if isinstance(decision, InvalidDecision): + self._counters = self._counters.with_parse_failure() + await self._write_step( + self._parse_failure_step( + call_index, parsed.history_text, decision, reply, chars, started + ), + None, + ) + stop = parse_failure_admission(self._counters, self._request.budget) + if stop is not None: + return await self._finish(stop) + continue # 这一支跳过完成判定：这一步没碰环境，完成信号不可能因为它改变。 + + self._counters = self._counters.with_parse_success() + + if isinstance(decision, FinalAnswer): + await self._write_step( + self._final_answer_step( + call_index, parsed.history_text, decision, reply, chars, started + ), + None, + ) + return await self._finish(StopReason.AGENT_FINISHED, final_answer=decision.text) + + # F 写动作意图，执行动作，无条件记步。 + outcome, spec_completes_run = await self._execute( + call_index, decision, write_intent=not skip_action_intent + ) + skip_action_intent = False + observation, is_synthetic, truncated = _project_observation( + outcome, self._definition.synthetic_observations + ) + if outcome.status is ActionStatus.EXECUTED: + self._counters = self._counters.with_action_executed() + await self._write_step( + self._action_step( + call_index, + parsed.history_text, + decision, + reply, + outcome, + observation, + is_synthetic, + truncated, + chars, + started, + ), + outcome, + ) + + # G 完成判定。 + stop = completion_verdict(outcome, spec_completes_run) + if stop is not None: + return await self._finish(stop) + # H 回 A。 + + async def _call_model( + self, call_index: int, messages: tuple[Message, ...], *, write_intent: bool + ) -> ModelReply | None: + """写意图、调模型、写结果。返回 `None` 表示这次调用失败了。 + + **失败也必须落一条结果记录。** 只写步不写结果的话，进程在写完步、还没写运行结束时 + 崩溃，恢复读到「意图有、结果无」会判为状态未知走重放策略，而这次调用的状态一点都不 + 未知——它明确地失败过。 + """ + result_id = _model_result_id(self._request.run_id, call_index) + if write_intent: + await self._definition.store.write_intent( + Intent( + run_id=self._request.run_id, + kind=IntentKind.MODEL_CALL, + call_index=call_index, + result_id=result_id, + replay_policy=self._request.model_replay_policy, + ) + ) + try: + reply = await self._definition.model_client.call( + ModelCall( + messages=messages, + call_index=call_index, + run_id=self._request.run_id, + result_id=result_id, + binding=self._request.model_binding, + ) + ) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 — 见 docstring：失败必须落一条结果记录 + await self._definition.store.write_model_call_result( + ModelCallResult( + run_id=self._request.run_id, + result_id=result_id, + reply=None, + # 存说明文本不存异常对象：异常对象没法可靠地序列化成任何一种持久形态， + # 而恢复只需要知道「失败过」以及失败的大致形态。 + failure=f"{type(exc).__name__}: {exc}", + ) + ) + return None + await self._definition.store.write_model_call_result( + ModelCallResult( + run_id=self._request.run_id, result_id=result_id, reply=reply, failure=None + ) + ) + return reply + + async def _execute( + self, call_index: int, action: Action, *, write_intent: bool + ) -> tuple[ActionOutcome, bool]: + """写动作意图，执行动作。返回结果与「这次执行的工具被标了完成标记吗」。 + + 重放策略从注册表查：没有工具的动作（模型输出的是一整段代码）问不出规格，取「绝不 + 重放」——当成可重放而其实不是会重复执行副作用且静默，反过来只是多停一次。 + """ + spec = ( + None + if action.tool_call is None + else self._request.tools.spec_for(action.tool_call.name) + ) + if write_intent: + await self._definition.store.write_intent( + Intent( + run_id=self._request.run_id, + kind=IntentKind.ACTION, + call_index=call_index, + result_id=_action_result_id(self._request.run_id, call_index), + replay_policy=ReplayPolicy.NEVER if spec is None else spec.replay_policy, + ) + ) + outcome = await self._request.action_executor.execute(action) + return outcome, bool(spec is not None and spec.completes_run) + + # -- 步记录 ------------------------------------------------------------- + + def _base_step(self, call_index: int, chars: int, started: float) -> dict[str, object]: + return { + "step_idx": call_index, + "prompt_chars": chars, + "step_wall_ms": int((time.monotonic() - started) * 1000), + } + + def _failed_call_step( + self, call_index: int, prompt_chars_used: int, started: float + ) -> StepRecord: + """模型调用失败那一步。 + + **`parse_error` 留空**，因为这一步压根没走到解释器。恢复那边正是靠这一点把它和解析 + 失败分开：解析失败必定带着回喂给模型的说明，它没有。 + """ + return StepRecord( + **self._base_step(call_index, prompt_chars_used, started), # type: ignore[arg-type] + raw_output="", + content_chars=0, + thinking_chars=0, + action=None, + parse_ok=False, + parse_error=None, + observation=self._definition.synthetic_observations.model_call_failed, + observation_is_synthetic=True, + observation_truncated_chars=0, + call_id=None, + ) + + def _parse_failure_step( + self, + call_index: int, + history_text: str, + decision: InvalidDecision, + reply: ModelReply, + chars: int, + started: float, + ) -> StepRecord: + return StepRecord( + **self._base_step(call_index, chars, started), # type: ignore[arg-type] + raw_output=history_text, + content_chars=len(reply.content), + thinking_chars=len(reply.thinking), + action=None, + parse_ok=False, + # 这段说明**就是**回喂给模型的那段观察，不是从一个固定串里取。压成一句会改掉模型 + # 收到的纠错信息，它的纠错行为也就跟着变。 + parse_error=decision.explanation, + observation=decision.explanation, + observation_is_synthetic=True, + observation_truncated_chars=0, + call_id=reply.call_id, + ) + + def _final_answer_step( + self, + call_index: int, + history_text: str, + decision: FinalAnswer, + reply: ModelReply, + chars: int, + started: float, + ) -> StepRecord: + return StepRecord( + **self._base_step(call_index, chars, started), # type: ignore[arg-type] + raw_output=history_text, + content_chars=len(reply.content), + thinking_chars=len(reply.thinking), + action=None, + parse_ok=True, + parse_error=None, + # 这一支不碰环境，所以没有观察。 + observation="", + observation_is_synthetic=False, + observation_truncated_chars=0, + call_id=reply.call_id, + ) + + def _action_step( + self, + call_index: int, + history_text: str, + action: Action, + reply: ModelReply, + outcome: ActionOutcome, + observation: str, + is_synthetic: bool, + truncated: int, + chars: int, + started: float, + ) -> StepRecord: + return StepRecord( + **self._base_step(call_index, chars, started), # type: ignore[arg-type] + raw_output=history_text, + content_chars=len(reply.content), + thinking_chars=len(reply.thinking), + action=action.text, + parse_ok=True, + parse_error=None, + observation=observation, + observation_is_synthetic=is_synthetic, + observation_truncated_chars=truncated, + call_id=reply.call_id, + tool_name=None if action.tool_call is None else action.tool_call.name, + tool_arguments=( + None + if action.tool_call is None + else _serialise_arguments(action.tool_call.arguments) + ), + action_status=outcome.status, + env_reported_completion=outcome.env_reported_completion, + ) + + +async def run(definition: AgentDefinition, request: RunRequest) -> RunResult: + """从头跑一次运行。 + + **这个运行标识已经有日志时直接报错**，不覆盖也不接着跑。覆盖会毁掉一次已经花完钱的运行 + 的留痕，接着跑是 `resume` 的事而两者的失败方式不同。 + + 取消时**不返回结果**，`CancelledError` 原样重抛——为的是不破坏调用方的结构化并发语义。 + 所以 `StopReason.CANCELLED` 只出现在两个地方：写进日志的那条结束记录里，以及恢复一个已 + 取消运行时重建出来的结果里。读代码的人容易写出「如果结果的停止原因是取消」这种永假分支。 + """ + log = await definition.store.read_log(request.run_id) + if log != RunLog(): + raise RunIdentityError( + f"运行标识 {request.run_id!r} 已经有日志了。要接着跑用 resume；" + "这里不覆盖，因为那会毁掉一次已经花完钱的运行的留痕" + ) + await definition.store.write_run_started( + RunStarted(run_id=request.run_id, parameter_snapshot=_merged_snapshot(definition, request)) + ) + return await _Driver(definition, request).drive() + + +async def resume(definition: AgentDefinition, request: RunRequest) -> RunResult: + """接着跑一次被打断的运行。 + + **读不到日志直接报错**：那说明这个标识对应的运行从没开始过，而 `resume` 的语义是「同一次 + 运行接着做」。 + + 比对参数快照，任何一项不一致直接报错。**这意味着续跑不能顺便改预算或换模型**——那不是 + 限制而是这条守卫的全部意义。 + """ + log = await definition.store.read_log(request.run_id) + if log == RunLog(): + raise RunIdentityError( + f"运行标识 {request.run_id!r} 读不到任何日志，这次运行从没开始过。要从头跑用 run" + ) + plan = plan_resume(log, request.model_replay_policy) + if plan.action is ResumeAction.START_FRESH: + raise CorruptLogError("日志里有记录却判成全新运行，这个状态不该出现") + + current = _merged_snapshot(definition, request) + stored = dict(log.started.parameter_snapshot) if log.started is not None else {} + drifted = sorted( + key for key in set(current) | set(stored) if current.get(key) != stored.get(key) + ) + if drifted: + raise ParameterDriftError( + f"续跑的装配与日志里那份对不上：{drifted}。" + "前几步与后几步来自两个不同的配置，而两段轨迹在文件里看起来是同一次运行" + ) + + if plan.action is ResumeAction.ALREADY_FINISHED: + # 这次运行早就跑完了，把存下来的结果原样交回去，不重跑最后一步。 + if plan.finished_result is None: + raise CorruptLogError("有结束标记却读不出结果，这条记录坏了") + return plan.finished_result + if plan.action is ResumeAction.STOP_UNKNOWN: + return await _Driver(definition, request).finish_resume_unknown(plan) + + return await _Driver(definition, request).drive(plan) + + +__all__ = [ + "AgentDefinition", + "ParameterDriftError", + "RunIdentityError", + "RunRequest", + "resume", + "run", +] diff --git a/src/polyloop/types/__init__.py b/src/polyloop/types/__init__.py index 5095766..3ecb225 100644 --- a/src/polyloop/types/__init__.py +++ b/src/polyloop/types/__init__.py @@ -182,6 +182,23 @@ class Budget: #: 单步装配出的提示词字符数上限。继任下游现有字段名，与步记录的 `prompt_chars` 同源。 max_prompt_chars: int + def __post_init__(self) -> None: + """四项都必须为正。 + + 零或负数不会当场炸，它会让第一档预算准入立刻命中，产出一次「零步、预算耗尽」的运行 + ——那和一次真的跑满了上限的运行在停止原因上完全一样，混进统计里分不出来。用显式异常 + 而不是 `assert`：`python -O` 会把断言整条移除（`CLAUDE.md` §6）。 + """ + for name in ( + "max_steps", + "max_actions", + "max_consecutive_parse_failures", + "max_prompt_chars", + ): + value = getattr(self, name) + if not isinstance(value, int) or isinstance(value, bool) or value < 1: + raise ValueError(f"{name} 必须是不小于 1 的整数，收到 {value!r}") + # --------------------------------------------------------------------------- # 停止原因 diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py new file mode 100644 index 0000000..1d35410 --- /dev/null +++ b/tests/unit/test_session.py @@ -0,0 +1,825 @@ +"""主循环的行为：停止判定顺序、写入序列、恢复守卫、取消收尾、并发隔离。 + +`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, + 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, + 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) -> None: + self.writes: list[object] = [] + self._log = log or RunLog() + + async def write_run_started(self, record: RunStarted) -> None: + self.writes.append(record) + self._log = RunLog( + started=record, + intents=self._log.intents, + model_results=self._log.model_results, + steps=self._log.steps, + finished=self._log.finished, + ) + + async def write_intent(self, record: Intent) -> None: + self.writes.append(record) + + async def write_model_call_result(self, record: ModelCallResult) -> None: + self.writes.append(record) + + async def write_step_completed(self, record: StepCompleted) -> None: + self.writes.append(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) + + 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) -> AgentDefinition: + return AgentDefinition( + model_client=model, + decision_parser=parser, + store=store, + event_sink=FakeSink(), + 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"] == "e1" + + +# --------------------------------------------------------------------------- +# 两条入口的失败方式 +# --------------------------------------------------------------------------- + + +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_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)]))) + store._log = RunLog( # noqa: SLF001 — 把刚写下的那些记录拼回一份可读的日志 + started=store.of_type(RunStarted)[0], # type: ignore[arg-type] + intents=tuple(store.of_type(Intent)), # type: ignore[arg-type] + model_results=tuple(store.of_type(ModelCallResult)), # type: ignore[arg-type] + steps=tuple(store.of_type(StepCompleted)), # type: ignore[arg-type] + finished=store.of_type(RunFinished)[0], # type: ignore[arg-type] + ) + + 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] + + +# --------------------------------------------------------------------------- +# 并发隔离 +# --------------------------------------------------------------------------- + + +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}