learn-claude-code 团队协议（Team Protocols）：用同一个 request_id 关联模式实现 Agent 间的关停与计划审批

本文围绕 learn-claude-code 仓库中旧版 12 课教程的 s10 团队协议文档 展开：在 s09 已经为 Agent 团队提供持久队友和异步邮箱之后，s10 用一套"请求-响应"结构化握手协议，解决队友关停（shutdown）与计划审批（plan approval）两个协调问题。读完本文，你将掌握如何用 request_id 关联请求与响应、如何设计 pending → approved | rejected 的共享状态机（FSM），并能在 agents/s10_team_protocols.py 中运行验证；同时本文会结合当前课程主线 s13_agent_teams 的 ProtocolState 实现和 tests/test_agent_teams_runtime.py 测试用例，展示这套协议在更完整团队运行时中的演进形态。

为什么需要结构化协议：s09 留下的两个协调缺口

s10 处于旧版课程链路的中间位置：s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > [ s10 ] > s11 > s12，其上一节 s09 Agent Teams 已经实现了三样东西：

1. 持久队友：TeammateManager.spawn() 用守护线程为每个队友跑一个完整的 agent loop；
2. 身份与生命周期：.team/config.json 记录成员名、角色和状态（working / idle / shutdown）；
3. 通信信道：MessageBus 为每个成员维护一个只追加的 JSONL 收件箱（.team/inbox/<name>.jsonl），send() 追加一行 JSON，read_inbox() 读取并清空。

但文档指出，"能通信"不等于"能协调"。s09 存在两个具体问题：

• 关停问题（Shutdown）：直接杀掉一个队友线程，会留下写到一半的文件，config.json 里的状态也不会更新。团队需要一个握手：lead 发起请求，队友审批（干完手头的事再退出）或拒绝（继续工作）。
• 计划审批问题（Plan approval）：lead 说一句"重构 auth 模块"，队友会立刻动手。对高风险变更，应该让队友先提交计划，lead 审阅通过后再执行。

文档给出的关键洞察是：这两个问题共享同一种结构——一方发送带唯一 ID 的请求，另一方引用该 ID 回复。这就是 s10 的全部内容：一个 FSM，两个应用（One FSM, two applications）。

协议设计：共享 FSM 与双追踪表

s10 的协议全景如下（直接继承自原文档）：

Shutdown Protocol            Plan Approval Protocol
==================           ======================

Lead             Teammate    Teammate           Lead
  |                 |           |                 |
  |--shutdown_req-->|           |--plan_req------>|
  | {req_id:"abc"}  |           | {req_id:"xyz"}  |
  |                 |           |                 |
  |<--shutdown_resp-|           |<--plan_resp-----|
  | {req_id:"abc",  |           | {req_id:"xyz",  |
  |  approve:true}  |           |  approve:true}  |

Shared FSM:
  [pending] --approve--> [approved]
  [pending] --reject---> [rejected]

Trackers:
  shutdown_requests = {req_id: {target, status}}
  plan_requests     = {req_id: {from, plan, status}}

三个要点：

1. 方向相反：shutdown 是 lead → 队友（lead 请求，队友决定）；plan approval 是队友 → lead（队友提交，lead 决定）。但消息形态完全对称：*_request / *_response，都携带 request_id 与 approve 布尔值。
2. 共享状态机：无论哪条协议，请求创建时状态都是 pending，收到响应后根据 approve 落到 approved 或 rejected。状态机不感知协议类型，因此可以复用到任何"请求-响应"协商上。
3. 双追踪表：shutdown_requests 记录 {req_id: {target, status}}，plan_requests 记录 {req_id: {from, plan, status}}。追踪表是内存中的"事实来源"，让 lead 不必解析自然语言回复就能查询进度。

在 agents/s10_team_protocols.py 中可以看到这两个追踪表的全局定义：

# -- Request trackers: correlate by request_id --
shutdown_requests = {}
plan_requests = {}
_tracker_lock = threading.Lock()

注意 _tracker_lock：因为队友在独立线程里运行，队友处理 shutdown_response 时会写 shutdown_requests，而 lead 主线程可能在同时读它。实现里所有对追踪表的读写都包在 with _tracker_lock: 中（见 agents/s10_team_protocols.py 和 L352-L360），这是多线程环境下协议状态一致性的最小保证。

关停协议（Shutdown Protocol）的完整链路

第一步：lead 发起关停请求

文档中的实现与源码一致。lead 生成一个 8 位 UUID 前缀作为 request_id，在追踪表中登记 pending，然后通过 MessageBus 把请求写进队友的收件箱：

shutdown_requests = {}

def handle_shutdown_request(teammate: str) -> str:
    req_id = str(uuid.uuid4())[:8]
    shutdown_requests[req_id] = {"target": teammate, "status": "pending"}
    BUS.send("lead", teammate, "Please shut down gracefully.",
             "shutdown_request", {"request_id": req_id})
    return f"Shutdown request {req_id} sent (status: pending)"

对应 agents/s10_team_protocols.py 的 handle_shutdown_request。注意两点实现细节：

• request_id 是请求方生成的，保证全局唯一且回复可回指；
• BUS.send 的第四个参数是消息类型 shutdown_request。MessageBus.send() 会校验类型必须属于白名单 VALID_MSG_TYPES = {"message", "broadcast", "shutdown_request", "shutdown_response", "plan_approval_response"}（见 agents/s10_team_protocols.py），非法类型直接返回错误。这个白名单让控制消息和普通聊天消息在协议层可区分。

写入邮箱的消息是一条 JSONL 行，形如：

{"type": "shutdown_request", "from": "lead", "content": "Please shut down gracefully.", "timestamp": 1756953600.0, "request_id": "a1b2c3d4"}

第二步：队友收到请求，回复 approve / reject

队友在自己的 agent loop 里每个循环先 BUS.read_inbox(name)，把收到的消息 JSON 注入上下文（见 agents/s10_team_protocols.py）。队友判断后调用 shutdown_response 工具：

if tool_name == "shutdown_response":
    req_id = args["request_id"]
    approve = args["approve"]
    shutdown_requests[req_id]["status"] = "approved" if approve else "rejected"
    BUS.send(sender, "lead", args.get("reason", ""),
             "shutdown_response",
             {"request_id": req_id, "approve": approve})

对应 agents/s10_team_protocols.py 的 _exec 分支。这里有两个值得注意的行为：

1. 更新追踪表 + 回发消息是成对出现的。追踪表让 lead 用 shutdown_response 工具（在 lead 侧它是"查询状态"的 handler，见 L394 的 _check_shutdown_status）随时轮询进度，回发消息则让 lead 的收件箱收到正式应答。
2. approve 触发真正的退出。在 _teammate_loop 中，工具执行后有一段关键判断（agents/s10_team_protocols.py）：

if block.name == "shutdown_response" and block.input.get("approve"):
    should_exit = True

循环结束后，成员状态按退出原因落盘（L218-L221）：

member["status"] = "shutdown" if should_exit else "idle"
self._save_config()

这就解决了 s09 的问题：线程不是被杀死的，而是自己走到一个安全的退出点——当前 LLM 回合的工具执行完成后才 break，文件写完、状态落盘。这正是"优雅关停"（graceful shutdown）的含义。

拒绝路径同样成立：队友回复 approve=false 加一段 reason（比如"正在写半个文件"），追踪表落到 rejected，线程继续工作，lead 可以在稍后重试或换人。

计划审批协议（Plan Approval）：同一 FSM 的第二个应用

计划审批方向相反：队友先提交计划，lead 后审查。

队友侧：提交计划

plan_requests = {}

def handle_plan_review(request_id, approve, feedback=""):
    req = plan_requests[request_id]
    req["status"] = "approved" if approve else "rejected"
    BUS.send("lead", req["from"], feedback,
             "plan_approval_response",
             {"request_id": request_id, "approve": approve})

以上是文档中 lead 侧的审查实现（对应 agents/s10_team_protocols.py）。队友侧的提交在 _exec 的 plan_approval 分支（agents/s10_team_protocols.py）：

if tool_name == "plan_approval":
    plan_text = args.get("plan", "")
    req_id = str(uuid.uuid4())[:8]
    with _tracker_lock:
        plan_requests[req_id] = {"from": sender, "plan": plan_text, "status": "pending"}
    BUS.send(
        sender, "lead", plan_text, "plan_approval_response",
        {"request_id": req_id, "plan": plan_text},
    )
    return f"Plan submitted (request_id={req_id}). Waiting for lead approval."

从源码结构看，s10 的实现刻意保持最小：提交计划后队友线程并不会阻塞，模型只是"被告知在等待"（返回字符串提示等待 lead 审批）。是否真正阻止执行，依赖队友的系统提示词（Submit plans via plan_approval before major work，见 L177-L182）和模型的自觉——这是一个"协议层"的约束，而非"执行层"的强制。这一点在下一节的 s13 演进中会得到硬化的对照。

lead 收到 request_id 后（通过收件箱中的消息），用 plan_approval 工具审批，handle_plan_review 校验 request_id 存在（不存在时返回 Error: Unknown plan request_id '...'），更新状态并把带 feedback 的响应发回队友。

工具面：从 9 个工具到 12 个工具

文档的"What Changed From s09"表格如下：

组件	s09	s10
工具数	9	12（+shutdown_req/resp +plan）
关停	仅自然退出	请求-响应握手
计划门禁	无	提交/审查，需审批
关联	无	每个请求一个 request_id
FSM	无	pending → approved/rejected

对照 agents/s10_team_protocols.py 的 lead 侧 TOOL_HANDLERS 分发表，可以精确核对这 12 个工具：

TOOL_HANDLERS = {
    "bash":              lambda **kw: _run_bash(kw["command"]),
    "read_file":         lambda **kw: _run_read(kw["path"], kw.get("limit")),
    "write_file":        lambda **kw: _run_write(kw["path"], kw["content"]),
    "edit_file":         lambda **kw: _run_edit(kw["path"], kw["old_text"], kw["new_text"]),
    "spawn_teammate":    lambda **kw: TEAM.spawn(kw["name"], kw["role"], kw["prompt"]),
    "list_teammates":    lambda **kw: TEAM.list_all(),
    "send_message":      lambda **kw: BUS.send("lead", kw["to"], kw["content"], kw.get("msg_type", "message")),
    "read_inbox":        lambda **kw: json.dumps(BUS.read_inbox("lead"), indent=2),
    "broadcast":         lambda **kw: BUS.broadcast("lead", kw["content"], TEAM.member_names()),
    "shutdown_request":  lambda **kw: handle_shutdown_request(kw["teammate"]),
    "shutdown_response": lambda **kw: _check_shutdown_status(kw.get("request_id", "")),
    "plan_approval":     lambda **kw: handle_plan_review(kw["request_id"], kw["approve"], kw.get("feedback", "")),
}

前 9 个继承自 s09（基础文件工具 4 个 + 团队工具 5 个），新增 3 个即协议工具。注意一个非对称设计：同名工具在 lead 侧和队友侧语义不同——shutdown_response 对队友是"回复关停请求"（L237-L247），对 lead 是"查询关停请求状态"（_check_shutdown_status，L377-L379）；plan_approval 对队友是"提交计划"，对 lead 是"审批计划"。工具集按角色裁剪，是团队运行时里避免越权的重要手法。

纵深对照：当前主线 s13 中的协议硬化

仓库 README 明确了双轨结构：agents/ 与 docs/ 是旧版 12 课的遗留轨，根级 s01_*~s17_* 是当前主线，旧 s10（Team Protocols）的内容并入新 s13 Agent Teams。当前主线 s13_agent_teams/code.py 把 s10 的"最小协议"升级成了一套带校验的团队运行时，值得作为对照阅读（s13_agent_teams/README.md 的第 11、12 节专门讲这部分）。

统一的 ProtocolState 取代两个字典

s13 用一个 dataclass 统一两种协议状态（s13_agent_teams/code.py）：

@dataclass
class ProtocolState:
    request_id: str
    type: str            # "shutdown" | "plan_approval"
    sender: str
    target: str
    status: str          # pending -> approved | rejected
    payload: str
    work_version: int | None = None
    task_id: str | None = None
    created_at: float = field(default_factory=time.time)

pending_requests: dict[str, ProtocolState] = {}

type 字段让一个追踪表同时承载两种协议，而 work_version / task_id 额外记录了计划提交时队友的工作上下文——这是 s10 没有的：s13 要求审批响应必须与"当时那份计划所属的任务和工作版本"匹配。

match_response：四重校验替代盲信

s10 中 lead 收到响应即更新状态；s13 的 match_response（s13_agent_teams/code.py）在更新前做四重校验：

1. request_id 必须存在于 pending_requests；
2. 响应类型必须与请求类型匹配（shutdown 请求只接受 shutdown_response，计划请求只接受 plan_approval_response）；
3. 响应方向必须匹配（from_agent 是原请求的 target，to_agent 是原请求的 sender）——防止 A 代 B 回复；
4. 状态必须还是 pending——防止重复响应被应用两次。

任意一条不满足就打印 [protocol] ... 日志并拒绝。consume_lead_inbox()（L901-L911）在把 Lead 邮箱内容注入模型上下文之前先跑一遍 match_response，即"运行时负责状态机，模型只负责决策"。

计划门禁从"提示词约束"变为"工具分发层拦截"

这是 s10 与 s13 最实质的差异。s10 里"先有计划再动手"写在队友系统提示词里，靠模型自觉；s13 把它做到了工具执行路径上（s13_agent_teams/code.py）：

def _run_teammate_tool(name: str, block, handlers: dict) -> str:
    gate = plan_gates.get(name, "not_required")
    if block.name in {"bash", "write_file", "edit_file"}:
        if gate != "approved":
            if gate != "not_required":
                return (f"Blocked: plan status is {gate}. Submit or revise the "
                        "plan and wait for approval before changing the workspace.")
    ...

plan_gates 的取值是 not_required / required / pending / approved / rejected：只要不在 not_required 和 approved 两态，队友的 bash、write_file、edit_file 一律返回 Blocked。读文件和重新提交计划不受阻，所以被拒的队友可以改完计划再交。此外 run_review_plan（L1408-L1426）还会检查 work_version 与 task_id 是否变化，任务切换会使旧审批失效，需要重新提交计划。

测试用例如何验证这些协议

tests/test_agent_teams_runtime.py 用假 anthropic 模块加载 s13 课程代码（不真正调用 API），对协议行为做了可复现的断言，其中与本文主题直接相关的几条：

• test_plan_rejection_requires_a_new_submission（tests/test_agent_teams_runtime.py）：完整走一遍"lead 要求计划 → 队友提交（pending）→ lead 驳回（rejected）→ 队友应用响应 → 重新提交拿到新的 request_id"，验证被拒后必须重新提交而不是复用旧请求；
• test_mismatched_plan_response_cannot_release_gate（L895-L924）：构造一条 request_id 不匹配的伪造审批消息，断言 apply_plan_response 返回 [Ignored plan response: request mismatch] 且门禁保持 pending——即使 request_id 匹配，如果 lead 尚未在追踪表中把状态置为已审查，同样不会放行；
• test_shutdown_response_must_come_from_requested_teammate（L926-L942）：bob 冒名替 alice 回复 shutdown_response，断言请求状态保持 pending；
• test_shutdown_request_must_match_active_protocol（L944-L972）：未知 request_id 的关停请求被忽略；合法请求使队友状态变为 stopping，同一请求重放第二次不再生效（幂等）；
• test_plan_gate_blocks_mutating_tools_until_approval（L672-L703）：门禁为 pending 时 write_file/edit_file 被拦截且 handler 未被调用，approved 后正常执行。

这些测试覆盖了协议设计的三条不变量：身份校验（响应方必须是请求的目标）、类型校验（响应类型必须匹配请求类型）、幂等性（已决请求不可二次变更）。

动手运行 s10

运行方式直接继承原文档的 Try It 部分（旧版轨脚本 agents/s10_team_protocols.py 独立可运行，需先安装 requirements.txt 并配置 ANTHROPIC_API_KEY / MODEL_ID 环境变量；脚本通过 load_dotenv 读取 .env，并支持 ANTHROPIC_BASE_URL 覆盖 API 端点，见 agents/s10_team_protocols.py）：

cd learn-claude-code
python agents/s10_team_protocols.py

按文档给出的步骤操作：

1. Spawn alice as a coder. Then request her shutdown. ——观察 lead 调用 shutdown_request 返回 request_id，随后 alice 的线程打印 [alice] shutdown_response: ... 并优雅退出；
2. List teammates to see alice's status after shutdown approval ——config.json 中 alice 的状态应为 shutdown；
3. Spawn bob with a risky refactoring task. Review and reject his plan. ——观察 bob 提交计划后，lead 用 plan_approval（approve=false + feedback）驳回；
4. Spawn charlie, have him submit a plan, then approve it. ——完整走通 approve 路径；
5. 输入 /team 随时查看团队名册与状态；另有 /inbox 命令可查看 lead 收件箱（agents/s10_team_protocols.py）。

运行产物会落在工作目录的 .team/ 下：config.json（名册）与 inbox/*.jsonl（邮箱）。建议运行结束后直接查看这些文件，能直观理解"文件即协议介质"的设计——协议状态不依赖内存共享，任何进程都能通过邮箱行 JSON 追溯一次握手的全过程。

小结：一个模式，两个领域

s10 的核心贡献可以压缩成一句话（也即 agents/s10_team_protocols.py 文档字符串里的 "Key insight"）：

Same request_id correlation pattern, two domains.

• 请求方生成唯一 request_id，把 pending 状态登记进追踪表，经文件邮箱投递结构化请求；
• 响应方引用同一 request_id 回复 approve 布尔值，状态机落到 approved / rejected；
• shutdown 用它实现优雅关停（线程自己走到安全退出点，名册落盘），plan approval 用它实现高风险变更的先审后做。

从当前主线 s13_agent_teams/code.py 的 ProtocolState + match_response + plan_gates 可以看到这套模式的工程化方向：追踪表统一化、响应四重校验、把计划门禁从提示词层下沉到工具分发层。若要继续深入，建议按顺序阅读 docs/en/s10-team-protocols.md 的上一节 s09 Agent Teams 理解邮箱基础，再读 s13_agent_teams/README.md 第 11、12 节及其 code.py 对应实现，最后用 tests/test_agent_teams_runtime.py 中的协议测试自证理解。