S13 Agent Teams：让多个 Agent 协作

{"name": "spawn_teammate",
 "description": "Spawn a persistent teammate.",
 "input_schema": {"type": "object",
                  "properties": {
                      "name": {"type": "string",
                               "pattern": "^[A-Za-z0-9_-]{1,64}$"},
                      "role": {"type": "string"},
                      "prompt": {"type": "string"},
                      "task_id": {"type": "string",
                                  "pattern": "^task_[0-9a-f]{8}$"},
                      "require_plan": {"type": "boolean"}},
                  "required": ["name", "role", "prompt"]}},

TOOL_HANDLERS = {
    "spawn_teammate": run_spawn_teammate,
    # 其余已注册工具省略
}

def run_spawn_teammate(name: str, role: str, prompt: str,
                       task_id: str | None = None,
                       require_plan: bool = False) -> str:
    return spawn_teammate_thread(name, role, prompt, task_id, require_plan)

def spawn_teammate_thread(name: str, role: str, prompt: str,
                          task_id: str | None = None,
                          require_plan: bool = False) -> str:
    """Claim an initial Task, then start one persistent teammate."""
    if not is_valid_agent_name(name):
        return ("Invalid teammate name: use 1-64 letters, digits, "
                "underscores, or dashes")
    if name.lower() in RESERVED_TEAMMATE_NAMES:
        return f"Invalid teammate name: '{name}' is reserved by the runtime"
    with team_lock:
        if any(existing.casefold() == name.casefold()
               for existing in active_teammates):
            return f"Teammate '{name}' already exists"
        active_teammates[name] = "working"
        plan_gates[name] = "required" if require_plan else "not_required"
        assignment_versions[name] = 0

    if task_id:
        try:
            claimed = claim_task(task_id, owner=name)
        except (FileNotFoundError, ValueError) as exc:
            claimed = f"Error: {exc}"
        if not claimed.startswith("Claimed "):
            with team_lock:
                active_teammates.pop(name, None)
                plan_gates.pop(name, None)
                assignment_versions.pop(name, None)
            return f"Cannot spawn teammate '{name}': {claimed}"

    runtime = TeammateRuntime(name, role, prompt, task_id, require_plan)
    thread = threading.Thread(target=runtime.run, daemon=True)
    with team_lock:
        teammate_threads[name] = thread
    thread.start()
    print(f"  [teammate] {name} spawned as {role}")
    assigned = f" for {task_id}" if task_id else " without an initial Task"
    return (
        f"Teammate '{name}' spawned as {role}{assigned}. "
        "End this turn; the runtime will deliver its events."
    )

def __init__(self, name: str, role: str, prompt: str,
             task_id: str | None, require_plan: bool):
    self.name = name
    self.system = (
        f"You are '{name}', a {role}. Use tools to complete the assigned "
        "Task, then call complete_task and report a concise result. "
        "If the first user message contains [Assigned task], that Task is "
        "already claimed; do not call claim_task for it again. "
        "When asked for a plan, call submit_plan and wait for approval "
        "before bash or file changes. File and shell tools use the Task's "
        "working directory; that directory is not a sandbox. The runtime "
        "delivers your final text to Lead. Use send_message only for "
        "intermediate coordination, and address the coordinator as 'lead'."
    )
    self.messages = [{"role": "user", "content": prompt}]
    if task_id:
        task = load_task(task_id)
        cwd = assignment_cwd(name)
        self.messages[0]["content"] += (
            f"\n\n[Assigned task {task.id}] {task.subject}\n"
            f"{task.description}\nWork directory: {cwd}"
        )
    if require_plan:
        self.messages[0]["content"] += (
            "\n\n[Plan required] Submit a plan and wait for Lead approval "
            "before changing files or using bash."
        )
    self.handlers = {
        "bash": self.bash,
        "read_file": self.read,
        "write_file": self.write,
        "edit_file": self.edit,
        "glob": self.glob,
        "send_message": lambda to, content: _teammate_send_message(
            name, to, content),
        "submit_plan": lambda plan: _teammate_submit_plan(name, plan),
        "list_tasks": run_list_tasks,
        "claim_task": self.claim,
        "complete_task": self.complete,
    }

def work(self) -> str:
    """Run one model turn. Return continue, idle, or stop."""
    if self.handle_inbox(BUS.read_inbox(self.name)):
        return "stop"
    with team_lock:
        active_teammates[self.name] = "working"
    try:
        response = client.messages.create(
            model=MODEL,
            system=self.system,
            messages=self.messages,
            tools=TEAMMATE_TOOLS,
            max_tokens=8000,
        )
    except Exception as exc:
        BUS.send(self.name, "lead",
                 f"{type(exc).__name__}: {exc}", "error")
        return "stop"

    self.messages.append({"role": "assistant",
                          "content": response.content})
    tool_calls = [
        block for block in response.content if block.type == "tool_use"
    ]
    if tool_calls:
        results = []
        for block in tool_calls:
            output = _run_teammate_tool(
                self.name, block, self.handlers
            )
            results.append({"type": "tool_result",
                            "tool_use_id": block.id,
                            "content": output})
        self.messages.append({"role": "user", "content": results})
        return "continue"

    summary = _last_assistant_text(response.content)
    gate = plan_gates.get(self.name, "not_required")
    if gate != "pending" and summary:
        BUS.send(self.name, "lead", summary, "result")
    if gate == "pending":
        with team_lock:
            active_teammates[self.name] = "waiting_approval"
    else:
        release_completed_assignment(self.name)
        with team_lock:
            active_teammates[self.name] = "idle"
        BUS.send(self.name, "lead", "Waiting for more work.",
                 "idle_notification")
    return "idle"

def run(self):
    try:
        state = "continue"
        while state != "stop":
            if state == "idle" and not self.wait_for_work():
                break
            state = self.work()
    except Exception as exc:
        try:
            BUS.send(self.name, "lead",
                     f"{type(exc).__name__}: {exc}", "error")
        except Exception:
            pass
    finally:
        try:
            release_teammate_assignment(self.name)
        except Exception as exc:
            try:
                BUS.send(
                    self.name, "lead",
                    f"Assignment cleanup failed: {type(exc).__name__}: {exc}",
                    "error",
                )
            except Exception:
                pass
        with team_lock:
            active_teammates.pop(self.name, None)
            plan_gates.pop(self.name, None)
            plan_request_ids.pop(self.name, None)
            teammate_threads.pop(self.name, None)
        print(f"  [teammate] {self.name} finished")

def _run_teammate_tool(name: str, block, handlers: dict) -> str:
    gate = plan_gates.get(name, "not_required")
    if block.name in {"bash", "write_file", "edit_file"}:
        if gate != "approved":
            if gate != "not_required":
                return (f"Blocked: plan status is {gate}. Submit or revise the "
                        "plan and wait for approval before changing the workspace.")
        blocked = check_permission(block, prompt_user=False)
        if blocked:
            return blocked
    handler = handlers.get(block.name)
    if not handler:
        return f"Unknown tool: {block.name}"
    trigger_hooks("PreToolUse", block, skip_permission=True)
    try:
        output = str(handler(**block.input))
    except Exception as exc:
        output = f"Error: {type(exc).__name__}: {exc}"
    trigger_hooks("PostToolUse", block, output)
    return output

def send(self, from_agent: str, to_agent: str, content: str,
         msg_type: str = "message", metadata: dict | None = None):
    msg = {"from": from_agent, "to": to_agent,
           "content": content, "type": msg_type,
           "ts": time.time(), "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()
    print(f"  [bus] {from_agent} -> {to_agent}: "
          f"({msg_type}) {content[:50]}")

def read_inbox(self, agent: str) -> list[dict]:
    with self._lock:
        return self._read_unlocked(agent)

def _read_unlocked(self, agent: str) -> list[dict]:
    inbox = self._path(agent)
    if not inbox.exists():
        return []
    msgs = [json.loads(line) for line in inbox.read_text(encoding="utf-8").splitlines()
            if line.strip()]
    inbox.unlink()
    return msgs

def peek(self, agent: str) -> bool:
    with self._lock:
        inbox = self._path(agent)
        return inbox.exists() and inbox.stat().st_size > 0

def wait_for_messages(self, agent: str,
                      timeout: float | None = None) -> list[dict]:
    """Block until the agent has messages or timeout expires."""
    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)

def handle_inbox(self, inbox: list[dict]) -> bool:
    """Append work messages and return True for a valid shutdown."""
    work_messages = []
    for msg in inbox:
        msg_type = msg.get("type", "message")
        if msg_type == "shutdown_request":
            accepted, notice = apply_shutdown_request(self.name, msg)
            if not accepted:
                work_messages.append(notice)
                continue
            BUS.send(self.name, "lead", "Shutdown acknowledged.",
                     "shutdown_response",
                     {"request_id": notice, "approve": True})
            return True
        if msg_type == "plan_approval_response":
            _, notice = apply_plan_response(self.name, msg)
            work_messages.append(notice)
            continue
        if msg_type == "plan_request":
            work_messages.append(f"[Plan required] {msg['content']}")
            continue
        work_messages.append(
            f"[Message from {msg['from']}] {msg['content']}"
        )
    if work_messages:
        self.messages.append({"role": "user",
                              "content": "\n".join(work_messages)})
    return False

def wait_for_cli_event() -> tuple[str, str | None]:
    prompt_visible = False
    while True:
        if BUS.peek("lead"):
            if prompt_visible:
                print()
            return "wake", None
        if not prompt_visible:
            print("s13 >> ", end="", flush=True)
            prompt_visible = True
        readable, _, _ = select.select([sys.stdin], [], [], 0.25)
        if readable:
            line = sys.stdin.readline()
            if line == "":
                return "quit", None
            return "user", line.rstrip("\n")

def consume_lead_inbox() -> list[dict]:
    """Consume Lead events and update protocol state before model delivery."""
    msgs = BUS.read_inbox("lead")
    for msg in msgs:
        metadata = msg.get("metadata", {})
        request_id = metadata.get("request_id", "")
        if request_id and msg.get("type", "").endswith("_response"):
            match_response(msg["type"], request_id,
                           metadata.get("approve", False),
                           msg.get("from", ""), msg.get("to", ""))
    return msgs

def format_team_events(msgs: list[dict]) -> str:
    lines = []
    for msg in msgs:
        metadata = msg.get("metadata", {})
        request_id = metadata.get("request_id")
        suffix = f" request_id={request_id}" if request_id else ""
        lines.append(
            f"[{msg['type']}{suffix}] {msg['from']}: {msg['content']}"
        )
    return "[Team events]\n" + "\n".join(lines)

def wait_for_work(self) -> bool:
    """Wait for a message or atomically claim the next ready Task."""
    while True:
        inbox = BUS.wait_for_messages(self.name, IDLE_SCAN_INTERVAL)
        if inbox:
            before = len(self.messages)
            if self.handle_inbox(inbox):
                return False
            if len(self.messages) > before:
                return True
            continue

        task = claim_next_task(self.name)
        if not task:
            continue
        cwd = assignment_cwd(self.name)
        self.messages.append({
            "role": "user",
            "content": (
                f"[Auto-claimed task {task.id}] {task.subject}\n"
                f"{task.description}\nWork directory: {cwd}"
            ),
        })
        print(f"  [idle] {self.name} claimed {task.id}: {task.subject}")
        return True

def scan_unclaimed_tasks() -> list[Task]:
    """Return ready tasks whose optional worktree binding is usable."""
    with task_lock:
        ready = []
        for task in list_tasks():
            if (task.status != "pending" or task.owner is not None
                    or not can_start(task.id)):
                continue
            _, error = task_worktree_cwd(task)
            if not error:
                ready.append(task)
        return ready

def claim_next_task(name: str) -> Task | None:
    """Claim the first still-available task, never a second assignment."""
    with task_lock:
        if teammate_assignments.get(name) or _owner_in_progress(name):
            return None
    for task in scan_unclaimed_tasks():
        result = claim_task(task.id, owner=name)
        if result.startswith("Claimed "):
            return load_task(task.id)
    return None

def claim_task(task_id: str, owner: str = "agent") -> str:
    """Atomically claim one task and bind the owner's filesystem cwd."""
    with task_store_lock():
        task = load_task(task_id)
        if task.status != "pending":
            return f"Task {task_id} is {task.status}, cannot claim"
        if task.owner:
            return f"Task {task_id} is already owned by {task.owner}"
        assignment = teammate_assignments.get(owner)
        if assignment:
            return (f"Owner {owner} must finish the current work turn for "
                    f"{assignment['task_id']} before claiming another task")
        current = _owner_in_progress(owner)
        if current:
            return (f"Owner {owner} must complete {current.id} before "
                    "claiming another task")
        if not can_start(task_id):
            return f"Blocked by: {_incomplete_dependencies(task)}"
        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}
        advance_assignment_version(owner)
    print(f"  [claim] {task.subject} -> in_progress (owner: {owner})")
    return f"Claimed {task.id} ({task.subject})"

def release_teammate_assignment(owner: str):
    """Return abandoned teammate work to the task board on thread exit."""
    with task_lock:
        try:
            task = _owner_in_progress(owner)
            if task:
                task.status = "pending"
                task.owner = None
                save_task(task)
        finally:
            teammate_assignments.pop(owner, None)
            advance_assignment_version(owner)
            if owner in globals().get("plan_gates", {}):
                globals()["plan_gates"][owner] = "not_required"

def agent_loop(messages: list):
    while True:
        try:
            response = client.messages.create(
                model=MODEL,
                system=SYSTEM,
                messages=messages,
                tools=TOOLS,
                max_tokens=8000,
            )
        except Exception as exc:
            messages.append({
                "role": "assistant",
                "content": [{
                    "type": "text",
                    "text": f"[Error] {type(exc).__name__}: {exc}",
                }],
            })
            release_completed_assignment("agent")
            trigger_hooks("Stop", messages)
            return

        messages.append({"role": "assistant", "content": response.content})
        tool_calls = [
            block for block in response.content if block.type == "tool_use"
        ]
        if not tool_calls:
            release_completed_assignment("agent")
            trigger_hooks("Stop", messages)
            return

        results = []
        for block in tool_calls:
            print(f"> {block.name}")
            output = execute_tool(block)
            print(output[:300])
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

# Lead CLI: two event sources, one ordinary Agent Loop
while cli_running:
    event = wait_for_cli_event()
    if event.kind == "quit":
        break
    if event.kind == "user":
        lead_history.append(user_message(event.text))
    else:
        messages = consume_lead_inbox()
        if not messages:
            continue
        lead_history.append(user_message(format_team_events(messages)))
    agent_loop(lead_history)

# Inside Lead's tool dispatch
# spawn -> claim initial Task if supplied -> start runtime thread
# immediately return paired tool_result; later events use MessageBus

# Each teammate has its own messages, system and tools
state = "continue"
while state != "stop":
    if state == "idle" and not runtime.wait_for_work():
        break
    state = runtime.work()
# finally: release unfinished assignment and remove runtime registration