任务系统：把目标拆成可依赖、可跨会话的图

上一章留下的问题：第5章的清单解决了”当前会话不跑偏”，但它有三个硬伤：会话一结束就没了；步骤之间没有先后依赖；没有归属人。

给 Agent 一个”做一个带用户认证的待办应用”这种目标，它会顺着往下做：建数据库表 → 写 API → 写测试。听起来合理，直到它在写 API 时发现表结构少了个字段，回头改表，然后之前写的部分又要返工。而且如果这时候你关掉终端，明天全部从头再来。

更远的问题是：**第15章开始会有多个 Agent 同时干活。**它们需要一个共享的、看得见的工作板，而不是各自脑子里的清单。

12.1 任务 = 文件 + 依赖 + 归属

每个任务是一个 JSON 文件，比如 .tasks/task_20260906_a3f1.json：

{
  "id": "task_20260906_a3f1",
  "subject": "设计数据库表结构",
  "description": "设计 users 与 todos 两张表，users 需要包含 email 唯一索引……",
  "status": "pending",
  "owner": "",
  "blockedBy": []
}

三个字段撑起了整套机制：

• status：pending → in_progress → completed，单向流转；
• owner：谁认领了它，防止两个 Agent 同时做同一件事；
• blockedBy：上游依赖的任务 ID 列表，形成一张有向无环图。

12.2 五个工具

import json, os, uuid
from datetime import datetime
from pathlib import Path

TASKS_DIR = Path(".tasks"); TASKS_DIR.mkdir(exist_ok=True)

def _path(task_id: str) -> Path:
    return TASKS_DIR / f"{task_id}.json"

def create_task(subject: str, description: str = "", blockedBy: list = None) -> str:
    task_id = f"task_{datetime.now():%Y%m%d}_{uuid.uuid4().hex[:6]}"
    _path(task_id).write_text(json.dumps({
        "id": task_id, "subject": subject, "description": description,
        "status": "pending", "owner": "", "blockedBy": blockedBy or [],
    }, ensure_ascii=False, indent=2))
    return f"已创建任务 {task_id}：{subject}"

def load_task(task_id: str) -> dict | None:
    p = _path(task_id)
    return json.loads(p.read_text()) if p.exists() else None

def can_start(task_id: str) -> bool:
    task = load_task(task_id)
    if not task:
        return False
    for dep_id in task["blockedBy"]:
        dep = load_task(dep_id)
        if dep is None or dep["status"] != "completed":
            return False              # 依赖缺失一律视为阻塞
    return True

def claim_task(task_id: str, owner: str) -> str:
    task = load_task(task_id)
    if not task:
        return f"任务 {task_id} 不存在"
    if task["status"] != "pending":
        return f"认领失败：任务当前状态是 {task['status']}"
    if not can_start(task_id):
        return "认领失败：上游依赖尚未全部完成"
    task["status"], task["owner"] = "in_progress", owner
    _path(task_id).write_text(json.dumps(task, ensure_ascii=False, indent=2))
    return f"Claimed {task_id} by {owner}"

def complete_task(task_id: str) -> str:
    task = load_task(task_id)
    task["status"] = "completed"
    _path(task_id).write_text(json.dumps(task, ensure_ascii=False, indent=2))

    unblocked = []                                   # 完成后主动找出被解锁的下游
    for p in TASKS_DIR.glob("*.json"):
        other = json.loads(p.read_text())
        if other["status"] == "pending" and task_id in other["blockedBy"] and can_start(other["id"]):
            unblocked.append(f"{other['id']}（{other['subject']}）")
    return f"已完成 {task_id}。" + (f"新解锁：{', '.join(unblocked)}" if unblocked else "暂无新解锁任务。")

can_start 里那句”依赖缺失一律视为阻塞”是有意的：如果 blockedBy 里引用了一个不存在的任务 ID（被删了、写错了），保守地认为不能开始，比乐观放行安全——宁可卡住等人来看，也不要在缺依赖的情况下闷头开工。

12.3 走一遍：一个四任务的项目

schema = create_task("设计数据库表结构")                        # task_A
api    = create_task("实现 REST 接口", blockedBy=["task_A"])     # task_B
tests  = create_task("编写集成测试", blockedBy=["task_B"])       # task_C
docs   = create_task("撰写 API 文档", blockedBy=["task_B"])      # task_D

注意最后一步：一次完成解锁了两个并行任务。到第15章，这两个任务就可以被两个不同的队友同时认领——依赖图天然表达了”哪些活可以并行”。

顺便，把终端关掉再打开，.tasks/ 目录还在。这就是”落盘”带来的东西：进度不再依附于某一次对话的存活。

12.4 清单和任务图的分工

两者并存。清单是 Agent 手边的便签，任务图是团队墙上的看板。一个 Agent 认领了一个任务之后，完全可以在自己的清单里把这个任务拆成八个步骤。

12.5 工程现实

教学版留了几个真实系统必须补上的洞：

• **没有环检测。**A 依赖 B、B 依赖 A，两个任务永远卡住，而且不会报错，只会安静地不可认领。创建任务时应该做一次可达性检查。
• **没有文件锁。**两个 Agent 同时读到 status == "pending"，然后同时写入认领——经典的 TOCTOU 竞态。生产实现会对任务文件加锁（配合短退避重试）。
• **认领之后没有释放路径。**认领了却做不完（进程崩了）的任务会永远停在 in_progress。需要一个超时回收机制。
• **完成时的扫描不是原子的。**在扫描”新解锁”的间隙，可能已经有别的 Agent 认领了。这不算致命，但要知道它存在。
• **ID 不能复用。**任务被删除后，ID 不应该被后来的任务重新用上，否则历史引用会指向错误的对象。

12.6 常见坑

12.7 建一张会解锁的依赖图

12.8 画出一张带分叉的依赖图