TodoWrite 适合记录一个 Agent 当前准备做哪些步骤，但复杂项目还需要回答更多问题：哪个任务依赖哪个任务？当前由谁负责？进程重启以后进度是否仍然存在？哪些任务现在可以开始？

Task System 把一次性的执行清单提升为持久化任务图：每个任务拥有稳定 ID、状态、负责人和前置依赖，并保存为独立 JSON 文件。

schema ──────→ API ──────→ tests
   │
   └─────────→ docs
 
箭头表示：右侧任务 blockedBy 左侧任务

这使 Harness 能判断某个任务是否具备开始条件，但任务图本身仍不负责真正执行、验证或调度工作。

TodoWrite 和 Task System 的边界

二者都包含 pending、in_progress 和 completed，但服务于不同层级：

维度TodoWriteTask System
目标当前任务的执行清单可恢复、可协调的任务集合
粒度一个 Agent 的步骤可独立认领的工作单元
存储当前进程或会话.tasks/*.json
更新通常整体替换清单按任务执行动作
依赖通常没有blockedBy 任务图
负责人通常没有owner
跨会话恢复不一定文件仍在即可恢复

一个任务内部仍然可以使用 TodoWrite 管理具体步骤。Task System 不必取代 Todo，而是管理更高一层的分工和依赖。

最小任务数据模型

当前课程使用：

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]

对应文件类似：

{
  "id": "task_a1b2c3d4",
  "subject": "实现 API",
  "description": "实现创建与查询端点，并补充错误处理",
  "status": "pending",
  "owner": null,
  "blockedBy": ["task_11223344"]
}

subject 用于列表摘要，description 承载完整工作契约。只看标题恢复任务容易丢失验收标准，所以跨会话继续前应读取完整记录。

现实系统通常还需要创建时间、更新时间、版本、完成证据、失败原因、优先级和标签。字段应围绕真实协调需求扩展，而不是为了“看起来像项目管理软件”无限增加。

当前 ID 是随机生成的

旧版笔记通过扫描 task_*.json 并提取最大数字来生成下一个 ID。当前实现已经改为：

task_id = f"task_{secrets.token_hex(4)}"

结果是 task_ 加 8 位随机十六进制字符，例如 task_a1b2c3d4。

创建文件时使用排他模式：

with path.open("x", encoding="utf-8") as handle:
    json.dump(task, handle)

若文件已存在，"x" 会抛出 FileExistsError，代码重新生成 ID。这避免了“扫描最大编号后两个进程同时选择下一个编号”的直接竞态。

随机 32 位 ID 在教学规模下足够，并通过排他创建处理碰撞。但大规模分布式系统更适合使用更宽的随机 ID、数据库生成 ID 或具备明确命名空间的标识符。

ID 校验也是路径边界

任务 ID 必须匹配：

^task_[0-9a-f]{8}$

Store 还会解析最终路径并确认它位于任务根目录内。这样，模型不能把 ../../secret 当作任务 ID，诱导 Store 读取任意 JSON 文件。

任务根目录自身也必须位于工作区内。这里的双重校验很有价值：格式白名单限制可表达的文件名，路径包含检查防止实现变化或符号链接带来的逃逸。

但任务描述仍然可能包含不可信文本，读取任务记录后不能把其中指令提升为高于系统规则或用户授权的内容。

为什么要先创建节点，再添加依赖

模型可以在同一次响应中调用多个 create_task。但这些 tool call 在任何 tool result 返回之前就已经全部生成，所以第二个调用无法知道第一个任务刚刚获得的随机 ID。

因此课程采用两阶段建图：

  1. 创建所有任务节点；
  2. 读取每个创建结果中的运行时 ID；
  3. 下一轮调用 update_task 添加依赖边。
schema = create_task("设计数据库结构")
api = create_task("实现 API")
 
update_task(api.id, addBlockedBy=[schema.id])

如果希望单轮创建整张图，可以让客户端先提供临时别名，再由服务端统一映射，或者提供一个接受节点和边的事务型 create_task_graph 工具。但这种接口必须定义整批失败时是否回滚。

依赖更新必须维持 DAG

blockedBy 表示任务开始前必须完成的任务 ID。更新依赖时，当前实现检查：

  • 目标任务仍为 pending 且无人认领；
  • 新依赖任务存在；
  • 任务不能依赖自身；
  • 添加边后不能形成环；
  • 重复依赖被去重。

环检查沿依赖边向前遍历：若候选依赖已经传递依赖目标任务，再让目标依赖候选，就会闭合一个环。

A blockedBy B
B blockedBy C
 
此时不能再添加：C blockedBy A

把依赖限制在任务仍未开始时修改，可以减少执行中途改变前置条件的歧义。但生产系统可能需要支持暂停、重新规划和移除依赖，这时必须定义正在执行任务如何处理。

can_start 只判断前置状态

一个任务可开始，当且仅当所有 blockedBy 任务都为 completed：

def can_start(task_id):
    return not incomplete_dependencies(load_task(task_id))

依赖文件缺失或记录非法时，当前实现保守地视为未完成。这比把未知依赖当成完成安全，因为损坏数据不会意外解锁下游任务。

但 can_start=True 只说明任务图中的前置状态满足，不代表：

  • 所需文件、凭据或外部服务已经就绪；
  • 当前 Agent 拥有执行权限；
  • 没有其他 Agent 正在抢占它；
  • 实际业务条件依然成立。

任务图是调度条件的一部分，不是完整的预检系统。

状态机由动作驱动

当前状态流转只有两步：

pending ── claim ──→ in_progress ── complete ──→ completed

claim_task 检查任务仍为 pending 且所有依赖完成，然后写入 owner 并切换到 in_progress。

complete_task 检查任务正在进行且 owner 匹配，再切换到 completed，最后扫描其他任务，报告刚刚解除阻塞的下游任务。

把状态变化封装成动作比允许任意 update(status=...) 更安全，因为每个动作可以携带自己的前置条件。

不过课程状态机没有取消、失败、阻塞、过期、重试或重新打开。遇到真实失败时，不应为了让图继续流动而错误标记 completed，而应扩展明确的失败和恢复语义。

completed 不等于工作通过验收

当前 handler 根据 Agent 调用把任务标记为完成，并不自动检查文件 diff、测试结果或部署状态。

因此完成动作最好要求证据：

  • 产物路径或变更集合；
  • 执行过的验证命令；
  • 测试退出码与摘要；
  • 评审或人工批准；
  • 无法验证的原因。

高风险任务可以把“执行完成”和“验收通过”拆成不同状态，或要求独立 verifier 确认后才能解锁下游。否则一个错误的 completed 会沿依赖图传播，导致后续任务建立在不存在的前提上。

文件持久化还不是并发安全存储

排他创建保证同一 ID 的初次文件不会被覆盖，但后续 save() 使用普通整文件写入，没有锁、版本号或原子替换。

两个 Agent 可能同时：

  1. 读取同一个 pending 任务；
  2. 都判断可以认领；
  3. 分别写入自己的 owner；
  4. 后写入者覆盖前写入者。

同样，依赖更新与完成操作也可能发生丢失更新；进程在写入中途崩溃，还可能留下截断 JSON。

如果需要真正多 Agent 协调，应使用具备原子条件更新的存储。例如用数据库执行：

UPDATE tasks
SET status = 'in_progress', owner = :owner, version = version + 1
WHERE id = :id
  AND status = 'pending'
  AND version = :expected_version;

然后检查实际更新行数。文件方案也可以加入文件锁、临时文件加原子替换和乐观版本检查，但跨主机协调通常更适合数据库或专门的任务队列。

Owner 需要可信身份

课程工具包装器固定使用 owner="agent"，主要用于演示所有权检查。真实系统不能只相信模型自由填写的 owner 字符串，否则任何调用者都能冒充其他 Agent 完成任务。

owner 应由运行时根据已认证执行身份注入，而不是作为普通模型参数。审计记录还应区分：

  • 谁创建任务；
  • 谁认领和执行；
  • 谁验证完成；
  • 哪个会话、进程或工作区产生了变更。

任务所有权是授权和可追责问题，不只是 UI 标签。

文件是事实源时要考虑生命周期

.tasks/ 保留在工作目录里，进程退出后仍可读取，因此具备最小跨会话恢复能力。但项目还要决定：

  • 任务文件是否进入 Git；
  • 临时本地任务如何清理；
  • 多个分支或 worktree 是否共享任务状态；
  • 文件损坏时如何恢复；
  • Schema 变化如何迁移旧记录；
  • 删除任务后下游依赖如何处理；
  • 已完成任务保留多久；
  • 敏感描述是否需要加密或脱敏。

如果任务状态是运行时协调数据，通常不应和源码提交混在一起；如果它是团队工作记录，则需要更明确的同步、冲突和审计机制。

列表顺序与旧版语法差异

旧稿解释了从 task_12.json 提取数字并按数字排序。当前随机 ID 不包含递增序号，sorted(root.glob("task_*.json")) 只是按文件名的字典序返回稳定结果，不代表创建时间、优先级或依赖拓扑顺序。

若 UI 需要“下一项可做任务”，应显式按 ready 状态、优先级和创建时间排序；若需要依赖执行顺序，应对 DAG 做拓扑排序，而不是依赖随机文件名。

JSON 读写的基础关系仍然成立：

  • json.dumps：Python 对象转 JSON 字符串；
  • json.loads：JSON 字符串转 Python 对象；
  • json.dump：写入文件对象；
  • json.load：从文件对象读取。

解析成功只说明 JSON 语法合法，不说明字段类型和业务约束正确。当前 Task(**data) 与状态检查是最小验证，生产系统应使用完整 Schema 验证每个字段、拒绝未知字段并提供迁移版本。

性能与调度边界

完成一个任务后，课程会列出全部任务并重新检查哪些任务刚刚被解锁。对于几十个本地任务很直观，但任务量增大后会产生大量文件读取和重复依赖遍历。

数据库可以为状态和依赖建立索引，任务队列可以主动投递 ready 任务。无论采用哪种存储，都要明确：

  • 任务图负责表达约束；
  • scheduler 负责选择何时运行；
  • worker 负责执行；
  • verifier 负责判断是否满足完成条件。

把四者混成一个“Agent 自己看列表决定”接口，在并发和故障场景下很难保持一致。

建议覆盖的测试

Task System 至少应验证：

  • 空标题、非法 ID 和任务根目录逃逸；
  • 随机 ID 碰撞后的排他重试；
  • JSON 缺字段、未知字段、错误类型和非法状态；
  • 创建所有节点后再添加运行时 ID 依赖；
  • 依赖不存在、自依赖、重复依赖和传递环；
  • 非 pending 或已认领任务不能修改依赖；
  • 缺失或损坏依赖保持阻塞；
  • 有依赖任务在前置完成前不能认领；
  • owner 不匹配时不能完成；
  • 完成任务后只报告新解锁的下游；
  • 同时认领、更新和完成时的竞态；
  • 写入中断后的原子性与恢复；
  • 完成证据和 verifier 失败；
  • 跨会话、分支、worktree 与 Schema 升级语义；
  • 大任务图的查询和拓扑性能。

小结

Task System 把 Agent 的工作从会话内清单提升为可恢复的任务图：随机稳定 ID 标识节点，blockedBy 表达依赖，claim 和 complete 驱动状态机，JSON 文件保存跨会话进度。

但它仍是教学级持久层，不是并发安全调度器。普通文件覆盖无法保证多个 Agent 原子认领，模型声明完成也不能替代验收。真实系统还需要条件更新、可信身份、原子存储、失败状态、完成证据，以及清楚分离的 scheduler、worker 和 verifier。

参考资料