Harness实战：后台任务——Fire-and-Forget，Agent不等了

用户: 帮我写一个 fizzbuzz.py（1到100），然后跑一下确认输出正确

Agent 的执行:
1. 写 fizzbuzz.py                     → 2s
2. task("跑 python fizzbuzz.py 验证")  → 15s  ← 阻塞等待
3. 告诉用户结果                        → 1s

同步 task / parallel_tasks:

  用户请求 → Agent → [子任务执行中...60s...] → 拿到结果 → 回复用户
                     ^^^^^^^^^^^^^^^^^^^^^^^^
                     这段时间 Agent 挂起，用户等待

后台 background_task:

  用户请求 → Agent → 提交后台任务 → 立即回复"已提交，task_id: abc123"
                          │
                          └→ [子任务在后台执行中...60s...]
                                                    │
  用户继续对话 → Agent 继续工作                      │
                          │                         │
  用户: "跑完了吗？"       → check_task("abc123") ←──┘
                          → "done, 全部通过"

import threading
import uuid
from concurrent.futures import ThreadPoolExecutor, Future

# 全局线程池：后台任务共享，避免每次创建/销毁
bg_pool = ThreadPoolExecutor(max_workers=4)

# {task_id: Future} — 注册表
bg_tasks: dict[str, Future] = {}
bg_lock = threading.Lock()

BACKGROUND_TASK_TOOL = {
    "name": "background_task",
    "description": (
        "Launch a subtask in the background. Returns immediately with a task_id. "
        "The task keeps running while you continue the conversation. "
        "Use for long-running work: tests, builds, large analysis. "
        "Check status later with check_task."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "prompt": {
                "type": "string",
                "description": "Task description for the background subagent",
            }
        },
        "required": ["prompt"],
    },
}

def run_background_task(prompt: str) -> str:
    """提交后台任务，立即返回 task_id。"""
    task_id = uuid.uuid4().hex[:8]

    future = bg_pool.submit(run_subagent, prompt)

    with bg_lock:
        bg_tasks[task_id] = future

    # 完成时在终端提醒用户
    def on_done(fut: Future, tid: str = task_id):
        status = "error" if fut.exception() else "done"
        print(f"\n  📬 Background task [{tid}] {status}!")

    future.add_done_callback(on_done)

    return json.dumps({"task_id": task_id, "status": "submitted"})

CHECK_TASK_TOOL = {
    "name": "check_task",
    "description": (
        "Check the status of a background task. "
        "Returns {status: 'running'} if still in progress, "
        "or {status: 'done', result: '...'} when finished, "
        "or {status: 'error', error: '...'} if it failed."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "task_id": {
                "type": "string",
                "description": "The task_id returned by background_task",
            }
        },
        "required": ["task_id"],
    },
}

def run_check_task(task_id: str) -> str:
    """查询后台任务状态。"""
    with bg_lock:
        future = bg_tasks.get(task_id)

    if future is None:
        return json.dumps({
            "status": "not_found",
            "error": f"No task with id '{task_id}'"
        })

    if not future.done():
        return json.dumps({"status": "running"})

    exc = future.exception()
    if exc:
        return json.dumps({"status": "error", "error": str(exc)})

    return json.dumps({"status": "done", "result": future.result()})

TOOLS = BASE_TOOLS + [
    TASK_TOOL,
    PARALLEL_TASKS_TOOL,
    BACKGROUND_TASK_TOOL,  # 新增
    CHECK_TASK_TOOL,       # 新增
]

TOOL_HANDLERS = {
    **BASE_HANDLERS,
    "task": lambda **kw: run_subagent(kw["prompt"]),
    "parallel_tasks": lambda **kw: run_parallel_tasks(kw["tasks"]),
    "background_task": lambda **kw: run_background_task(kw["prompt"]),  # 新增
    "check_task": lambda **kw: run_check_task(kw["task_id"]),          # 新增
}

❯ 帮我写一个 fizzbuzz.py（1到100），写完后在后台跑一下确认输出正确，
  同时帮我写一个 README.md 说明用法

> write_file: fizzbuzz.py
> background_task: {"prompt": "Run python fizzbuzz.py and verify output"}
  🚀 Background task [e4a91c3f] submitted

> write_file: README.md

fizzbuzz.py 和 README.md 都已写好。
测试在后台跑着 (task_id: e4a91c3f)，你可以继续问我别的。

❯ 帮我看一下 pyproject.toml 里的 Python 版本要求对不对

> read_file: pyproject.toml

pyproject.toml 里写的 requires-python = ">=3.10"，和代码里用的
match/case 语法一致，没问题。

  📬 Background task [e4a91c3f] done!

❯ fizzbuzz 跑得怎么样？

> check_task: {"task_id": "e4a91c3f"}

运行正确。输出 100 行：Fizz 27次、Buzz 14次、FizzBuzz 6次，
其余为数字，总计 100 行，无异常。

if __name__ == "__main__":
    # ... 主循环 ...

    # 优雅退出：等待后台任务完成
    with bg_lock:
        pending = {tid: f for tid, f in bg_tasks.items() if not f.done()}
    if pending:
        print(f"\n⏳ Waiting for {len(pending)} background task(s)...")
        for tid, future in pending.items():
            try:
                future.result(timeout=300)
                print(f"  ✓ {tid} done")
            except Exception as e:
                print(f"  ✗ {tid} error: {e}")
    bg_pool.shutdown(wait=False)

submit()          → Future 创建，状态: running
on_done()        → Future 完成，终端通知
check_task()      → 读取 Future 结果
程序退出           → bg_pool.shutdown() 清理

SYSTEM = f"""You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain.

When a task can be broken into subtasks, use `parallel_tasks` to run them concurrently.
Put ALL tasks (including dependent ones) in a SINGLE `parallel_tasks` call — the scheduler handles ordering.
..."""

SYSTEM = f"""You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain.

You have several task delegation strategies:
- `task`: Run a subtask synchronously. Blocks until done, returns result.
- `parallel_tasks`: Run multiple subtasks concurrently with dependency ordering. Blocks until ALL complete.
- `background_task`: Fire-and-forget. Returns a task_id immediately. The task runs while you continue the conversation.
- `check_task`: Poll a background task by task_id. Returns running/done/error.

For background_task: launch it, tell the user the task_id, and keep working. Check later with check_task."""

PLAN_PROMPT = (
    "Analyze the task and outline a step-by-step plan. Do NOT execute anything yet. "
    "For each step, note which other steps it depends on (if any). "
    "Mark steps that can run in parallel."
)

PLAN_PROMPT = (
    "Analyze the task and outline a step-by-step plan. Do NOT execute anything yet. "
    "For each step, note which other steps it depends on (if any). "
    "Mark steps that can run in parallel. "
    "Identify long-running steps that could run in the background."
)

1. 写 fizzbuzz.py（同步，需要确认写入成功）
2. 在后台跑 python fizzbuzz.py 验证输出（后台，耗时长，不阻塞）
3. 写 README.md（同步，可以和 2 同时进行）

EXECUTE_PROMPT = (
    "Now execute it step by step. "
    "Use `parallel_tasks` to run multiple subtasks concurrently with dependency ordering. "
    "Only use sequential `task` calls when the next step depends on seeing the previous result."
)

EXECUTE_PROMPT = (
    "Now execute it step by step. "
    "Use `parallel_tasks` to run multiple subtasks concurrently with dependency ordering. "
    "Use `background_task` for long-running work that doesn't block progress. "
    "Use `check_task` to retrieve background results when needed. "
    "Only use sequential `task` calls when the next step depends on seeing the previous result to decide what to do."
)