Agent Harness 工程：Task System 与持久化任务图
Agent Harness 工程：Task System 与持久化任务图
晚上十一点，Agent 已经为搜索功能改完接口，正在补最后两条测试。
这时终端意外退出了。
第二天重新启动，模型仍然会分析代码，也仍然会调用工具。但它不知道昨天完成了什么，不知道测试跑到哪里，更不知道还有一个前端任务正在等待这个接口。
如果我们只保存了 messages，也许还能找回一部分对话；如果只保存了 Todo，或许还能看见几条待办。但它们都很难回答：
- 这项工作是否已经被某个 Agent 认领；
- 哪些任务可以并行，哪些必须等待；
- 上一个执行者的租约是否仍然有效；
- 中断前已经产生了哪些文件和结果；
- 新进程从哪里接手才不会重复劳动。
上一篇解决的是“一次运行里的错误恢复”。这一章继续向外走一步：即使整个运行过程消失，工作本身也不能跟着消失。
我们需要一个 Task System。
Todo 帮 Agent 记住，Task 帮系统记住
第 6 章实现的 TodoWrite 很有用。它能让 Agent 在当前任务里先列步骤，再逐项推进：
[x] 阅读搜索接口
[x] 修复分页边界
[ ] 补充空查询测试
[ ] 运行完整测试
但 Todo 更像工作台上的便签。它服务于当前会话、当前 Agent 和当前思路。
Task 则是一份可以交接的工作记录。它要跨会话、跨进程，甚至跨 Agent 保存：
目标是什么
依赖谁
现在是什么状态
由谁执行
执行权何时过期
上一次做到哪里
留下了哪些产物
最后如何完成或失败
所以两者不是替代关系：
Task System 管理一项工作从创建到完成的生命；Todo 管理某个执行者眼前的几步路。
Agent 认领 Task 之后，仍然可以用 TodoWrite 规划内部步骤。Task 被另一个进程接手时，新的 Agent 会根据 checkpoint 重新生成自己的 Todo，而不是继承一段已经过期的思考过程。
复杂工作不是清单，而是一张图
假设要交付一套搜索功能：
analyze 明确接口和验收条件
prepare-data 准备测试数据
build-api 实现 API，依赖 analyze
build-ui 实现界面，依赖 analyze
integration-test 集成测试，依赖 prepare-data、build-api、build-ui
build-api 和 build-ui 可以并行，integration-test 却必须等待三个前置任务全部完成。
任务图让可以并行的工作自然展开，也让尚未满足依赖的工作保持安静。
这是一张有向无环图，也就是 DAG。最小实现先坚持四条规则：
- Task 不能依赖自己；
- 依赖必须真实存在；
- 依赖关系不能形成环；
- 只有前置 Task 进入
completed，依赖才算满足。
为了让文件型实现保持简单，我们还可以加一条约束：Task 创建后不再修改依赖，新 Task 只能依赖已经存在的 Task。
只要按拓扑顺序创建任务，新节点永远只指向旧节点，环就不会出现。以后要支持批量创建或动态改图，再用拓扑排序完整检查 DAG。
只保存事实，不保存可以计算的答案
Task 的持久化状态可以保持得很少：
from typing import Literal
TaskStatus = Literal[
"pending",
"in_progress",
"completed",
"failed",
"cancelled",
]
TASK_STATUSES = {
"pending",
"in_progress",
"completed",
"failed",
"cancelled",
}
你可能会问：ready 和 blocked 去哪里了？
它们不需要写进文件。
pending且所有依赖已完成，当前视图就是ready；pending且仍有依赖未完成，当前视图就是blocked；in_progress但租约已经过期，当前视图就是reclaimable。
文件只保存稳定状态；是否就绪、阻塞或可接管，则根据依赖和租约实时推导。
这样做能避开一个很隐蔽的问题。
如果我们把 blocked 也写进文件，那么前置任务完成时，Harness 必须找到所有下游任务并逐个改成 ready。只要中间一次写入失败，同一张图就会同时出现“依赖已经完成”和“任务仍然阻塞”两个互相矛盾的事实。
派生状态不落盘，前置任务完成后也不必“解锁”任何文件。下一次查询时，下游任务自然变成 ready。
一条 Task 需要记住什么
最小 Task 可以这样定义：
from dataclasses import asdict, dataclass, field
@dataclass
class Task:
id: str
title: str
description: str
status: TaskStatus = "pending"
dependencies: list[str] = field(default_factory=list)
claimed_by: str | None = None
claim_token: str | None = None
lease_expires_at: str | None = None
checkpoint: str | None = None
artifacts: list[str] = field(default_factory=list)
result: str | None = None
error: str | None = None
created_at: str = ""
updated_at: str = ""
schema_version: int = 1
@classmethod
def from_dict(cls, payload: dict) -> "Task":
if payload.get("schema_version") != 1:
raise ValueError("unsupported task schema version")
task = cls(**payload)
if task.status not in TASK_STATUSES:
raise ValueError(f"invalid task status: {task.status}")
if not all(
isinstance(item, str) and item
for item in task.dependencies
):
raise ValueError("dependencies must contain task ids")
return task
def to_dict(self) -> dict:
return asdict(self)
这些字段分成三层：
id、description、dependencies描述工作本身；status、claimed_by、claim_token、lease_expires_at描述执行所有权；checkpoint、artifacts、result、error帮助恢复和交接。
checkpoint 不保存模型的完整推理。它只保留一个新的执行者真正需要的事实：
已完成什么
当前卡在哪里
下一步建议做什么
相关文件、Job 或外部资源在哪里
隐藏推理既不稳定，也不是恢复任务所需的业务事实。
先把 Task 安稳地写到磁盘
最小版本可以把每个 Task 保存成独立 JSON：
.agent/
└── tasks/
├── analyze.json
├── build-api.json
├── build-ui.json
└── integration-test.json
但不要直接用 Path.write_text() 覆盖目标文件。进程可能在写到一半时退出，留下半截 JSON。
更稳妥的写法是：在同一目录写临时文件，fsync() 后再原子替换。
import json
import os
import tempfile
from pathlib import Path
def atomic_write_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temp_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=path.parent,
prefix=f".{path.name}.",
suffix=".tmp",
delete=False,
) as temp_file:
json.dump(
payload,
temp_file,
ensure_ascii=False,
indent=2,
sort_keys=True,
)
temp_file.write("\n")
temp_file.flush()
os.fsync(temp_file.fileno())
temp_path = Path(temp_file.name)
os.replace(temp_path, path)
finally:
if temp_path is not None and temp_path.exists():
temp_path.unlink()
认领 Task 又是一次“先读后写”。两个 Agent 如果同时读到 pending，都可能认为自己认领成功。因此单机多进程还需要一把文件锁：
import fcntl
import re
from contextlib import contextmanager
from typing import Iterator
TASK_ID_PATTERN = re.compile(
r"^[a-z0-9][a-z0-9-]{0,63}$"
)
class TaskStoreError(RuntimeError):
pass
class TaskUnavailableError(TaskStoreError):
pass
class TaskStore:
def __init__(self, root: Path):
self.root = root.resolve()
self.root.mkdir(parents=True, exist_ok=True)
self.lock_path = self.root / ".lock"
def _task_path(self, task_id: str) -> Path:
if not TASK_ID_PATTERN.fullmatch(task_id):
raise TaskStoreError(f"invalid task id: {task_id}")
return self.root / f"{task_id}.json"
@contextmanager
def locked(self) -> Iterator[None]:
with self.lock_path.open("a+") as lock_file:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
def _read_unlocked(self, task_id: str) -> Task:
path = self._task_path(task_id)
if not path.exists():
raise TaskStoreError(f"unknown task: {task_id}")
try:
payload = json.loads(path.read_text("utf-8"))
task = Task.from_dict(payload)
except (OSError, TypeError, ValueError) as exc:
raise TaskStoreError(
f"cannot read {path.name}: {exc}"
) from exc
if task.id != task_id:
raise TaskStoreError("task id does not match file name")
return task
def _list_unlocked(self) -> list[Task]:
return [
self._read_unlocked(path.stem)
for path in sorted(self.root.glob("*.json"))
]
def _write_unlocked(self, task: Task) -> None:
task.updated_at = utc_now_text()
atomic_write_json(
self._task_path(task.id),
task.to_dict(),
)
def get(self, task_id: str) -> Task:
with self.locked():
return self._read_unlocked(task_id)
def list(self) -> list[Task]:
with self.locked():
return self._list_unlocked()
TASK_ID_PATTERN 不只是命名偏好，也是一条路径安全边界。Task ID 不能包含 /、.. 或其它能逃出任务目录的内容。
flock 适合 macOS 和 Linux 上的单机实现。多个 Agent 如果运行在不同机器上，就要换成 SQLite、PostgreSQL 或其它支持事务与条件更新的存储。文件锁不是分布式事务。
创建 Task 时，稳定 ID 同时充当幂等键：相同 ID、相同定义返回原记录；相同 ID 对应不同内容则明确冲突。
def create_task(
store: TaskStore,
task_id: str,
title: str,
description: str,
dependencies: list[str] | None = None,
) -> Task:
raw_dependencies = dependencies or []
if not all(
isinstance(item, str)
and TASK_ID_PATTERN.fullmatch(item)
for item in raw_dependencies
):
raise TaskStoreError(
"dependencies must contain valid task ids"
)
dependency_ids = sorted(
set(raw_dependencies)
)
title = title.strip()
description = description.strip()
if not title or not description:
raise TaskStoreError(
"title and description are required"
)
with store.locked():
target = store._task_path(task_id)
if target.exists():
existing = store._read_unlocked(task_id)
if (
existing.title == title
and existing.description == description
and existing.dependencies
== dependency_ids
):
return existing
raise TaskStoreError(
"task id has another definition"
)
existing = {
task.id: task
for task in store._list_unlocked()
}
if task_id in dependency_ids:
raise TaskStoreError(
"task cannot depend on itself"
)
missing = [
item
for item in dependency_ids
if item not in existing
]
if missing:
raise TaskStoreError(
f"unknown dependencies: {missing}"
)
now = utc_now_text()
task = Task(
id=task_id,
title=title,
description=description,
dependencies=dependency_ids,
created_at=now,
updated_at=now,
)
store._write_unlocked(task)
return task
如果 Task 文件已经落盘、工具响应却在返回途中丢失，模型重试会得到同一条记录，而不是创建重复工作。
依赖决定能否开始，租约决定谁能继续
先根据任务图计算未完成依赖：
def unresolved_dependencies(
task: Task,
tasks: dict[str, Task],
) -> list[str]:
return [
dependency_id
for dependency_id in task.dependencies
if dependency_id not in tasks
or tasks[dependency_id].status != "completed"
]
当多个 Agent 同时看到同一个 ready Task 时，真正的执行权必须通过存储锁竞争。
只记录 claimed_by 还不够。Agent 崩溃后，这项工作会永远停在 in_progress。所以每次认领还要带一份会过期的租约：
from datetime import datetime, timedelta, timezone
UTC = timezone.utc
def utc_now() -> datetime:
return datetime.now(UTC)
def utc_now_text() -> str:
return utc_now().isoformat(timespec="seconds")
def lease_is_active(
task: Task,
now: datetime | None = None,
) -> bool:
if not task.lease_expires_at:
return False
expires_at = datetime.fromisoformat(
task.lease_expires_at
)
if expires_at.tzinfo is None:
raise TaskStoreError("lease timestamp needs timezone")
current = now or utc_now()
if current.tzinfo is None:
raise TaskStoreError("current time needs timezone")
return expires_at.astimezone(UTC) > current.astimezone(UTC)
认领、依赖检查和状态写入必须在同一个临界区：
def claim_task_unlocked(
store: TaskStore,
tasks: dict[str, Task],
task: Task,
agent_id: str,
claim_token: str,
lease_seconds: int = 300,
) -> Task:
if not claim_token.strip():
raise TaskStoreError("claim token is required")
lease_seconds = max(30, min(lease_seconds, 3600))
if task.status in {"completed", "failed", "cancelled"}:
raise TaskUnavailableError(
f"task is already {task.status}"
)
if task.status == "in_progress" and lease_is_active(task):
same_claim = (
task.claimed_by == agent_id
and task.claim_token == claim_token
)
if same_claim:
return task
raise TaskUnavailableError(
f"task is owned by {task.claimed_by}"
)
blocked_by = unresolved_dependencies(task, tasks)
if blocked_by:
raise TaskUnavailableError(
f"task is blocked by {blocked_by}"
)
task.status = "in_progress"
task.claimed_by = agent_id
task.claim_token = claim_token
task.lease_expires_at = (
utc_now() + timedelta(seconds=lease_seconds)
).isoformat(timespec="seconds")
store._write_unlocked(task)
return task
def claim_task(
store: TaskStore,
task_id: str,
agent_id: str,
claim_token: str,
lease_seconds: int = 300,
) -> Task:
with store.locked():
tasks = {
task.id: task
for task in store._list_unlocked()
}
task = tasks.get(task_id)
if task is None:
raise TaskUnavailableError("task does not exist")
return claim_task_unlocked(
store=store,
tasks=tasks,
task=task,
agent_id=agent_id,
claim_token=claim_token,
lease_seconds=lease_seconds,
)
这里有两个身份：
agent_id是稳定的逻辑身份，例如backend；claim_token是本次进程生成的随机令牌。
同一个 agent_id 的旧进程不能拿着过期现场继续写 Task，因为新的执行者会使用不同 token。后续 checkpoint、release、complete 和 fail 都必须同时校验这两个值。
不过，随机 claim_token 只保护 Task Store。它不是外部数据库、部署平台或支付接口的 fencing token。外部副作用仍然需要自己的条件更新和幂等机制。
Checkpoint 是留给下一位执行者的路标
长 Task 要定期保存 checkpoint，并顺便续租：
def ensure_task_owner(
task: Task,
agent_id: str,
claim_token: str,
) -> None:
if task.status != "in_progress":
raise TaskStoreError("task is not in progress")
if task.claimed_by != agent_id:
raise TaskStoreError("task belongs to another agent")
if task.claim_token != claim_token:
raise TaskStoreError("claim token does not match")
if not lease_is_active(task):
raise TaskStoreError("task lease has expired")
def checkpoint_task(
store: TaskStore,
task_id: str,
agent_id: str,
claim_token: str,
summary: str,
artifacts: list[str] | None = None,
renew_seconds: int = 300,
) -> Task:
summary = summary.strip()
if not summary or len(summary) > 4000:
raise TaskStoreError("invalid checkpoint summary")
new_artifacts = artifacts or []
if not all(
isinstance(item, str) and item.strip()
for item in new_artifacts
):
raise TaskStoreError(
"artifacts must contain paths"
)
with store.locked():
task = store._read_unlocked(task_id)
ensure_task_owner(task, agent_id, claim_token)
task.checkpoint = summary
task.artifacts = list(
dict.fromkeys(
[
*task.artifacts,
*(
item.strip()
for item in new_artifacts
),
]
)
)[:20]
task.lease_expires_at = (
utc_now()
+ timedelta(
seconds=max(30, min(renew_seconds, 3600))
)
).isoformat(timespec="seconds")
store._write_unlocked(task)
return task
def renew_task_lease(
store: TaskStore,
task_id: str,
agent_id: str,
claim_token: str,
renew_seconds: int = 300,
) -> Task:
with store.locked():
task = store._read_unlocked(task_id)
ensure_task_owner(task, agent_id, claim_token)
task.lease_expires_at = (
utc_now()
+ timedelta(
seconds=max(30, min(renew_seconds, 3600))
)
).isoformat(timespec="seconds")
store._write_unlocked(task)
return task
其余状态操作沿用同一模式：
task_release
校验所有权 -> 回到 pending -> 保留 checkpoint
task_complete
校验所有权 -> 写入 result -> 进入 completed
task_fail
校验所有权 -> 写入 error -> 进入 failed
task_retry
确认原状态为 failed -> 清除 error -> 回到 pending
claim_task() 会拒绝直接认领 failed Task。只有显式调用 task_retry，经过重试次数、权限和副作用检查后，它才会重新回到 pending；失败不会触发自动重跑。
完成、失败和释放后都要清空 claimed_by、claim_token 与租约。
这些操作还应具备幂等语义。比如同一个 Task 已经以相同 result 完成，重复的 task_complete 可以返回现有记录；如果 result 不同，则必须报告冲突，而不是覆盖第一次完成。
中断之后，新的 Agent 怎样接手
Harness 重启时不需要恢复模型的完整对话。它只要恢复事实：
- 读取 Task Store；
- 查找自己仍在有效租约内的 Task；
- 否则优先选择自己以前留下、但租约已经过期的 Task；
- 再选择普通
readyTask； - 在锁内重新认领，处理“刚刚被别人抢走”的竞争；
- 读取 description、checkpoint 和 artifacts；
- 检查真实文件与外部状态，再生成新的 Todo。
租约过期只意味着“原执行者不再拥有写入权”，不意味着它做过的事情全部无效。
例如 checkpoint 写着：
搜索接口已经修改。
待运行 tests/test_search.py。
候选文件：src/search.py、tests/test_search.py。
新的 Agent 应先查看这两个文件和 Git 状态，确认现场仍然存在，再继续测试。恢复不是相信旧摘要，而是用旧摘要快速找到需要重新验证的地方。
模型只能通过 Task 工具改状态
不要提供一个可以任意覆盖字段的 task_update。更清楚的工具边界是：
task_create
task_list
task_get
task_claim
task_checkpoint
task_release
task_complete
task_fail
task_retry
模型只提供业务参数。agent_id 与 claim_token 由 Harness 注入，也不能出现在工具返回值里。
.agent/tasks 还必须对通用文件工具和 Shell 保持只读。否则模型可以绕过依赖检查和所有权校验，直接把任意 JSON 改成 completed。
这条保护不能只写在 write_file 里。bash、脚本执行器和后台进程同样可能通过重定向修改 Task Store。真正可靠的边界是：只有 Task Store 服务拥有写权限，任务沙箱看到的是只读挂载。
一次任务如何穿过 Task System
回到开头的搜索功能。
Lead 先创建 analyze，再创建依赖它的 build-api 和 build-ui，最后创建 integration-test。
当 analyze 完成后，两个实现任务自然变成 ready。Backend Agent 认领 build-api，得到一份五分钟租约，随后用 TodoWrite 规划阅读代码、修改接口、补测试和运行验证。
每完成一段工作，它就保存 checkpoint 并续租。终端退出后，Task 仍然停在 in_progress，代码与 artifact 也仍在原位置。租约过期后，新的 Agent 可以重新认领，根据 checkpoint 检查现场并继续。
只有当 build-api 真正进入 completed，依赖它的后续任务才会解锁。一条“接口写好了”的聊天消息，不会改变任务图。
Task System 因此做了一件很朴素、也很关键的事：
它让工作不再依附于某一次对话，而是成为系统中可以查询、交接和恢复的事实。
几条不能省略的边界
实现最小 Task System 时，最容易出错的地方可以收束成六条：
ready、blocked和reclaimable应实时计算，不要重复持久化；- 认领必须在同一事务或锁内完成“检查并写入”；
- 只有租约没有 checkpoint，接手者仍然不知道做到哪里；
- 只有
agent_id没有claim_token，旧进程仍可能继续写入； - Task 文件要原子替换，不能冒险留下半截 JSON；
- Task Store 必须与模型可写的普通工作区隔离。
本章的 JSON 加 flock 适合本地 Harness。真正的分布式团队会把这套模型迁到支持事务、唯一约束和条件更新的数据库中，但 Task 的核心语义不需要随之改变。
小结
这一章把会话里的 Todo 升级成了可以穿过进程与时间的 Task System：
- DAG 表达任务依赖；
- 稳定状态落盘，派生状态实时计算；
- 原子 JSON 和存储锁保护单机并发；
claimed_by + claim_token + lease建立执行所有权；- checkpoint、artifact、result 和 error 支持恢复与交接；
- 专用 Task 工具守住状态转换边界。
现在，即使 Agent 中途离开，任务也仍然知道自己在哪里。
不过，Task 只描述“要完成什么”。如果下一步是运行二十分钟测试，当前 Agent Loop 仍然会一直等在那里，既不能推进其它工作，也很容易让租约悄悄过期。
下一章继续处理这个等待问题：把一次耗时执行抽成 后台 Job，让命令先启动、Agent 先离开，等结果出现时再回来。