learn-claude-code s08 背景任务：守护线程跑慢命令，通知队列把结果注入 Agent Loop

本篇技术指南基于 learn-claude-code 课程的 s08 章节文档（docs/en/s08-background-tasks.md），完整讲解「后台执行（Background Execution）」这一 harness 层能力：为什么 npm install 这类分钟级命令不能阻塞 agent loop、如何用守护线程 + 线程安全通知队列让模型在命令运行期间继续思考、以及结果如何以 <background-results> 消息的形式在每次 LLM 调用前被排空注入。读完你可以独立实现并运行 agents/s08_background_tasks.py，并理解其在完整版 harness 中如何演进。

一、问题：阻塞式循环让模型空等

课程文档首先给出动机：某些命令天然耗时——npm install、pytest、docker build 动辄几分钟。在阻塞式（blocking）agent loop 中，工具调用不返回，模型就只能空等；如果用户要求「装依赖，同时创建配置文件」，agent 只能串行执行，而不是并行。

s08 章节给出的答案是：慢操作交给后台线程，主循环保持单线程，只有子进程 I/O 是并行的。文档用一张时序图概括了整个设计：

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 -- the agent doesn't block while the command runs."

二、核心实现：BackgroundManager 的四个环节

整个机制由 agents/s08_background_tasks.py 中的 BackgroundManager 一个类承载，配合 agent loop 的排空逻辑。下面按文档的四个步骤展开，并补齐源码中比文档代码片段更完整的细节。

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

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

见 agents/s08_background_tasks.py。两个数据结构各司其职：

• self.tasks：所有任务的状态注册表，task_id 映射到 {status, result, command}，供 check_background 工具主动查询；
• self._notification_queue：完成事件的「待投递队列」，由 _lock 保护，是后台线程与主循环之间唯一的跨线程通信通道。

模块级单例 BG = BackgroundManager() 在 L111 创建，被工具处理器和 agent loop 共享。

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

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

见 agents/s08_background_tasks.py。三个设计点：

• task_id 取 UUID 前 8 位，短且唯一，模型在后续对话中用它查询状态；
• daemon=True：守护线程不阻止进程退出——用户 q 退出 REPL 时，挂着的后台子进程不会卡住主程序关闭；
• 立即返回：run() 本身不做任何子进程操作，工具调用的 tool_result 只是 Background task xxxx started: <命令前 80 字符>，loop 随即继续处理下一轮工具调用或模型响应。

3. _execute()：子进程执行、输出截断、状态入队

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],
        })

见 agents/s08_background_tasks.py。这里比文档摘要多了两层值得注意的细节：

• 三级状态机：completed / timeout / error。文档只展示了超时分支，而源码中通用 Exception 也兜底为 error 状态，保证任何异常都能进入队列而非让守护线程静默死亡；
• 两级截断：完整输出截断到 50000 字符（存 tasks 注册表），而通知队列里只放前 500 字符——注入进 LLM 消息的是短摘要，完整结果模型可以再用 check_background 查询（check 返回时读的是注册表里的完整结果）；
• cwd=WORKDIR：与同步 bash 工具一致，后台命令也在当前工作目录执行；
• 空输出统一替换为 (no output)，避免向模型注入空字符串。

4. drain_notifications()：每次 LLM 调用前排空队列

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

见 agents/s08_background_tasks.py。list(...) + clear() 在同一把锁内完成，保证「取出并清空」是原子操作，不会丢失或重复投递通知。

排空动作发生在 agent loop 的每一次 LLM 调用之前：

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,
        )
        ...

见 agents/s08_background_tasks.py。要点：

• 通知被包装成一条 role: "user" 的消息，用 <background-results>...</background-results> 包裹，每行格式为 [bg:<task_id>] <status>: <result 摘要>；
• 完成的任务不会自己唤醒 agent——它只在主循环下一次迭代、下一次 LLM 调用前才被收集。这与后续章节（如 s11）中 <task_notification> 的「占位 tool_result + 后续轮次收集」模式一脉相承；
• 文档特别强调：loop 保持单线程，只有子进程 I/O 被并行化。主循环没有任何竞争条件，所有并发被隔离在线程池（守护线程）那一侧。

三、工具面变化：从 s07 到 s08

s08 在 s07 的基础上把工具集精简为 6 个（4 个基础工具 + 2 个后台工具），并注册进 TOOL_HANDLERS：

工具	说明
bash	阻塞式 shell 执行，120s 超时，输出截断 50000 字符，内置危险命令黑名单（rm -rf /、sudo、shutdown、reboot、> /dev/）
read_file / write_file / edit_file	文件操作，路径经 safe_path() 校验必须落在 WORKDIR 内
background_run	新增。在后台线程运行命令，立即返回 task_id
check_background	新增。查询单个任务（传 task_id）或列出全部任务状态

其中 check_background 的实现见 agents/s08_background_tasks.py：不传 task_id 时返回 task_id: [status] command 的多行清单；传了未知 id 返回 Error: Unknown task <id>；仍在运行则结果位置显示 (running)。

工具 schema 定义见 agents/s08_background_tasks.py。系统提示也同步更新为 You are a coding agent at {WORKDIR}. Use background_run for long-running commands.（L46），显式引导模型对长命令使用后台工具——因为 s08 阶段 harness 不做关键词推断，由模型自己决定哪个工具。

文档给出的 s07 → s08 差异总表：

组件	s07	s08
工具数	8	6（base + background_run + check）
执行模式	仅阻塞	阻塞 + 后台线程
通知机制	无	每轮循环排空队列
并发	无	守护线程

四、在完整版 harness 中的演进与测试佐证

在课程最终的整合版 agents/s_full.py（SECTION: background (s08)）中，同一机制有若干可观察的演进：

• 通知队列从「锁 + 列表」换成标准库 queue.Queue，drain() 用 get_nowait() 循环取出，线程安全性不再依赖手写锁；
• background_run 暴露了 timeout 参数（默认 120 秒），工具处理器注册为 BG.run(kw["command"], kw.get("timeout", 120))，见 agents/s_full.py；
• 异常分支统一为 error 状态（不再区分 timeout），check() 对 result is None 的 running 任务返回 [running] (running) 占位。

这一行为有单元测试直接验证：tests/test_s_full_background.py 中的 test_check_returns_running_placeholder_when_result_is_none 构造 status="running", result=None 的任务，断言 check("abc123") == "[running] (running)"。该测试通过 mock anthropic/dotenv 模块并 chdir 到临时目录加载被测模块（load_s_full_module），不依赖真实 API Key，可以在本地直接跑。

五、动手运行

依赖与环境（依据 requirements.txt）：

pip install anthropic python-dotenv   # 另需 pyyaml（课程其他章节使用）

程序要求 MODEL_ID 环境变量（模块加载时 os.environ["MODEL_ID"] 为硬依赖），并支持通过 .env（load_dotenv(override=True)）与 ANTHROPIC_BASE_URL 指定网关地址；若设置了 ANTHROPIC_BASE_URL，源码会主动弹出 ANTHROPIC_AUTH_TOKEN（agents/s08_background_tasks.py）。

按文档「Try It」启动：

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

文档推荐的三个验证提示词：

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 的 tool_result 是否立即返回 task_id（而非命令输出）；后续轮次消息中是否出现 <background-results> 块；同一批通知是否只被注入一次（drain 的清空语义保证不会重复投递）。

六、小结：这个模式值得记住的三件事

1. 单线程循环 + 无锁投递点：并发被压缩到「后台线程写队列、主循环读队列」两个接触面上，agent loop 本身没有任何竞态，可推理性远好于让多个线程直接改 messages；
2. 通知是拉（pull）而非推（push）：完成事件不会中断模型正在进行的轮次，而是在下一次 LLM 调用前随 <background-results> 消息一并送达——这与课程后续 s11 章节（新章节体系中的 s11_background_tasks，改为由模型在 bash 工具中显式传 run_in_background 参数触发，并以 <task_notification> 注入）是同族设计，s08 是其最早、最简的版本；
3. 所有跨线程数据都截断：50000 字符存注册表、500 字符进通知、命令名截断 80 字符——harness 对 LLM 上下文的每一字节都有预算意识。

掌握这一章后，你具备的能力是：为任意 agent loop 增加「慢命令不阻塞、结果可追踪、完成必通知」的后台执行层，并且知道它如何在完整版 harness（agents/s_full.py）中被进一步打磨与测试覆盖。