learn-claude-code 后台任务机制详解：慢命令进后台线程，通知队列注入 Agent Loop

本篇基于 learn-claude-code 仓库的后台任务章节文档 docs/zh/s08-background-tasks.md，完整讲解"后台执行"这一 harness 层机制的动机、架构与实现：BackgroundManager 如何用守护线程运行耗时命令、如何用线程安全的通知队列在每轮 LLM 调用前注入结果，并结合 agents/s08_background_tasks.py 的源码给出可运行的配置与实操步骤。读完你能掌握在不阻塞 Agent Loop 的前提下并行执行 npm install、pytest 等慢命令的完整方案。

问题：阻塞式循环里模型只能干等

文档开头给出的问题是：有些命令要跑好几分钟——npm install、pytest、docker build。在阻塞式循环中，工具调用必须等命令返回才能继续，模型只能干等；用户说"装依赖，顺便建个配置文件"，Agent 却只能一个一个来。

learn-claude-code 的口号是 "Bash is all you need"，它用 17 课渐进式地把一个 claude code 风格的 agent harness 从 0 搭到 1。后台任务（文档中的 s08，现 17 课体系中的 s11）要解决的正是其中"运行长任务"阶段的问题。仓库 README 给这一课的定位是：「慢操作丢后台，agent 继续想下一步」—— 后台线程跑命令，完成后注入通知，属于 Harness 层的机制：模型继续思考，harness 负责等待。

解决方案：主线程 + 后台线程 + 通知队列

文档给出的架构示意（主线程跑 agent loop，后台线程跑子进程，完成后把结果排入队列）：

Main thread                Background thread
+-----------------+        +-----------------+
| agent loop      |        | subprocess runs |
| ...             |        | ...             |
| [LLM call] <---+------- | enqueue(result) |
|  ^drain queue   |        +-----------------+
+-----------------+

Timeline:
Agent --[spawn A]--[spawn B]--[other work]----
             |          |
             v          v
          [A runs]   [B runs]      (parallel)
             |          |
             +-- results injected before next LLM call --+

关键洞察就一句话：Fire and forget —— 命令在跑的时候，agent 不阻塞。只有子进程 I/O 被并行化，agent loop 本身保持单线程。

工作原理：BackgroundManager 源码逐段解析

文档给出了四步工作原理，下面结合 agents/s08_background_tasks.py 的完整实现逐一展开。

1. 线程安全的任务注册表 + 通知队列

BackgroundManager（agents/s08_background_tasks.py#L50-L54）只维护三个状态：

class BackgroundManager:
    def __init__(self):
        self.tasks = {}  # task_id -> {status, result, command}
        self._notification_queue = []  # completed task results
        self._lock = threading.Lock()

• tasks：任务注册表，task_id -> {status, result, command}，供 check_background 工具随时查询状态；
• _notification_queue：已完成任务的结果队列，等待在 LLM 调用前被排空；
• _lock：保护队列的 threading.Lock，因为后台线程写、主线程读。

2. run()：启动守护线程，立即返回

def run(self, command: str) -> str:
    """Start a background thread, return task_id immediately."""
    task_id = str(uuid.uuid4())[:8]
    self.tasks[task_id] = {"status": "running", "result": None, "command": command}
    thread = threading.Thread(
        target=self._execute, args=(task_id, command), daemon=True
    )
    thread.start()
    return f"Background task {task_id} started: {command[:80]}"

实现上有三个值得注意的细节：

• task_id 取 uuid4 的前 8 位（agents/s08_background_tasks.py#L58），足够区分并发任务，又便于模型在后续轮次里引用；
• daemon=True：守护线程不会阻止进程退出——即使主循环结束，残留的后台命令线程也随进程一起被回收；
• 立即返回字符串 "Background task xxxx started: ..."，这个字符串就是回给 LLM 的 tool_result，模型拿到 task_id 后可以继续干别的。

3. _execute()：子进程执行、超时保护、结果截断

线程目标是 _execute（agents/s08_background_tasks.py#L66-L89），它是整个机制中防御性最强的部分：

def _execute(self, task_id: str, command: str):
    """Thread target: run subprocess, capture output, push to queue."""
    try:
        r = subprocess.run(
            command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=300
        )
        output = (r.stdout + r.stderr).strip()[:50000]
        status = "completed"
    except subprocess.TimeoutExpired:
        output = "Error: Timeout (300s)"
        status = "timeout"
    except Exception as e:
        output = f"Error: {e}"
        status = "error"
    self.tasks[task_id]["status"] = status
    self.tasks[task_id]["result"] = output or "(no output)"
    with self._lock:
        self._notification_queue.append({
            "task_id": task_id,
            "status": status,
            "command": command[:80],
            "result": (output or "(no output)")[:500],
        })

关键参数与行为：

参数 / 行为	取值	作用
shell=True	—	支持 &&、管道等 shell 语法，与 bash 工具一致
cwd=WORKDIR	进程启动时的当前目录	后台任务与主循环共享工作区
timeout=300	300 秒	超时被捕获为 timeout 状态，不会无限挂起
结果存储上限	[:50000] 字符	防止 pytest 全量输出撑爆 tasks 注册表
通知队列 preview	[:500] 字符	注入对话的只是结果摘要，控制 token 开销
空输出	"(no output)" 占位	保证 LLM 一定能读到有意义的反馈
状态机	running / completed / timeout / error	异常也走统一状态，模型可据此决策

注意截断的两级设计：完整输出（≤50000 字符）留在 tasks[task_id]["result"] 里，模型可以后续用 check_background task_id 查询；进入通知队列的只有 500 字符的摘要。

4. check() 与 drain_notifications()：两种结果获取方式

def check(self, task_id: str = None) -> str:
    """Check status of one task or list all."""
    if task_id:
        t = self.tasks.get(task_id)
        if not t:
            return f"Error: Unknown task {task_id}"
        return f"[{t['status']}] {t['command'][:60]}\n{t.get('result') or '(running)'}"
    lines = []
    for tid, t in self.tasks.items():
        lines.append(f"{tid}: [{t['status']}] {t['command'][:60]}")
    return "\n".join(lines) if lines else "No background tasks."

def drain_notifications(self) -> list:
    """Return and clear all pending completion notifications."""
    with self._lock:
        notifs = list(self._notification_queue)
        self._notification_queue.clear()
    return notifs

• check() 是拉模式：不带 task_id 时列出全部任务，带 task_id 时返回单个任务的状态与结果（运行中显示 (running)）；
• drain_notifications() 是推模式的触发点：加锁、复制、清空，原子地完成"取出并清空"，保证同一条通知只被注入一次。

Agent Loop 集成：每次 LLM 调用前排空队列

文档第 4 步是集成点。在 agents/s08_background_tasks.py#L188-L215 中，agent_loop 在每一轮调用 LLM 之前先排空通知队列，并把结果包装成一条 <background-results> 消息追加进对话：

def agent_loop(messages: list):
    while True:
        # Drain background notifications and inject as system message before LLM call
        notifs = BG.drain_notifications()
        if notifs and messages:
            notif_text = "\n".join(
                f"[bg:{n['task_id']}] {n['status']}: {n['result']}" for n in notifs
            )
            messages.append({"role": "user", "content": f"<background-results>\n{notif_text}\n</background-results>"})
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        ...

这个注入位置是设计的核心：

1. 不唤醒模型——后台完成不会打断正在进行的推理，而是"搭车"下一次 client.messages.create；
2. XML 标签包裹 <background-results>...</background-results>，让模型能区分这是系统事件而非用户输入；
3. 每行通知带 task_id 与状态，模型能把它和之前 background_run 返回的 task_id 对应起来。

配合系统提示 SYSTEM = "You are a coding agent at {WORKDIR}. Use background_run for long-running commands."（agents/s08_background_tasks.py#L46），引导模型主动把慢命令分给后台。

工具集：6 个工具与分发表

s08 的工具面是 6 个：4 个基础文件/命令工具 + 2 个后台专用工具，通过 TOOL_HANDLERS 分发表注册（agents/s08_background_tasks.py#L163-L170）：

TOOL_HANDLERS = {
    "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"]),
    "background_run":   lambda **kw: BG.run(kw["command"]),
    "check_background": lambda **kw: BG.check(kw.get("task_id")),
}

工具	说明	关键约束（源码可查）
bash	阻塞式 shell 命令	120s 超时；危险命令黑名单 rm -rf /、sudo、shutdown 等直接拦截
read_file	读文件，可选 limit 行数	路径必须 resolve 后仍在 WORKDIR 内（safe_path 校验）
write_file	写文件，自动建父目录	输出上限 50000 字符
edit_file	精确替换一段文本，找不到即报错	只替换第一处（replace(..., 1)）
background_run	后台线程跑命令，立即返回 task_id	300s 超时，结果截断见上文
check_background	查询单个任务或列出全部	task_id 可省略

bash 与 background_run 形成对照：前者 120 秒超时、同步返回输出；后者 300 秒超时、异步返回 task_id。模型可以按命令预期耗时自行分流。

相对 s07（Task System）的变更

文档给出的对比表（继承原文档）：

组件	之前 (s07)	之后 (s08)
Tools	8	6 (基础 + background_run + check)
执行方式	仅阻塞	阻塞 + 后台线程
通知机制	无	每轮排空的队列
并发	无	守护线程

s07 是任务系统（agents/s07_task_system.py，task_create / task_update / task_list / task_get 四个任务工具 + 4 个基础工具 = 8 个），而 s08 用 background_run / check_background 两个后台工具换掉了任务工具，回到 6 个工具。两课解决的是不同问题：s07 让"目标"在压缩后存活（落盘 JSON + 依赖图），s08 让"慢命令"不阻塞循环。

实操：运行与推荐 prompt

环境配置

运行前提（见 requirements.txt 与 .env.example）：

pip install -r requirements.txt   # anthropic>=0.25.0, python-dotenv>=1.0.0, pyyaml>=6.0
cp .env.example .env

.env 中需要配置：

ANTHROPIC_API_KEY=sk-ant-xxx        # 必填
MODEL_ID=claude-sonnet-4-6          # 必填，也可换 Anthropic 兼容服务商的模型
# ANTHROPIC_BASE_URL=...            # 可选，指向兼容端点；设置后会清掉 ANTHROPIC_AUTH_TOKEN

运行

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

程序以交互式 REPL 启动（提示符 s08 >>），输入 q / exit 退出。文档推荐的三个测试 prompt（英文 prompt 对 LLM 效果更好，也可以用中文）：

1. Run "sleep 5 && echo done" in the background, then create a file while it runs —— 验证"后台跑命令的同时创建文件"的并行能力；
2. Start 3 background tasks: "sleep 2", "sleep 4", "sleep 6". Check their status. —— 验证多任务并发与 check_background 的状态查询；
3. Run pytest in the background and keep working on other things —— 用真实慢命令验证通知注入时机。

观察要点：background_run 是否立即返回 task_id；后续轮次的 <background-results> 里是否出现对应 task_id 的完成通知；check_background 不带参数时是否列出全部任务。

延伸：从 legacy s08 到现 17 课体系的 s11

需要说明一点版本关系：本文档属于 legacy 12 课轨道（agents/ + docs/）。仓库 README 给出了映射表——old s08 对应 new s11（Background Tasks）。现行 17 课实现位于 s11_background_tasks/（s11_background_tasks/README.zh.md、s11_background_tasks/code.py），机制在两点上演进：

1. 显式参数取代独立工具：不再有 background_run 工具，而是给 bash 的 schema 增加 run_in_background 布尔参数，should_run_background() 只在 tool_name == "bash" 且参数明确为 True 时进入后台路径，不做关键词猜测；
2. 通知不复用 tool_use_id：后台命令先返回带 bg_id 的占位 tool_result（保持一个 tool_use 只对应一个 tool_result），完成结果在后续轮次以独立的 <task_notification> 事件注入，格式为 <task_id>...</task_id><status>completed</status>。

这些行为有自动化测试背书：tests/test_background_tasks.py 验证了"后台执行必须显式声明"、"权限检查先于后台分发"（rm -rf 类命令即使带 run_in_background: true 也会被 Permission 拦截、且不会创建后台任务）、"完成结果只在后续 LLM 调用前被收集一次"（collect_background_results() 第二次调用返回空列表）。

小结

• 机制：BackgroundManager 用守护线程 + 锁保护的通知队列，把子进程 I/O 并行化，agent loop 保持单线程；
• 关键参数：后台命令 300s 超时、结果存储 50000 字符、通知摘要 500 字符、task_id 为 8 位 UUID 前缀；
• 集成点：每次 client.messages.create 之前 drain_notifications()，以 <background-results> 消息注入；
• 可验证路径：文档 docs/zh/s08-background-tasks.md，legacy 实现 agents/s08_background_tasks.py，现行实现 s11_background_tasks/code.py，测试 tests/test_background_tasks.py。

这一课给出的模式可以抽象为通用结论：在 LLM 驱动的循环里，"并行化"应该发生在 I/O 层（子进程、网络请求），而不是循环本身；结果回收统一挂到"下一次模型调用前"这个天然同步点，既能避免阻塞，又不需要引入回调、事件总线等更复杂的并发设施。