feat(_stopping): 落成停止判定四档，0008 转已接受

四个纯函数各管一档，顺序本身由主循环走：预算准入（A）、提示词规模（B）、连续解析失败（D）、
完成判定（G）。加一个不可变的三计数结构。

三处刻意钉死的边界，都写了测试：
- 两个预算上界同时耗尽报步数那一个。理由是兼容性不是原理（某下游按步数耗尽的占比告警），
  正因为推不出来才必须被测试钉死。
- 预算「达到即拦」用 >=，提示词规模「超过才拦」用 >。两处不是同一件事：一个数已经用掉
  几个额度，一个量一个东西有多大。
- 环境故障排在完成判定前面；未执行不做完成判定；完成信号恒为假不是故障（初稿在这里
  写错过，会让某个下游的每一次运行都在第一步终止）。

不 import polyloop.tools，五个逻辑层模块互不 import，所以「这次执行的工具被标了完成标记吗」
由调用方查好注册表传一个布尔进来。

research-wiki/design/0008-tool-handlers.md

@@ -1,6 +1,6 @@
 # Design 0008 · 工具的实现怎么挂进注册表
 
-**日期** 2026-08-10 · **状态** 待确认
+**日期** 2026-08-10 · **状态** 已接受（2026-08-10 项目负责人确认）
 
 **补充** `0006-public-names-and-signatures.md` 决策六。那一条定了 `ToolSpec` 的五个字段与
 `ToolRegistry` 的五个方法，本文补上其中缺的那一样：工具本身的实现挂在哪里。

src/polyloop/_stopping/__init__.py

@@ -1,7 +1,147 @@
-"""停止判定与预算结算。内部模块。
+"""停止判定：每一档在什么条件下给出哪个停止原因。
 
-判定顺序是有序的，不是一组独立条件——「恰好在最后一步做完」和「预算耗尽」的轨迹长度
+**内部模块**（下划线开头，不进 `polyloop/__init__.py`）。它是
-一模一样，顺序错了两者会互换，而且不会有任何地方报错。
+`research-wiki/design/0004-stopping-and-step-record.md` 决策三那套顺序的唯一实现；顺序本身由
+`polyloop.session` 的主循环走，这里只提供每一档的判定。
 
-**纯逻辑，约束同 `_assembly`。**
+**纯逻辑，无 I/O、无事件循环。** 这一条由 `pyproject.toml` 的一条 import 契约断言（禁止
+import `asyncio` 与 `pathlib`）。它是烟雾报警不是纯度证明——`os`、`subprocess` 都能绕过去。
+
+**不 import `polyloop.tools`**，五个逻辑层模块互不 import。所以「这次执行的工具被标了完成
+标记吗」由调用方查好注册表、把答案作为一个布尔传进来。
+
+判定顺序的完整论证不在这里复述，见那份 design doc。这里的 docstring 只写「读这段代码的人
+不知道就会写错什么」。
 """
+
+from dataclasses import dataclass, replace
+
+from polyloop.types import ActionOutcome, ActionStatus, Budget, StopReason
+
+
+@dataclass(frozen=True, slots=True)
+class RunCounters:
+    """一次运行走到此刻的三个计数。
+
+    **步数与已执行动作数是两个独立计数，不是一个标量。** 两个消费者要的不是同一个量：一个数
+    的是追加进轨迹的步记录条数（解析失败、模型调用失败、环境故障的步都算），一个数的是动作
+    执行接缝返回「已执行」的次数。合成一个的后果是其中一方的语义被改写，而且「模型反复调不
+    存在的工具烧光预算」与「真的做了五十步没做完」在轨迹上就分不开了。
+
+    不可变，每次推进返回新实例。并发的多次运行各持一份，不共享。
+    """
+
+    #: 追加进轨迹的步记录条数。
+    steps_appended: int = 0
+    #: 动作执行接缝返回「已执行」的次数。
+    actions_executed: int = 0
+    #: 连续解释不出有效决策的次数。任何一个有效决策把它清零。
+    consecutive_parse_failures: int = 0
+
+    def with_step_appended(self) -> "RunCounters":
+        return replace(self, steps_appended=self.steps_appended + 1)
+
+    def with_action_executed(self) -> "RunCounters":
+        """动作执行接缝返回「已执行」时加一。
+
+        **只有「已执行」加**：未执行与环境故障都不算真的做了事。步数那一侧另算——那一步
+        照样被记进轨迹，所以两个计数在同一步里可能一个加一个不加。
+        """
+        return replace(self, actions_executed=self.actions_executed + 1)
+
+    def with_parse_failure(self) -> "RunCounters":
+        return replace(self, consecutive_parse_failures=self.consecutive_parse_failures + 1)
+
+    def with_parse_success(self) -> "RunCounters":
+        """任何一个有效决策（动作或最终回答）把连续失败计数清零。
+
+        不清零的话，一次运行里零散的几次解析失败会累加到上限，然后一次「连续失败」被报成
+        停止原因——而它根本没有连续过。
+        """
+        return replace(self, consecutive_parse_failures=0)
+
+
+def budget_admission(counters: RunCounters, budget: Budget) -> StopReason | None:
+    """A 档：预算准入。每次迭代**开头**问一次，不是上一次迭代的结尾。
+
+    放在开头，一次「恰好用满预算完成」的运行走到完成判定、记成目标达成；放在结尾，它会先撞
+    上预算上限记成预算耗尽。**两者的轨迹长度一模一样**，事后从数据里分不出来，而按停止原因
+    分层的整批统计会因此失真——本该算作成功的那些运行被计进了「预算不够」那一档。
+
+    **两个上限同时命中报步数那一个。** 这条的理由是兼容性而不是原理：某个下游的告警判据按
+    步数耗尽的占比统计，报另一个会让那条判据在这种情况下漏掉。两个上界同时耗尽时，两个原因
+    描述的其实是同一件事——所以它必须被固定下来并被测试断言，而不是留给实现随手决定
+    （`research-wiki/design/0004-stopping-and-step-record.md` 决策一）。
+
+    上限是**可以取到**的：已追加步数**达到**上限就不许再走一步。
+    """
+    if counters.steps_appended >= budget.max_steps:
+        return StopReason.STEP_BUDGET
+    if counters.actions_executed >= budget.max_actions:
+        return StopReason.ACTION_BUDGET
+    return None
+
+
+def prompt_size_admission(prompt_chars: int, budget: Budget) -> StopReason | None:
+    """B 档：装配出的提示词规模。**超过**上限才拦，正好等于上限是允许的。
+
+    和 A 档那个「达到即拦」不是一回事：那里数的是已经用掉几个额度，这里量的是一个东西有多
+    大。两处都把上限本身算作允许，只是「用掉 N 个」和「有 N 那么大」在同一条界线上落到了
+    不同的比较符上。
+
+    **超限必须显式终止，不许静默截断。** 截断是一次前缀破坏操作，会让后续每一步重新全价
+    计费，而且被截断的运行表现成一批低分，看起来像模型能力不足。
+
+    命中这一档时**不产生步记录、也不写任何意图记录**——模型还没被调用、没花钱、没有调用
+    标识需要对账。这是唯一一种「真的一步都没走」的终止。伪造一条空步会在轨迹里多出一条永远
+    连不上账目的记录。这条约束由调用方遵守，这个函数管不着。
+    """
+    if prompt_chars > budget.max_prompt_chars:
+        return StopReason.CONTEXT_OVERFLOW
+    return None
+
+
+def parse_failure_admission(counters: RunCounters, budget: Budget) -> StopReason | None:
+    """D 档：连续解析失败。计数**加过之后**问，达到上限就停。
+
+    这一支**跳过完成判定**：这一步没碰环境，环境的完成信号不可能因为它改变。
+    """
+    if counters.consecutive_parse_failures >= budget.max_consecutive_parse_failures:
+        return StopReason.PARSE_FAILED_REPEATEDLY
+    return None
+
+
+def completion_verdict(outcome: ActionOutcome, tool_completes_run: bool) -> StopReason | None:
+    """G 档：动作执行之后的完成判定。返回 `None` 表示接着跑。
+
+    `tool_completes_run` 是调用方查注册表得来的答案：这次执行的工具被标了完成标记吗。没有
+    工具的动作（模型输出的是一整段代码）传假。这个模块不 import 工具模块，所以查不了。
+
+    **状态是「未执行」时不做完成判定**：动作根本没进入真实执行，环境状态没变，完成条件不可能
+    因为它成立。
+
+    **完成信号恒为「未完成」不是故障。** 没有环境完成信号的环境就是这么返回的，走「接着跑」
+    那一支，靠完成标记或预算收尾。初稿在这里写错过，错法值得记下来：那时完成信号被定成
+    「布尔或空、空表示取不到」，这一档写着「取不到就是环境故障」——而有一个下游每一步都返回
+    空，于是它的每一次运行都会在第一步撞环境故障终止。不是边缘情况，是全部。修法是把完成
+    信号收成布尔、把「查询失败」挪到状态字段上，两件事从此不共用一个取值。
+
+    **环境故障排在完成判定前面。** 环境坏了就没有下一步可走，继续跑只会产出一串同样的故障，
+    把预算烧光而轨迹上全是噪声。
+    """
+    if outcome.status is ActionStatus.ENV_ERROR:
+        return StopReason.ENV_ERROR
+    if outcome.status is not ActionStatus.EXECUTED:
+        return None
+    if outcome.env_reported_completion or tool_completes_run:
+        return StopReason.TASK_COMPLETED
+    return None
+
+
+__all__ = [
+    "RunCounters",
+    "budget_admission",
+    "completion_verdict",
+    "parse_failure_admission",
+    "prompt_size_admission",
+]

tests/unit/test_stopping.py

@@ -0,0 +1,229 @@
+"""停止判定每一档的行为。
+
+这个模块的输入空间小到可以穷举，所以这里就穷举：三个计数各自在上限的下方、正好、上方，
+四种动作状态叉乘两种完成信号与两种完成标记。**边界那一格是重点**——「恰好用满预算完成」
+和「预算耗尽」的轨迹长度一模一样，判错了事后从数据里分不出来。
+"""
+
+import pytest
+
+from polyloop._stopping import (
+    RunCounters,
+    budget_admission,
+    completion_verdict,
+    parse_failure_admission,
+    prompt_size_admission,
+)
+from polyloop.types import ActionOutcome, ActionStatus, Budget, StopReason
+
+pytestmark = pytest.mark.unit
+
+BUDGET = Budget(
+    max_steps=10,
+    max_actions=4,
+    max_consecutive_parse_failures=3,
+    max_prompt_chars=1000,
+)
+
+
+def _outcome(
+    status: ActionStatus,
+    *,
+    env_reported_completion: bool = False,
+) -> ActionOutcome:
+    return ActionOutcome(
+        status=status,
+        observation="",
+        observation_is_synthetic=False,
+        env_reported_completion=env_reported_completion,
+        observation_truncated_chars=0,
+    )
+
+
+# ---------------------------------------------------------------------------
+# 计数
+# ---------------------------------------------------------------------------
+
+
+def test_counters_start_at_zero() -> None:
+    assert RunCounters() == RunCounters(
+        steps_appended=0, actions_executed=0, consecutive_parse_failures=0
+    )
+
+
+def test_advancing_a_counter_leaves_the_original_alone() -> None:
+    """不可变，每次推进返回新实例。并发的多次运行各持一份。"""
+    start = RunCounters()
+
+    advanced = start.with_step_appended().with_action_executed()
+
+    assert start == RunCounters()
+    assert advanced == RunCounters(steps_appended=1, actions_executed=1)
+
+
+def test_a_step_and_an_executed_action_are_counted_separately() -> None:
+    """同一步里两个计数可能一个加一个不加。
+
+    步数数的是追加进轨迹的条数（解析失败、模型调用失败、环境故障的步都算），已执行动作数
+    只数动作执行接缝返回「已执行」的次数。合成一个的话，「模型反复调不存在的工具烧光预算」
+    和「真的做了五十步没做完」在轨迹上就分不开了。
+    """
+    only_a_step = RunCounters().with_step_appended()
+
+    assert only_a_step.steps_appended == 1
+    assert only_a_step.actions_executed == 0
+
+
+def test_a_parse_success_clears_the_consecutive_counter() -> None:
+    """不清零的话，一次运行里零散的几次解析失败会累加到上限，然后被报成「连续失败」。"""
+    counters = RunCounters().with_parse_failure().with_parse_failure()
+
+    assert counters.with_parse_success().consecutive_parse_failures == 0
+
+
+# ---------------------------------------------------------------------------
+# A 档：预算准入
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("steps", [0, 1, 9])
+def test_below_the_step_limit_the_run_continues(steps: int) -> None:
+    assert budget_admission(RunCounters(steps_appended=steps), BUDGET) is None
+
+
+@pytest.mark.parametrize("steps", [10, 11])
+def test_reaching_the_step_limit_stops_the_run(steps: int) -> None:
+    """上限是可以取到的：已追加步数**达到**上限就不许再走一步。"""
+    assert budget_admission(RunCounters(steps_appended=steps), BUDGET) is StopReason.STEP_BUDGET
+
+
+@pytest.mark.parametrize(("actions", "expected"), [(3, None), (4, StopReason.ACTION_BUDGET)])
+def test_the_action_limit_has_its_own_stop_reason(actions: int, expected: object) -> None:
+    assert budget_admission(RunCounters(actions_executed=actions), BUDGET) is expected
+
+
+def test_when_both_limits_are_hit_the_step_one_wins() -> None:
+    """两个上界同时耗尽时报步数那一个。
+
+    理由是兼容性不是原理——某个下游的告警判据按步数耗尽的占比统计，报另一个会让那条判据在
+    这种情况下漏掉。正因为它不是从原理推出来的，才必须被测试钉死，而不是留给实现随手决定。
+    """
+    both = RunCounters(steps_appended=10, actions_executed=4)
+
+    assert budget_admission(both, BUDGET) is StopReason.STEP_BUDGET
+
+
+def test_the_budget_is_checked_before_the_step_that_would_exceed_it() -> None:
+    """走完第 9 步（还差一步到上限）时不停，走完第 10 步才停。
+
+    这一条守的是「预算结算在下一次迭代的开头」：一次恰好用满预算完成的运行走的是完成判定，
+    记成目标达成；结算放在本次结尾的话，它会先撞上预算上限记成预算耗尽。两者的轨迹长度
+    一模一样，事后分不出来。
+    """
+    assert budget_admission(RunCounters(steps_appended=9), BUDGET) is None
+    assert budget_admission(RunCounters(steps_appended=10), BUDGET) is StopReason.STEP_BUDGET
+
+
+# ---------------------------------------------------------------------------
+# B 档：提示词规模
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("chars", "expected"),
+    [(0, None), (999, None), (1000, None), (1001, StopReason.CONTEXT_OVERFLOW)],
+)
+def test_the_prompt_size_limit_itself_is_allowed(chars: int, expected: object) -> None:
+    """**超过**上限才拦，正好等于上限是允许的。
+
+    和 A 档那个「达到即拦」不是一回事：那里数的是已经用掉几个额度，这里量的是一个东西有
+    多大。
+    """
+    assert prompt_size_admission(chars, BUDGET) is expected
+
+
+# ---------------------------------------------------------------------------
+# D 档：连续解析失败
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("failures", "expected"),
+    [(0, None), (2, None), (3, StopReason.PARSE_FAILED_REPEATEDLY)],
+)
+def test_consecutive_parse_failures_stop_the_run_at_the_limit(
+    failures: int, expected: object
+) -> None:
+    assert (
+        parse_failure_admission(RunCounters(consecutive_parse_failures=failures), BUDGET)
+        is expected
+    )
+
+
+# ---------------------------------------------------------------------------
+# G 档：完成判定
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("tool_completes_run", [True, False])
+@pytest.mark.parametrize("env_reported_completion", [True, False])
+def test_an_env_error_stops_the_run_whatever_else_is_true(
+    env_reported_completion: bool, tool_completes_run: bool
+) -> None:
+    """环境故障排在完成判定前面，其余两个信号都盖不过它。
+
+    环境坏了就没有下一步可走，继续跑只会产出一串同样的故障，把预算烧光而轨迹上全是噪声。
+    """
+    outcome = _outcome(ActionStatus.ENV_ERROR, env_reported_completion=env_reported_completion)
+
+    assert completion_verdict(outcome, tool_completes_run) is StopReason.ENV_ERROR
+
+
+@pytest.mark.parametrize("tool_completes_run", [True, False])
+@pytest.mark.parametrize("env_reported_completion", [True, False])
+def test_a_rejected_action_never_completes_the_run(
+    env_reported_completion: bool, tool_completes_run: bool
+) -> None:
+    """状态是「未执行」时不做完成判定。
+
+    动作根本没进入真实执行，环境状态没变，完成条件不可能因为它成立。这一档回到预算准入
+    接着跑。
+    """
+    outcome = _outcome(ActionStatus.NOT_EXECUTED, env_reported_completion=env_reported_completion)
+
+    assert completion_verdict(outcome, tool_completes_run) is None
+
+
+@pytest.mark.parametrize(
+    ("env_reported_completion", "tool_completes_run", "expected"),
+    [
+        (False, False, None),
+        (True, False, StopReason.TASK_COMPLETED),
+        (False, True, StopReason.TASK_COMPLETED),
+        (True, True, StopReason.TASK_COMPLETED),
+    ],
+)
+def test_either_completion_channel_ends_an_executed_action(
+    env_reported_completion: bool, tool_completes_run: bool, expected: object
+) -> None:
+    """两条完成通路任一成立都收尾，可信度的差别不体现在停止原因上。
+
+    一条是环境状态里真的留下了记录，一条是 agent 调了一个被标为完成标记的工具、环境状态
+    一点没变。它们共用一个停止原因，因为「这次运行为什么停」的答案是同一个；要区分是哪一种
+    看那一步的步记录。
+    """
+    outcome = _outcome(ActionStatus.EXECUTED, env_reported_completion=env_reported_completion)
+
+    assert completion_verdict(outcome, tool_completes_run) is expected
+
+
+def test_a_completion_signal_that_is_always_false_is_not_a_failure() -> None:
+    """没有环境完成信号的环境就是这么返回的，走「接着跑」那一支。
+
+    初稿在这里写错过：那时完成信号被定成「布尔或空、空表示取不到」，这一档写着「取不到就是
+    环境故障」——而有一个下游每一步都返回空，于是它的每一次运行都会在第一步撞环境故障终止。
+    不是边缘情况，是全部。
+    """
+    outcome = _outcome(ActionStatus.EXECUTED, env_reported_completion=False)
+
+    assert completion_verdict(outcome, tool_completes_run=False) is None