⭐ 后台任务 (Background Tasks)

s01 > s02 > s03 > s04 > s05 > s06 | [[s07-任务系统|s07]] > [ s08 ] s09 > s10 > s11 > s12

"慢操作丢后台，agent 继续想下一步" — 后台线程跑命令，完成后注入通知。Harness 层的核心能力：分离 I/O 等待与思考。

问题

有些命令要跑好几分钟：npm install、pytest、docker build。[[s07-任务系统|s07]] 的任务图虽然解决了依赖编排，但每个任务本身还是阻塞式的 — 模型只能干等子进程结束。

更糟糕的是，用户说"装依赖，顺便建个配置文件"，智能体却只能一个一个来，白白浪费 LLM 调用的等待时间。

解决方案

在 harness 中引入 BackgroundManager：用守护线程执行子进程，把结果排队，在每次 LLM 调用前注入。主循环保持单线程，只有子进程 I/O 被并行化。

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

时间线:
Agent --[spawn A]--[spawn B]--[other work]----
             |          |
             v          v
          [A runs]   [B runs]      (并行)
             |          |
             +-- 结果在下次 LLM 调用前注入 --+

工作原理

1. BackgroundManager

用线程安全的队列追踪后台任务：

class BackgroundManager:
    def __init__(self):
        self.tasks = {}
        self._notification_queue = []
        self._lock = threading.Lock()

2. 非阻塞启动

run() 启动守护线程后立即返回，不阻塞主循环：

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

3. 结果排队

子进程完成后结果进入通知队列，带上截断的输出：

def _execute(self, task_id, command):
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=300)
        output = (r.stdout + r.stderr).strip()[:50000]
    except subprocess.TimeoutExpired:
        output = "Error: Timeout (300s)"
    with self._lock:
        self._notification_queue.append({
            "task_id": task_id, "result": output[:500]})

4. 注入机制

每次 LLM 调用前排空通知队列，把结果包装为 system 消息注入上下文：

def agent_loop(messages: list):
    while True:
        notifs = BG.drain_notifications()
        if notifs:
            notif_text = "\n".join(
                f"[bg:{n['task_id']}] {n['result']}" for n in notifs)
            messages.append({"role": "user",
                "content": f"<background-results>\n{notif_text}\n"
                           f"</background-results>"})
            messages.append({"role": "assistant",
                "content": "Noted background results."})
        response = client.messages.create(...)

循环的主线程从未阻塞。守护线程在后台安静执行，LLM 继续思考下一组操作。

与 s07 的关系

后台任务与 [[s07-任务系统|任务系统 (s07)]] 是互补而非替代关系：

维度	s07 任务图	s08 后台任务
用途	长期目标编排与依赖管理	单步命令的非阻塞执行
粒度	一个任务 = 一个结构化工作单元	一个后台任务 = 一条 shell 命令
并发	任务图描述并行可能性，由 agent 调度	守护线程真正并行执行
持久化	JSON 文件存磁盘，跨会话存活	纯内存，随会话销毁

实际使用时：在 s07 任务图中标记一个任务为 in_progress，然后用 s08 后台执行 去跑它的 shell 命令，agent 不必等它完成就可以开始下一个任务。

相对 s07 的变更

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

⭐ 核心路径标记

后台任务是整个课程的分水岭能力：

• 之前 (s01-s07)：所有操作都是阻塞的，agent 线性执行
• 从此 (s08 开始)：I/O 与思考分离，agent 可以同时"做"和"想"
• s09 [[s09-agent-teams|Agent 团队]]、s10 [[s10-team-protocols|团队协议]]、s12 [[s12-worktree-task-isolation|Worktree 隔离]] 都依赖此机制实现真正的并行

试一试

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

试试这些 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.
3. Run pytest in the background and keep working on other things

延展思考

• Timeout：_execute 中的 300s 超时是硬编码的 — 实际产品中应作为参数传入
• 输出截断：50000 → 500 字符的两级截断防止了上下文爆炸，但复杂任务可能需要流式输出
• 线程安全：self._lock 保护通知队列的读写 — 多线程环境中不加锁的队列是竞态条件的温床
• 与 s09 的桥梁：后台任务为 [[s09-agent-teams|多 agent 并行]] 提供了执行原语 — 每个子 agent 都可以有自己的后台任务