learn-claude-code 自主 Agent 机制：空闲轮询、任务自动认领与身份重注入的完整实现解析

在 learn-claude-code 的课程体系中，s11（Legacy 12 课时轨道）解决一个多智能体协作中的规模化瓶颈：让 teammate 不再等待 lead 逐个分派工作，而是自己扫描任务板、认领无主任务、做完后继续寻找下一份工作。本文基于 docs/en/s11-autonomous-agents.md 与可运行实现 agents/s11_autonomous_agents.py，完整讲清 WORK/IDLE 双阶段生命周期、空闲轮询参数、原子任务认领与上下文压缩后的身份重注入，并结合当前 17 课时轨道 s13_agent_teams/code.py 中的演进实现，帮助读者掌握一套可直接落地的"自治 Agent"运行时设计。

问题：Lead 逐一分派工作无法规模化

在 s09–s10 的实现中，teammate 只有被明确指派时才工作：lead 必须为每个 teammate 生成一个带具体 prompt 的 spawn 调用。假设任务板上有 10 个无主任务，lead 就要手动逐一分派——这在多 Agent 场景下不可扩展。

s11 引入的"真自治"模式是：

• teammate 主动扫描任务板（task board），认领无主（unclaimed）任务；
• 处理完后继续寻找下一份任务，而不是直接退出；
• 一个微妙问题：经历上下文压缩（s06 课程的主题）后，agent 可能"忘记自己是谁"。s11 用**身份重注入（identity re-injection）**修复这一点。

解决方案：带空闲周期的 teammate 生命周期

文档给出的核心生命周期是一个 spawn → WORK → IDLE →（循环回 WORK 或 SHUTDOWN）的状态机：

Teammate lifecycle with idle cycle:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+---+---+                +-------+
    |
    | stop_reason != tool_use (or idle tool called)
    v
+--------+
|  IDLE  |  poll every 5s for up to 60s
+---+----+
    |
    +---> check inbox --> message? ----------> WORK
    |
    +---> scan .tasks/ --> unclaimed? -------> claim -> WORK
    |
    +---> 60s timeout ----------------------> SHUTDOWN

Identity re-injection after compression:
  if len(messages) <= 3:
    messages.insert(0, identity_block)

状态机的四个关键转移条件：

1. WORK → IDLE：LLM 的 stop_reason != "tool_use"（自然停止），或 teammate 显式调用 idle 工具；
2. IDLE → WORK（消息恢复）：轮询期间发现收件箱（inbox）有消息；
3. IDLE → WORK（自动认领）：扫描 .tasks/ 发现无主任务，认领后进入工作；
4. IDLE → SHUTDOWN：60 秒超时，期间既无消息也无任务。

两个可调参数在源码中定义于 agents/s11_autonomous_agents.py：

POLL_INTERVAL = 5    # 空闲轮询间隔（秒）
IDLE_TIMEOUT = 60    # 空闲超时（秒），超时后自动 shutdown

轮询次数由 IDLE_TIMEOUT // POLL_INTERVAL 计算，即 60s / 5s = 12 次轮询机会。

工作机制一：双阶段 teammate 主循环

源码中 teammate 的循环（TeammateManager._loop）严格对应文档描述的两个阶段，见 agents/s11_autonomous_agents.py：

def _loop(self, name, role, prompt):
    while True:
        # -- WORK PHASE --
        messages = [{"role": "user", "content": prompt}]
        for _ in range(50):
            response = client.messages.create(...)
            if response.stop_reason != "tool_use":
                break
            # execute tools...
            if idle_requested:
                break

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

实现上有几个值得注意的工程细节（对照 源码 L224-L302）：

• WORK 阶段每次迭代都先排空收件箱：调用 LLM 前先执行 BUS.read_inbox(name)，若发现 shutdown_request 消息则立即置为 shutdown 状态并返回，保证关闭握手（s10 引入的协议）在忙碌时也能被及时响应；
• idle 工具的特殊处理：idle 不是一个真实执行的工具，而是在工具分发时设置 idle_requested = True 标志并返回固定文本 "Entering idle phase. Will poll for new tasks."（L251-L253）；
• 异常降级：LLM 调用抛异常时，teammate 被置为 idle 并安全返回，而不是让整个团队进程崩溃（L241-L243）；
• WORK 阶段有 50 轮工具调用上限，防止单任务无限循环烧 token。

teammate 的系统提示词（L217-L220）明确告诉模型自治行为契约：

sys_prompt = (
    f"You are '{name}', role: {role}, team: {team_name}, at {WORKDIR}. "
    f"Use idle tool when you have no more work. You will auto-claim new tasks."
)

工作机制二：空闲轮询 inbox 与任务板

IDLE 阶段的轮询逻辑（对应文档第 2 点的 _idle_poll，实现见 源码 L266-L301）：

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)
            messages.append({"role": "user",
                "content": f"<auto-claimed>Task #{unclaimed[0]['id']}: "
                           f"{unclaimed[0]['subject']}</auto-claimed>"})
            return True
    return False  # timeout -> shutdown

从源码实现看，轮询顺序是"先 inbox、后任务板"：消息（来自 lead 或队友的协作信息）优先级高于自取任务；且每次轮询失败（认领冲突等）会 continue 继续下一轮，直到 12 轮耗尽才真正 shutdown。

inbox 的底层存储是 MessageBus（L81-L121）：每个 teammate 对应 .team/inbox/{name}.jsonl 一个 JSONL 文件，send 追加一行、read_inbox 读取后清空（drain 语义），合法消息类型包括 message、broadcast、shutdown_request、shutdown_response、plan_approval_response（L65-L71）。这种基于文件邮箱的异步通信，是 s09/s10 建立的协议层在 s11 中的直接复用。

工作机制三：任务板扫描与原子认领

任务板以 .tasks/task_*.json 文件为持久化载体。扫描函数返回满足三个条件的任务（L128-L137）：

def scan_unclaimed_tasks() -> list:
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed

三个条件分别保证：任务未开始（pending）、无主（owner 为空）、且没有被依赖任务阻塞（blockedBy 为空）——这意味着 teammate 会自动尊重任务 DAG 的依赖顺序，这是文档"Try It"第 3 条"创建带依赖的任务，观察 teammate 遵守阻塞顺序"的行为来源。

认领函数 claim_task 则通过全局 _claim_lock 线程锁实现临界区保护（L140-L155）：

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 existing_owner := task.get("owner"):
            return f"Error: Task {task_id} has already been claimed by {existing_owner}"
        if (status := task.get("status")) != "pending":
            return f"Error: Task {task_id} cannot be claimed because its status is '{status}'"
        if task.get("blockedBy"):
            return f"Error: Task {task_id} is blocked by other task(s) and cannot be claimed yet"
        task["owner"] = owner
        task["status"] = "in_progress"
        path.write_text(json.dumps(task, indent=2))
    return f"Claimed task #{task_id} for {owner}"

注意它是读—校验—写在同一把锁内完成：多个 teammate 同时轮询到同一个无主任务时，只有第一个写入 owner 的成功，其余拿到 "already been claimed by ..." 错误。IDLE 轮询中若认领返回 Error: 前缀会 continue 重试下一轮（L284-L286），从而平滑处理认领竞争。这个原子性契约在当前轨道的测试套件 tests/test_agent_teams_runtime.py 中有专门验证，例如并发认领同一任务时只有一个 owner 成功（test_inbox_delivery_is_runtime_owned、claim_task 相关断言）。

工作机制四：上下文压缩后的身份重注入

这是 s11 独有的一个防御性机制。当上下文压缩发生后，messages 列表会缩短；teammate 可能丢失"我是谁、我的角色是什么"的认知。s11 的判据非常朴素：消息数小于等于 3 条即认为发生过压缩，此时在消息列表头部插入身份块：

if len(messages) <= 3:
    messages.insert(0, identity_block)

身份块构造逻辑见 make_identity_block（L159-L163）：

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

在实际调用点（自动认领任务恢复 WORK 时，L291-L295），注入的是"一问一答"两条消息，让角色设定以对话形式成立：

messages.insert(0, make_identity_block(name, role, team_name))
messages.insert(1, {"role": "assistant", "content": f"I am {name}. Continuing."})

这个手法可以推广为通用经验：长生命周期 agent 的关键状态（身份、角色约束）不应只存在于 system prompt 或对话开头，而应在上下文被重写后主动重新锚定。

相对 s10 的变更清单

文档中的对照表完整如下，是理解 s11 增量边界的最快方式：

Component	Before (s10)	After (s11)
Tools	12	14 (+idle, +claim_task)
Autonomy	Lead-directed	Self-organizing
Idle phase	None	Poll inbox + task board
Task claiming	Manual only	Auto-claim unclaimed tasks
Identity	System prompt	+ re-injection after compress
Timeout	None	60s idle -> auto shutdown

工具增量在源码中可精确核对：teammate 侧工具池（_teammate_tools，L342-L365）在 s02 的四个基础文件/终端工具（bash、read_file、write_file、edit_file）与 s09/s10 的协作工具（send_message、read_inbox、shutdown_response、plan_approval）之上，新增 idle 与 claim_task 两个工具；lead 侧分发表 TOOL_HANDLERS（L469-L484）共 14 个条目，其中 lead 调用 idle 会返回 "Lead does not idle."（lead 常驻不空闲），lead 调用 claim_task 则把自己当作 owner 认领。

新增工具	输入参数	语义
idle	无	声明"我手头没有工作了"，触发进入 IDLE 轮询阶段
claim_task	task_id: integer（必填）	按 ID 原子认领任务板上的任务

运行与验证

运行方式（需先配置 MODEL_ID 等环境变量，参见 requirements.txt 与 README 的 Quick Start）：

cd learn-claude-code
python agents/s11_autonomous_agents.py

进入交互 REPL 后，文档给出的 5 条验证路径：

1. Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim. —— 验证空闲轮询 + 自动认领；
2. Spawn a coder teammate and let it find work from the task board itself —— 验证 spawn 时只给角色、不给具体任务；
3. Create tasks with dependencies. Watch teammates respect the blocked order. —— 验证 blockedBy 过滤；
4. 输入 /tasks 查看任务板（含 owner 标记）；
5. 输入 /team 监控各 teammate 处于 working 还是 idle。

两个斜杠命令的实现可直接阅读：/team 打印 TEAM.list_all()（团队名 + 每个成员的名字/角色/状态，L564-L566）；/tasks 遍历 .tasks/task_*.json，用 [ ]/[>]/[x] 标记 pending/in_progress/completed 并附带 @owner（L570-L577）。团队状态持久化在 .team/config.json，_set_status 每次状态迁移都会落盘，因此进程外也能观察 teammate 生命周期。

演进视角：当前 17 课时轨道中的自治认领

需要说明的适用前提：README 指出本仓库存在两条轨道——docs/ 与 agents/ 是保留旧链接的 Legacy 12 课时轨道，根级 s01_*–s17_* 目录是当前轨道。按 Legacy-to-Current 映射表，旧 s11 的"autonomous task claiming"并入新 s13 Agent Teams。当前实现 s13_agent_teams/code.py 将同样的 WORK/IDLE 思想工程化得更完整，可作为进阶参照：

• 空闲扫描间隔缩短为 IDLE_SCAN_INTERVAL = 2.0 秒（L1053）；
• wait_for_work 用阻塞式 BUS.wait_for_messages 等待消息，无消息时调用 claim_next_task 原子认领（L1252-L1276），认领成功即以 [Auto-claimed task ...] 消息注入对话并附带该任务绑定的工作目录；
• claim_next_task 在 task_lock 下额外保证一个 teammate 同时只持有一个任务（teammate_assignments 已有分配则直接返回 None，L1070-L1079），并叠加了 can_start 依赖检查与 worktree 可用性校验（L1056-L1067）。

从源码结构看，两条轨道共享同一套核心契约——scan_unclaimed_tasks（pending、无主、未阻塞）+ 锁保护的 claim_task + 空闲轮询恢复——差异主要在超时策略（s11 硬编码 60s 后 shutdown；新轨道把生命周期交给 lead/用户管理）和并发语义的加固。

设计要点回顾

1. 自治不来自新工具，来自循环结构：idle 和 claim_task 只是两个小工具，真正的机制是 IDLE 阶段的"轮询 → 恢复/超时"状态机；
2. 文件即协调介质：.tasks/*.json 与 .team/inbox/*.jsonl 让多个线程（未来甚至多进程）通过共享磁盘状态协作，_claim_lock 保证单进程内原子性；
3. 认领必须幂等且可失败：认领函数返回 Error: 前缀的软错误，轮询方以 continue 消化竞争失败，而不是抛异常；
4. 身份是需要在压缩后重新注入的状态，len(messages) <= 3 是一个低成本、可解释的压缩探测启发式。

延伸阅读：前序文档 s10 团队协议 解释了 shutdown/plan 两种 request-response 握手（s11 的 IDLE 轮询中同样会响应 shutdown_request），以及 s09 Agent Teams 中 MessageBus 与 teammate 持久化的基础设计。