第9课：Agent Teams——多Agent协作：团队与邮箱系统

                      .team/
                      ├── config.json        ← 团队名册
                      └── inbox/
                          ├── coder.jsonl    ← coder 的收件箱
                          ├── tester.jsonl   ← tester 的收件箱
                          └── leader.jsonl   ← leader 的收件箱

+----------+   spawn()    +---------+
|  Leader  | ───────────→ |  Coder  |  ← 独立线程，独立 Agent Loop
|  Agent   |   spawn()    +---------+
|          | ───────────→ | Tester  |  ← 独立线程，独立 Agent Loop
+----+-----+              +---------+
     |                         ↑
     +── MessageBus ───────────+
          send() / broadcast()
          (append JSONL, drain-on-read)

class MessageBus:
    def __init__(self, inbox_dir=".team/inbox"):
        self.inbox_dir = inbox_dir
        os.makedirs(inbox_dir, exist_ok=True)

    def send(self, from_name: str, to_name: str, content: str):
        """往目标队友的收件箱追加一条消息"""
        msg = {"from": from_name, "to": to_name,
               "content": content, "timestamp": time.time()}
        path = os.path.join(self.inbox_dir, f"{to_name}.jsonl")
        with open(path, "a") as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            f.write(json.dumps(msg, ensure_ascii=False) + "\n")
            fcntl.flock(f, fcntl.LOCK_UN)

    def broadcast(self, from_name: str, content: str, team: dict):
        for name in team:
            if name != from_name:
                self.send(from_name, name, content)

    def read_inbox(self, name: str) -> list[dict]:
        """读取并清空收件箱（drain-on-read）"""
        path = os.path.join(self.inbox_dir, f"{name}.jsonl")
        if not os.path.exists(path):
            return []
        with open(path, "r+") as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            lines = f.readlines()
            f.seek(0)
            f.truncate()              # 读完即清
            fcntl.flock(f, fcntl.LOCK_UN)
        return [json.loads(l) for l in lines if l.strip()]

@dataclass
class TeammateConfig:
    name: str
    system_prompt: str
    status: str = "idle"    # idle | working | shutdown

class TeammateManager:
    def __init__(self):
        self.team = load_team_config()   # 从 .team/config.json 加载
        self.bus = MessageBus()
        self.threads: dict[str, threading.Thread] = {}

    def spawn(self, name: str, system_prompt: str) -> str:
        config = TeammateConfig(name=name, system_prompt=system_prompt)
        self.team[name] = config
        save_team_config(self.team)

        thread = threading.Thread(
            target=self._teammate_loop, args=(name,), daemon=True)
        self.threads[name] = thread
        thread.start()
        return f"队友 {name} 已创建并启动"

    def send_task(self, to_name: str, task: str) -> str:
        self.bus.send("leader", to_name, task)
        self.team[to_name].status = "working"
        save_team_config(self.team)
        return f"任务已发送给 {to_name}"

    def shutdown(self, name: str) -> str:
        self.team[name].status = "shutdown"
        save_team_config(self.team)
        return f"队友 {name} 已关闭"

{
  "coder": {
    "name": "coder",
    "system_prompt": "你是代码编写专家，专注于写出干净、可测试的代码。",
    "status": "working"
  },
  "tester": {
    "name": "tester",
    "system_prompt": "你是测试工程师，专注于编写全面的测试用例。",
    "status": "idle"
  }
}

def _teammate_loop(self, name: str):
    config = self.team[name]
    while config.status != "shutdown":
        messages = self.bus.read_inbox(name)
        if not messages:
            time.sleep(1)
            config = self.team.get(name, config)
            continue

        config.status = "working"
        save_team_config(self.team)

        for msg in messages:
            result = self._run_teammate_agent(
                system=config.system_prompt, task=msg["content"])
            self.bus.send(name, msg["from"], f"[{name} 完成] {result}")

        config.status = "idle"
        save_team_config(self.team)

def _run_teammate_agent(self, system: str, task: str) -> str:
    """执行一轮独立的 Agent Loop——跟第1课完全一样"""
    msgs = [{"role": "user", "content": task}]
    for _ in range(20):
        resp = client.messages.create(
            model=MODEL, system=system,
            messages=msgs, tools=TOOLS, max_tokens=8000)
        msgs.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason != "tool_use":
            return next((b.text for b in resp.content
                         if hasattr(b, "text")), "(无文本回复)")
        results = []
        for b in resp.content:
            if b.type == "tool_use":
                results.append({"type": "tool_result",
                    "tool_use_id": b.id,
                    "content": dispatch_tool(b.name, b.input)})
        msgs.append({"role": "user", "content": results})
    return "(达到最大轮次)"

def leader_loop(messages: list):
    manager = TeammateManager()
    while True:
        inbox = manager.bus.read_inbox("leader")
        if inbox:
            reports = "\n".join(
                f"[来自 {m['from']}] {m['content']}" for m in inbox)
            messages.append({"role": "user",
                "content": f"<teammate-reports>\n{reports}\n</teammate-reports>"})

        response = client.messages.create(
            model=MODEL, system=LEADER_SYSTEM,
            messages=messages, tools=ALL_TOOLS, max_tokens=8000)
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return
        # ... 标准工具执行逻辑（同前）

s09 >> 实现一个计算器模块，要有测试

领导: 这个任务需要两个角色
  ↓
spawn_teammate("coder", "你是代码编写专家...")  → coder 启动
spawn_teammate("tester", "你是测试工程师...")   → tester 启动
  ↓
send_to_teammate("coder", "实现 calc.py，支持加减乘除")
领导: "已分配任务给 coder，等待完成..."
  ↓
[coder 在后台独立工作: 创建 calc.py → 验证 → 结果发回 leader]
  ↓
leader 收件箱收到报告
领导: "coder 完成了，把代码发给 tester"
  ↓
send_to_teammate("tester", "为 calc.py 编写单元测试并运行")
  ↓
[tester 在后台独立工作: 读取 → 写测试 → pytest → 结果发回 leader]
  ↓
领导汇总: "计算器模块已完成，4个测试全部通过。"

# 进入项目目录
cd learn-claude-code

# 启动第九课
python agents/s09_agent_teams.py

s09 >> Spawn alice (coder) and bob (tester). Have alice send bob a message.

> spawn_teammate: Spawned 'alice' (role: coder)
> spawn_teammate: Spawned 'bob' (role: tester)

  [alice] send_message: Sent message to bob    ← alice 主动给 bob 发消息

  [bob] read_inbox: [{
    "type": "message",
    "from": "alice",
    "content": "Hey Bob! 👋 I'm Alice, one of the coders on the team..."
  }]

  [bob] bash: 列出项目文件...
  [bob] read_file: hello.py, utils.py, test_utils.py ...
  [bob] bash: pytest → ===== test session starts =====
  [bob] send_message: Sent message to alice    ← bob 把测试报告发给 alice

s09 >> Broadcast "status update: phase 1 complete" to all teammates

> broadcast: Broadcast to 2 teammates
  → 📩 alice (coder)
  → 📩 bob (tester)

s09 >> Check the lead inbox for any messages
> read_inbox: []       ← 收件箱为空，但团队还在

s09 >> /team           ← 内置命令查看团队状态
Team: default
  alice (coder): idle
  bob (tester): idle

s09 >> /inbox          ← 内置命令查看收件箱
[]

# 预告：s10 的自治协作
rules = {
    "coder": {"on_complete": ["tester"]},
    "tester": {"on_complete": ["reviewer"]},
    "reviewer": {"on_complete": ["leader"]},
}
# coder 完成 → 自动触发 tester → 自动触发 reviewer → 汇报领导