Harness实战：Autonomous Agents——从等指令到自驱动

Lead: spawn("alice", "coder", "你负责写代码")
Lead: send("alice", "写 auth 模块")          ← Lead 手动派活
Alice: send("lead", "auth 写完了")
Lead: send("alice", "写 user 模块")          ← Lead 又手动派活
Alice: send("lead", "user 写完了")
Lead: shutdown_request("alice")

Lead: create_team_task("写 auth 模块")        ← 贴到看板
Lead: create_team_task("写 user 模块")        ← 贴到看板
Lead: create_team_task("写测试", blocked_by=[1,2])
Lead: spawn("alice", "coder", "你是开发者")

Alice: (自动认领 #1) → 干活 → complete_task(1)
Alice: (idle → 扫描看板 → 自动认领 #2) → 干活 → complete_task(2)
Alice: (idle → 扫描看板 → #3 已解锁 → 自动认领 #3) → 干活 → complete_task(3)
Alice: (idle → 60s 无新任务 → 自动关机)

原有 harness-agent-teams.py (保留全部)
├── Base Tools (4): bash, read_file, write_file, edit_file
├── Team Comm (2): send_message, read_inbox
├── Lead Tools: compact, task, parallel_tasks, background_task, check_task,
│              spawn_teammate, list_teammates, broadcast,
│              shutdown_request, shutdown_status, plan_review
└── Mate Tools: shutdown_response, plan_approval

+ 新增
├── Task Board: .tasks/ 目录，任务 JSON 文件，原子化认领
└── 自治循环: WORK → IDLE 双阶段 + 自动认领 + 身份重注入

TEAMMATE_TOOLS = BASE_TOOLS + TEAM_COMM_TOOLS + MATE_TOOLS    # 11
TOOLS = BASE_TOOLS + TEAM_COMM_TOOLS + LEAD_TOOLS              # 19

.tasks/
  task_1.json    ← {"id":1, "subject":"写auth模块", "status":"pending", "owner":null}
  task_2.json    ← {"id":2, "subject":"写user模块", "status":"in_progress", "owner":"alice"}
  task_3.json    ← {"id":3, "subject":"写测试", "status":"pending", "blockedBy":[1,2]}

def create_team_task(subject: str, description: str = "",
                blocked_by: list = None) -> str:
    """创建任务到看板。"""
    TASKS_DIR.mkdir(exist_ok=True)
    existing = list(TASKS_DIR.glob("task_*.json"))
    next_id = max((int(f.stem.split("_")[1]) for f in existing), default=0) + 1
    task = {
        "id": next_id,
        "subject": subject,
        "description": description,
        "status": "pending",
        "owner": None,
        "blockedBy": blocked_by or [],
    }
    path = TASKS_DIR / f"task_{next_id}.json"
    path.write_text(json.dumps(task, indent=2))
    return f"Created task #{next_id}: {subject}"

_claim_lock = threading.Lock()

def claim_task(task_id: int, owner: str) -> str:
    """原子化认领任务，加锁防止竞争。"""
    with _claim_lock:
        path = TASKS_DIR / f"task_{task_id}.json"
        if not path.exists():
            return f"Error: Task {task_id} not found"
        task = json.loads(path.read_text())
        if task.get("owner"):
            return f"Error: Task {task_id} already claimed by {task['owner']}"
        if task.get("status") != "pending":
            return f"Error: Task {task_id} status is '{task['status']}', not pending"
        if task.get("blockedBy"):
            return f"Error: Task {task_id} is blocked by {task['blockedBy']}"
        task["owner"] = owner
        task["status"] = "in_progress"
        path.write_text(json.dumps(task, indent=2))
    return f"Claimed task #{task_id} for {owner}"

def complete_task(task_id: int, owner: str) -> str:
    """标记任务完成，并解除对其他任务的阻塞。"""
    with _claim_lock:
        path = TASKS_DIR / f"task_{task_id}.json"
        task = json.loads(path.read_text())
        if task.get("owner") != owner:
            return f"Error: Task {task_id} is not owned by {owner}"
        task["status"] = "completed"
        path.write_text(json.dumps(task, indent=2))
        # 解除依赖：移除其他任务 blockedBy 中的本任务
        for f in TASKS_DIR.glob("task_*.json"):
            if f == path:
                continue
            other = json.loads(f.read_text())
            blocked = other.get("blockedBy", [])
            if task_id in blocked:
                blocked.remove(task_id)
                other["blockedBy"] = blocked
                f.write_text(json.dumps(other, indent=2))
    return f"Task #{task_id} completed by {owner}"

[pending] ──claim──> [in_progress] ──complete──> [completed]
                                          │
                                          └──> 解除其他任务的 blockedBy

spawn → WORK (max 50 turns) → 结束

spawn → WORK → IDLE → WORK → IDLE → ... → SHUTDOWN

def _teammate_loop(self, name: str, role: str, prompt: str):
    """队友的自治 agent loop: WORK → IDLE → WORK → ... → SHUTDOWN"""
    messages = [{"role": "user", "content": prompt}]
    should_shutdown = False

    while True:
        # ── WORK PHASE ────────────────────────
        for _ in range(50):
            inbox = BUS.read_inbox(name)
            for msg in inbox:
                messages.append({"role": "user", "content": json.dumps(msg)})

            response = client.messages.create(
                model=MODEL, system=sys_prompt,
                messages=messages, tools=TEAMMATE_TOOLS,
            )
            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason != "tool_use":
                break

            results = []
            idle_requested = False
            for block in response.content:
                if block.type == "tool_use":
                    if block.name == "idle":
                        idle_requested = True
                        output = "Entering idle phase. Will poll for new tasks."
                    else:
                        output = self._exec(name, block.name, block.input)
                    results.append({"type": "tool_result", ...})
                    # 后置检查: 队友批准关机
                    if block.name == "shutdown_response" and block.input.get("approve"):
                        should_shutdown = True
            messages.append({"role": "user", "content": results})

            if idle_requested or should_shutdown:
                break

        if should_shutdown:
            self._set_status(name, "shutdown")
            return

        # ── IDLE PHASE ────────────────────────
        self._set_status(name, "idle")
        resume = self._idle_poll(name, role, messages)
        if not resume:
            self._set_status(name, "shutdown")
            return
        self._set_status(name, "working")

def _idle_poll(self, name: str, role: str, messages: list) -> bool:
    """空闲阶段: 轮询收件箱和任务看板。返回 True 恢复工作，False 关机。"""
    polls = IDLE_TIMEOUT // max(POLL_INTERVAL, 1)    # 60 / 5 = 12 次

    for _ in range(polls):
        time.sleep(POLL_INTERVAL)                     # 每 5 秒一次

        # 1. 检查收件箱
        inbox = BUS.read_inbox(name)
        if inbox:
            for msg in inbox:
                if msg.get("type") == "shutdown_request":
                    # 空闲时收到关机请求 → 自动批准
                    BUS.send(name, "lead", "Auto-approved (was idle)",
                             "shutdown_response", ...)
                    return False
                messages.append({"role": "user", "content": json.dumps(msg)})
            return True                               # 有消息 → 恢复工作

        # 2. 扫描任务看板
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            task = unclaimed[0]
            result = claim_task(task["id"], name)
            if result.startswith("Error:"):
                continue                              # 被抢了，继续轮询
            # 身份重注入 (如果上下文被压缩过)
            if len(messages) <= 3:
                messages.insert(0, make_identity_block(name, role, team_name))
            messages.append({"role": "user", "content": f"<auto-claimed>Task #{task['id']}: ..."})
            return True                               # 认领成功 → 恢复工作

    return False                                      # 超时 → 关机

def make_identity_block(name: str, role: str, team_name: str) -> dict:
    return {
        "role": "user",
        "content": (
            f"<identity>You are '{name}', role: {role}, "
            f"team: {team_name}. Continue your work.</identity>"
        ),
    }

# _idle_poll 中认领新任务时
if len(messages) <= 3:
    messages.insert(0, make_identity_block(name, role, team_name))
    messages.insert(1, {
        "role": "assistant",
        "content": f"I am {name}. Continuing.",
    })

# ── 所有角色共用 (4) ─────────────────────────────────
BASE_TOOLS = [
    bash, read_file, write_file, edit_file
]

# ── 团队通信: Lead + 队友共用 (2) ────────────────────
TEAM_COMM_TOOLS = [
    send_message, read_inbox
]

# ── 队友专属 (5) ─────────────────────────────────────
MATE_TOOLS = [
    # 协议
    shutdown_response, plan_approval,
    # 自治
    idle, claim_task, complete_task,
]

# ── Lead 专属 (13) ───────────────────────────────────
LEAD_TOOLS = [
    # 上下文压缩
    compact,
    # 子任务委派
    task, parallel_tasks, background_task, check_task,
    # 团队管理
    spawn_teammate, list_teammates, broadcast,
    # 协议控制
    shutdown_request, shutdown_status, plan_review,
    # 任务看板
    create_team_task, list_team_tasks,
]

REPL (main thread)
  │
  ├─ agent_loop (Lead, tools=19)
  │   ├─ task/parallel_tasks/background_task
  │   │   └→ run_subagent (子 Agent, tools=4)
  │   │       └→ bash/read_file/write_file/edit_file
  │   │
  │   ├─ spawn_teammate
  │   │   └→ _teammate_loop (队友, daemon thread, tools=11)
  │   │       ├─ WORK: LLM → idle / _exec() / shutdown check
  │   │       │   └→ _exec: bash/read_file/.../claim_task/complete_task/...
  │   │       └─ IDLE: poll inbox → poll task board → timeout
  │   │
  │   ├─ create_team_task → 贴到 .tasks/
  │   ├─ send_message → 写入 .team/inbox/
  │   └─ compact → auto_compact (无 tools)
  │
  └─ plan_then_execute (无 tools → agent_loop)

elif query.strip() == "/tasks":
    print(list_team_tasks())
    continue

Task Board:
  [x] #1: 写 auth 模块 @alice
  [>] #2: 写 user 模块 @alice
  [ ] #3: 写测试 (blocked by: [2])