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

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

【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code

在 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 增量边界的最快方式：

工具增量在源码中可精确核对：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 认领。

运行与验证

运行方式（需先配置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 持久化的基础设计。

【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code