从一个Agent到一支团队：Claude Code多Agent协作架构全解——源码精读（三）

.tasks/
  task_1.json  {"id":1, "status":"completed", "blockedBy":[]}
  task_2.json  {"id":2, "status":"pending",   "blockedBy":[1]}
  task_3.json  {"id":3, "status":"pending",   "blockedBy":[1]}
  task_4.json  {"id":4, "status":"pending",   "blockedBy":[2,3]}

def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

class BackgroundManager:
    def run(self, command: str) -> str:
        task_id = str(uuid.uuid4())[:8]
        thread = threading.Thread(
            target=self._execute, args=(task_id, command), daemon=True)
        thread.start()
        return f"Background task {task_id} started"  # 立即返回

    def _execute(self, task_id, command):
        r = subprocess.run(command, shell=True, ...)
        with self._lock:
            self._notification_queue.append({
                "task_id": task_id, "result": r.stdout[:500]})

def agent_loop(messages):
    while True:
        notifs = BG.drain_notifications()  # 先排空
        if notifs:
            messages.append({"role": "user",
                "content": f"<background-results>\n{...}\n</background-results>"})
        response = client.messages.create(...)

def spawn(self, name: str, role: str, prompt: str) -> str:
    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)
    thread.start()
    return f"Spawned teammate '{name}' (role: {role})"

class MessageBus:
    def send(self, sender, to, content, msg_type="message"):
        msg = {"type": msg_type, "from": sender,
               "content": content, "timestamp": time.time()}
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")  # append-only

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        msgs = [json.loads(l) for l in path.read_text().strip().splitlines() if l]
        path.write_text("")  # drain（读取后清空）
        return json.dumps(msgs, indent=2)

# 关机协议：领导发请求
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"

# 队友响应（approve=True 则优雅退出，False 则拒绝继续干）
if tool_name == "shutdown_response":
    req_id = args["request_id"]
    shutdown_requests[req_id]["status"] = "approved" if args["approve"] else "rejected"

def _idle_poll(self, name, messages):
    for _ in range(IDLE_TIMEOUT // POLL_INTERVAL):  # 60s / 5s = 12
        time.sleep(POLL_INTERVAL)
        inbox = BUS.read_inbox(name)
        if inbox:
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
            return True
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            claim_task(unclaimed[0]["id"], name)
            return True
    return False  # timeout → shutdown

if len(messages) <= 3:  # 压缩发生了
    messages.insert(0, {"role": "user",
        "content": f"<identity>You are '{name}', role: {role}, "
                   f"team: {team_name}. Continue your work.</identity>"})
    messages.insert(1, {"role": "assistant",
        "content": f"I am {name}. Continuing."})

def create(self, name: str, task_id: int = None):
    # 创建独立 git 分支目录
    subprocess.run(
        ["git", "worktree", "add", "-b", f"wt/{name}",
         f".worktrees/{name}", "HEAD"])
    # 绑定任务，同时推进到 in_progress
    if task_id:
        self.tasks.bind_worktree(task_id, name)

WORKTREES.keep("ui-login")                         # 保留供后续使用
WORKTREES.remove("auth-refactor", complete_task=True)  # 删除 + 完成任务，一步到位