learn-claude-code s13 精讲：Agent Teams 团队运行时与协调协议——持久队友、原子 Claim、worktree 与计划门禁

本文围绕 learn-claude-code 课程第 13 章文档 s13_agent_teams/README.ja.md 展开，讲解当单个 Agent 无法承载整份工作时的团队化方案：Lead 与持久 teammates 如何分工、如何通过文件信箱通信、如何对共享任务板做原子 Claim、如何用可选 worktree 隔离并行编辑，以及 shutdown 与计划批准如何成为可追踪、可强制执行的类型化协议。读完本文，你将理解 code.py 中团队运行时的完整设计，并能实际运行该课程验证每个机制。

1. 问题：单 Agent 无法承载的整块工作

假设让 Agent 重构整个后端，工作横跨配置加载、认证、测试等多个领域。单个 Agent 也能顺序完成，但耗时长，且早期细节会逐渐滑出上下文。这类工作天然适合并行化，但用户通常只描述目标而非设计团队：

Refactor this sample backend. Clean up configuration loading,
authentication, and tests, preserve the existing interfaces,
and make sure the tests pass.

因此 Harness 必须回答一组相互关联的问题：

1. 谁来判断并行工作有价值、谁来批准追加 Agent；
2. 每个 teammate 如何在多次任务分配之间保持身份与上下文；
3. 结果如何回到 Lead，而不要求模型轮询信箱；
4. IDLE 的 teammate 能否不等新分配就领取就绪任务；
5. 并行编辑可能冲突时，任务应该使用哪个工作目录；
6. shutdown 与计划批准如何成为可追踪、可强制的协议。

2. 解决方案总览：在 s10 基础上叠加 Lead 管理团队运行时

s13 复用 s10 的基础工具、Hooks、Permission 检查与 Task System，再叠加一层 Lead 管理团队运行时：

• Lead：拥有用户会话，提出分工方案并等待确认；
• Teammates：各自运行独立的 agent loop，在 WORK 与 IDLE 之间交替；
• MessageBus：通过文件信箱（.mailboxes/<name>.jsonl）承载普通消息、结果与控制事件；
• 运行时投递（Runtime delivery）：消费 Lead 的信箱，把团队事件注入下一轮对话；
• 共享任务板：让 IDLE 的 teammate 发现就绪工作并在锁下 Claim；
• 可选 worktree：需要时为任务绑定另一个工作目录，未绑定的任务使用常规仓库目录；
• 类型化协议与计划门禁：让 shutdown 与批准状态显式化，并在必需计划被批准前阻止变更类工具。

需要强调的两个设计边界：s11 的 background task 与 s12 的 scheduled task 没有被带入本章——两者对 teammate 通信、任务 Claim、计划批准都不是必需的；并且这些机制都是 Team Harness 层的一部分，teammate 不需要为任务发现另建循环，worktree 也不会创造一种新的 Agent。这一取舍在测试 tests/test_agent_teams_runtime.py 的 test_agent_teams_builds_on_tasks_not_background_or_cron 中也有验证。

3. 机制详解

3.1 Lead 提出团队方案并等待用户确认

启动 teammate 会改变成本、并发度以及“谁可以编辑工作区”这一主体集合。Lead 的系统提示词把这条边界写得很明确（见 code.py 的 PROMPT 部分）：

"When parallel work would help, first propose a small team with clear "
"responsibilities and wait for the user's confirmation. Do not call "
"spawn_teammate before the user confirms."

面对第一个请求，Lead 只给出分工草案：

I suggest three parallel areas:
- config: clean up configuration loading
- auth: refactor authentication
- tests: add regression coverage

I will start the teammates after you confirm.

用户回复“Go ahead”之后，Lead 才能调用 spawn_teammate。Lead 先创建 Task，再把初始 task_id 传给 teammate——用户陈述目标、Lead 设计团队、用户确认执行边界，三者职责分明。

3.2 每个 teammate 拥有独立的持久循环

s06 的 subagent 是一次性调用，teammate 则是持久执行单元：

	s06 Subagent	s13 Teammate
生命周期	一次调用后结束	直到 shutdown 前持续 WORK → IDLE → WORK
上下文	只存在于一个任务中	跨多次分配保持
通信	返回一次结果	接收消息并发出事件
协调	单向委派	与 Lead 双向协作

TeammateRuntime（code.py#L1085）为每个 teammate 维护专属的 system prompt、messages、handlers 与当前 Task，并在 daemon thread 中运行 WORK / IDLE 循环；teammate 工作时 Lead 可以继续协调。lead 与 agent 作为运行时身份被保留，MessageBus 仍接受 lead 作为协调者信箱；保留名校验对应测试 test_reserved_teammate_names_do_not_shadow_runtime_identities。

关键细节：spawn_teammate 在启动线程之前先 Claim 初始 Task，Claim 失败则 teammate 根本不启动（code.py#L1314-L1353，测试见 test_spawn_claims_the_initial_task_before_starting_the_thread）。没有 Task 时，workspace 与 Shell 工具会要求先 Claim，而不是回退到仓库目录。

3.3 MessageBus：把通信移出模型上下文

Lead 与 teammates 不能共享同一个 messages 数组，否则某个 teammate 的工具结果会污染另一个 teammate 的推理。MessageBus 为每个 Agent 准备 .mailboxes/<name>.jsonl 信箱（code.py#L783-L850）：

class MessageBus:
    def send(self, from_agent, to_agent, content,
             msg_type="message", metadata=None):
        msg = {
            "from": from_agent,
            "to": to_agent,
            "content": content,
            "type": msg_type,
            "metadata": metadata or {},
        }
        with self._changed:
            MAILBOX_DIR.mkdir(parents=True, exist_ok=True)
            with self._path(to_agent).open("a", encoding="utf-8") as handle:
                handle.write(json.dumps(msg, ensure_ascii=True) + "\n")
            self._changed.notify_all()

    def wait_for_messages(self, agent, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        with self._changed:
            while not self.peek(agent):
                remaining = (None if deadline is None
                             else deadline - time.monotonic())
                if remaining is not None and remaining <= 0:
                    return []
                self._changed.wait(remaining)
            return self._read_unlocked(agent)

锁保护信箱文件的并发访问；Condition 既能在消息到达时唤醒 teammate，也支持 IDLE 期间的短超时扫描。消息体是 JSONL：from、to、content、type（message / result / idle_notification / shutdown_request / plan_approval_request 等）、metadata（携带 request_id、approve）。

3.4 运行时投递信箱事件，模型不轮询

read_inbox() 通过读取并删除信箱文件来消费消息，因此 Lead 侧只保留单一消费者 consume_lead_inbox()（code.py#L901-L911）：

def consume_lead_inbox():
    messages = BUS.read_inbox("lead")
    for message in messages:
        if message["type"].endswith("_response"):
            match_response(...)
    return messages

CLI 主循环用 select 同时等待终端输入与 Lead 信箱（wait_for_cli_event，code.py#L1743-L1758，输入轮询间隔 0.25 秒）。消息到达后的投递链路：

MessageBus → consume_lead_inbox
           → 更新协议状态
           → 将 [Team events] 注入 history
           → 开启 Lead 的下一轮

Lead 在 spawn teammate 之后会直接结束当前 turn，而不是反复调用 list_teammates、get_task 等待；团队事件到达时由运行时开启下一轮。check_inbox 不是模型工具——消息到达属于运行时职责，模型只处理已被投递进上下文的事件。这一职责划分有专门测试 test_inbox_delivery_is_runtime_owned 保证。

3.5 result 与 IDLE 是两个独立事件

teammate 完成一次分配后，运行时按顺序发送两个事件（code.py#L1237-L1250）：

result:            "Authentication refactored; related tests pass."
idle_notification: "Waiting for more work."

result 回答“这个分配产出了什么”，idle_notification 回答“这个 teammate 现在能否接受新工作”。一个含糊的“完成”无法同时表达两种状态。IDLE 的 teammate 并不退出：收到直接消息或 ready task 会回到 WORK；收到 shutdown_request 则开始优雅关闭握手。

3.6 IDLE 优先查信箱，再找 ready task

IDLE 循环（code.py#L1252-L1276，IDLE_SCAN_INTERVAL = 2.0 秒）先等消息、后查任务板：

while True:
    inbox = BUS.wait_for_messages(name, IDLE_SCAN_INTERVAL)
    if inbox:
        should_stop = handle_messages(inbox)
        if should_stop or messages[-1]["role"] == "user":
            break
        continue

    task = claim_next_task(name)
    if task:
        messages.append({
            "role": "user",
            "content": f"[Auto-claimed task {task.id}] {task.subject}",
        })
        break

shutdown、计划批准、Lead 的直达指令都应当先于“顺手捡活”被处理。既无消息又无就绪任务时，teammate 保持 IDLE；被阻塞的任务也可能因另一 teammate 完成前置任务而转为 ready（依赖检查由 can_start(task_id) 基于 blockedBy 完成）。

3.7 发现与 Claim 分离，Claim 是原子操作

扫描只产生候选、不改变状态（code.py#L1056-L1067）：

def scan_unclaimed_tasks() -> list[Task]:
    return [
        task for task in list_tasks()
        if task.status == "pending"
        and task.owner is None
        and can_start(task.id)
    ]

候选列表只是某时刻的快照：另一个 teammate、甚至另一个使用同一任务目录的 Harness 进程，都可能看到同一个任务。因此所有权变更发生在 claim_task() 内，由 task_store_lock() 串行化——它组合了进程内 threading.RLock 与 fcntl.flock 文件锁（code.py#L70-L89）：

def claim_task(task_id: str, owner: str) -> str:
    with task_store_lock():
        task = load_task(task_id)
        if task.status != "pending" or task.owner is not None:
            return "Task is no longer available"
        if _owner_in_progress(owner):
            return "Owner must complete its current task first"
        if not can_start(task_id):
            return "Task is blocked"
        cwd, error = task_worktree_cwd(task)
        if error:
            return f"Cannot claim {task_id}: {error}"
        task.owner = owner
        task.status = "in_progress"
        save_task(task)
        teammate_assignments[owner] = {"task_id": task.id, "cwd": cwd}
        return f"Claimed {task.id}"

多个 teammate 可能发现同一候选，但只有一个 Claim 能把它推进到 in_progress。任务文件通过临时文件写入并在同一把 store lock 持有期间原子替换。teammate 必须完成当前任务才能领取下一个任务；worktree 绑定损坏时 Claim 失败关闭（fail closed），而不是悄悄回退到仓库目录。

3.8 被 Claim 的工作复用同一个 WORK 循环

Claim 成功后，运行时把任务 ID、主题、描述注入 teammate 的 messages（[Auto-claimed task ...]），随后状态流转为：

ready task appears
  → IDLE teammate discovers it
  → claim_task writes owner and in_progress
  → task enters teammate messages
  → WORK
  → complete_task
  → result + idle_notification
  → IDLE

teammate 在此复用与 Lead 直接分配完全相同的模型调用、文件工具、Shell、计划门禁、结果上报与 shutdown 协议。任务发现只是进入既有 WORK 循环的另一条入口——这正是“不需要为任务发现单独建循环”的含义。

3.9 由任务选择工具的工作目录

Task 的 worktree 是可选字段（code.py#L112-L120）：

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]
    worktree: str | None = None

需要把并行编辑分到其他目录时，Lead 可以创建并绑定 worktree：

create_worktree(name="auth-refactor", task_id="task_1a2b3c4d")

create_worktree 是 Lead 专属工具（code.py#L445-L522）：只接受 pending、无 owner、未绑定 worktree 的任务，依次校验名称（^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$、禁止 ..）、路径、分支（wt/<name>）与 Git 注册表；全部通过后才执行 git worktree add，checkout 创建成功后才写入任务绑定。如果 Git 失败但残留了分支或已注册 checkout，运行时报告 partial operation、保持任务未绑定，并把残留物保留供人工恢复。teammate 只拥有任务工具与文件工具。

Claim 时解析出的目录存入 teammate_assignments（code.py#L64-L67），该 teammate 的 bash、read_file、write_file、edit_file、glob wrapper 全部从 assignment 读取 cwd（code.py#L1129-L1155）；无 worktree 的任务解析到 WORKDIR，没有 Claim 过 Task 的 teammate 无法使用这些 workspace 工具（测试 test_teammate_workspace_tools_require_a_claimed_task）：

cwd, error = task_worktree_cwd(task)
if not error:
    teammate_assignments[owner] = {
        "task_id": task.id,
        "cwd": cwd,
    }

complete_task(task_id, owner) 会校验调用者确实是该 in-progress 任务的 owner。成功时记录结果，但不立即解除 assignment 目录——让同一模型轮次的后续 tool call 仍停留在该任务的目录中；teammate 回到 IDLE 时由运行时解除。失败则保留目录，便于修复后重试。进程重启后，assignment_cwd()（code.py#L390-L410）能基于持久化的 task owner 与 worktree 绑定重建 in-progress 的 assignment；binding 缺失或无效时同样失败关闭，不会悄悄把工作路由到仓库目录。

注意：worktree 隔离的是 Git 工作目录与分支，不是 sandbox——Shell 命令仍可访问父进程被允许的路径与资源。

3.10 worktree 删除权属于 host

模型可以创建 task-bound worktree，但不能删除它。清理保留为 host helper（remove_worktree(name, discard_changes)，code.py#L524-L606），以便 user 或 host 先检查任务所有权、assignment lease 与 Git 状态。helper 拒绝 pending / in-progress 的任务绑定与当前轮次的 lease；除非显式选择破坏性删除，tracked、untracked、ignored 文件都会阻止删除（测试 test_worktree_removal_is_host_only）。remove_worktree(name, discard_changes=True) 只能由已取得用户明确确认的 host 代码调用。两种删除路径都保留 wt/<name> 分支（包括无 upstream 的 clean 本地提交）；删除成功后解除任务绑定，因为 checkout 已不存在：

clean worktree   → host may remove directory and retain wt/<name> branch
changed worktree → user decides how to preserve or discard it
pending/running task → refuse removal

任务完成与 worktree 清理也是解耦的：complete_task 只记录任务结果；teammate 回到 IDLE 后，user 或 host 再决定检查、合并、保留还是删除 worktree。

3.11 控制消息使用类型与 request_id

普通协作可以用自由文本，但 shutdown 与批准不应依赖“猜意图”。它们使用结构化消息与显式状态机（code.py#L853-L898）：

@dataclass
class ProtocolState:
    request_id: str
    type: str
    sender: str
    target: str
    status: str
    payload: str
    work_version: int | None = None
    task_id: str | None = None


pending_requests: dict[str, ProtocolState] = {}

shutdown 的完整路径：

Lead creates a pending shutdown request
  → shutdown_request(request_id) enters the teammate inbox
  → the teammate finishes its current step
  → shutdown_response(request_id) returns to Lead
  → request_id locates the original request
  → pending becomes approved and the teammate loop exits

match_response() 做三重校验：request_id 把一个响应关联到一个请求；类型防止不匹配的响应改变状态（shutdown 只接受 shutdown_response，plan_approval 只接受 plan_approval_response）；status 防止同一响应被重复应用（非 pending 直接拒绝）；发送方/接收方与 sender/target 不一致也会被记录并丢弃。

3.12 计划批准不仅追踪，还约束执行

计划协议方向相反：

Lead → plan_request
teammate → plan_approval_request(request_id, plan)
Lead → plan_approval_response(request_id, approve, feedback)

若 Lead 一开始就知道某 teammate 必须先出计划，使用 spawn_teammate(..., task_id=task.id, require_plan=True)——运行时先 Claim Task、激活 gate，再启动 teammate 线程；对已在运行的 teammate 则可用 request_plan 补提要求。

工具 dispatch 在门禁处强制（code.py#L969-L982）。源码中 gate 的实际取值是 not_required / required / pending / rejected / approved，bash、write_file、edit_file 只有在前两者之外被放行：

def _run_teammate_tool(name, block, handlers):
    gate = plan_gates.get(name, "not_required")
    if block.name in {"bash", "write_file", "edit_file"}:
        if gate != "approved" and gate != "not_required":
            return f"Blocked: plan status is {gate}."
        # 之后还会走 check_permission 的 deny list 与路径检查
        ...

状态处于 required、pending、rejected 期间，teammate 可以读文件、可以提交或修订计划，但不能运行 Shell 命令、写文件、编辑文件（测试 test_plan_gate_blocks_mutating_tools_until_approval）。计划提交时记录 teammate 当前的 task 与 work version（assignment_versions）：Claim 或 release Task 会推进 version，从而使旧批准失效；普通 message 既不改变任务身份也不改变批准状态（测试 test_plain_message_does_not_change_assignment_or_plan_version）。Lead 端 run_review_plan 在批准前还会复核 work_version 与 task_id 是否仍与提交时一致，不一致会返回 “belongs to an earlier assignment”。另外，teammate 不从后台线程读取用户输入：危险命令或 workspace 外的路径直接返回 permission error，交由 Lead 与用户决策。

4. 一次完整运行示例

s13 >> Put the backend refactor on a shared task board. Clean up
       configuration, authentication, and tests in parallel where possible.
       Use a worktree for authentication, preserve existing interfaces,
       and make sure the tests pass.

Lead: I suggest config, auth, and tests as three areas.
      Shall I start the team?

s13 >> Go ahead.

[task] config created
[task] auth created → worktree auth-refactor
[task] tests created
[claim] alice → config (cwd: repository)
[claim] bob → auth (cwd: .worktrees/auth-refactor)
[teammate] alice spawned
[teammate] bob spawned
[complete] auth
[bus] bob → lead (result) ...
[bus] bob → lead (idle_notification) ...
[wake: 2 team events → new turn]
Lead: I received the authentication result and will coordinate the rest.

终端完整暴露了：用户请求、Lead 提案、任务状态、Claim、所选目录、结果、IDLE 迁移与控制事件。用户不需要指名 Lead，也不需要请求它“检查一下信箱”——投递全部由运行时完成。

5. 与 s10 的差异对照

组件	s10	s13
Agent	单个 Agent	1 个 Lead + 持久 teammates
用户流程	直接执行请求	先提出团队方案，再确认启动
通信	无	文件信箱 + 运行时投递
生命周期	一个循环	teammate 的 WORK / IDLE / shutdown
共享工作	一个 Agent 使用任务工具	IDLE 扫描 + teammate 原子 Claim
工作目录	仓库 WORKDIR	已 Claim 的 Task，可选 worktree
结果通知	当前 Agent 的输出	result 与 idle_notification 分离
控制	无	类型化 shutdown 与计划批准协议
强制	无团队约束	必需计划会门控变更类工具

工具面上，Lead 可用 spawn_teammate（name 限 ^[A-Za-z0-9_-]{1,64}$、task_id 限 ^task_[0-9a-f]{8}$）、list_teammates、send_message、request_shutdown、request_plan、review_plan、create_worktree，外加全部 TASK_TOOLS；teammate 的 TEAMMATE_TOOLS 则是基础文件/Shell 工具加 send_message、submit_plan、list_tasks、claim_task、complete_task（code.py#L1493-L1562）。

6. 动手运行

运行方式（需要 pip install anthropic python-dotenv，并在 .env 中配置 ANTHROPIC_API_KEY；程序读取环境变量 MODEL_ID，且要求当前目录是 Git 仓库根，create_worktree 依赖 git worktree）：

cd learn-claude-code
python s13_agent_teams/code.py

输入一条普通请求：

Put the backend refactor on a shared task board. Complete configuration,
authentication, and tests in parallel where dependencies allow. Use a
worktree for authentication, preserve existing interfaces, and summarize
the result.

Lead 提出团队方案后，回复：

Go ahead.

观察点：.tasks/ 中状态从 pending 到 in_progress 再到 completed 的流转；.mailboxes/ 中 result 与 idle_notification 的投递；.worktrees/ 只为绑定了 worktree 的任务生成。还可以验证：直接消息优先于任务板扫描；complete_task 失败后 teammate 的工作目录不会被重置。

7. 本章定位与后续

code.py 中 Lead 与 teammate 能调用的，只有直接定义在文件里的工具（TOOL_HANDLERS 注册表）。要连接 Jira、部署平台或知识库，仍需要为每个外部系统单独编写 tool schema 与 handler，外部工具的增删改也要同步修改课程代码。这正是下一章的动机：s14 MCP Tools 将通过统一的发现与调用协议，在运行时接入外部服务并将其工具加入工具池（参见 s14_mcp_plugin/README.md）。

相关资源：s13 英文版文档、s13 实现代码、团队运行时测试、s10 任务系统文档。