07 — 任务系统：持久化的有向无环图

"大目标要拆成小任务，排好序，记在磁盘上" — 任务图是所有多步协作的骨架。

Harness 层：持久化任务 — 比任何一次对话都长命的目标。

课程路径：[[05-技能加载|s05 技能加载]] → [[06-上下文压缩|s06 上下文压缩]] → ⭐ s07 任务系统

---

为什么需要任务图？

回顾 [[03-待办写入|s03 的 TodoManager]]，它只是一个内存中的扁平清单：没有顺序、没有依赖、状态只有"做完/没做完"两种。真实世界中的目标是有结构的：

• 任务 B 依赖 任务 A
• 任务 C 和 D 可以并行
• 任务 E 必须等待 C 和 D 都完成

没有显式的依赖关系，智能体分不清什么能做、什么被卡住、什么能同时跑。而且清单只活在内存里——[[06-上下文压缩|s06 的上下文压缩]]一跑就没了。

任务系统要回答三个根本问题：

问题	含义
什么可以做？	状态为 pending 且 blockedBy 为空的任务
什么被卡住？	等待前置任务完成的任务
什么做完了？	状态为 completed 的任务，完成时自动解锁后续任务

从扁平清单到任务图

核心升级：把内存中的列表变为磁盘上的有向无环图（DAG）。

DAG 是什么？

有向无环图（Directed Acyclic Graph）是一种数据结构，用"节点"表示任务、"边"表示依赖关系，且不存在循环依赖。在这个图中：

• 有向：箭头从前置任务指向后续任务（A → B 意思是"先做 A，再做 B"）
• 无环：不存在 A→B→C→A 的死循环
• 可以直观看到：哪些任务可以并行（没有箭头相连）、哪些被阻塞（箭头指向未完成的任务）

.tasks/
  task_1.json  {"id":1, "status":"completed"}
  task_2.json  {"id":2, "blockedBy":[1], "status":"pending"}
  task_3.json  {"id":3, "blockedBy":[1], "status":"pending"}
  task_4.json  {"id":4, "blockedBy":[2,3], "status":"pending"}

对应的 DAG 可视化：

                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

从这个图中可以读出的信息：

观察点	含义
顺序	task 1 必须先完成，task 2 和 3 才能开始
并行	task 2 和 3 没有互相依赖，可以同时执行
依赖	task 4 必须等待 task 2 和 task 3 都完成
状态	pending → in_progress → completed
阻塞	task 2、3、4 都在等待前置任务完成

为什么持久化到磁盘？

对比	内存清单 (s03)	磁盘任务图 (s07)
生命周期	一次会话	跨会话、跨压缩
依赖关系	无	blockedBy + blocks 边
容错	崩溃即丢失	重启后恢复
协作	单智能体	多智能体共享

这个任务图是 s07 之后所有机制的协调骨架：[[08-后台任务|后台执行 (s08)]]、[[09-智能体团队|多 agent 团队 (s09+)]]、[[12-Worktree隔离|worktree 隔离 (s12)]] 都读写这同一个结构。

工作原理

TaskManager 核心

每个任务是一个独立的 JSON 文件，TaskManager 提供 CRUD + 依赖图操作：

class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self._next_id = self._max_id() + 1

    def create(self, subject, description=""):
        task = {"id": self._next_id, "subject": subject,
                "status": "pending", "blockedBy": [],
                "blocks": [], "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)

依赖解除

完成任务时，自动将其 ID 从其他任务的 blockedBy 中移除，解锁后续任务：

def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

状态变更 + 依赖关联

update 统一处理状态转换和依赖边：

def update(self, task_id, status=None,
           add_blocked_by=None, add_blocks=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    if add_blocked_by:
        task["blockedBy"].extend(add_blocked_by)
    if add_blocks:
        task["blocks"].extend(add_blocks)
    self._save(task)

集成到工具系统

四个任务工具加入 dispatch map，与 [[02-工具使用|s02 的工具系统]] 完全一致：

TOOL_HANDLERS = {
    # ...base tools...
    "task_create": lambda **kw: TASKS.create(kw["subject"]),
    "task_update": lambda **kw: TASKS.update(kw["task_id"], kw.get("status")),
    "task_list":   lambda **kw: TASKS.list_all(),
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),
}

从此，模型可以通过自然语言创建、查询、更新任务，task manager 自动维护依赖图。

相对 s06 的变更

组件	之前 (s06)	之后 (s07)
Tools	5	8（+ task_create/update/list/get）
规划模型	扁平清单（仅内存）	带依赖关系的任务图（磁盘）
关系	无	blockedBy + blocks 边
状态追踪	做完/没做完	pending → in_progress → completed
持久化	压缩后丢失	压缩和重启后存活
协作基础	不支持	多智能体共享任务图

⭐ 核心要点

1. 任务即文件 — 每个任务是一个独立 JSON 文件，天然支持持久化、版本控制和多进程访问
2. DAG 建模依赖 — blockedBy/blocks 双向边精确表达任务间的顺序、并行和阻塞关系
3. 自动解锁机制 — 完成任务时自动清理后续任务的依赖，无需手动操作
4. 共识数据层 — 任务图是所有后续机制（后台执行、多 agent、worktree 隔离）的共享状态
5. 工具接口不变 — 新能力通过新工具表达，agent loop 和 dispatch map 结构保持不变

试一试

cd learn-claude-code
python agents/s07_task_system.py

试试这些 prompt（英文 prompt 对 LLM 效果更好，也可以用中文）：

1. Create 3 tasks: "Setup project", "Write code", "Write tests". Make them depend on each other in order.
2. List all tasks and show the dependency graph
3. Complete task 1 and then list tasks to see task 2 unblocked
4. Create a task board for refactoring: parse -> transform -> emit -> test, where transform and emit can run in parallel after parse

思考题

1. 如果任务 A 依赖 B、B 又依赖 A（循环依赖），系统应该怎么处理？Harness 应该在创建时还是执行时报错？
2. 任务状态的粒度够吗？是否需要 blocked（被阻塞）、failed（失败）、cancelled（取消）等状态？
3. 多个智能体同时修改同一个任务文件可能产生竞态条件，如何保证一致性？

---

[[06-上下文压缩|← 上一课：上下文压缩]] | [[08-后台任务|下一课：后台任务 →]] | [[术语表|术语表]]