learn-claude-code s09 解析：用持久化队友与 JSONL 邮箱构建 Agent 团队运行时

本文围绕 learn-claude-code 教程第 09 章（Agent Teams）展开，讲解如何在单个进程内构建"领导 + N 个持久化队友"的多 Agent 协作系统：每个队友在独立线程里运行完整 agent loop，通过磁盘上的 JSONL 收件箱互相通信，状态持久化在 config.json 中。读完后，你将掌握持久化 Agent 的生命周期管理、基于文件的团队邮箱设计，以及如何在现有 harness 中安全地引入多 Agent 并行。

问题：为什么一次性 Subagent 不够

要理解 s09 解决的问题，需要先回顾前序章节的两种能力边界：

• Subagent（s04/s06） 是一次性的：生成、干活、返回摘要、消亡。它没有身份，没有跨调用的记忆，无法被"再找一次"。
• Background Tasks（s08） 能后台跑 shell 命令，但执行的是固定命令，做不了需要 LLM 持续引导的决策。

真正的团队协作需要三样东西：(1) 能跨多轮对话存活的持久 Agent；(2) 身份和生命周期管理；(3) Agent 之间的通信通道。s09 用"持久化队友 + 文件邮箱"这套最朴素但完整的机制回答了这三个问题。源码文件头部的注释（agents/s09_agent_teams.py）把两种 Agent 的对比写得很直白：

Subagent (s04):  spawn -> execute -> return summary -> destroyed
Teammate (s09):  spawn -> work -> idle -> work -> ... -> shutdown

总体架构：团队目录布局与通信模型

整个团队系统只依赖一个目录 .team/，其中保存名册和所有收件箱：

Teammate lifecycle:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Communication:
  .team/
    config.json           <- team roster + statuses
    inbox/
      alice.jsonl         <- append-only, drain-on-read
      bob.jsonl
      lead.jsonl

通信模型的核心是 append-only + drain-on-read（读后清空）：发送方只追加一行 JSON，接收方在自己的 agent loop 里读取全部消息并清空文件。源码中目录的定义见 agents/s09_agent_teams.py：

WORKDIR = Path.cwd()
TEAM_DIR = WORKDIR / ".team"
INBOX_DIR = TEAM_DIR / "inbox"

这意味着通信完全不经过内存中的共享队列，而是走文件系统——队友之间甚至理论上可以跨进程协作，因为收件箱本身就是持久化状态。

MessageBus：JSONL 收件箱实现

通信通道由 MessageBus 类实现，它只有三个方法（agents/s09_agent_teams.py）：

send：向收件箱追加一行

def send(self, sender: str, to: str, content: str,
         msg_type: str = "message", extra: dict = None) -> str:
    if msg_type not in VALID_MSG_TYPES:
        return f"Error: Invalid type '{msg_type}'. Valid: {VALID_MSG_TYPES}"
    msg = {
        "type": msg_type,
        "from": sender,
        "content": content,
        "timestamp": time.time(),
    }
    if extra:
        msg.update(extra)
    inbox_path = self.dir / f"{to}.jsonl"
    with open(inbox_path, "a") as f:
        f.write(json.dumps(msg) + "\n")
    return f"Sent {msg_type} to {to}"

每行是一条独立的 JSON 记录，包含 type、from、content、timestamp 四个字段，extra 可以合并任意附加字段（如协议请求的 request_id）。注意 msg_type 受 VALID_MSG_TYPES 白名单约束，非法类型直接返回错误字符串而不是抛异常——错误会被作为工具结果回传给模型，模型有机会自我纠正。

read_inbox：读取并清空（drain）

def read_inbox(self, name: str, clear: bool = True) -> list:
    inbox_path = self.dir / f"{name}.jsonl"
    if not inbox_path.exists():
        return []
    messages = []
    for line in inbox_path.read_text().strip().splitlines():
        if line:
            messages.append(json.loads(line))
    if clear:
        inbox_path.write_text("")
    return messages

clear=True 时读完即清空，保证每条消息只被消费一次；clear=False 则只读不清空，这正是 CLI 中 /inbox 命令使用的模式——你可以反复查看领导的收件箱而不破坏消息。

broadcast：群发到所有队友

def broadcast(self, sender: str, content: str, teammates: list) -> str:
    count = 0
    for name in teammates:
        if name != sender:
            self.send(sender, name, content, "broadcast")
            count += 1
    return f"Broadcast to {count} teammates"

广播并不引入新机制，只是循环调用 send 并跳过发送者自己。

五种消息类型：为后续协议预留

VALID_MSG_TYPES 声明了五种消息类型（agents/s09_agent_teams.py）：

类型	用途
message	普通文本消息
broadcast	群发给所有队友
shutdown_request	请求平滑关机（s10 实现处理）
shutdown_response	同意/拒绝关机（s10 实现处理）
plan_approval_response	同意/拒绝计划审批（s10 实现处理）

源码注释明确说"5 message types (all declared, not all handled here)"——s09 只保证这些类型可以合法收发，真正的请求/响应握手协议（request_id 匹配、状态机）在 s10 中完成。这种"先声明类型、再分期实现处理逻辑"的做法让邮箱的线格式（wire format）保持稳定，后续章节无需改动存储层。

TeammateManager：名册、spawn 与生命周期

TeammateManager 负责团队名册和队友线程（agents/s09_agent_teams.py）：

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.dir.mkdir(exist_ok=True)
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads = {}

名册持久化在 .team/config.json，初始结构为 {"team_name": "default", "members": []}（见 _load_config，agents/s09_agent_teams.py）。每个成员记录包含 name、role、status 三个字段，status 取值 working / idle / shutdown。

spawn：创建队友并启动线程

def spawn(self, name: str, role: str, prompt: str) -> str:
    member = self._find_member(name)
    if member:
        if member["status"] not in ("idle", "shutdown"):
            return f"Error: '{name}' is currently {member['status']}"
        member["status"] = "working"
        member["role"] = role
    else:
        member = {"name": name, "role": role, "status": "working"}
        self.config["members"].append(member)
    self._save_config()
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt),
        daemon=True,
    )
    self.threads[name] = thread
    thread.start()
    return f"Spawned '{name}' (role: {role})"

三个关键设计值得注意：

1. 重名即复用而非新建。如果 alice 已存在且处于 idle 或 shutdown 状态，再次 spawn 会复用其名册记录、更新角色并重新置为 working——这就是"持久化"的含义：身份跨调用保留，只有状态在流转。正在 working 的队友重复 spawn 会收到错误提示，避免并发冲突。
2. 先落盘再启动线程。_save_config() 在 thread.start() 之前执行，即使线程立即异常，名册也始终与磁盘一致。
3. daemon=True。队友线程是守护线程，主进程退出时不会阻塞等待队友收尾。

队友的 Agent Loop：每轮先读收件箱

每个队友在自己的线程里运行一个完整的 agent loop（agents/s09_agent_teams.py）：

def _teammate_loop(self, name: str, role: str, prompt: str):
    sys_prompt = (
        f"You are '{name}', role: {role}, at {WORKDIR}. "
        f"Use send_message to communicate. Complete your task."
    )
    messages = [{"role": "user", "content": prompt}]
    tools = self._teammate_tools()
    for _ in range(50):
        inbox = BUS.read_inbox(name)
        for msg in inbox:
            messages.append({"role": "user", "content": json.dumps(msg)})
        try:
            response = client.messages.create(
                model=MODEL, system=sys_prompt,
                messages=messages, tools=tools, max_tokens=8000,
            )
        except Exception:
            break
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            break
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = self._exec(name, block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output),
                })
        messages.append({"role": "user", "content": results})
    member = self._find_member(name)
    if member and member["status"] != "shutdown":
        member["status"] = "idle"
        self._save_config()

要点：

• 收件箱注入位置：每次 LLM 调用之前先 BUS.read_inbox(name)，有新消息则以 user 角色逐条追加进 messages。文档中的 <inbox>...</inbox> 包装在 lead 侧使用（见下文 agent_loop），队友侧直接以 JSON 序列化形式注入——两者效果相同，都是把文件事件转化为对话上下文。
• 循环上限 50 轮，防止单个队友无限消耗 API 额度。
• 状态收尾：循环退出（无工具调用、API 异常或达到上限）后，若状态不是 shutdown 就写回 idle。这就是文档中"每次 LLM 调用前检查收件箱"所支撑的 WORKING -> IDLE -> WORKING 生命周期。
• 队友也有身份化的 system prompt："You are 'alice', role: coder, at /path..."，配合 s04 子 Agent 的思路，但队友的身份在 config.json 中是持久的。

Lead 侧：9 个工具与工具分发

领导（当前主线程）拥有 9 个工具，比 s08 多出 spawn_teammate、send_message、read_inbox 中的前两者以及 list_teammates、broadcast（agents/s09_agent_teams.py）：

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()),
}

注意 lead 发送消息时 from 字段硬编码为 "lead"——发送者身份由分发层注入而非由模型填写，模型无法冒充其他队友说话。

主循环 agent_loop（agents/s09_agent_teams.py）在每一轮 LLM 调用前消费 lead 自己的收件箱：

def agent_loop(messages: list):
    while True:
        inbox = BUS.read_inbox("lead")
        if inbox:
            messages.append({
                "role": "user",
                "content": f"<inbox>{json.dumps(inbox)}</inbox>",
            })
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS, max_tokens=8000,
        )
        ...

这形成一个自然的消息泵：lead 每等待/处理一轮用户输入，都会顺带把队友发来的消息注入上下文，因此队友的结果无需轮询即可被"看见"。

队友的工具集与权限边界

队友只有 6 个工具（agents/s09_agent_teams.py）：bash、read_file、write_file、edit_file、send_message、read_inbox。对比 lead 的 9 个，队友不能 spawn 新队友、不能群发、不能列出名册——团队拓扑是星型的，只有 lead 能改变它。

_exec 分发中还有一处细节（agents/s09_agent_teams.py）：send_message 和 read_inbox 的 sender 参数都传当前队友名 name，同样由运行时注入，防止身份伪造。

内建的安全护栏

s09 继承了 s02 的基础工具并保留了两道护栏：

• 路径逃逸检查 _safe_path：read_file/write_file/edit_file 先把相对路径 resolve 到 WORKDIR 下，不在工作区内则抛错（agents/s09_agent_teams.py）。
• 危险命令黑名单 _run_bash：拦截 rm -rf /、sudo、shutdown、reboot 等模式，超时 120 秒，输出截断到 50000 字符（agents/s09_agent_teams.py）。

在多 Agent 场景下这些护栏更重要，因为现在有多个并行执行者共享同一个 shell。

相对 s08 的变更

组件	之前 (s08)	之后 (s09)
Tools	6	9 (+spawn/send/read_inbox)
Agent 数量	单一	领导 + N 个队友
持久化	无	config.json + JSONL 收件箱
线程	后台命令	每线程完整 agent loop
生命周期	一次性	idle -> working -> idle
通信	无	message + broadcast

这张表浓缩了 s09 的增量：s08 的线程里跑的是无状态 shell 命令，s09 的线程里跑的是有身份、有持久状态、可被再次唤醒的完整 agent loop。

试一试：运行与验证

环境准备

依赖来自 requirements.txt：

pip install anthropic python-dotenv pyyaml

程序通过 load_dotenv(override=True) 加载环境变量（agents/s09_agent_teams.py），运行时至少需要 MODEL_ID；若使用兼容端点，还可设置 ANTHROPIC_BASE_URL（设置该变量时代码会自动移除 ANTHROPIC_AUTH_TOKEN 以避免认证冲突）。

启动

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

启动后进入 s09 >> 交互式提示符，输入 q 或 exit 退出。

建议的验证 prompt

1. Spawn alice (coder) and bob (tester). Have alice send bob a message. —— 验证 spawn + 点对点消息；
2. Broadcast "status update: phase 1 complete" to all teammates —— 验证广播（观察返回的 Broadcast to N teammates）；
3. Check the lead inbox for any messages —— 验证 lead 收件箱读取。

两个内建命令

• /team：调用 TEAM.list_all()，打印团队名册及每个成员的 name (role): status；
• /inbox：调用 BUS.read_inbox("lead", False)——注意 clear=False，查看不清空，消息仍会在下一轮 lead 调用时被正常消费。

观察什么

运行期间留意三点：终端上 [alice] tool_name: ... 前缀的行是队友线程的实时动作（队友工具执行时会打印，见 agents/s09_agent_teams.py）；.team/config.json 中各成员的 status 从 working 变为 idle 的时机；.team/inbox/*.jsonl 在消息被读走后被清空——这三者共同构成 drain-on-read 语义的可观察证据。

从 s09 到 s10、s13：演进方向

s09 刻意把"消息类型声明"和"协议处理"拆开：shutdown_request/plan_approval_response 等类型已能合法收发，但握手逻辑（request_id 匹配、审批闸门）留待 s10（Team Protocols）实现。仓库中更完整的团队运行时见 s13_agent_teams/code.py，它在 s09 的邮箱模式之上增加了共享任务板（原子认领）、类型化控制协议、计划审批闸门和可选 worktree 隔离，并有完整的运行时测试覆盖，例如收件箱投递 tests/test_agent_teams_runtime.py 中的 test_inbox_delivery_is_runtime_owned。如果只想学习"持久化队友 + 文件邮箱"这个最小闭环，s09 的约 400 行实现是最清晰的切入点。