feat(session): 落成主循环、两个装配对象与两条入口

A 到 H 八档由这里唯一执行，每档的判定住在 _stopping。三处容易写错的地方都有测试钉着：
恰好用满预算完成记成目标达成而不是预算耗尽（结算在下一次迭代开头）；规模超限不写步也不写
意图（唯一一种真的一步都没走的终止）；未执行与环境故障两档的观察取库合成的那段，执行器
给的不进历史。

写入序列断言成一条流水：模型意图 → 模型结果 → 动作意图 → 步记录，模型调用失败也落一条
结果记录（不落的话恢复会把一次已知的失败判成状态未知走重放）。结果 ID 从运行标识与序号
推出来而不是随机数——重放同一步拿到同一个 ID，轨迹里也少一列每次都不同的值。

写这块时修掉的两个自己的坑：
- 重放时会重复写意图，而同一步两条同种意图被 _recovery 判成「日志被并发写过」，一次成功的
  重放反倒把日志弄坏。加了两个一次性开关跳过已经落过盘的那条。
- _recovery 数连续解析失败时把模型调用失败那一步也算进去了。它压根没走到解释器，判据改成
  「解释器给了一段回喂文本」——解析失败必定带着那段说明，模型调用失败没有。

取消：CancelledError 原样重抛、run 不返回结果，但结束标记要在宽限期内尽力写下去，写不完
只记日志不再抛（再抛会把取消这件事本身盖掉）。并发隔离：全部可变状态住在每次运行一个的
_Driver 里，定义与请求都是 frozen 的，有一条并发跑两次的测试。

Budget 加了构造期校验（四项都必须为正）：零或负数会产出一次「零步、预算耗尽」的运行，
那和一次真的跑满上限的运行在停止原因上完全一样，混进统计里分不出来。

事件出口收下了但没有调用点——Event 还没有字段，发一条内容为空的事件既没用又会变成一份
要兼容的形状。

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

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",
+]

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}")
+
 
 # ---------------------------------------------------------------------------
 # 停止原因

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}