Harness实战：Agent Teams——从单兵到团队协作

Lead: "分析技术栈" → 派出子 Agent → 子 Agent 干完 → 返回结果 → 子 Agent 消失

Lead: spawn("alice", "coder", "负责实现功能")
Lead: spawn("bob", "tester", "负责写测试")

Lead → send("alice", "开始写 auth 模块")
Alice → send("lead", "auth 模块写完了")
Lead → send("bob", "alice 写完了，你开始写测试")
Bob → plan_approval("计划: 写3个测试用例...")
Lead → plan_review(approve=True)
Bob → send("lead", "测试全过了")
Lead → shutdown_request("alice")
Lead → shutdown_request("bob")

原有 agentic-demo.py (保留全部)
├── Base Tools (5): bash, read_file, write_file, edit_file, compact
├── Subagent Tools (4): task, parallel_tasks, background_task, check_task
├── Context Compact: micro_compact, auto_compact, manual compact
└── Plan Mode

+ 新增
├── MessageBus: JSONL 收件箱通信
├── TeammateManager: 持久化命名队友
└── Protocols: 关机握手 + 计划审批

.team/
  inbox/
    alice.jsonl    ← Alice 的收件箱
    bob.jsonl      ← Bob 的收件箱
    lead.jsonl     ← Lead 的收件箱

class MessageBus:
    """每个队友一个 JSONL 文件，append-only 写入，drain-on-read。"""

    def __init__(self, inbox_dir: Path):
        self.dir = inbox_dir
        self.dir.mkdir(parents=True, exist_ok=True)

    def send(self, sender: str, to: str, content: str,
             msg_type: str = "message", extra: dict = None) -> str:
        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}"

    def read_inbox(self, name: str) -> 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))
        inbox_path.write_text("")  # drain
        return messages

    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"

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads: dict[str, threading.Thread] = {}

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

{
  "team_name": "default",
  "members": [
    {"name": "alice", "role": "coder", "status": "working"},
    {"name": "bob", "role": "tester", "status": "idle"}
  ]
}

def _teammate_loop(self, name: str, role: str, prompt: str):
    sys_prompt = (
        f"You are '{name}', role: {role}, at {WORKDIR}. "
        f"Submit plans via plan_approval before major work. "
        f"Respond to shutdown_request with shutdown_response. "
        f"Use send_message to communicate with teammates and lead."
    )
    messages = [{"role": "user", "content": prompt}]
    should_exit = False

    for _ in range(50):
        # ★ 差异1: 每轮检查收件箱
        inbox = BUS.read_inbox(name)
        for msg in inbox:
            messages.append({"role": "user", "content": json.dumps(msg)})
        if should_exit:
            break
        response = client.messages.create(
            model=MODEL,
            system=sys_prompt,
            messages=messages,
            tools=self._teammate_tools(),  # ★ 差异2: 队友专属工具集
            max_tokens=4096,
        )
        # ... 执行工具 ...

        # ★ 差异3: 关机协议
        if block.name == "shutdown_response" and block.input.get("approve"):
            should_exit = True

bash, read_file, write_file, edit_file     ← 干活用
send_message, read_inbox                   ← 通信用
shutdown_response, plan_approval           ← 协议用

Lead             Teammate
  |                 |
  |--shutdown_req-->|    Lead: "请优雅关机"
  | {req_id:"abc"}  |
  |<--shutdown_resp-|    Teammate: "好的/再等等"
  | {req_id:"abc",  |
  |  approve:true}  |

def handle_shutdown_request(teammate: str) -> str:
    req_id = str(uuid.uuid4())[:8]
    with _tracker_lock:
        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 to '{teammate}' (status: pending)"

if tool_name == "shutdown_response":
    req_id = args["request_id"]
    approve = args["approve"]
    with _tracker_lock:
        if req_id in shutdown_requests:
            shutdown_requests[req_id]["status"] = (
                "approved" if approve else "rejected"
            )
    BUS.send(sender, "lead", args.get("reason", ""),
             "shutdown_response",
             {"request_id": req_id, "approve": approve})
    return f"Shutdown {'approved' if approve else 'rejected'}"

Teammate           Lead
   |                 |
   |--plan_req------>|    Teammate: "我打算这样做..."
   | {req_id:"xyz"}  |
   |<--plan_resp-----|    Lead: "同意/否决"
   | {req_id:"xyz",  |
   |  approve:true}  |

if tool_name == "plan_approval":
    plan_text = args.get("plan", "")
    req_id = str(uuid.uuid4())[:8]
    with _tracker_lock:
        plan_requests[req_id] = {
            "from": sender, "plan": plan_text, "status": "pending",
        }
    BUS.send(sender, "lead", plan_text, "plan_approval_response",
             {"request_id": req_id, "plan": plan_text})
    return f"Plan submitted (request_id={req_id}). Waiting for lead approval."

def handle_plan_review(request_id: str, approve: bool, feedback: str = "") -> str:
    with _tracker_lock:
        req = plan_requests.get(request_id)
    if not req:
        return f"Error: Unknown plan request_id '{request_id}'"
    with _tracker_lock:
        req["status"] = "approved" if approve else "rejected"
    BUS.send("lead", req["from"], feedback, "plan_approval_response",
             {"request_id": request_id, "approve": approve, "feedback": feedback})
    return f"Plan {req['status']} for '{req['from']}'"

[pending] ──approve──> [approved]
[pending] ──reject───> [rejected]

shutdown_requests: dict[str, dict] = {}   # {req_id: {target, status}}
plan_requests: dict[str, dict] = {}       # {req_id: {from, plan, status}}
_tracker_lock = threading.Lock()

def agent_loop(messages: list):
    while True:
        micro_compact(messages)
        if estimate_tokens(messages) > THRESHOLD:
            messages[:] = auto_compact(messages)

        # ★ 新增: Lead 每轮检查收件箱
        inbox = BUS.read_inbox("lead")
        if inbox:
            messages.append({
                "role": "user",
                "content": f"<inbox>{json.dumps(inbox, indent=2)}</inbox>",
            })

        response = client.messages.create(...)
        # ... 其余不变

# 领导 Agent 拥有全部工具
TOOLS = BASE_TOOLS + SUBAGENT_TOOLS + TEAM_TOOLS + PROTOCOL_TOOLS

# 完整 dispatch map
TOOL_HANDLERS = {
    # base (5)
    "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"]),
    "compact":          lambda **kw: "Manual compression requested.",
    # subagent (4)
    "task":             lambda **kw: run_subagent(kw["prompt"]),
    "parallel_tasks":   lambda **kw: run_parallel_tasks(kw["tasks"]),
    "background_task":  lambda **kw: run_background_task(kw["prompt"]),
    "check_task":       lambda **kw: run_check_task(kw["task_id"]),
    # team (5)
    "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"], ...),
    "read_inbox":       lambda **kw: json.dumps(BUS.read_inbox("lead"), indent=2),
    "broadcast":        lambda **kw: BUS.broadcast("lead", kw["content"], TEAM.member_names()),
    # protocols (3)
    "shutdown_request": lambda **kw: handle_shutdown_request(kw["teammate"]),
    "shutdown_status":  lambda **kw: handle_shutdown_status(kw["request_id"]),
    "plan_review":      lambda **kw: handle_plan_review(kw["request_id"], kw["approve"], ...),
}

elif query.strip() == "/team":
    print(TEAM.list_all())
    continue
elif query.strip() == "/inbox":
    msgs = BUS.read_inbox("lead")
    if msgs:
        print(json.dumps(msgs, indent=2, ensure_ascii=False))
    else:
        print("  (inbox empty)")
    continue

# 优雅退出：等待后台任务 + 关机队友
with bg_lock:
    pending = {tid: f for tid, f in bg_tasks.items() if not f.done()}
active_threads = {n: t for n, t in TEAM.threads.items()
                  if t.is_alive()}

if pending or active_threads:
    print(f"⏳ Waiting for {len(pending)} background task(s) "
          f"+ {len(active_threads)} teammate(s)...")
    for tid, future in pending.items():
        future.result(timeout=300)
    for name, thread in active_threads.items():
        thread.join(timeout=10)

SYSTEM = f"""You are a team lead agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain.

You have several task delegation strategies:
- `task`: Run a subtask synchronously. Blocks until done, returns result.
- `parallel_tasks`: Run multiple subtasks concurrently with dependency ordering.
- `background_task`: Fire-and-forget. Returns a task_id immediately.
- `check_task`: Poll a background task by task_id.

Team management:
- `spawn_teammate`: Spawn a persistent named teammate.
- `list_teammates`: See all teammates with name, role, status.
- `send_message`: Send a message to a teammate's inbox.
- `read_inbox`: Read and drain the lead's inbox.
- `broadcast`: Send a message to all teammates.

Protocols:
- `shutdown_request`: Request a teammate to shut down gracefully.
- `shutdown_status`: Check shutdown request status.
- `plan_review`: Approve or reject a teammate's plan."""

"Use `spawn_teammate` for persistent agents that need ongoing communication. "