Agent Harness 工程：后台任务与异步通知

上一篇实现了 Task System。任务不再只是当前会话里的 Todo，而是有依赖、有状态、可持久化的 DAG；不同 Agent 可以认领已经就绪的任务，通过 checkpoint 保存进度，并在中断后继续执行。

但具体的工具调用仍然是同步的。

假设 Agent 通过 bash 执行下面这些命令：

pip install -r requirements.txt
npm run build
pytest
docker compose up

它们可能运行几十秒，也可能持续几分钟。同步工具会一直等待命令结束，模型在此期间拿不到工具结果，Agent Loop 也无法处理其它已经就绪的任务。

更合理的方式是把耗时操作放到后台：Harness 启动进程后立即返回 job_id，Agent 可以继续读取文件、处理其它 Task，或者先向用户报告当前进度；当后台进程结束时，运行时再把完成事件交给 Agent。

本章解决三个问题：

模型如何决定一个操作是否应该后台执行；
Harness 如何启动、监控、超时和取消后台 Job；
Job 完成后，如何通知一个可能已经结束当前 Agent Loop 的 Agent。

Task 不是 Job

后台任务很容易和上一章的 Task System 混在一起。先把两个概念分开：

对比维度	Task	Job
表达什么	一个需要完成的工作目标	一次具体的程序或外部操作
生命周期	可以跨会话、等待依赖、失败后重试	从启动到进程退出
示例	“完成搜索接口并验证”	“运行 `pytest tests/search`”
结果	业务结果、artifact、checkpoint	退出码、stdout、stderr
关系	一个 Task 可以启动多个 Job	一个 Job 通常服务于一个 Task

Task 负责协调“要完成什么”，Job 负责记录“某次耗时操作执行得怎么样”。

Job 成功也不应该自动把 Task 标记为 completed。例如测试命令退出码为 0，只能说明测试通过；Agent 还需要确认代码修改、文档和其它验证是否已经完成，再调用 task_complete。

哪些操作适合后台执行

不能简单地把所有 bash 调用都放进后台。

操作类型	建议模式	原因
`pwd`、`git status`、短文件查询	前台	结果马上用于下一步判断
安装依赖、完整构建、大型测试	后台	预计耗时较长，可以同时处理其它工作
启动开发服务器、监听器	后台	进程本来就不会主动结束
需要交互输入的命令	拒绝	最小 Job Runner 没有交互式终端
修改系统或需要高权限的命令	先审批	后台模式不能绕过原有权限边界
下一步必须立即依赖其输出的命令	前台	转入后台只会增加轮询与上下文切换成本

最小改动是在原来的 bash 工具中增加显式参数：

BASH_TOOL = {
    "type": "function",
    "function": {
        "name": "bash",
        "description": "Run a shell command in foreground or background.",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string"},
                "background": {
                    "type": "boolean",
                    "description": "Start a long-running command and return a job id.",
                },
                "timeout_seconds": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 3600,
                },
            },
            "required": ["command"],
        },
    },
}

是否后台执行由模型显式表达，Harness 不需要通过 if "pip install" in command 之类的关键词规则猜测。

但模型的选择仍然只是请求。权限 Hook、命令策略、最大并发数和超时上限必须在 Harness 中再次校验。一个不允许执行的命令，不能因为 background=true 就变得可执行。

后台执行的完整流程

后台执行不是简单地给 subprocess.run() 加一个参数。它至少需要六个环节：

模型请求后台执行命令；
Harness 完成权限与参数校验；
Job Manager 启动子进程并立即返回 job_id；
Agent 继续处理其它工作；
Job Manager 监控进程状态并生成完成事件；
Runtime Event Loop 收到事件，重新唤醒 Agent。

子进程和 Agent Loop 独立推进。Job 完成后先进入运行时事件队列，而不是伪装成用户消息；Agent 再按需读取完整输出。

这里有两个不同的循环：

Agent Loop 负责模型推理、工具调用和结果回传；
Runtime Event Loop 负责用户输入、后台 Job 事件和其它外部事件。

如果只有 Agent Loop，Agent 在给出最终回复后就停止了。此后即使后台进程完成，也没有新的模型调用能够接收通知。因此后台任务必须接在更外层的运行时循环上。

定义 Job 生命周期

一个最小 Job 可以使用下面几种状态：

from typing import Literal


JobStatus = Literal[
    "running",
    "succeeded",
    "failed",
    "cancelled",
    "timed_out",
    "lost",
]

Job 启动后进入 running；退出码决定 succeeded 或 failed，超时和主动取消是独立终态。Harness 重启后无法确认结果的进程标记为 lost，不自动假设失败。

lost 需要特别解释。

本章的最小实现由当前 Harness 进程保存 Popen 句柄。Harness 重启后，即使磁盘上还有 Job JSON，也无法可靠取得原进程的退出码。此时不能直接把 Job 标记为 failed，更不能自动重跑，因为原命令可能已经产生副作用。

因此，启动恢复器应该读取已有 Job JSON：已经处于终态的 Job 只恢复查询能力；仍然标记为 running、却找不到受管进程句柄的 Job 改为 lost，并生成一条内部事件等待 Agent 检查。lost 不是由下面的日常轮询产生的，而是重启恢复阶段对“不确定结果”的显式表达。

真正需要跨进程恢复的后台 Job，应该交给独立 Worker、系统服务或支持持久化状态的队列执行。

定义 Job 数据

Job 需要保存足够的诊断信息，但不应把完整日志塞进 JSON：

from dataclasses import asdict, dataclass


@dataclass
class BackgroundJob:
    id: str
    command_summary: str
    cwd: str
    status: JobStatus
    stdout_path: str
    stderr_path: str
    timeout_seconds: int
    task_id: str | None = None
    task_claim_token: str | None = None
    pid: int | None = None
    returncode: int | None = None
    created_at: str = ""
    started_at: str | None = None
    finished_at: str | None = None
    error: str | None = None

    def to_dict(self) -> dict:
        return asdict(self)

目录结构可以沿用上一章的 .agent：

.agent/
└── jobs/
    ├── job-a81d3f.json
    ├── job-a81d3f.stdout.log
    └── job-a81d3f.stderr.log

Job JSON 保存状态和路径，stdout、stderr 分开写入日志文件。这样命令输出再大，也不会直接撑满任务文件或模型上下文。

启动后台进程

Job Manager 维护 Job 元数据和当前进程的 Popen 句柄：

import json
import os
import signal
import subprocess
import time
import uuid
from pathlib import Path
from threading import RLock


def atomic_write_json(path: Path, data: dict) -> None:
    temp_path = path.with_suffix(path.suffix + ".tmp")
    temp_path.write_text(
        json.dumps(data, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temp_path, path)


class BackgroundJobManager:
    def __init__(self, root: Path):
        self.root = root.resolve()
        self.root.mkdir(parents=True, exist_ok=True)
        self.jobs: dict[str, BackgroundJob] = {}
        self.processes: dict[str, subprocess.Popen] = {}
        self.deadlines: dict[str, float] = {}
        self.lock = RLock()

    def _save(self, job: BackgroundJob) -> None:
        path = self.root / f"{job.id}.json"
        atomic_write_json(path, job.to_dict())

    def start(
        self,
        command: str,
        command_summary: str,
        cwd: Path,
        timeout_seconds: int = 300,
        task_id: str | None = None,
        task_claim_token: str | None = None,
    ) -> BackgroundJob:
        timeout_seconds = max(1, min(timeout_seconds, 3600))
        job_id = f"job-{uuid.uuid4().hex[:12]}"
        stdout_path = self.root / f"{job_id}.stdout.log"
        stderr_path = self.root / f"{job_id}.stderr.log"

        with stdout_path.open("ab", buffering=0) as stdout_file, stderr_path.open(
            "ab", buffering=0
        ) as stderr_file:
            process = subprocess.Popen(
                ["/bin/zsh", "-lc", command],
                cwd=cwd,
                stdin=subprocess.DEVNULL,
                stdout=stdout_file,
                stderr=stderr_file,
                start_new_session=True,
            )

        job = BackgroundJob(
            id=job_id,
            command_summary=command_summary,
            cwd=str(cwd),
            status="running",
            stdout_path=str(stdout_path),
            stderr_path=str(stderr_path),
            timeout_seconds=timeout_seconds,
            task_id=task_id,
            task_claim_token=task_claim_token,
            pid=process.pid,
            created_at=utc_now_text(),
            started_at=utc_now_text(),
        )
        with self.lock:
            self.jobs[job_id] = job
            self.processes[job_id] = process
            self.deadlines[job_id] = time.monotonic() + timeout_seconds
            self._save(job)
        return job

这里复用了上一章的 atomic_write_json() 思路：先写临时文件，再通过 os.replace() 替换 Job JSON。

command 只用于启动进程，写入磁盘的是经过审计 Hook 脱敏的 command_summary。如果命令中可能出现 token、密码或私有地址，不能把原始字符串直接保存到 Job JSON。

这个最小示例在 Popen() 成功后才保存 Job，二者之间仍有一个很小的崩溃窗口。要求严格恢复时，可以增加 starting 状态：先持久化 Job，再启动进程，最后写入 PID 并切换到 running。如果启动中途崩溃，恢复器同样把结果不确定的 Job 标记为 lost。

stdin 被设置为 DEVNULL，因此后台命令不能等待用户输入。需要交互式终端的工具必须使用专门的 PTY 实现，不能直接塞进这个 Job Manager。

start_new_session=True 让子进程拥有独立进程组，取消或超时时可以终止整个进程组，而不只是最外层 shell。

轮询、超时与结束事件

子进程启动后，Harness 不需要创建线程读取全部输出。stdout 和 stderr 已经由操作系统直接写入文件，Job Manager 只需要定期调用 poll()：

def tail_text(path: str, max_bytes: int = 2000) -> str:
    try:
        with open(path, "rb") as file:
            file.seek(0, os.SEEK_END)
            size = file.tell()
            file.seek(max(0, size - max_bytes))
            return file.read().decode("utf-8", errors="replace").strip()
    except OSError:
        return ""


def stop_process_group(process: subprocess.Popen) -> None:
    try:
        os.killpg(process.pid, signal.SIGTERM)
        process.wait(timeout=3)
    except subprocess.TimeoutExpired:
        os.killpg(process.pid, signal.SIGKILL)
        process.wait()
    except ProcessLookupError:
        pass

然后实现状态收集：

def poll_jobs(manager: BackgroundJobManager) -> list[dict]:
    events = []

    with manager.lock:
        for job_id, process in list(manager.processes.items()):
            job = manager.jobs[job_id]
            timed_out = time.monotonic() >= manager.deadlines[job_id]

            if process.poll() is None and timed_out:
                stop_process_group(process)
                job.status = "timed_out"

            returncode = process.poll()
            if returncode is None:
                continue

            if job.status == "running":
                job.status = "succeeded" if returncode == 0 else "failed"

            job.returncode = returncode
            job.finished_at = utc_now_text()
            if job.status != "succeeded":
                job.error = redact_sensitive_text(tail_text(job.stderr_path))
            manager._save(job)

            summary_path = (
                job.stdout_path if job.status == "succeeded" else job.stderr_path
            )
            events.append(
                {
                    "event_id": f"job-finished:{job.id}",
                    "kind": "background_job_finished",
                    "job_id": job.id,
                    "task_id": job.task_id,
                    "status": job.status,
                    "returncode": job.returncode,
                    "summary": redact_sensitive_text(tail_text(summary_path)),
                }
            )

            del manager.processes[job_id]
            del manager.deadlines[job_id]

    return events

通知只携带有限长度的日志尾部。完整输出仍然保留在文件里，由 Agent 通过 job_output 按需读取。

event_id 应该稳定。运行时采用至少一次投递时，同一个事件可能因为重试再次出现，Agent 可以通过 event_id 去重；相比静默丢失完成通知，偶尔重复一次更容易恢复。

这里的 redact_sensitive_text() 代表已有的脱敏 Hook。stderr 摘要会写入 Job JSON，完成摘要还会进入模型上下文，两者都不能绕过敏感信息过滤。完整日志也需要限制文件权限、保留时间和读取者。

暴露 Job 工具

除了带 background 参数的 bash，还需要几个只操作 Job 的工具：

工具	作用
`job_status`	查看状态、PID、退出码和关联 Task
`job_output`	按 stdout 或 stderr 读取有限长度的日志
`job_cancel`	终止运行中的进程组
`job_list`	列出当前运行或最近完成的 Job

取消操作只改变 Job 的执行状态，不改变关联 Task：

def cancel_job(manager: BackgroundJobManager, job_id: str) -> BackgroundJob:
    with manager.lock:
        job = manager.jobs[job_id]
        process = manager.processes.get(job_id)

        if job.status == "running" and process and process.poll() is None:
            stop_process_group(process)
            job.status = "cancelled"
            manager._save(job)

        return job

监控器会在下一次轮询时补齐退出码和结束时间，并发布同样的完成事件。

bash 处理函数可以保持一条清晰分支：

def run_bash(
    command: str,
    background: bool = False,
    timeout_seconds: int = 300,
) -> str:
    if background:
        task_id = RUNTIME.current_task_id
        job = JOB_MANAGER.start(
            command=command,
            command_summary=redact_sensitive_text(command)[:500],
            cwd=WORKDIR,
            timeout_seconds=timeout_seconds,
            task_id=task_id,
            task_claim_token=RUNTIME.claim_token if task_id else None,
        )
        return json.dumps(
            {
                "job_id": job.id,
                "status": job.status,
                "task_id": job.task_id,
            },
            ensure_ascii=False,
        )

    return run_bash_foreground(command, timeout_seconds)

这个函数默认仍然前台执行，避免所有命令突然改变语义。

权限检查应该发生在 run_bash() 之前，因此前台和后台走同一套参数校验、审批、沙箱和审计 Hook。job_cancel 也需要确认 Job 属于当前运行空间，不能仅凭模型提供的 job_id 终止任意系统进程。job_status 和 job_list 返回模型前还要移除 task_claim_token，保持与上一章 Task 工具相同的 fencing token 边界。

不要把通知伪装成用户消息

Job 完成事件来自 Harness，不是用户输入。

如果把它直接追加成普通 user 消息，模型可能误以为“测试完成了”是用户说的，也会让后续摘要和审计记录失真。

更合理的做法是把事件放进运行时状态，再由 Prompt Assembler 渲染为短暂的动态 Section：

def render_runtime_events(events: list[dict]) -> str:
    if not events:
        return ""

    lines = ["Runtime events:"]
    for event in events:
        lines.append(
            "- "
            f"job={event['job_id']} "
            f"status={event['status']} "
            f"task={event.get('task_id') or 'none'} "
            f"summary={event.get('summary') or 'none'}"
        )
    return "\n".join(lines)

事件被成功送入一次模型调用后再确认消费。如果模型请求失败，事件仍然保留，下一次调用继续投递。在当前 Harness 进程内，这是一种简单的至少一次通知语义；如果还要跨重启保证通知不丢失，就需要把待投递事件写入持久化 Outbox。

模型收到事件后可以：

调用 job_output 查看更多日志；
根据退出码判断是否需要修复或重试；
更新 Task checkpoint；
继续执行验证；
在整个 Task 真正完成后调用 task_complete。

接入 Runtime Event Loop

Agent Loop 仍然负责一轮模型与工具交互，外层再增加一个 Runtime Event Loop：

from queue import Empty, Queue
from threading import Thread


def monitor_background_jobs(state: HarnessState, event_queue: Queue[dict]) -> None:
    while not state.shutdown_requested:
        for event in poll_jobs(state.job_manager):
            event_queue.put(event)
        time.sleep(0.5)


def run_runtime(state: HarnessState) -> None:
    event_queue: Queue[dict] = Queue()
    monitor = Thread(
        target=monitor_background_jobs,
        args=(state, event_queue),
        daemon=True,
    )
    monitor.start()

    while not state.shutdown_requested:
        try:
            event = event_queue.get(timeout=0.5)
        except Empty:
            continue

        state.runtime_events.append(event)
        try:
            run_agent_turn(state)
        except Exception:
            # 重新入队，并通过退避避免连续失败时形成热循环。
            event_queue.put(event)
            time.sleep(1)
        finally:
            state.runtime_events.remove(event)

用户消息也可以由同一个 Runtime Event Loop 监听，但必须保留独立的事件类型和提示词角色：用户输入仍然渲染为 user 消息，Job 通知只进入运行时事件 Section。调度入口可以统一，语义不能混在一起。

监控线程只负责检查 Job 和投递事件，不直接修改 Agent 对话状态；主循环仍然一次只执行一个模型回合。如果 Job 在 Agent 忙碌时结束，事件会先留在队列中，等当前回合结束后再处理，从而避免两个模型回合并发修改同一份状态。

Queue 本身是线程安全的，Job Manager 的共享字典则由 RLock 保护。示例采用 0.5 秒轮询间隔，便于理解；真实服务可以改用异步进程监控、系统事件或消息队列。

与 Task System 配合

当后台 Job 关联了 task_id，Task 仍然保持 in_progress。这个关联不应该由模型填写：Harness 根据当前 Task 自动注入 task_id，并把本次运行的 claim_token 一起保存到 Job 元数据中，但不会把 token 返回给模型。

Job Manager 或 Harness 应该使用 agent_id + task_claim_token 定期为这个 Task 续租，并写入类似下面的 checkpoint：

已启动后台测试 job-a81d3f，等待执行结果。
stdout: .agent/jobs/job-a81d3f.stdout.log
stderr: .agent/jobs/job-a81d3f.stderr.log

这里要遵守三条规则：

第一，不要在 Job 启动后立即释放 Task。 否则另一个 Agent 可能重新认领同一 Task，再次启动相同命令。

如果续租时所有权校验失败，说明 Task 已经被重新认领。旧 Job 可以继续记录进程结果，但不能再更新该 Task；Harness 应该发布一条冲突事件，让当前所有者决定复用结果、忽略结果还是重新执行。

第二，Job 成功不等于 Task 完成。 Agent 必须读取结果、执行必要验证，再决定是否完成 Task。

第三，重启后不要自动重跑 lost Job。 原命令可能已经完成写入、部署或发送请求。新的 Agent 应该先检查 artifact、日志和外部系统状态，再决定是否重试。

有了后台 Job，一个 Agent 可以在等待测试时认领另一项就绪 Task。但这也意味着它可能同时持有多个任务，Prompt Assembler 应该只注入简短任务摘要，避免运行态迅速膨胀。

资源与权限边界

后台执行比前台执行更容易失控，因为 Agent 不再直接感受到等待成本。

Harness 至少应该限制：

每个 Agent 和整个进程的并发 Job 数；
每个 Job 的最大运行时间；
stdout、stderr 的磁盘配额；
可以使用的工作目录和环境变量；
后台服务器允许监听的端口；
取消和强制终止的权限；
Harness 退出时如何处理仍在运行的进程。

还要考虑敏感信息。命令行参数和环境变量可能包含 token、密码或私有地址，不应该原样写入 Job JSON、通知摘要和审计日志。

后台模式只是调度方式，不是新的安全级别。原有 Permission、Hook、路径限制和沙箱仍然全部有效。

常见坑

第一，让 Harness 通过命令关键词猜后台模式。 关键词只能作为提示，最终应该由工具参数和执行策略明确表达。

第二，所有命令都默认后台运行。 很多短命令的结果马上用于下一步，后台化只会增加状态切换和工具调用。

第三，把完整 stdout 注入通知。 大型构建日志会迅速占满上下文。通知只带摘要，完整日志按需读取。

第四，只启动进程，不保存 Job 元数据。 Agent 拿到 job_id 后无法查询状态，也无法在日志中定位问题。

第五，只轮询，不保留完成事件。 模型调用失败时通知会丢失。事件至少要保留到一次成功投递之后；需要跨重启恢复时，还要写入持久化 Outbox。

第六，只终止 shell，不终止子进程组。 构建工具和开发服务器可能留下孤儿进程。

第七，Job 成功就自动完成 Task。 一个命令成功通常只是 Task 的一个步骤。

第八，Harness 重启后自动重跑未知 Job。 结果不确定的写操作可能产生重复副作用。

第九，后台执行绕过权限审批。 是否等待结果与命令是否安全是两个独立问题。

小结

本文在 Task System 之后，为 Agent Harness 增加了最小后台执行机制：

区分长期工作目标 Task 和一次具体执行 Job；
通过 background 参数让模型显式选择执行模式；
使用 Popen 启动独立进程，并立即返回稳定 job_id；
将 stdout、stderr 写入文件，通知只携带有限摘要；
通过轮询管理成功、失败、取消和超时，并在重启恢复时把结果未知的 Job 标记为 lost；
用 Runtime Event Loop 接收 Job 完成事件并重新唤醒 Agent；
通过 Prompt Assembler 注入内部事件，不把通知伪装成用户消息；
保持 Job 与 Task 状态分离，由 Agent 验证结果后再完成 Task；
对并发数、运行时间、磁盘输出和进程终止设置明确边界。

至此，Agent 不必再守着一个耗时命令等待结果。它可以启动后台 Job，继续推进其它任务，并在操作结束时回到正确的上下文中处理结果。

后台 Job 解决的是“启动以后如何不阻塞”，但它仍然需要用户或 Agent 在某个时刻主动发起。如果我们希望 Agent 每天早上 8 点检查服务器运行状态，或者每天晚上自动运行一轮测试，就还缺少一个能够按时间唤醒 Harness 的调度层。

下一章讲 定时任务与 Scheduler：把 Cron 表达式、时区和下一次执行时间持久化，在触发时创建 Task 或启动 Job，并处理错过执行、重复触发、并发冲突和失败重试，让 Agent 可以稳定执行每日巡检与夜间自动化测试。