learn-claude-code s13 深解：Agent Teams 的 Lead-Teammate 运行时、任务认领与协调协议

当单个 Agent 无法独自承载一整个任务时，learn-claude-code 课程在 s13 引入「Agent 团队」：一个 Lead 负责与用户对话、拆解工作并确认执行边界，多个常驻 Teammate 各自运行独立的 Agent 循环，通过文件邮箱（.mailboxes/）交换消息、通过共享任务板（.tasks/）原子认领工作、通过可选的 Git worktree（.worktrees/）隔离并行编辑。本文基于章节文档 s13_agent_teams/README.md 和约 1800 行的完整实现 s13_agent_teams/code.py 展开，覆盖团队启动确认、WORK/IDLE 生命周期、消息投递、原子认领、worktree 绑定与类型化协议的全部细节，读完即可理解多 Agent 协作中「状态归运行时、模型只处理事件」这一核心设计。

问题：一个 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. 空闲的 Teammate 能否不等指派就主动领取就绪任务？
5. 并行编辑可能冲突时，任务应使用哪个目录？
6. 关闭与计划审批如何变成可追溯、可强制的协议？

s13 复用 s10 的基础工具、hooks、权限检查与 Task System（参见 s10_task_system/README.md），在此之上增加由 Lead 管理的团队运行时：

• Lead 拥有与用户的对话，提出分工方案并等待确认；
• Teammate 运行独立的 Agent 循环，在 WORK 与 IDLE 之间交替；
• MessageBus 通过文件邮箱承载普通消息、结果与控制事件；
• 运行时投递消费 Lead 的邮箱，把团队事件注入下一轮；
• 共享任务板让空闲 Teammate 在锁保护下发现并认领就绪任务；
• 可选 worktree 把任务绑定到另一个工作目录，未绑定的任务使用正常仓库目录；
• 类型化协议与计划门（plan gate） 让关闭与审批状态显式化，并在计划获批前拦截变更类工具。

需要说明：s11 后台任务与 s12 定时任务没有被带入本章——Teammate 通信、任务认领与计划审批都不依赖这两类机制。这些组件同属 Team harness 层：Teammate 不需要额外的任务发现循环，worktree 也不会产生新种类的 Agent。

团队启动：Lead 提出方案，用户确认边界

启动 Teammate 会改变成本、并发度以及可编辑工作区的角色集合。s13 把这条边界直接写进 Lead 的系统提示词（code.py 中 PROMPT_SECTIONS 的 "teams" 段）：

"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 设计团队，用户确认执行边界。提示词同时要求 Lead 在 spawn 之后结束当前轮次、不要轮询 Teammate 状态——团队事件由运行时投递并唤醒 Lead（这一点在 code.py 主循环 中落实）。

Teammate：从一次性子调用到常驻执行单元

s06 的 subagent 是一次性调用，而 Teammate 是常驻执行单元：

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

TeammateRuntime（code.py）给每个 Teammate 独立的 system prompt、messages、工具集与当前 Task，并在守护线程（threading.Thread(..., daemon=True)）中运行 WORK/IDLE 循环（见 spawn_teammate_thread），Lead 因此可以在 Teammate 工作期间继续协调。

两条源码层面的约束值得注意：

• 名字保留：lead 与 agent 被 RESERVED_TEAMMATE_NAMES 保留为运行时身份，不能作为 Teammate 名字（code.py），对应测试 test_reserved_teammate_names_do_not_shadow_runtime_identities；
• 认领先于启动：spawn_teammate 在线程启动之前就认领初始 Task（claim_task(task_id, owner=name)），认领失败会回滚注册并阻止线程启动，测试 test_spawn_claims_the_initial_task_before_starting_the_thread 固化了这一行为。没有 Task 的 Teammate 使用工作区与 Shell 工具时，会收到「请先认领一个 Task」的错误，而不是回退到仓库目录（测试 test_teammate_workspace_tools_require_a_claimed_task）。

Teammate 的初始 user 消息会携带 [Assigned task ...] 与工作目录，并在 require_plan=True 时追加 [Plan required] 指令（TeammateRuntime.init）；其系统提示还明确告知：文件与 Shell 工具使用 Task 的工作目录，该目录不是沙箱；最终文本由运行时投递给 Lead，send_message 只用于中间协调。

MessageBus：把通信放在模型上下文之外

Lead 和 Teammate 不能共享同一个 messages 数组——否则一个 Teammate 的工具结果会泄漏进另一个 Teammate 的推理。MessageBus（code.py）为每个 Agent 提供 .mailboxes/<name>.jsonl 收件箱：

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)

从源码看有两个工程细节：

• 收件人是受检资源：_path() 用正则 ^[A-Za-z0-9_-]{1,64}$ 校验名字，并要求解析后的路径仍位于 .mailboxes/ 目录内，测试 test_message_bus_rejects_unregistered_or_unsafe_recipients 验证了非法或越界收件人的拒绝；
• 锁 + Condition：一把 RLock 保护邮箱文件免受并发读写；Condition 让运行时在有消息时唤醒 Teammate，同时支持 IDLE 时的短超时等待（wait_for_messages）。

消息实际落盘时还会带上时间戳（"ts": time.time()），并在终端打印 [bus] from -> to: (type) ... 摘要，这正是文档「One Complete Run」中 [bus] bob -> lead (result) 这类行的来源。

运行时投递：Lead 保持单一消费者

read_inbox() 的语义是「读取并删除邮箱文件」（破坏性读取），所以 Lead 侧只保留一个消费者 consume_lead_inbox()：

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

实现见 code.py：凡是带 request_id 且类型以 _response 结尾的消息，都会先走 match_response() 更新协议状态，再原样返回。CLI 主循环用 select 同时等待终端输入和 Lead 邮箱（wait_for_cli_event()，stdin 轮询间隔 0.25 秒）：一旦 BUS.peek("lead") 为真就返回 "wake"，主循环随即消费邮箱、用 format_team_events() 生成 [Team events] 注入历史，并开启新的一轮 Lead（主循环 中的 [wake: N team event(s) -> new turn] 输出）：

MessageBus → consume_lead_inbox
           → update protocol state
           → inject [Team events] into history
           → start another Lead turn

spawn 之后 Lead 结束当前轮次，不再反复调用 list_teammates 或 get_task；下一轮由团队事件到达时由运行时启动。check_inbox 因此不是模型工具——消息到达属于运行时职责，模型只在运行时把事件送进上下文之后处理它们。测试 test_inbox_delivery_is_runtime_owned 明确断言这一点。

结果与 IDLE 是两类事件

Teammate 完成一次指派后，运行时按顺序发送两个事件：

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

result 回答「这次指派产出了什么」，idle_notification 回答「这个 Teammate 现在能不能接活」。一个含糊的 "done" 无法同时表达两个事实。对应实现是 TeammateRuntime.work() 的收尾段（code.py）：取最后一轮的 assistant 文本，非空则发送 result；若计划门处于 pending，状态置为 waiting_approval，否则调用 release_completed_assignment()、置状态 idle 并发送 idle_notification。

空闲不等于退出：一条直接消息或一个就绪任务都能把 Teammate 拉回 WORK；shutdown_request 则触发优雅关闭握手（见下文协议一节）。

IDLE 循环：先查邮箱，再找就绪任务

IDLE 阶段消息优先，其次才是共享任务板（wait_for_work）：

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

代码中 IDLE_SCAN_INTERVAL = 2.0 秒（code.py）：每次等待邮箱最多 2 秒，超时后去任务板扫一遍。这样设计的原因很直接——来自 Lead 的 shutdown、计划审批与直接指令应当先于机会主义干活到达；没有消息也没有就绪任务时，Teammate 保持 IDLE。一个被依赖阻塞的任务可能在别的 Teammate 完成前置任务后转为就绪，complete_task 成功时也会列出新解锁的任务（code.py 的 Unblocked: 提示）。

发现与认领分离，认领是原子的

扫描只产生候选（scan_unclaimed_tasks()）：

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() 下完成——该锁组合了进程内 RLock 与 fcntl.flock 文件锁（task_store_lock），跨线程与跨进程都能串行化：

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}"

源码中的完整 claim_task 还比文档片段多做两件事：拒绝在本地仍有当前指派（teammate_assignments）时二次认领；以及认领成功后调用 advance_assignment_version(owner) 推进工作版本（用于使旧计划审批失效，见协议一节）。任务文件通过「写临时文件 + os.replace 原子替换」持久化（save_task），且全程持有同一把 store 锁。

由此得到的并发性质是：很多 Teammate 可能发现同一个候选，但只有一个认领能把它推进 in_progress；一个 Teammate 必须先完成当前任务才能认领下一个；worktree 绑定损坏时fail closed，而不是回退到仓库目录。

认领后的工作复用同一条 WORK 回路

认领成功后，运行时把任务 ID、主题、描述与工作目录注入 Teammate 的消息（[Auto-claimed task ...]），然后走与 Lead 直接指派完全相同的 WORK 回路：

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 使用的是同一套模型调用、文件工具、Shell、计划门、结果汇报与关闭协议——任务发现只是既有 WORK 回路的另一个入口。其工具集由 TEAMMATE_TOOLS 定义：5 个基础工具（bash/read_file/write_file/edit_file/glob）+ send_message + submit_plan + list_tasks/claim_task/complete_task，共 10 个；而 Lead 的 TOOLS 是 5 个基础工具 + 5 个任务工具 + 7 个团队工具（spawn_teammate、list_teammates、send_message、request_shutdown、request_plan、review_plan、create_worktree），共 17 个。

任务决定工具的默认目录：可选 worktree

Task 结构中的 worktree 字段是可选的（code.py）：

@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，其校验链非常严格：任务必须 pending、无主、未绑定；worktree 名字必须匹配 ^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$ 且不得含 ..；目标分支 wt/<name> 不得已存在；工作目录必须是 Git 仓库根；.worktrees/<name> 路径不得已存在，且该路径不能已在 Git worktree 注册表中。全部通过后才执行 git worktree add -b wt/<name> .worktrees/<name> HEAD 并写入任务绑定。

对部分失败的处置是显式的「fail closed + 保留现场」：如果 Git 在留下分支或已注册 checkout 之后才报错，运行时返回 Partial operation 报告，任务保持未绑定，不删除任何 Git 数据，提示用户手动 git worktree list 检查并处置；反过来，如果 checkout 创建成功但任务绑定写入失败，则返回 Partial success 并保留 checkout 供手动恢复。

认领成功后，解析出的目录被存入 teammate_assignments[owner] = {"task_id": ..., "cwd": ...}；该 Teammate 的 bash、read_file、write_file、edit_file、glob 包装器（TeammateRuntime.current_cwd）每次都从这个注册表取目录。没有 worktree 的任务解析到 WORKDIR；没有已认领 Task 的 Teammate 根本不能使用这些工作区工具。

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

complete_task(task_id, owner)（code.py）会校验调用者确实拥有这个 in_progress 任务；成功完成只记录结果，但指派目录保留到该模型轮次结束（work() 收尾时才调用 release_completed_assignment），这让同一响应里的后续工具调用仍停留在任务的 worktree 中。运行时在 Teammate 回到 IDLE 时释放指派；而 complete_task 失败则保留指派，让 Teammate 可以修复任务后重试。重启之后，assignment_cwd() 能从持久的任务 owner 与 worktree 绑定重建进行中的指派，并在同一 owner 已换任务时替换过期的本地租约；绑定缺失或非法时同样 fail closed，而不是把活儿悄悄路由到仓库目录。

注意：worktree 只隔离 Git 工作目录与分支，它不是沙箱——Shell 命令仍能访问父进程被允许的路径与资源。s13 的权限检查（DENY_LIST 与破坏性命令关键词，见 code.py）依然生效，且 Teammate 后台线程不读用户输入：危险命令或工作区外路径会返回权限错误，交由 Lead 与用户决策。

worktree 的移除权留在宿主持有

模型可以创建任务绑定的 worktree，但不能删除它。清理是宿主持有者（host）的辅助能力（remove_worktree），让用户或宿主先检查任务所有权、指派租约与 Git 状态。该辅助函数拒绝 pending 或 in_progress 的任务绑定以及当前轮次的租约；没有显式破坏性选择时，任何 tracked、untracked 或 ignored 文件都会阻止移除（通过 git status --porcelain --ignored 判断）。测试 test_worktree_removal_is_host_only 固化了这一边界。

remove_worktree(name, discard_changes=True) 仅留给已取得用户显式确认的宿主代码。无论哪条路径，wt/<name> 分支都会保留——包括没有上游的干净本地提交；成功移除会清除任务绑定，因为 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 之后，用户或宿主可以检视、合并、保留或移除 worktree。

类型化协议：关闭与审批不靠猜意图

自由文本适合日常协作，但关闭与审批不应依赖意图猜测。s13 使用结构化消息，核心是 ProtocolState：

@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] = {}

（源码中还有 created_at 时间戳字段。）关闭路径的完整时序：

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

三个字段各司其职：request_id 把一个回复关联到一个请求；type 防止错位回复改变状态（match_response 校验 shutdown ↔ shutdown_response、plan_approval ↔ plan_approval_response 的配对，以及应答者/接收者方向）；status 防止重复响应被应用两次（match_response）。Teammate 侧的 apply_shutdown_request 只接受「来自 lead、发给自己、pending 状态、自己未处于 stopping」的请求，随后置 stopping、回发 shutdown_response 并退出循环；run() 的 finally 块会清理注册表、释放指派租约（code.py）。

计划审批：在执行层强制「先规划后动手」

计划协议方向相反（run_request_plan / _teammate_submit_plan / run_review_plan）：

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) 会在 Teammate 线程启动前就认领 Task 并激活门（plan_gates[name] = "required"）；request_plan 则可以给已在运行的 Teammate 追加计划要求（直接把 plan_gates[teammate] 置为 required 并发送 plan_request）。

工具分发在门控上强制执行（文档版片段）：

def _run_teammate_tool(name, block, handlers):
    gate = plan_gates.get(name, "not_required")
    if block.name in {"bash", "write_file", "edit_file"} and gate not in {
        "not_required", "approved"
    }:
        return f"Blocked: plan status is {gate}."
    try:
        return handlersblock.name
    except Exception as error:
        return f"Error: {type(error).__name__}: {error}"

源码实现（_run_teammate_tool）与之等价并叠加了 s04 的 hooks 与权限检查。门控状态机为：not_required → required → pending → approved / rejected。状态处于 required、pending、rejected 时，Teammate 可以读文件、提交或修改计划，但不能执行 Shell 命令、写文件或编辑文件——测试 test_plan_gate_blocks_mutating_tools_until_approval 验证了拦截，test_plan_rejection_requires_a_new_submission 验证拒绝后必须重新提交。

一个容易忽略的细节是版本失效：提交计划时会记录 Teammate 当前的任务 ID 与工作版本（work_version/task_id 写入 ProtocolState）。认领或释放 Task 会经由 advance_assignment_version()（code.py）递增版本、清掉旧计划请求并把非 not_required 的门重置回 required，从而使旧审批失效；普通消息则既不改变任务身份也不改变审批状态（测试 test_plain_message_does_not_change_assignment_or_plan_version）。Lead 侧 review_plan 在批准前同样复核 work_version 与 task_id（code.py），Teammate 侧 apply_plan_response 要求请求 ID、方向、类型、状态与 approve 标志全部匹配才放行，错位响应无法解除门控（测试 test_mismatched_plan_response_cannot_release_gate）。

Teammate 不读用户输入：危险命令或工作区外路径会返回权限错误，由 Lead 与用户共同处理。

一次完整运行

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 的方案、任务状态、认领、所选目录、结果、IDLE 迁移与控制事件——用户不需要点名 Lead，也不需要要求它检查收件箱。

与 s10 的差异总览

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

动手运行

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

运行前提（见 code.py 文件头注释）：pip install anthropic python-dotenv，并在 .env 中配置 ANTHROPIC_API_KEY；模型 ID 从环境变量 MODEL_ID 读取（支持 ANTHROPIC_BASE_URL 指向兼容端点）。

从一个普通请求开始：

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/ 只为绑定任务出现。再重点核对两条不变量：直接消息优先于任务板扫描（IDLE 先查邮箱）；complete_task 失败时不会重置 Teammate 的工作目录。可进一步参考 tests/test_agent_teams_runtime.py 中覆盖上述行为的用例，以及中文/日文版章节文档 README.zh.md、README.ja.md。

小结与下一步

s13 给出的答案可以浓缩成三句话：状态归运行时（邮箱、任务板、协议状态都在进程与文件系统层面，由锁串行化），事件驱动轮次（Lead 不轮询，团队事件唤醒新轮次），约束在工具分发层强制执行（认领前置、计划门、fail closed 的目录解析）。这些机制都不依赖 s11 后台任务或 s12 定时任务，是 s10 Task System 之上的纯粹「团队层」扩展。

Lead 与其 Teammates 目前只能调用 code.py 中直接定义的工具；要接入 Jira、部署平台或知识库，仍需要为每个外部系统单独编写工具 schema 与 handler。下一章 s14 的 MCP 工具（s14_mcp_plugin/README.md）将展示如何通过一套发现与调用协议在运行时接入外部服务，并把它们的工具加入工具池。