第8课：后台执行——异步任务与通知队列

用户: "装依赖顺便建个配置文件"
  ↓
模型思考: 先装依赖
  ↓
工具调用: bash → npm install         ← 阻塞 120 秒
  ↓
(模型干等…… 什么也做不了)
  ↓
工具结果: added 847 packages
  ↓
模型思考: 现在建配置文件
  ↓
工具调用: bash → echo "DB_HOST=..." > .env
  ↓
完成。总耗时: 122 秒

主线程（Agent Loop）               后台线程（守护）
─────────────────────              ─────────────────
│ LLM 调用                         │
│   ↓                              │
│ 工具: run_in_background          │
│   ↓                              │
│ run() → 启动守护线程 ──────────→ │ _execute()
│   ↓ 立即返回 task_id             │   ↓
│                                  │ subprocess.run("npm install")
│ 工具: bash → 建 .env             │   (跑了 120 秒)
│   ↓                              │   ↓
│ LLM 调用                         │ 结果 → notification_queue
│   ↓                              │
│ drain_notifications() ←──────────┘
│   ↓
│ 注入 <background-results> 标签
│   ↓
│ LLM 看到后台任务完成

import threading
import queue
import subprocess
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class BackgroundTask:
    task_id: str
    command: str
    status: str = "running"        # running | completed | failed
    result: Optional[str] = None

class BackgroundManager:
    def __init__(self):
        self.tasks: dict[str, BackgroundTask] = {}
        self.notification_queue: queue.Queue = queue.Queue()
        self.lock = threading.Lock()

def run(self, command: str) -> str:
    task_id = f"bg_{len(self.tasks)}_{int(time.time())}"

    task = BackgroundTask(task_id=task_id, command=command)
    with self.lock:
        self.tasks[task_id] = task

    thread = threading.Thread(
        target=self._execute,
        args=(task_id, command),
        daemon=True,
    )
    thread.start()

    return task_id  # 立即返回，不等子进程

def _execute(self, task_id: str, command: str):
    try:
        r = subprocess.run(
            command, shell=True,
            capture_output=True, text=True,
            timeout=600,
        )
        output = (r.stdout + r.stderr).strip()
        result = output[:50000] if output else "(no output)"
        status = "completed"
    except subprocess.TimeoutExpired:
        result = "Error: Background task timed out (600s)"
        status = "failed"
    except Exception as e:
        result = f"Error: {e}"
        status = "failed"

    # 更新任务状态
    with self.lock:
        self.tasks[task_id].status = status
        self.tasks[task_id].result = result

    # 通知主线程
    self.notification_queue.put({
        "task_id": task_id,
        "command": command,
        "status": status,
        "result": result,
    })

def drain_notifications(self) -> list[dict]:
    notifications = []
    while True:
        try:
            n = self.notification_queue.get_nowait()
            notifications.append(n)
        except queue.Empty:
            break
    return notifications

bg_manager = BackgroundManager()

def agent_loop(messages: list):
    while True:
        # ——— 新增：排空后台通知 ———
        notifications = bg_manager.drain_notifications()
        if notifications:
            bg_results = format_background_results(notifications)
            messages.append({"role": "user", "content": bg_results})

        # 原有逻辑不变
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = dispatch_tool(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

def format_background_results(notifications: list[dict]) -> str:
    parts = ["<background-results>"]
    for n in notifications:
        parts.append(f"[{n['task_id']}] {n['command']}\n"
                     f"Status: {n['status']}\nResult: {n['result'][:2000]}")
    parts.append("</background-results>")
    return "\n".join(parts)

{"name": "run_in_background",
 "description": "Run a shell command in the background. Returns immediately "
                "with a task ID. Use for long-running commands like npm install, "
                "pytest, docker build. Results delivered as background notifications.",
 "input_schema": {"type": "object",
                  "properties": {"command": {"type": "string"}},
                  "required": ["command"]}}

模型: npm install 很慢 → run_in_background（立即返回 task_id）
模型: .env 很快 → bash → echo "DB_HOST=..." > .env → 完成
模型: ".env 已创建。npm install 在后台运行中。"
      …… 继续做其他事 ……
下一轮: drain_notifications() → [bg_0] npm install completed
模型: "依赖也安装完了，added 847 packages。"

# 进入项目目录
cd learn-claude-code

# 启动第八课
python agents/s08_background_tasks.py

s08 >> Run "sleep 5 && echo done" in the background, then create a file while it runs

> background_run:
  Background task 223025f0 started: sleep 5 && echo done

> write_file: Wrote 65 bytes          ← sleep 还在跑，Agent 已经在写文件了

> check_background:
  [running] sleep 5 && echo done      ← 轮询发现还在跑

> bash: (no output)                   ← 继续做其他事

  → 后台任务完成，输出: done

s08 >> Start 3 background tasks: "sleep 2", "sleep 4", "sleep 6". Check their status.

> background_run: Background task fa70445d started: sleep 2
> background_run: Background task ec97d9ed started: sleep 4
> background_run: Background task eafcece7 started: sleep 6

> check_background:                   ← 第一次检查
  fa70445d: [completed] sleep 2       ← 2秒的已完成
  ec97d9ed: [running] sleep 4         ← 4秒的还在跑
  eafcece7: [running] sleep 6         ← 6秒的还在跑

> check_background: ec97d9ed [completed]  ← 4秒的完成了
> check_background: eafcece7 [completed]  ← 6秒的也完成了

| Task ID    | Command   | Duration | Status       |
|------------|-----------|----------|--------------|
| fa70445d   | sleep 2   | 2s       | ✅ Completed |
| ec97d9ed   | sleep 4   | 4s       | ✅ Completed |
| eafcece7   | sleep 6   | 6s       | ✅ Completed |

s08 >> Run pytest in the background and keep working on other things

> background_run:
  Background task da186af7 started: pytest 2>&1

# 预告：s09 的团队协作
team = TeamManager()
team.add_agent("coder", system="你是一个代码编写专家")
team.add_agent("tester", system="你是一个测试工程师")
team.add_agent("reviewer", system="你是一个代码审查专家")

team.run("实现一个用户注册功能，包含测试和代码审查")