架构定位

Agent 核心运行环

围绕目标和状态选择动作、执行工具、观察并验证结果，再决定停止或重新规划。

系统如何选择动作、执行工具、观察结果并决定停止或重新规划？

受这些层约束

概念校准

Agent

它是什么: 围绕目标和状态，在约束下选择动作、执行、观察并验证结果的运行系统。
它不是什么: 不是独立人格，也不是把模型放进无限循环。

工具

它是什么: Agent 可通过明确参数、权限和结果合同调用的受约束能力。
它不是什么: 工具不等于 MCP；本地函数、命令或 HTTP 客户端也可以是工具。

验证与评测

它是什么: 用测试、Schema、引用、状态和验收规则检查单次结果，并用数据集衡量系统表现。
它不是什么: 不等于让模型对自己的输出再给一次意见；验证必须落到外部可检查证据。

可观测性

它是什么: 通过结构化日志、Trace、指标和版本信息解释系统在一次运行中发生了什么。
它不是什么: 不只是保存终端输出；没有关联 ID、状态转换和敏感信息边界的日志难以诊断。

单 Agent 与工具调用循环

实现工具选择、结构化参数、流式输出、重试、幂等和停止条件。

本章只做一件事让 Forge 完成第一条可观测的单 Agent 交付循环。

开始前，Forge 已经具备已完成 Module 06 的“Skill、Prompt 与 Agent 能力包”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够实现第一个可运行的单 Agent 交付循环。

卡住时的最小恢复点限制为两个只读工具和最多三轮循环。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01根据目标与工具能力选择唯一动作，并在歧义时请求澄清02把工具结果记录为有序观察，并区分可继续错误与终止状态03在调用工具前验证参数合同，并把返回值包装为明确的执行结果04仅对明确的瞬时故障进行有上限的重试，并保留尝试次数05用幂等键与请求指纹防止重复副作用，同时拒绝键冲突06根据可验证成功、失败、预算和重复动作做出显式停止决定07将动作、执行、观察和停止控制组合成有界且可审计的单 Agent 循环

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### Agent agent\n\n- 它是什么: 围绕目标和当前状态，在约束下选择动作、执行、观察并验证结果的运行系统。核心运行环是：目标/状态 → 选择动作 → 执行工具 → 观察结果 → 验证 → 停止或重新规划。\n- 它不是什么: 它不是独立人格，也不是让模型无期限循环。模型只是决策组件，动作、权限、状态和停止条件由外部运行环境共同约束。\n- 与相邻概念的关系: Context 提供可见信息，Tool 提供动作，Harness 控制权限和副作用，Verification 判断结果，Planning 只在复杂任务需要时加入。\n- 在 HeatStack Forge 中的位置: Forge 读取任务合同与状态，选择一个受许可动作，记录工具结果，再用测试、Schema 或验收规则决定成功、失败、预算耗尽、重复动作、人工介入或重新规划。\n- 典型误用与修正: 误用是把“模型继续生成”当成进展。修正方法是让每轮产生可观察的状态变化，并为所有停止原因建立确定性检查。\n\n### 工具 tool\n\n- 它是什么: Agent 可通过明确参数、权限、执行边界和结果合同调用的受约束能力，例如函数、命令、数据库查询或 HTTP 客户端。\n- 它不是什么: 工具不等于 MCP，也不等于任意脚本。协议可以暴露能力，但 Harness 仍决定是否允许调用及如何处理副作用。\n- 与相邻概念的关系: Agent Loop 选择工具，Schema 约束参数，Harness 执行权限检查和隔离，Verification 检查结果是否推进目标。\n- 在 HeatStack Forge 中的位置: Forge 的 tool registry 保存合同与风险等级；执行前生成变更计划，高风险写操作等待确认，执行后记录结果与恢复信息。\n- 典型误用与修正: 误用是让模型根据工具名称自由拼参数。修正方法是严格 Schema、确定性校验、超时、幂等键和拒绝分支测试。\n\n### 验证与评测 evaluation\n\n- 它是什么: Verification 用测试、Schema、引用、状态或验收规则检查一次结果；Evaluation 用数据集和指标衡量多次运行表现。\n- 它不是什么: 它不是让模型对自己的回答再给一次意见；通过与否必须落到外部可检查证据。\n- 与相邻概念的关系: Agent Loop 依赖 Verification 决定停止，RAG 需要分层评测，Planning 需要节点验收，Observability 提供运行记录。\n- 在 HeatStack Forge 中的位置: Forge 为单次运行保存合同测试与 evidence，再用固定场景集比较质量、延迟、成本、权限和恢复能力。\n- 典型误用与修正: 误用是把一个综合分数当成全部结论。修正方法是定义分层指标、失败样本和版本基线，并结合产品风险评审。\n\n### 可观测性 observability\n\n- 它是什么: 通过结构化日志、Trace、指标、状态转换和版本信息解释一次运行发生了什么，并支持定位失败与比较版本。\n- 它不是什么: 它不只是终端输出，也不是无限记录所有输入；缺少关联 ID、状态转换或敏感信息边界的日志既难排查又可能泄露数据。\n- 与相邻概念的关系: Harness 产生运行事件，Agent Loop 与工具调用形成 Trace，Evaluation 聚合指标，Memory 与恢复依赖版本记录。\n- 在 HeatStack Forge 中的位置: Forge 为每次任务生成 operation_id，记录阶段、工具、耗时、成本、错误类别、状态变化和证据路径，同时过滤敏感信息。\n- 典型误用与修正: 误用是等失败后再临时加日志。修正方法是在合同设计时定义事件与关联字段，并用故障注入确认能重建时间线。\n\n

STAGE 01

根据目标与工具能力选择唯一动作，并在歧义时请求澄清

先看一个具体问题

当两个工具能力得分并列时，选择器不能默认取第一个

在单 Agent 循环的第一步，系统需要根据目标所需的能力集合从已注册工具列表中选择唯一一个工具来执行。当前 forge/agent/actions.py 文件尚不存在，测试在导入阶段就因 ModuleNotFoundError: No module named 'forge.agent' 而中断收集。

本阶段的核心决策不仅是创建该文件，更关键的是在实现 select_action 时正确处理能力得分并列的情况。如果两个工具匹配所需能力的数量完全相同，排序后取第一个并不能证明该选择是明确的，此时必须抛出 ValueError("ambiguous tool choice requires clarification") 以请求外部澄清。

故障实验中提供的 faultySource 已经实现了目标检查、能力排序和无匹配检查，但故意移除了并列检查分支。这导致 test_ambiguous_tools_require_clarification 测试失败，因为选择器静默地返回了排序后的第一个工具而非抛出异常。

先做判断

当 select_action 对工具列表按能力匹配数降序排序后，发现 ranked[0][0] == ranked[1][0]（即前两个工具得分相同），此时正确的做法是什么？

直接返回 ranked[0] 对应的工具，因为排序已经决定了优先级
抛出 ValueError 并在消息中包含 “ambiguous”，表示选择不明确需要澄清
返回一个特殊的 Action 对象标记为待定状态
重新排序并按工具名字母顺序选择第一个

判断依据: 排序只能给出一个序列，但当最高分有多个工具并列时，排序本身无法区分哪个工具才是正确选择。正确做法是在返回 Action 之前检查并列并抛出 ValueError，让外部循环知道需要澄清。

它是什么

动作选择是能力匹配与唯一性验证的组合

动作选择函数 select_action 接收三个输入：目标字符串、所需能力集合 required_capabilities、以及工具规格列表 tools。它通过计算每个工具的 capabilities 与 required_capabilities 的交集大小来排序，选出匹配数最高的工具。

唯一性验证是动作选择不可跳过的步骤：即使排序产生了第一名，如果存在并列，选择器就不能断言该选择是确定的。并列检查 ranked[0][0] == ranked[1][0] 必须在返回 Action 之前执行，否则选择器会将排序顺序误认为选择依据。

当没有任何工具匹配任何所需能力时，ranked[0][0] == 0，函数应抛出包含 “no tool” 的 ValueError，表示当前工具列表无法满足目标需求。

它不是什么

动作选择不执行工具也不决定停止

动作选择只负责从候选工具中选出一个并返回 Action 对象，它不调用工具函数、不管理 Agent 状态、不决定循环是否终止。这些职责属于 Harness 和 Agent Loop 的后续阶段。

动作选择也不是简单的排序取第一个。排序只是中间步骤，排序结果必须经过并列检查和无匹配检查两道验证后才能转化为 Action 返回值。

动作选择不处理工具的权限、副作用或隔离。这些由 Harness 在实际执行 Action 时负责，select_action 只依据能力匹配做决策。

它与相邻概念的关系

动作选择在 Agent 循环中的位置

在 Agent 核心循环（目标/状态 → 选择动作 → 执行工具 → 观察结果 → 验证 → 停止或重新规划）中，select_action 对应“选择动作”这一步。它的输出 Action(tool_name, reason) 会被传递给执行层来调用对应工具。

ToolSpec 定义了工具的静态能力描述（名称和能力集合），是 select_action 的输入。Action 是选择结果的数据载体，包含 tool_name 和 reason，是执行层的输入。两者都是 frozen=True 的 dataclass，保证不可变。

当 select_action 抛出 ValueError 时，Agent Loop 应捕获异常并决定是请求用户澄清（ambiguous 情况）还是报告能力缺失（no tool 情况），而不是继续执行。

本阶段的边界决定: 动作选择只负责根据能力交集大小排序并验证唯一性，在并列时抛出 ValueError 而非猜测；它不执行工具、不管理状态、不决定停止条件，这些职责由外部 Harness 和 Agent Loop 承担。

实验 ID: single-agent-loop-01-action-selection
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/01-action-selection/
验收: 待验收

教师示范

目标: 在软件场景中，给定目标 “update the file”、所需能力 {read, write} 和两个工具 ToolSpec(“reader”, {read}) 与 ToolSpec(“editor”, {read, write})，演示 select_action 返回 Action(tool_name=“editor”)，因为 editor 匹配全部两个能力。

首先创建 forge/agent/__init__.py 和 forge/agent/actions.py 两个文件。__init__.py 负责导出公共 API，actions.py 定义 ToolSpec、Action 和 select_action。

在 select_action 中，先检查目标非空，然后对每个工具计算 len(required_capabilities & set(tool.capabilities)) 得到匹配数，按降序排序。对于 reader，匹配数为 1（只有 read）；对于 editor，匹配数为 2（read 和 write）。排序后 ranked = [(2, "editor"), (1, "reader")]。

接下来验证：ranked[0][0] 为 2，不等于 0，所以不触发无匹配错误。然后检查并列：ranked 长度为 2，但 ranked[0][0]（2）不等于 ranked[1][0]（1），所以不触发并列错误。最终返回 Action(tool_name="editor", reason="matches 2 required capabilities")。

实施步骤

创建 forge/agent/__init__.py，写入 from .actions import Action, ToolSpec, select_action 和 __all__ = ["Action", "ToolSpec", "select_action"]
创建 forge/agent/actions.py，定义 ToolSpec 和 Action 两个 frozen dataclass，并实现 select_action 函数
在 select_action 中实现目标检查、能力交集排序、无匹配检查和并列检查四个步骤，确保在无并列时返回 Action

真实 solution 代码: forge/agent/__init__.py

from .actions import Action, ToolSpec, select_action

__all__ = ["Action", "ToolSpec", "select_action"]

真实 solution 代码: forge/agent/actions.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class ToolSpec:
    name: str
    capabilities: frozenset[str]

@dataclass(frozen=True)
class Action:
    tool_name: str
    reason: str

def select_action(goal: str, required_capabilities: set[str], tools: list[ToolSpec]) -> Action:
    if not goal.strip():
        raise ValueError("goal is required")
    ranked = sorted(((len(required_capabilities & set(tool.capabilities)), tool.name) for tool in tools), reverse=True)
    if not ranked or ranked[0][0] == 0:
        raise ValueError("no tool satisfies the required capabilities")
    if len(ranked) > 1 and ranked[0][0] == ranked[1][0]:
        raise ValueError("ambiguous tool choice requires clarification")
    return Action(tool_name=ranked[0][1], reason=f"matches {ranked[0][0]} required capabilities")

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

在这个例子中，editor 的能力集合 {read, write} 完全覆盖了所需能力 {read, write}，交集大小为 2，是所有工具中最大的。reader 只匹配 read 一个能力，交集大小为 1。两者得分不同，不存在并列，所以选择是明确的。

排序使用 reverse=True 确保最高分排在前面。sorted 在 Python 中是稳定排序，但当两个工具得分相同时，稳定性不能作为选择依据——这正是并列检查必须存在的原因。

带提示练习

解释当两个工具 ToolSpec(“a”, {read}) 和 ToolSpec(“b”, {read}) 能力得分并列时，select_action 为何必须抛出 ValueError(“ambiguous”)，并补全并列检查分支。

在 select_action 函数中，找到无匹配检查 if not ranked or ranked[0][0] == 0 之后、return Action(...) 之前的位置
添加并列检查条件：当 ranked 列表长度大于 1 且 ranked[0][0] 等于 ranked[1][0] 时，抛出 ValueError("ambiguous tool choice requires clarification")
运行 python -m pytest -q tests/test_stage.py::test_ambiguous_tools_require_clarification 验证测试通过

逐级提示

当两个工具的匹配数相同时，ranked[0][0] 和 ranked[1][0] 的值会相等，此时无法仅凭排序确定唯一工具
并列检查的条件需要同时满足两个前提：列表中有至少两个工具，且前两个工具的得分相同
抛出的异常消息中必须包含 “ambiguous” 子串，因为测试用 pytest.raises(ValueError, match="ambiguous") 来匹配

完成标准

test_ambiguous_tools_require_clarification 测试通过，即当两个工具得分并列时 select_action 抛出 ValueError 且消息匹配 “ambiguous”
并列检查代码位于无匹配检查之后、return Action(...) 之前，保证逻辑顺序正确

故障注入与排查

故障实验使用 fixtures/failure.json 中定义的 ambiguous-action 故障。faultySource 中的 select_action 已实现目标检查、能力排序和无匹配检查，但故意移除了并列检查分支。当运行 test_ambiguous_tools_require_clarification 测试时，两个工具 ToolSpec("a", frozenset({"read"})) 和 ToolSpec("b", frozenset({"read"})) 得分均为 1，选择器静默返回了排序后的第一个工具而非抛出异常。

本阶段专门教授动作选择的唯一性验证机制。通过分析 faultySource 中缺失的并列检查，学习者能理解排序顺序不等于选择依据。修复方法是在 select_action 的 return Action(...) 之前添加 if len(ranked) > 1 and ranked[0][0] == ranked[1][0]: raise ValueError("ambiguous tool choice requires clarification")，恢复被移除的并列守卫。

已验证的故障合同

目标测试: test_ambiguous_tools_require_clarification
故障文件: forge/agent/actions.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_ambiguous_tools_require_clarification，观察输出 Failed: DID NOT RAISE ValueError，说明选择器在并列情况下未抛出异常
检查 faultySource 中 select_action 的实现，确认在无匹配检查 if not ranked or ranked[0][0] == 0 之后直接 return Action(...)，中间缺少并列检查
对比 codeChanges 中 after 的完整实现，发现缺失的行是 if len(ranked) > 1 and ranked[0][0] == ranked[1][0]: raise ValueError("ambiguous tool choice requires clarification")

修复原则: 在 forge/agent/actions.py 的 select_action 函数中，在无匹配检查之后、return Action(...) 之前，添加并列检查：if len(ranked) > 1 and ranked[0][0] == ranked[1][0]: raise ValueError("ambiguous tool choice requires clarification")。这确保当两个工具得分相同时，选择器不会静默选择排序后的第一个工具，而是抛出异常请求外部澄清。

python -m pytest -q ../tests/test_stage.py::test_ambiguous_tools_require_clarification

迁移挑战

办公研究场景：目标 “search the spreadsheet”，所需能力 {search}，工具列表为 ToolSpec(“finder”, frozenset({“search”})) 和 ToolSpec(“reader”, frozenset({“read”}))。

改变的条件: 将场景数据从软件交付切换为办公研究，目标变为搜索电子表格，所需能力变为 {search}，工具列表替换为 finder 和 reader 两个不同能力的工具

在 tests/test_stage.py 的一个可丢弃副本中，将 test_selects_the_only_best_matching_tool 的输入数据替换为：目标 “search the spreadsheet”，所需能力 {“search”}，工具列表 [ToolSpec(“finder”, frozenset({“search”})), ToolSpec(“reader”, frozenset({“read”}))]，断言 action.tool_name == "finder"。不修改 forge/agent/actions.py 中的任何代码，只替换测试中的数据值。

完成标准

测试输入中目标为 “search the spreadsheet”，所需能力集合为 {“search”}，工具列表包含 finder 和 reader 两个工具且各自能力集合不变
select_action 返回 Action(tool_name="finder")，因为 finder 匹配 1 个所需能力而 reader 匹配 0 个，不存在并列

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么在 select_action 中排序后取第一个不能保证选择是明确的？并列检查应该在返回 Action 之前还是之后执行，如果放在之后会发生什么？

一个有力回答应包含

排序只按匹配数排列工具，当两个工具匹配数相同时排序无法区分哪个更优，因此取第一个不等于选择是明确的
并列检查必须在返回 Action 之前执行，因为一旦返回了 Action 对象，调用方就会认为选择已确定并可能直接执行该工具
如果并列检查放在返回 Action 之后，则永远不会触发，因为函数已在并列检查之前返回，导致歧义工具被静默选择

掌握证据

select_action 在最佳匹配场景中返回 Action(tool_name="editor")，因为 editor 的能力集合覆盖了 {read, write} 全部所需能力且无并列
select_action 在两个工具得分并列时抛出 ValueError 且消息匹配 “ambiguous”，对应 test_ambiguous_tools_require_clarification 测试通过
select_action 在无工具满足所需能力时抛出 ValueError 且消息匹配 “no tool”，对应 test_missing_capability_is_rejected 测试通过
python -m pytest -q tests/test_stage.py 退出码为 0 且输出 3 passed，证明三个测试全部通过

STAGE 02

把工具结果记录为有序观察，并区分可继续错误与终止状态

先看一个具体问题

工具结果记录与可恢复错误的混淆

在单 Agent 循环中，工具执行后的结果必须被记录为有序的观察，以便后续的动作选择和验证环节能够回溯完整的运行轨迹。当前 forge/agent/observations.py 文件不存在，测试在收集阶段就因 ModuleNotFoundError: No module named 'forge.agent.observations' 而中断，导致整个测试套件无法运行。

更关键的是，工具调用的结果状态分为 ok、retryable_error 和 fatal_error 三种，其中可恢复的瞬时错误（如 HTTP 503）不应导致循环终止，而致命错误（如权限拒绝）才应设置终止标志。如果将瞬时传输错误与循环停止状态混为一谈，Agent 将在遇到任何暂时性故障时永久停摆，无法进行重试或重新规划。

先做判断

当 LoopState.record 方法接收到 status="retryable_error" 时，state.terminal 应该被设置为什么值？

True，因为发生了错误就应该停止
False，因为瞬时错误可被重试，不应终止循环
None，因为状态未定义
抛出异常，因为错误无法处理

判断依据: 正确答案是 False。retryable_error 表示可恢复的瞬时故障（如网络超时或 503），Agent 循环应有机会重试或选择替代动作。只有 fatal_error 才代表不可恢复的终止条件，应将 terminal 设为 True。

它是什么

LoopState 的职责边界

LoopState 是一个数据类，负责按调用顺序记录每次工具执行产生的 Observation，并通过 terminal 布尔标志管理循环是否可以继续接受新观察。Observation 是不可变的冻结数据类，包含 sequence（从 1 开始递增的序号）、tool_name、status 和 payload，确保每条观察一旦创建就不会被篡改。

record 方法是 LoopState 的核心入口：它先检查 terminal 标志，若已终止则抛出 ValueError；然后校验 status 是否在合法集合中；接着用 len(self.observations) + 1 计算新序号并创建 Observation；最后根据 status 决定是否将 terminal 设为 True。这一流程将状态管理与动作选择解耦——LoopState 不决定下一步做什么，只记录发生了什么以及是否还能继续。

它不是什么

LoopState 不做的事

LoopState 不执行工具调用、不选择下一个动作、不进行重试逻辑，也不对 payload 的内容做语义解释。它只是一个被动的状态容器和事件序列记录器，由外层的 Agent 循环驱动其 record 方法。

LoopState 也不是日志系统或可观测性平台——它不记录时间戳、调用链 ID 或指标，只维护有序观察列表和终止标志。如果需要 Trace 或指标聚合，那是 Observability 模块的职责，而非 LoopState 的功能。

它与相邻概念的关系

与 Agent 循环其他组件的关系

LoopState 与 Stage 01 中的 select_action 和 ToolSpec 形成上下游关系：select_action 基于 LoopState 中的观察历史决定下一个动作，工具执行后结果通过 record 回写到 LoopState，形成闭环。terminal 标志直接影响循环的退出条件——当它为 True 时，外层循环应停止调用 record 并结束运行。

Observation 的 sequence 字段为外层的验证与评测提供了确定性的顺序依据，使得测试可以精确断言第 N 次工具调用的结果，也使得 Trace 回放成为可能。status 字段的三值分类（ok/retryable_error/fatal_error）将错误严重性显式化，让循环控制逻辑可以据此区分重试与终止。

本阶段的边界决定: LoopState 只记录观察和管理终止标志，不选择动作、不执行工具、不做重试；retryable_error 保持 terminal=False，fatal_error 设置 terminal=True，terminal=True 后 record 抛出 ValueError。

实验 ID: single-agent-loop-02-observation-state
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/02-observation-state/
验收: 待验收

教师示范

目标: 在软件场景中，创建 LoopState()，依次调用 record("read", "ok", {"text": "main.py"}) 和 record("search", "retryable_error", {"code": 503})，验证序号分别为 1 和 2 且 terminal 为 False。

假设 Agent 正在为前端仓库添加功能，第一步用 read 工具读取 main.py 文件内容，返回 ok 状态；第二步用 search 工具搜索代码库，但因后端服务暂时返回 503 而产生 retryable_error。这两次调用都应被记录为有序观察，且 503 不应终止循环。

我们需要验证 LoopState 正确地为第一次观察分配序号 1，为第二次分配序号 2，并且在遇到 retryable_error 后 terminal 仍为 False，使得 Agent 有机会在下一轮循环中重试搜索或选择替代工具。

实施步骤

创建 forge/agent/observations.py 文件，定义 Observation 冻结数据类，包含 sequence、tool_name、status、payload 四个字段
定义 LoopState 数据类，包含 observations 列表（默认空）和 terminal 布尔值（默认 False）
实现 record 方法：检查 terminal、校验 status、用 len(self.observations) + 1 生成序号、追加观察、仅在 status == "fatal_error" 时设 terminal = True

真实 solution 代码: forge/agent/observations.py

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

@dataclass(frozen=True)
class Observation:
    sequence: int
    tool_name: str
    status: str
    payload: Any

@dataclass
class LoopState:
    observations: list[Observation] = field(default_factory=list)
    terminal: bool = False

    def record(self, tool_name: str, status: str, payload: Any) -> Observation:
        if self.terminal:
            raise ValueError("terminal loop cannot accept observations")
        if status not in {"ok", "retryable_error", "fatal_error"}:
            raise ValueError("unknown observation status")
        observation = Observation(len(self.observations) + 1, tool_name, status, payload)
        self.observations.append(observation)
        if status == "fatal_error":
            self.terminal = True
        return observation

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

序号生成使用 len(self.observations) + 1 而非单独的计数器字段，这保证了序号始终与观察列表的长度一致，且在追加观察后立即正确递增。第一次调用时列表长度为 0，序号为 1；第二次调用时列表长度为 1，序号为 2。

retryable_error 不触发 terminal = True，因为该方法中只有 status == "fatal_error" 才设置终止标志。这使得 Agent 循环在遇到瞬时网络故障后仍能继续运行，符合可恢复错误的设计语义。

带提示练习

解释为什么 record("write", "fatal_error", {"code": "denied"}) 后再调用 record 会抛出 ValueError，并验证 fatal_error 设置 terminal=True 的分支逻辑。

在已创建的 LoopState 上调用 state.record("write", "fatal_error", {"code": "denied"})，观察返回的 Observation 的 status 字段
检查 state.terminal 的值是否已变为 True
再次调用 state.record("read", "ok", {})，确认抛出 ValueError 且异常消息匹配 "terminal"

逐级提示

查看 record 方法末尾的条件分支：if status == "fatal_error": 做了什么操作？
当 terminal 为 True 时，record 方法的第一行代码 if self.terminal: 会执行什么？
测试 test_fatal_error_is_terminal 使用 pytest.raises(ValueError, match="terminal") 来断言异常，确认你的错误消息包含字符串 "terminal"

完成标准

state.record("write", "fatal_error", {"code": "denied"}) 后 state.terminal 为 True
在 terminal=True 后调用 record 抛出 ValueError 且消息包含 "terminal"

故障注入与排查

故障实验室中的 forge/agent/observations.py 存在一个注入缺陷：record 方法在处理 retryable_error 时错误地将其与 fatal_error 一起设置了 terminal = True。运行 test_retryable_error_does_not_end_the_loop 测试时，断言 state.terminal is False 失败，因为实际值为 True。这模拟了一个常见的工程错误：将传输层故障的严重性与循环停止状态混为一谈。

本阶段已经建立了 LoopState 的完整心智模型和 record 方法的实现经验，学习者能够区分 retryable_error（可恢复，不终止）和 fatal_error（不可恢复，终止）的语义差异。通过检查 record 方法中 if status in {"fatal_error", "retryable_error"}: 这行代码，学习者可以定位到缺陷根源——条件集合不应包含 retryable_error，修复方式是将条件改为 if status == "fatal_error":。

已验证的故障合同

目标测试: test_retryable_error_does_not_end_the_loop
故障文件: forge/agent/observations.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_retryable_error_does_not_end_the_loop，观察失败信息 AssertionError: assert True is False，确认 state.terminal 在 retryable_error 后被错误地设为 True
打开 forge/agent/observations.py，定位到 record 方法中设置 terminal 的条件分支 if status in {"fatal_error", "retryable_error"}:
分析该条件：retryable_error 被包含在终止集合中，导致瞬时错误也触发了终止，这与可恢复错误应保持循环继续的设计意图矛盾

修复原则: 将 record 方法中的条件从 if status in {"fatal_error", "retryable_error":} 修改为 if status == "fatal_error":，确保只有致命错误才设置 terminal = True。修复后 retryable_error 仅被记录为观察，循环可以继续接受后续工具调用结果。

python -m pytest -q ../tests/test_stage.py::test_retryable_error_does_not_end_the_loop

迁移挑战

音乐创作场景：Agent 正在整理曲目发行计划，先用 lookup 工具查询曲目信息，再用 publish 工具发布但遇到瞬时 503 错误。

改变的条件: 将工具名和 payload 数据从软件场景替换为音乐场景的工具调用数据，保持两次 record 调用的 status 序列为 ok 后跟 retryable_error

创建新的 LoopState()，调用 state.record("lookup", "ok", {"track": "song-a"}) 和 state.record("publish", "retryable_error", {"code": 503})，验证第一条观察的 sequence 为 1，第二条为 2，且 state.terminal 为 False。不修改 LoopState 的 API 或测试固定断言，仅替换场景数据。

完成标准

第一次 record 返回的 Observation 的 sequence 等于 1，tool_name 等于 “lookup”，status 等于 “ok”
第二次 record 返回的 Observation 的 sequence 等于 2，且 state.terminal 为 False，state.observations 长度为 2

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么 retryable_error 不应该设置 terminal=True，而 fatal_error 必须设置？请从 Agent 循环的重试能力和不可恢复错误的语义边界两个角度解释这一设计决策的因果权衡。

一个有力回答应包含

retryable_error 表示瞬时故障（如 503），Agent 可以在下一轮循环中重试同一工具或选择替代动作，若设置 terminal=True 会剥夺重试机会
fatal_error 表示不可恢复条件（如权限拒绝），继续循环无意义且可能产生副作用，必须设置 terminal=True 阻止后续 record
terminal=True 后 record 抛出 ValueError 是一种防御机制，防止在已终止的循环中继续写入不一致的观察

掌握证据

LoopState.record 返回的 Observation 具有从 1 开始连续递增的 sequence 序号，验证了 test_observations_receive_contiguous_sequence_numbers
retryable_error 不设置 terminal=True，验证了 test_retryable_error_does_not_end_the_loop 中 state.terminal is False 的断言
fatal_error 设置 terminal=True 且后续 record 抛出 ValueError 匹配 "terminal"，验证了 test_fatal_error_is_terminal
python -m pytest -q tests/test_stage.py 退出码为 0，3 个测试全部通过

STAGE 03

在调用工具前验证参数合同，并把返回值包装为明确的执行结果

先看一个具体问题

未声明参数必须在调用 handler 之前被拒绝

在单 Agent 循环中，execute_tool 负责在调用具体 handler 之前验证参数合同。当前 forge/agent/execution.py 尚不存在，测试文件 tests/test_stage.py 在导入时直接抛出 ModuleNotFoundError，导致三个测试全部无法收集。

驱动问题是：为什么必须在调用 handler 之前拒绝未声明的参数？如果多余参数被直接透传给 handler，handler 可能因为意外的关键字参数崩溃，或者在更危险的情况下执行了未授权的副作用（例如传入 sudo=True）。

本阶段的任务是从零创建 forge/agent/execution.py，定义 ToolContract 和 ExecutionResult 数据类，并实现 execute_tool 函数，使其在参数缺失或包含未声明字段时抛出 ValueError，在参数完全合法时返回包装好的 ExecutionResult。

先做判断

当 execute_tool 接收到参数 {"path": "a.txt", "sudo": True}，但 ToolContract 只声明了 required=frozenset({"path"}) 且没有 optional 集合时，如果代码只检查了 missing 参数就直接调用 handler(**arguments)，会发生什么？

handler 正常执行并忽略多余参数
handler 抛出 TypeError 因为收到了意外的关键字参数 sudo
execute_tool 抛出 ValueError 并提示 unexpected
系统自动过滤掉 sudo 参数并继续执行

判断依据: 正确答案是 handler 抛出 TypeError。因为 Python 的 **arguments 解包会把所有键作为关键字参数传入，handler 签名不接受 sudo 就会崩溃。这证明了必须在调用前主动检查 unexpected 参数，而不是依赖 handler 自己防御。

它是什么

参数合同验证边界

execute_tool 是 Agent Loop 与具体工具实现之间的隔离层。它的唯一职责是：根据 ToolContract 声明的 required 和 optional 参数集合，验证传入的 arguments 字典是否严格匹配，匹配则调用 handler 并包装结果，不匹配则抛出 ValueError。

集合运算是验证的核心：missing = set(contract.required) - set(arguments) 找出缺失的必填参数；unexpected = set(arguments) - set(contract.required) - set(contract.optional) 找出既不在必填也不在可选集合中的多余参数。这两个检查必须在调用 handler 之前完成。

它不是什么

execute_tool 的职责边界

execute_tool 不做重试逻辑，不管理 LoopState，不做幂等性检查，也不关心 handler 内部的业务逻辑。它只是一个严格的结构化网关。

它也不是一个通用的参数过滤器——不会自动剥离 unexpected 参数然后调用 handler，因为静默丢弃参数会掩盖调用方的错误意图，可能导致未授权能力被执行。

它与相邻概念的关系

与 Agent Loop 和 Verification 的关系

Agent Loop 通过 select_action 决定调用哪个工具后，把控制权交给 execute_tool。execute_tool 保证只有符合合同约定的参数才能到达 handler，从而为后续的 Verification 提供确定性基础。

ExecutionResult 将 handler 的返回值包装为带有 tool_name 的结构化结果，使 Observation 层可以统一记录和追踪工具调用的输出，而不需要关心原始返回值的具体类型。

本阶段的边界决定: execute_tool 只在参数合同验证通过后才调用 handler 并返回 ExecutionResult；任何 missing 或 unexpected 参数都必须在 handler 被调用之前以 ValueError 终止，绝不静默过滤或透传。

实验 ID: single-agent-loop-03-structured-tool-execution
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/03-structured-tool-execution/
验收: 待验收

教师示范

目标: 在软件场景中，用 execute_tool(ToolContract("sum", frozenset({"a", "b"})), lambda a, b: a + b, {"a": 2, "b": 3}) 验证返回 ExecutionResult(tool_name="sum", output=5)。

这个示例展示了 execute_tool 的正常路径：参数完全匹配合同声明，验证通过后 handler 被调用，返回值被包装为 ExecutionResult。

关键在于理解集合运算如何判定参数合法性：missing 集合为空且 unexpected 集合为空时，参数合同才算通过。

实施步骤

定义 ToolContract 为 frozen dataclass，包含 name、required 和默认为 frozenset() 的 optional。
定义 ExecutionResult 为 frozen dataclass，包含 tool_name 和 output。
在 execute_tool 中计算 missing 和 unexpected 集合，两者都为空时调用 handler(**arguments) 并返回 ExecutionResult。

真实 solution 代码: forge/agent/execution.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Callable

@dataclass(frozen=True)
class ToolContract:
    name: str
    required: frozenset[str]
    optional: frozenset[str] = frozenset()

@dataclass(frozen=True)
class ExecutionResult:
    tool_name: str
    output: Any

def execute_tool(contract: ToolContract, handler: Callable[..., Any], arguments: dict[str, Any]) -> ExecutionResult:
    missing = set(contract.required) - set(arguments)
    unexpected = set(arguments) - set(contract.required) - set(contract.optional)
    if missing:
        raise ValueError("missing arguments: " + ", ".join(sorted(missing)))
    if unexpected:
        raise ValueError("unexpected arguments: " + ", ".join(sorted(unexpected)))
    return ExecutionResult(contract.name, handler(**arguments))

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

当 arguments = {"a": 2, "b": 3} 且 required = frozenset({"a", "b"}) 时，missing = set() 且 unexpected = set()，两个 if 分支都不触发，handler 被正常调用并返回 5。

ExecutionResult(contract.name, handler(**arguments)) 将 handler 的原始返回值包装为结构化结果，使上层调用者可以通过 result.tool_name 知道是哪个工具产生了这个输出。

带提示练习

解释为什么 execute_tool(ToolContract("read", frozenset({"path"})), handler, {"path": "a.txt", "sudo": True}) 必须在调用 handler 之前抛出 ValueError("unexpected")，并补全 unexpected 参数检查分支。

查看 faulty 源码中 execute_tool 的实现，注意它计算了 unexpected 集合但没有用它做任何判断。
在 if missing: 分支之后、return ExecutionResult(...) 之前，添加 if unexpected: 分支。
在 unexpected 分支中抛出 ValueError("unexpected arguments: " + ", ".join(sorted(unexpected)))，确保错误消息包含字符串 unexpected。

逐级提示

faulty 源码已经计算了 unexpected = set(arguments) - set(contract.required) - set(contract.optional)，但没有对它做任何检查。想想如果 unexpected 集合非空意味着什么。
测试用 pytest.raises(ValueError, match="unexpected") 来断言异常，所以你的错误消息中必须包含 unexpected 这个子串。
注意检查的顺序：先检查 missing，再检查 unexpected，最后才调用 handler。如果 unexpected 检查在 handler 调用之后，handler 可能已经产生了副作用。

完成标准

test_unexpected_arguments_are_rejected_before_execution 通过，证明 handler 未被调用（called is False）。
test_missing_arguments_are_reported 通过，证明空参数字典触发 ValueError 且消息包含 missing。

故障注入与排查

在 fixtures/failure.json 描述的故障场景中，forge/agent/execution.py 的 faulty 版本已经计算了 unexpected 集合，但没有添加 if unexpected: 分支来拒绝多余参数。当测试 test_unexpected_arguments_are_rejected_before_execution 传入 {"path": "a.txt", "sudo": True} 时，execute_tool 直接调用 handler(**arguments)，handler 因不接受 sudo 参数而抛出 TypeError，而非预期的 ValueError。

本阶段教授的参数合同验证机制正是解决此问题的关键：在 execute_tool 中，unexpected 集合的计算结果已经存在，只需添加 if unexpected: raise ValueError(...) 分支即可在调用 handler 之前拦截多余参数。修复后的代码确保 test_unexpected_arguments_are_rejected_before_execution 中的 called 变量保持 False，因为 handler 根本不会被调用。

已验证的故障合同

目标测试: test_unexpected_arguments_are_rejected_before_execution
故障文件: forge/agent/execution.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_unexpected_arguments_are_rejected_before_execution，观察输出中的 TypeError: handler() got an unexpected keyword argument 'sudo'。
打开 forge/agent/execution.py，定位到 execute_tool 函数，发现 unexpected 变量已被计算但在后续代码中未被使用。
对比 faulty 源码和 verified solution，确认差异仅在于缺少 if unexpected: raise ValueError("unexpected arguments: " + ...) 分支。

修复原则: 在 execute_tool 函数中，if missing: 分支之后、return ExecutionResult(...) 之前，插入 if unexpected: raise ValueError("unexpected arguments: " + ", ".join(sorted(unexpected)))。这样当 arguments 包含未在 contract.required 或 contract.optional 中声明的键时，execute_tool 会在调用 handler 之前抛出 ValueError，确保 handler 不会因意外参数而产生副作用。

python -m pytest -q ../tests/test_stage.py::test_unexpected_arguments_are_rejected_before_execution

迁移挑战

办公研究场景：从文档系统中查找一份备忘录的内容。

改变的条件: 将工具名称从 sum 改为 lookup，参数键从 a/b 改为 doc_id，参数值改为 memo-42，handler 逻辑改为字符串拼接

在办公场景中，使用 execute_tool(ToolContract("lookup", frozenset({"doc_id"})), lambda doc_id: f"content-{doc_id}", {"doc_id": "memo-42"})，验证返回 ExecutionResult(tool_name="lookup", output="content-memo-42")。不修改 ToolContract 或 execute_tool 的任何代码，只替换调用时的数据值。

完成标准

调用 execute_tool 时传入的参数字典为 {"doc_id": "memo-42"}，且 ToolContract 的 required 为 frozenset({"doc_id"})
返回的 ExecutionResult 的 tool_name 为 "lookup" 且 output 为 "content-memo-42"

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么 unexpected 参数检查必须在调用 handler 之前完成，而不是在 handler 内部捕获 TypeError 后再处理？请从副作用隔离和调用方反馈两个角度解释这个设计决策的因果权衡。

一个有力回答应包含

handler 可能在抛出 TypeError 之前已经执行了部分副作用（如文件写入、网络请求），在调用前拦截可以保证零副作用
在 execute_tool 层统一抛出 ValueError 可以给调用方提供结构化的错误消息（包含哪些参数是 unexpected），而 TypeError 的消息依赖 handler 的具体签名
如果依赖 handler 自己防御，每个 handler 都要重复实现参数过滤逻辑，违反单一职责原则
调用前验证使得 ToolContract 成为唯一的参数合同来源，handler 只需关心业务逻辑

掌握证据

execute_tool 在参数合法时返回 ExecutionResult 且 output 值正确，例如 execute_tool(ToolContract("sum", frozenset({"a", "b"})), lambda a, b: a + b, {"a": 2, "b": 3}) 返回 result.output == 5
execute_tool 在存在多余参数时于调用 handler 之前抛出 ValueError 且消息匹配 unexpected，例如传入 {"path": "a.txt", "sudo": True} 时 called is False
execute_tool 在缺少必填参数时抛出 ValueError 且消息匹配 missing，例如传入空字典 {} 时触发
python -m pytest -q tests/test_stage.py 退出码为 0 且输出 3 passed

STAGE 04

仅对明确的瞬时故障进行有上限的重试，并保留尝试次数

先看一个具体问题

为什么不能对所有异常都无差别重试

在单 Agent 循环中，工具执行可能遇到两类性质完全不同的故障：一类是瞬时故障（如服务暂时繁忙），另一类是永久故障（如权限被拒绝）。如果重试策略不区分这两类异常，把所有异常都当作可重试处理，就会导致权限不足的操作被反复尝试三次，白白浪费重试预算并掩盖真实错误。

本阶段需要新建 forge/agent/retry.py 文件，实现 run_with_retry 函数。该函数必须只捕获 TransientToolError 进行有上限重试，遇到 PermanentToolError 时立即重新抛出。当前测试套件在 tests/test_stage.py 中定义了三个测试，由于 forge.agent.retry 模块尚不存在，测试在收集阶段直接抛出 ModuleNotFoundError。

先做判断

当 run_with_retry 接收到一个始终抛出 PermanentToolError 的操作时，正确的实现应该调用该操作几次？

1次，然后立即重新抛出异常
3次，直到耗尽 max_attempts 预算
0次，直接返回失败结果
2次，留一次预算给后续恢复

判断依据: 正确答案是1次。PermanentToolError 表示永久性故障，重试不会改变结果，因此必须在首次捕获后立即重新抛出，不消耗剩余重试预算。

它是什么

有上限重试的精确语义

run_with_retry 是一个只负责重试策略的独立函数，它接收一个无参数的 operation 可调用对象和 max_attempts 上限，返回包含最终值和实际尝试次数的 RetryResult。它的核心逻辑是在 for 循环中调用 operation，仅当捕获到 TransientToolError 且当前尝试次数未达上限时才继续循环。

RetryResult 使用 frozen=True 的 dataclass，确保返回值不可变，调用方可以安全地检查 result.value 和 result.attempts 来验证执行轨迹。当 max_attempts 耗尽且最后一次仍抛出 TransientToolError 时，该异常会被重新抛出给调用方处理。

它不是什么

重试策略的边界与职责限制

run_with_retry 不负责选择工具、管理 Agent 循环状态或检查操作幂等性，它只是一个纯粹的执行包装器。它不记录日志、不维护 Trace、不与外部可观测性系统交互，这些职责属于 Harness 和 Observability 模块。

重试策略不是让模型无期限循环直到成功，max_attempts 提供了硬性上限。当 max_attempts < 1 时，函数直接抛出 ValueError，拒绝执行任何无意义的尝试。

它与相邻概念的关系

异常分类与重试决策的因果关系

TransientToolError 和 PermanentToolError 都继承自 RuntimeError，但它们在重试循环中的处理路径完全不同。except 子句只捕获 TransientToolError，这使得 PermanentToolError 和其他未列出的异常会自然穿透循环，立即终止重试。

这种基于异常类型的分类策略将故障性质与重试决策直接绑定：瞬时故障意味着重试可能成功，永久故障意味着重试必然失败。Agent Loop 依赖这种精确分类来决定是继续重试还是向上层报告不可恢复的错误。

本阶段的边界决定: run_with_retry 只对 TransientToolError 做有上限重试，遇到 PermanentToolError 或任何其他异常时立即重新抛出，不执行工具选择、不管理循环状态、不做幂等检查。

实验 ID: single-agent-loop-04-retry-policy
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/04-retry-policy/
验收: 待验收

教师示范

目标: 在软件交付场景中，定义一个前两次抛出 TransientToolError("busy")、第三次返回 "ok" 的操作，用 run_with_retry(operation, max_attempts=3) 验证返回 RetryResult(value="ok", attempts=3)。

假设前端仓库的构建服务在前两次调用时返回繁忙状态，第三次恢复正常。我们需要用计数器模拟这种行为：每次调用 operation 时递增 attempts，当 attempts < 3 时抛出 TransientToolError，否则返回 "ok"。

run_with_retry 在循环中第一次调用 operation 捕获到 TransientToolError，因为 attempt=1 不等于 max_attempts=3，继续循环。第二次同理。第三次调用成功，返回 RetryResult(value="ok", attempts=3)，循环终止。

实施步骤

定义 TransientToolError 和 PermanentToolError 两个异常类，均继承 RuntimeError
定义 RetryResult 为 frozen=True 的 dataclass，包含 value 和 attempts 字段
实现 run_with_retry：校验 max_attempts >= 1，在 for attempt in range(1, max_attempts + 1) 循环中调用 operation，仅捕获 TransientToolError，在最后一次尝试时重新抛出

真实 solution 代码: forge/agent/retry.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, TypeVar

T = TypeVar("T")

class TransientToolError(RuntimeError):
    pass

class PermanentToolError(RuntimeError):
    pass

@dataclass(frozen=True)
class RetryResult:
    value: object
    attempts: int

def run_with_retry(operation: Callable[[], T], max_attempts: int = 3) -> RetryResult:
    if max_attempts < 1:
        raise ValueError("max_attempts must be positive")
    for attempt in range(1, max_attempts + 1):
        try:
            return RetryResult(operation(), attempt)
        except TransientToolError:
            if attempt == max_attempts:
                raise
    raise AssertionError("unreachable")

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

operation 在前两次抛出 TransientToolError("busy")，run_with_retry 的 except TransientToolError 分支捕获后检查 attempt == max_attempts，因为 1 和 2 都不等于 3，循环继续。

第三次调用返回 "ok"，RetryResult(operation(), attempt) 将值和当前尝试次数 3 封装返回。测试断言 (result.value, result.attempts) == ("ok", 3) 通过，证明瞬时故障被正确重试且尝试次数被准确记录。

带提示练习

解释为什么 run_with_retry 对 PermanentToolError("permission denied") 只调用一次 operation 就重新抛出，并补全只捕获 TransientToolError 的 except 分支。

审查 test_permanent_error_is_not_retried 测试：它断言 attempts == 1 且期望 PermanentToolError 被抛出
在 run_with_retry 的 try 块中，确认 except 子句只列出 TransientToolError，不包含 PermanentToolError
验证当 operation 抛出 PermanentToolError 时，异常不被 except 捕获，直接穿透到调用方，循环不会继续

逐级提示

except 子句捕获的异常类型决定了哪些错误会被重试，如果同时捕获了 PermanentToolError，权限错误也会被当作可重试处理
PermanentToolError 不在 except TransientToolError 的捕获范围内，因此它会直接穿透 try 块，终止函数执行
测试中 attempts 计数器验证了 operation 只被调用一次，这要求异常在首次发生时就立即向上传播

完成标准

test_permanent_error_is_not_retried 测试通过，attempts 断言为 1
run_with_retry 在遇到 PermanentToolError 时立即重新抛出，不进入下一次循环迭代

故障注入与排查

故障实验室中的 forge/agent/retry.py 存在一个注入缺陷：except 子句同时捕获了 TransientToolError 和 PermanentToolError。运行 test_permanent_error_is_not_retried 测试时，始终抛出 PermanentToolError 的 operation 被调用了 3 次而非 1 次，导致 assert attempts == 1 失败，输出 assert 3 == 1。

本阶段的核心任务就是区分瞬时故障和永久故障的重试策略。通过将 except 子句从 (TransientToolError, PermanentToolError) 修改为仅捕获 TransientToolError，PermanentToolError 将直接穿透 try 块不被重试，test_permanent_error_is_not_retried 测试中的 attempts 断言自然通过。

已验证的故障合同

目标测试: test_permanent_error_is_not_retried
故障文件: forge/agent/retry.py

诊断顺序

查看 faultySource 中 run_with_retry 的 except 子句，发现它捕获 (TransientToolError, PermanentToolError) 两种异常
分析 test_permanent_error_is_not_retried 的断言 assert attempts == 1，期望 operation 只被调用一次
对比 failureContract 中的 repair 字段：Catch only TransientToolError inside the retry loop，确认需要从 except 子句中移除 PermanentToolError

修复原则: 将 forge/agent/retry.py 中 run_with_retry 函数的 except (TransientToolError, PermanentToolError) 修改为 except TransientToolError。这样当 operation 抛出 PermanentToolError 时，异常不被捕获，直接穿透到调用方，循环终止，operation 只被调用一次。最终验证命令 python -m pytest -q tests/test_stage.py 输出 3 passed。

python -m pytest -q ../tests/test_stage.py::test_permanent_error_is_not_retried

迁移挑战

音乐创作场景：发布服务在前两次调用时返回繁忙，第三次成功发布。

改变的条件: 将场景数据从软件构建的 “ok” 替换为音乐发布的 “published”，操作错误消息从 “busy” 替换为 “publishing service busy”

定义一个 operation，前两次抛出 TransientToolError("publishing service busy")，第三次返回 "published"。用 run_with_retry(operation, max_attempts=3) 验证返回 RetryResult(value="published", attempts=3)。只替换数据值，不改变 run_with_retry 的 API 或测试固定断言结构。

完成标准

验证 result.value == "published" 且 result.attempts == 3，确认前两次瞬时故障被重试且第三次成功
验证 operation 总共被调用 3 次，重试预算未被超支，run_with_retry 的公共 API 和 RetryResult 结构保持不变

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 当 max_attempts 耗尽且最后一次仍抛出 TransientToolError 时，run_with_retry 选择重新抛出该异常而不是返回一个包含 None 的 RetryResult，这种设计在调用方需要区分成功与失败时有什么因果优势？

一个有力回答应包含

重新抛出异常让调用方能够通过 try/except 结构明确区分成功返回值和失败情况，避免将失败伪装成正常返回值
返回包含 None 的 RetryResult 会导致调用方必须额外检查 value is None，且无法区分业务返回的合法 None 值与重试耗尽的失败
重新抛出保留了原始异常的消息和类型信息，调用方可以据此决定是向上传播还是执行降级策略

掌握证据

run_with_retry 对 TransientToolError 进行有上限重试，返回包含正确值和尝试次数的 RetryResult，验证了 test_transient_failure_is_retried_with_a_bound 通过
run_with_retry 不重试 PermanentToolError，首次捕获后立即重新抛出且 attempts=1，验证了 test_permanent_error_is_not_retried 通过
run_with_retry 在 max_attempts 耗尽且最后一次仍为 TransientToolError 时重新抛出该异常，验证了 test_retry_budget_is_enforced 通过
执行 python -m pytest -q tests/test_stage.py 退出码为 0，输出 3 passed

STAGE 05

用幂等键与请求指纹防止重复副作用，同时拒绝键冲突

先看一个具体问题

相同幂等键附带不同请求指纹时的注册表行为

在单 Agent 循环中，工具调用可能因网络重试或用户重复提交而被触发多次。如果每次调用都执行写操作，就会产生重复的副作用，例如重复写入文件或重复发送邮件。幂等注册表通过记录幂等键和请求指纹来防止这种重复执行。当同一个幂等键再次到达时，注册表必须检查请求指纹是否与已存储的指纹完全一致。如果指纹不同，说明这是一个不同的请求复用了同一个键，注册表必须拒绝执行并抛出异常。当前 forge/agent/idempotency.py 文件尚不存在，测试在导入阶段就因 ModuleNotFoundError 而失败。你需要创建该文件并实现 IdempotencyRegistry 类，使三个测试全部通过。

先做判断

当 execute_once("op-1", "hash-b", operation) 被调用，而注册表中已存储了键 “op-1” 对应的指纹 “hash-a” 时，注册表应该怎么做？

直接执行 operation 并用新结果覆盖旧结果
返回已缓存的第一次结果并标记为重复
抛出 ValueError 表示请求指纹不匹配
忽略新请求且不返回任何值

判断依据: 正确答案是抛出 ValueError。幂等键只标识一个操作槽位，请求指纹才标识具体的请求内容。如果键相同但指纹不同，说明两个不同的请求意外复用了同一个键，注册表必须拒绝以防止语义错误。

它是什么

幂等注册表是什么

IdempotencyRegistry 是一个用 dataclass 实现的内存注册表，通过 _entries 字典存储幂等键到 (fingerprint, value) 元组的映射。它的 execute_once 方法接收幂等键、请求指纹和一个可调用的操作函数，返回 (value, bool) 元组。布尔值 False 表示这是首次执行并真正调用了操作函数，True 表示命中缓存且未执行操作。这个机制让 Agent 循环在重试或重复调用时安全地避免重复副作用。

它不是什么

幂等注册表不是什么

IdempotencyRegistry 不负责选择工具、不执行重试逻辑、也不管理 Agent 循环的状态转换。它不验证操作函数的返回类型是否符合工具合同，也不持久化到磁盘或数据库。它只是一个局部的、进程内的去重守卫，生命周期与单个注册表实例绑定。

它与相邻概念的关系

与 Agent 循环其他组件的关系

在 Agent 循环中，IdempotencyRegistry 位于工具执行路径上，通常在 execute_tool 调用实际副作用之前介入。它与 retry.py 的重试机制互补：重试处理瞬时失败，幂等注册表处理重复请求。ToolContract 定义工具的参数和结果合同，而幂等注册表确保同一合同下的重复调用不会产生多次副作用。可观测性组件可以记录每次 execute_once 的命中或拒绝事件，用于追踪 Agent 的执行轨迹。

本阶段的边界决定: 幂等注册表的边界是：只用键和指纹防止重复副作用，不做工具选择、不做重试、不管理循环状态；空键或空指纹必须被拒绝，因为它们无法唯一标识一个操作槽位。

实验 ID: single-agent-loop-05-idempotent-actions
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/05-idempotent-actions/
验收: 待验收

教师示范

目标: 在软件交付场景中，创建 IdempotencyRegistry()，用 execute_once("op-1", "hash-a", lambda: "written") 验证第一次返回 ("written", False)，第二次相同键和指纹返回 ("written", True)，且操作只执行一次。

假设 Agent 正在为前端仓库写入一个配置文件，用户可能因网络抖动触发两次相同的提交。你需要确保写操作只发生一次，第二次调用直接返回缓存结果。这要求注册表在键已存在且指纹匹配时跳过操作执行，同时返回 True 标记命中缓存。

通过维护一个 _entries 字典，注册表在首次调用时存储 (fingerprint, value) 并返回 (value, False)。第二次调用发现键已存在，比较指纹一致后直接返回 (value, True)，操作函数不会被调用。

实施步骤

创建 forge/agent/idempotency.py 文件，定义 IdempotencyRegistry dataclass，包含 _entries 字典字段。
实现 execute_once 方法：首先检查 key 和 fingerprint 是否为空，为空则抛出 ValueError。
检查 key 是否已在 _entries 中：若存在且指纹匹配，返回 (value, True)；若不存在，调用 operation()，存储结果并返回 (value, False)。

真实 solution 代码: forge/agent/idempotency.py

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable, TypeVar

T = TypeVar("T")

@dataclass
class IdempotencyRegistry:
    _entries: dict[str, tuple[str, object]] = field(default_factory=dict)

    def execute_once(self, key: str, fingerprint: str, operation: Callable[[], T]) -> tuple[T, bool]:
        if not key or not fingerprint:
            raise ValueError("key and fingerprint are required")
        if key in self._entries:
            stored_fingerprint, value = self._entries[key]
            if stored_fingerprint != fingerprint:
                raise ValueError("idempotency key reused with a different request")
            return value, True
        value = operation()
        self._entries[key] = (fingerprint, value)
        return value, False

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

第一次调用 execute_once("op-1", "hash-a", operation) 时，"op-1" 不在 _entries 中，因此 operation() 被执行一次，calls 变为 1，结果 ("written", False) 被返回并存储。

第二次调用使用相同的 "op-1" 和 "hash-a"，键已存在且指纹匹配，注册表直接返回 ("written", True) 而不调用 operation()，因此 calls 保持为 1。

布尔值 False 与 True 的区分让调用方能够区分首次执行和缓存命中，这对可观测性日志和结果验证很重要。

带提示练习

解释为什么 execute_once("op-1", "hash-b", lambda: "second") 在已存储 "hash-a" 时必须抛出 ValueError，并补全指纹比较分支。

回顾已实现的缓存命中路径：当 key in self._entries 时，取出 stored_fingerprint 和 value。
在返回缓存结果之前，添加指纹比较：如果 stored_fingerprint != fingerprint，抛出 ValueError。
确保异常消息包含 "different request" 字样，以匹配测试中的 pytest.raises(ValueError, match="different request") 断言。
运行目标测试验证修复效果。

逐级提示

测试期望 ValueError 的消息匹配正则 "different request"，所以异常消息必须包含这个子串。
在 if key in self._entries: 分支内部，取出 stored_fingerprint 后，添加一个 if stored_fingerprint != fingerprint: 条件判断。
抛出的异常类型是 ValueError，消息可以是 "idempotency key reused with a different request"。
注意指纹比较必须在返回缓存值之前执行，否则不同的请求会错误地收到第一次的结果。

完成标准

test_reused_key_with_different_fingerprint_is_rejected 测试通过，验证不同指纹抛出 ValueError。
test_duplicate_request_returns_cached_result_without_side_effect 仍然通过，验证相同指纹正确返回缓存。
test_empty_identity_is_rejected 通过，验证空键或空指纹被拒绝。

故障注入与排查

在 fixtures/failure.json 描述的故障场景中，forge/agent/idempotency.py 的 execute_once 方法在键已存在时直接返回缓存结果，完全忽略了请求指纹的比较。这导致 test_reused_key_with_different_fingerprint_is_rejected 测试失败，因为当用 "hash-b" 调用已存储 "hash-a" 的键 "op-1" 时，注册表没有抛出 ValueError，而是错误地返回了第一次的结果 ("first", True)。

本阶段实现的 IdempotencyRegistry 已经具备键存在检查和指纹存储机制，只需在返回缓存值之前增加一个指纹比较分支即可修复此缺陷。execute_once 方法中已经从 _entries[key] 取出了 stored_fingerprint，只需将其与传入的 fingerprint 进行比较，不匹配时抛出 ValueError 即可。

已验证的故障合同

目标测试: test_reused_key_with_different_fingerprint_is_rejected
故障文件: forge/agent/idempotency.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_reused_key_with_different_fingerprint_is_rejected，观察测试失败信息 DID NOT RAISE ValueError。
打开 forge/agent/idempotency.py，定位到 execute_once 方法中 if key in self._entries: 分支。
检查该分支内的逻辑：发现它直接返回 (value, True) 而没有比较 stored_fingerprint 与传入的 fingerprint。
确认缺陷位置：在 stored_fingerprint, value = self._entries[key] 之后缺少 if stored_fingerprint != fingerprint: raise ValueError(...) 分支。

修复原则: 在 execute_once 方法的 if key in self._entries: 分支中，在返回缓存值之前添加指纹比较：if stored_fingerprint != fingerprint: raise ValueError("idempotency key reused with a different request")。这确保只有键和指纹都匹配时才返回缓存结果，指纹不匹配时拒绝执行并抛出异常。修复后运行 python -m pytest -q tests/test_stage.py 验证三个测试全部通过。

python -m pytest -q ../tests/test_stage.py::test_reused_key_with_different_fingerprint_is_rejected

迁移挑战

办公研究场景：Agent 从文档和表格生成带引用的决策简报，用户可能重复提交相同的简报请求。

改变的条件: 将幂等键从 “op-1” 改为 “brief-1”，指纹从 “hash-a”/“hash-b” 改为 “fp-alpha”/“fp-beta”，操作结果从 “written” 改为 “brief-draft”。

创建 IdempotencyRegistry()，调用 execute_once("brief-1", "fp-alpha", lambda: "brief-draft") 验证第一次返回 ("brief-draft", False)，第二次相同键和指纹返回 ("brief-draft", True)。然后调用 execute_once("brief-1", "fp-beta", lambda: "other") 验证抛出 ValueError 匹配 "different request"。只替换数据值，不改变 IdempotencyRegistry 的 API 或测试固定断言。

完成标准

使用 execute_once("brief-1", "fp-alpha", lambda: "brief-draft") 第一次调用返回 ("brief-draft", False)，第二次相同键和指纹返回 ("brief-draft", True)，操作只执行一次。
使用 execute_once("brief-1", "fp-beta", lambda: "other") 调用时抛出 ValueError 且异常消息匹配 "different request"。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么幂等键单独不足以唯一标识一个请求？如果注册表只比较键而不比较指纹，在什么具体场景下会导致错误的副作用？

一个有力回答应包含

幂等键标识一个操作槽位，但同一操作可能因参数不同而对应不同的请求内容，指纹用于区分这些不同的请求。
如果只比较键，当两个不同的写请求意外复用同一键时，第二个请求会错误地收到第一个请求的结果，导致数据不一致。
例如用户用同一幂等键发送两封内容不同的邮件，只比较键会导致第二封邮件不被发送，收件人收到的是第一封的内容。
指纹比较确保键冲突时拒绝执行，迫使调用方使用不同的键或修正请求，从而维护操作的语义正确性。

掌握证据

execute_once 首次调用返回 (value, False)，相同键和指纹的重复调用返回 (value, True) 且操作只执行一次，验证了 test_duplicate_request_returns_cached_result_without_side_effect。
execute_once 在键已存在但指纹不匹配时抛出 ValueError 且消息匹配 "different request"，验证了 test_reused_key_with_different_fingerprint_is_rejected。
execute_once 在 key 为空字符串或 fingerprint 为空时抛出 ValueError，验证了 test_empty_identity_is_rejected。
python -m pytest -q tests/test_stage.py 退出码为 0 且输出 3 passed，确认三个测试全部通过。

STAGE 06

根据可验证成功、失败、预算和重复动作做出显式停止决定

先看一个具体问题

产出答案本身不等于验证成功

在单 Agent 循环中，模型可能已经生成了 “answer” 动作，但这并不意味着任务目标已经被外部验收通过。当前需要创建 forge/agent/stopping.py 文件，实现 StopController 类，让循环根据外部传入的 verified 标志、fatal_error、step_count 预算和 action_history 中的重复动作模式做出显式停止决定。

在未实现该文件时，测试模块因找不到 forge.agent.stopping 而直接报 ModuleNotFoundError。而在存在缺陷的版本中，代码错误地将 action_history 最后一个动作等于 “answer” 视为 verified_success，导致循环在验收通过前就提前停止。

本阶段的驱动问题是：为什么产出答案本身不足以触发 verified_success 停止？核心在于产出是模型行为，而验证是外部可检查证据，两者不能混为一谈。

先做判断

当 decide(step_count=1, verified=False, fatal_error=False, action_history=["answer"]) 被调用时，StopController 应该返回什么结果？

返回 stop=True, reason=“verified_success”，因为模型已经产出了答案
返回 stop=False, reason=“continue”，因为外部验证尚未通过
返回 stop=True, reason=“step_budget_exhausted”，因为步数已耗尽
返回 stop=True, reason=“fatal_error”，因为未验证即失败

判断依据: 正确答案是返回 stop=False。verified 参数为 False 意味着外部验收检查尚未通过，即使动作历史中出现了 “answer”，循环也不应停止。产出答案和验证成功是两个独立阶段。

它是什么

StopController 的职责边界

StopController 是一个只负责停止决策的组件，它接收 step_count、verified、fatal_error 和 action_history 四个外部输入，返回一个包含 stop 布尔值和 reason 字符串的 StopDecision 数据类。

停止条件的优先级顺序是：首先检查 verified 是否为 True，然后检查 fatal_error，接着检查 step_count 是否达到 max_steps 预算上限，最后检查 action_history 中最后一个动作的重复次数是否达到 repeat_limit。如果都不满足，返回 stop=False 和 reason="continue"。

它不是什么

StopController 不做什么

StopController 不执行任何工具调用，不管理观察结果，也不负责重试逻辑。它不读取模型输出本身，也不判断答案内容的质量，它只依赖外部 Harness 传入的布尔标志。

它不是让模型无期限循环的机制，也不是独立人格。停止决策的权力属于外部运行环境，StopController 只是将这些约束条件汇总为一个明确的决策对象。

它与相邻概念的关系

与 Agent Loop 其他组件的关系

StopController 在 Agent Loop 中位于验证之后、动作选择之前。Harness 执行工具并观察结果后，通过 Verification 组件检查结果是否推进目标，然后将 verified 布尔值传给 StopController。

action_history 由 LoopState 维护，记录了到目前为止的所有动作序列。step_count 也是 LoopState 的一部分。StopController 通过检查 action_history 中最后一个动作的出现次数来判断是否陷入重复循环。

本阶段的边界决定: StopController 只根据外部传入的 verified、fatal_error、step_count 和 action_history 做停止决策，不执行工具、不管理观察、不做重试，也不自行推断答案是否正确。

实验 ID: single-agent-loop-06-stop-conditions
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/06-stop-conditions/
验收: 待验收

教师示范

目标: 在软件交付场景中，验证 StopController 在 verified=False 时不停止，在 verified=True 时以 verified_success 原因停止。

假设在软件交付场景中，Agent 在第一步产出了 “answer” 动作，但外部测试尚未运行通过，此时 verified=False。StopController 不应因为历史中存在 “answer” 就停止循环，而应继续执行让 Harness 运行验收测试。

当第二步外部验收测试通过后，Harness 将 verified=True 传入 decide 方法，此时 StopController 返回 stop=True 和 reason="verified_success"，循环正常终止。

实施步骤

创建 StopController 实例，使用默认参数 max_steps=8 和 repeat_limit=2。
调用 decide(step_count=1, verified=False, fatal_error=False, action_history=["answer"])，验证返回的 StopDecision.stop 为 False。
调用 decide(step_count=2, verified=True, fatal_error=False, action_history=["answer"])，验证返回的 (stop, reason) 等于 (True, "verified_success")。

真实 solution 代码: forge/agent/stopping.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class StopDecision:
    stop: bool
    reason: str

class StopController:
    def __init__(self, max_steps: int = 8, repeat_limit: int = 2) -> None:
        self.max_steps = max_steps
        self.repeat_limit = repeat_limit

    def decide(self, *, step_count: int, verified: bool, fatal_error: bool, action_history: list[str]) -> StopDecision:
        if verified:
            return StopDecision(True, "verified_success")
        if fatal_error:
            return StopDecision(True, "fatal_error")
        if step_count >= self.max_steps:
            return StopDecision(True, "step_budget_exhausted")
        if action_history and action_history.count(action_history[-1]) >= self.repeat_limit:
            return StopDecision(True, "repeated_action")
        return StopDecision(False, "continue")

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

第一条 if verified: 分支确保只有当外部验收检查通过时才触发成功停止，这与 action_history 中是否包含 “answer” 无关。

当 verified=False 且其他停止条件均不满足时，代码执行到最后一行返回 StopDecision(False, "continue")，循环继续运行以允许 Harness 执行后续验收。

带提示练习

解释并补全 StopController(repeat_limit=2) 在 action_history=["search", "search"] 时返回 reason="repeated_action" 的重复动作检查分支。

创建 StopController(repeat_limit=2) 实例。
调用 decide(step_count=2, verified=False, fatal_error=False, action_history=["search", "search"])。
检查返回的 StopDecision.reason 是否等于 "repeated_action"。

逐级提示

注意 action_history 的最后一个元素是什么，以及它在整个列表中出现了几次。
action_history.count(action_history[-1]) 的返回值与 self.repeat_limit 之间应该满足什么比较关系？
确保重复动作检查分支放在 step_count 预算检查之后，但在 return StopDecision(False, "continue") 之前。

完成标准

当 action_history 最后一个动作的出现次数大于等于 repeat_limit 时，decide 返回 stop=True 和 reason="repeated_action"。
当 action_history 为空或最后一个动作未达到重复阈值时，不触发 repeated_action 停止。

故障注入与排查

在 fixtures/failure.json 描述的缺陷场景中，forge/agent/stopping.py 的 decide 方法错误地将 action_history 最后一个动作等于 “answer” 视为验证成功。当运行 test_only_verified_success_stops_as_success 测试时，测试期望 verified=False 时 stop 为 False，但缺陷代码返回了 stop=True。这个缺陷暴露了产出与验证被混淆的核心问题。

本阶段通过实现正确的 StopController 解决了这个问题。正确的实现中，decide 方法只在 verified=True 时返回 verified_success，完全移除了对 action_history 中 “answer” 动作的检查。这样，产出答案和验证成功就被明确分离为两个独立阶段，循环只有在外部验收通过后才会停止。

已验证的故障合同

目标测试: test_only_verified_success_stops_as_success
故障文件: forge/agent/stopping.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_only_verified_success_stops_as_success，观察失败信息。
查看失败信息中 decide(step_count=1, verified=False, fatal_error=False, action_history=['answer']) 返回了 StopDecision(stop=True, reason='verified_success')。
打开 forge/agent/stopping.py，检查 decide 方法的第一条 if 语句，发现条件为 if verified or (action_history and action_history[-1] == "answer")。
分析该条件：即使 verified=False，只要 action_history 最后一个动作是 “answer”，就会触发 verified_success，这就是产出与验证混淆的根因。

修复原则: 修复方法是将 decide 方法中的第一条 if 语句从 if verified or (action_history and action_history[-1] == "answer") 修改为 if verified。这样 StopController 只依赖外部传入的 verified 标志来判断是否以 verified_success 停止，不再检查 action_history 的内容。这与 failureContract 中的修复描述一致：只有当验证器提供 verified=True 时才以 verified_success 停止。

python -m pytest -q ../tests/test_stage.py::test_only_verified_success_stops_as_success

迁移挑战

在音乐创作场景中，Agent 正在执行作曲、混音和母带处理三个步骤，需要验证步数预算耗尽时的停止行为。

改变的条件: 将 action_history 的数据值从默认场景替换为音乐创作的三个不同动作 [“compose”, “mix”, “master”]，验证步数预算耗尽时的停止行为。

创建 StopController(max_steps=3)，调用 decide(step_count=3, verified=False, fatal_error=False, action_history=["compose", "mix", "master"])，验证返回的 reason 等于 "step_budget_exhausted"。只替换 action_history 数据，不改变 StopController 的 API 或测试固定断言。

完成标准

decide 调用时 action_history 包含恰好三个元素 [“compose”, “mix”, “master”]，且 step_count=3 等于 max_steps=3。
返回的 StopDecision.reason 必须等于 "step_budget_exhausted"，且 stop 为 True，验证步数预算检查在重复动作检查之前生效。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么在 verified=False 时，即使 action_history 包含 “answer” 也不应该停止？请从产出与验证的因果关系角度解释这一设计决策的边界。

一个有力回答应包含

产出答案是模型的内部行为，不保证结果正确或满足外部验收标准。
验证是外部可检查证据，必须由独立的验收机制（如测试通过、Schema 校验）提供 verified=True。
如果将产出等同于验证，循环会在未经验证的答案上停止，导致不可靠的交付。
StopController 的设计边界是只依赖外部传入的 verified 标志，不自行推断答案正确性。

掌握证据

decide 在 verified=False 时返回 stop=False，即使 action_history 包含 “answer”。
decide 仅在 verified=True 时返回 (True, "verified_success")。
decide 在最后一个动作重复达到 repeat_limit 次时返回 reason="repeated_action"。
decide 在 step_count >= max_steps 时返回 reason="step_budget_exhausted"。
python -m pytest -q tests/test_stage.py 退出码为 0 且 3 个测试通过。

STAGE 07

将动作、执行、观察和停止控制组合成有界且可审计的单 Agent 循环

先看一个具体问题

验证通过后循环未停止导致多余工具调用

在当前阶段，forge/agent/runner.py 文件尚不存在，测试文件 tests/test_stage.py 在导入 from forge.agent.runner import run_agent 时直接抛出 ModuleNotFoundError，导致收集阶段中断且退出码为 2。你需要创建该文件并实现 run_agent 函数，将前几个阶段已实现的 LoopState 和 StopController 组合成一个有界循环。

在故障实验中，run_agent 的实现已经存在但包含一个精确缺陷：当 StopController.decide 返回 decision.stop 为 True 且 reason 为 verified_success 时，代码没有返回而是继续循环。这导致 test_runner_stops_immediately_after_verified_output 断言失败，因为 step 回调被调用了 3 次而非预期的 2 次，且 result.reason 变成了 repeated_action 而非 verified_success。

核心工程决策是：run_agent 必须对任何 decision.stop 为 True 的决策立即返回，无论 reason 是 verified_success、repeated_action 还是 step_budget_exhausted。忽略任何停止条件都会破坏循环的有界性保证，并可能在验证通过后触发不必要的副作用。

先做判断

当 verify(output) 返回 True 且 StopController.decide 返回 decision.stop == True 时，run_agent 应该立即做什么？

立即返回 LoopResult，不再调用 step
继续循环直到 max_steps 耗尽
抛出异常终止循环
重新调用 step 确认结果

判断依据: 正确选择是立即返回。验证通过意味着目标已达成，继续循环不仅浪费步骤预算，还可能因为重复动作触发错误的 repeated_action 停止原因。

它是什么

有界 Agent 循环的组成

run_agent 是一个组合器，它接收一个 step 回调和一个 verify 验证函数，在 max_steps 限制内反复执行：调用 step 获取动作与输出、记录到 LoopState、用 StopController 判断是否停止。run_agent 自身不做工具选择、不做重试、不做幂等检查，这些职责由 step 回调和已实现的前置模块承担。

循环的每一步都产生一个 StopDecision，其中 stop 布尔值是唯一的终止信号。run_agent 的核心职责是忠实地执行这个信号：一旦 stop 为 True，无论原因如何都必须立即退出循环并返回当前状态。

它不是什么

run_agent 不做什么

run_agent 不是自主决策器，它不决定调用哪个工具——这由外部 step 回调根据步骤编号 number 决定。run_agent 也不做错误重试或幂等性检查，这些属于 run_with_retry 和 IdempotencyRegistry 的职责。

run_agent 不是无界循环或独立人格。它不会因为模型觉得需要更多信息就继续运行；停止条件完全由 StopController 和 max_steps 参数约束，任何停止决策都必须被立即执行。

它与相邻概念的关系

模块间的职责边界

LoopState 负责记录动作历史和观察结果，StopController 负责根据步骤计数、验证结果和动作历史判断是否停止。run_agent 将两者串联：每步先记录再判断，判断结果驱动循环控制流。

step 回调是动作来源，verify 是验收来源，StopController 是停止仲裁者，run_agent 是执行者。这四者的分离使得 run_agent 的逻辑可以保持简单：执行、记录、判断、返回或继续。

本阶段的边界决定: run_agent 的边界是：对任何 decision.stop == True 的决策无条件返回，不区分 reason；循环步数严格受 max_steps 限制；不承担工具选择、重试或幂等检查职责。

实验 ID: single-agent-loop-07-agent-loop-runner
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/single-agent-loop/07-agent-loop-runner/
验收: 待验收

教师示范

目标: 在软件场景中，定义 step 回调使第 1 步返回未完成的 inspect，第 2 步返回已完成的 answer，验证 run_agent 在第 2 步以 verified_success 停止。

首先创建 forge/agent/runner.py 文件，定义 LoopResult 数据类和 run_agent 函数的基本骨架。LoopResult 包含 reason、steps 和 observations 三个字段，分别记录停止原因、执行步数和观察序列。

然后实现循环主体：用 range(1, max_steps + 1) 遍历步骤编号，每步调用 step(number) 获取动作名和输出对象，将动作追加到 history 列表，调用 state.record 记录观察，再调用 controller.decide 获取停止决策。

最后实现停止逻辑：当 decision.stop 为 True 时立即返回 LoopResult，否则循环结束后返回 step_budget_exhausted。这个示例教会你循环的基本骨架和停止决策的处理位置。

实施步骤

创建 forge/agent/runner.py，导入 LoopState 和 StopController，定义 LoopResult 数据类
实现 run_agent 函数：初始化 LoopState 和 StopController，创建 history 列表
在 for number in range(1, max_steps + 1) 循环中调用 step(number) 获取动作和输出
将动作追加到 history，调用 state.record(action, "ok", output) 记录观察
调用 controller.decide(step_count=number, verified=verify(output), fatal_error=False, action_history=history)
检查 decision.stop，为 True 时返回 LoopResult(decision.reason, number, tuple(state.observations))

真实 solution 代码: forge/agent/runner.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable

from .observations import LoopState
from .stopping import StopController

@dataclass(frozen=True)
class LoopResult:
    reason: str
    steps: int
    observations: tuple[object, ...]

def run_agent(step: Callable[[int], tuple[str, object]], verify: Callable[[object], bool], max_steps: int = 5) -> LoopResult:
    state = LoopState()
    controller = StopController(max_steps=max_steps)
    history: list[str] = []
    for number in range(1, max_steps + 1):
        action, output = step(number)
        history.append(action)
        state.record(action, "ok", output)
        decision = controller.decide(step_count=number, verified=verify(output), fatal_error=False, action_history=history)
        if decision.stop:
            return LoopResult(decision.reason, number, tuple(state.observations))
    return LoopResult("step_budget_exhausted", max_steps, tuple(state.observations))

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

在这个示例中，step 回调在第 1 步返回 ("inspect", {"complete": False})，verify 返回 False，StopController.decide 返回 stop=False，循环继续。第 2 步返回 ("answer", {"complete": True})，verify 返回 True，StopController.decide 返回 stop=True, reason="verified_success"，run_agent 立即返回。

关键在于 if decision.stop: 这一行不区分 reason。如果加上 and decision.reason != "verified_success" 的条件，就会忽略验证通过的停止信号，导致第 3 步被错误执行。这正是故障实验中注入的缺陷。

带提示练习

解释为什么 run_agent 在 decision.stop 为 True 且 reason 为 verified_success 时必须立即返回，并确认返回分支覆盖所有 stop=True 决策。

审查 run_agent 中的 if decision.stop: 分支，确认它没有排除任何 reason
思考如果 reason 为 verified_success 时不返回会发生什么：step 会被第 3 次调用，calls 列表变为 [1, 2, 3]
思考如果 reason 为 repeated_action 时不返回会发生什么：循环会继续执行重复动作，浪费步骤预算
确认循环结束后的 return LoopResult("step_budget_exhausted", max_steps, ...) 只在所有步骤都未触发 stop=True 时执行

逐级提示

查看 StopController.decide 的返回值：stop 为 True 时有哪些可能的 reason 值？
如果 verified_success 被排除在停止条件外，第 3 步的 step 调用会产生什么 action？它与之前的 action 是否相同？
test_runner_stops_repeated_actions 期望 steps=2，这意味着 repeated_action 也必须在第 2 步立即返回。
test_runner_obeys_step_budget 期望 steps=3 且 reason="step_budget_exhausted"，这意味着当 max_steps=3 且无停止条件触发时，循环自然结束。

完成标准

if decision.stop: 不包含任何排除 reason 的条件
所有三个测试通过：verified_success 在第 2 步停止、repeated_action 在第 2 步停止、step_budget_exhausted 在第 3 步停止

故障注入与排查

在故障实验中，forge/agent/runner.py 已存在但 run_agent 的停止条件写成了 if decision.stop and decision.reason != "verified_success":。这导致验证通过后循环不返回，而是继续执行第 3 步。运行 test_runner_stops_immediately_after_verified_output 时，step 回调被调用了 3 次（calls == [1, 2, 3]），而测试期望只调用 2 次（calls == [1, 2]）。同时 result.reason 变成了 repeated_action 而非 verified_success，因为第 3 步的 action 与第 2 步相同（都是 "answer"），触发了重复动作检测。

本阶段的核心学习目标就是理解 run_agent 必须对任何 stop=True 的决策无条件返回。StopController 已经封装了所有停止判断逻辑（验证通过、重复动作、步数耗尽），run_agent 的职责是忠实执行这些决策。通过诊断这个故障，学习者能直接观察到忽略 verified_success 停止信号如何导致多余的工具调用和错误的停止原因。修复方法是将 if decision.stop and decision.reason != "verified_success": 改为 if decision.stop:，使所有停止决策都被立即执行。

已验证的故障合同

目标测试: test_runner_stops_immediately_after_verified_output
故障文件: forge/agent/runner.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_runner_stops_immediately_after_verified_output，观察断言失败信息：assert ('repeated_ac... 3, [1, 2, 3]) == ('verified_su...s', 2, [1, 2])
分析失败信息：result.reason 是 repeated_action 而非 verified_success，result.steps 是 3 而非 2，calls 是 [1, 2, 3] 而非 [1, 2]
打开 forge/agent/runner.py，定位到 if decision.stop and decision.reason != "verified_success": 这一行，识别出 and decision.reason != "verified_success" 条件排除了验证通过的停止决策
理解因果链：第 2 步 verify 返回 True，StopController.decide 返回 stop=True, reason="verified_success"，但由于排除条件，if 分支不执行，循环继续到第 3 步
第 3 步 step(3) 返回 ("answer", {"complete": True})，action 为 "answer" 与第 2 步相同，StopController.decide 检测到重复动作返回 stop=True, reason="repeated_action"，此时 if 分支执行并返回

修复原则: 将 forge/agent/runner.py 中 run_agent 函数的停止条件从 if decision.stop and decision.reason != "verified_success": 修改为 if decision.stop:。这样当 StopController.decide 返回 stop=True 时，无论 reason 是 verified_success、repeated_action 还是其他停止原因，run_agent 都会立即返回 LoopResult，不再继续调用 step。修复后第 2 步验证通过时直接返回 reason="verified_success", steps=2，calls 列表只包含 [1, 2]。

python -m pytest -q ../tests/test_stage.py::test_runner_stops_immediately_after_verified_output

迁移挑战

办公研究场景：从文档和网页生成决策简报

改变的条件: 替换 step 回调的动作名称和输出数据字段，将软件场景的 inspect/answer 改为办公场景的 research/draft，同时将验证字段从 complete 改为 done

在办公场景中定义 step 回调：number==1 时返回 ("research", {"done": False})，number==2 时返回 ("draft", {"done": True})，verify 为 lambda output: output["done"]，用 run_agent(step, verify, max_steps=5) 验证 (result.reason, result.steps) == ("verified_success", 2) 且 len(result.observations) == 2。只替换 step 回调数据和 verify 逻辑，不改变 run_agent API 或测试固定断言。

完成标准

step 回调第 1 步返回 ("research", {"done": False})，第 2 步返回 ("draft", {"done": True})，verify 为 lambda output: output["done"]
run_agent 返回 result.reason == "verified_success" 且 result.steps == 2 且 len(result.observations) == 2

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么 run_agent 在 decision.stop 为 True 时无论 reason 是什么都必须返回？如果允许 verified_success 时不返回，会对循环的有界性和副作用产生什么后果？

一个有力回答应包含

StopController 已封装所有停止判断逻辑，run_agent 的职责是忠实执行停止决策而非重新判断
忽略 verified_success 会导致验证通过后继续调用 step，产生多余的工具调用和潜在副作用
继续循环可能触发 repeated_action，使 result.reason 从 verified_success 变为 repeated_action，掩盖真正的停止原因
有界循环的保证依赖于每个 stop=True 信号被立即执行，任何条件排除都破坏这个保证

掌握证据

run_agent 在第 2 步 verify 返回 True 时以 reason="verified_success" 和 steps=2 停止，calls 列表为 [1, 2]
run_agent 在相同 action 重复时以 reason="repeated_action" 和 steps=2 停止，对应 test_runner_stops_repeated_actions
run_agent 在 max_steps=3 且无验证通过时以 reason="step_budget_exhausted" 和 steps=3 停止，对应 test_runner_obeys_step_budget
python -m pytest -q tests/test_stage.py 退出码为 0 且 3 个测试全部通过

完成本章

实现第一个可运行的单 Agent 交付循环。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

Agent

工具

验证与评测

可观测性

单 Agent 与工具调用循环

先看一个具体问题

当两个工具能力得分并列时，选择器不能默认取第一个

先做判断

它是什么

动作选择是能力匹配与唯一性验证的组合

它不是什么

动作选择不执行工具也不决定停止

它与相邻概念的关系

动作选择在 Agent 循环中的位置

先看一个具体问题

工具结果记录与可恢复错误的混淆

先做判断

它是什么

LoopState 的职责边界

它不是什么

LoopState 不做的事

它与相邻概念的关系

与 Agent 循环其他组件的关系

先看一个具体问题

未声明参数必须在调用 handler 之前被拒绝

先做判断

它是什么

参数合同验证边界

它不是什么

execute_tool 的职责边界

它与相邻概念的关系

与 Agent Loop 和 Verification 的关系

先看一个具体问题

为什么不能对所有异常都无差别重试

先做判断

它是什么

有上限重试的精确语义

它不是什么

重试策略的边界与职责限制

它与相邻概念的关系

异常分类与重试决策的因果关系

先看一个具体问题

相同幂等键附带不同请求指纹时的注册表行为

先做判断

它是什么

幂等注册表是什么

它不是什么

幂等注册表不是什么

它与相邻概念的关系

与 Agent 循环其他组件的关系

先看一个具体问题

产出答案本身不等于验证成功

先做判断

它是什么

StopController 的职责边界

它不是什么

StopController 不做什么

它与相邻概念的关系

与 Agent Loop 其他组件的关系

先看一个具体问题

验证通过后循环未停止导致多余工具调用

先做判断

它是什么

有界 Agent 循环的组成

它不是什么

run_agent 不做什么

它与相邻概念的关系

模块间的职责边界