learn-claude-code s07 任务系统：从内存 Todo 清单到磁盘持久化的任务图

本篇围绕 learn-claude-code 课程第 s07 章讲解 Task System（任务系统）：它把 s03 中只存在于内存里的扁平 Todo 清单，升级为落在 .tasks/ 目录下、带 blockedBy 依赖边的磁盘持久化任务图（DAG）。读完本文，你将理解任务图如何回答"什么可执行、什么被阻塞、什么已完成"这三个问题，掌握 TaskManager 的创建、依赖解除与状态迁移实现，并能直接运行 agents/s07_task_system.py 验证依赖自动解锁的全过程。

一、问题：内存里的 Todo 清单撑不起多步目标

s03 中的 TodoManager（见 agents/s03_todo_write.py）本质上只是一个内存里的扁平检查列表：没有顺序、没有依赖关系，状态只有"完成/未完成"两种。而真实目标是带结构的——任务 B 依赖任务 A；任务 C 和 D 可以并行；任务 E 必须等 C 和 D 都完成。

缺少显式关系时，Agent 无法判断什么可以执行、什么被阻塞、什么能同时跑。更致命的是，清单只存在于内存中，一旦触发上下文压缩（s06），它就随之消失。

s07 的解决思路一句话概括：把扁平清单升格为持久化到磁盘的任务图。每个任务是一个 JSON 文件，携带状态（status）和前置依赖（blockedBy）。任务图始终回答三个问题：

• 什么可执行？——pending 状态且 blockedBy 为空的任务；
• 什么被阻塞？——还在等待未完成依赖的任务；
• 什么完成了？——completed 的任务，完成时自动解锁下游任务。

.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 必须早于 2 和 3 完成
并行:       task 2 和 3 可以同时进行
依赖:       task 4 等待 2 和 3 两者
状态:       pending -> in_progress -> completed

这个任务图是 s07 之后所有机制的协调中枢：后台任务（s08）、多 Agent 团队（s09+）、worktree 隔离（s12）都读写同一份 .tasks/ 结构。从仓库源码可以验证这一点——agents/s11_autonomous_agents.py 会扫描 .tasks/ 寻找"无主且未被阻塞"的任务去认领，agents/s12_worktree_task_isolation.py 则把任务绑定到独立 worktree（bind_worktree）。

二、TaskManager：每个任务一个 JSON 文件

完整实现位于 agents/s07_task_system.py，核心是一个 TaskManager 类，目录在模块初始化时就固定为工作区下的 .tasks/：

TASKS_DIR = WORKDIR / ".tasks"   # agents/s07_task_system.py L41
TASKS = TaskManager(TASKS_DIR)  # 全局单例，L121

2.1 创建任务与 ID 分配

ID 采用自增整数：初始化时扫描目录内已有文件取最大 ID，避免重启后 ID 冲突。

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 _max_id(self) -> int:
        ids = [int(f.stem.split("_")[1]) for f in self.dir.glob("task_*.json")]
        return max(ids) if ids else 0

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

每个任务的文件结构为 {"id", "subject", "description", "status", "blockedBy", "owner"}。注意与原始课程文档中的简化代码相比，仓库实际实现多带了 description 字段（用于跨会话恢复上下文），且写盘使用 ensure_ascii=False 以正确保存非 ASCII 文本——见 agents/s07_task_system.py。

2.2 依赖解除：_clear_dependency

任务完成时，TaskManager 会遍历所有任务文件，把已完成任务的 ID 从其他任务的 blockedBy 列表中移除，从而自动解锁下游：

def _clear_dependency(self, completed_id: int):
    """Remove completed_id from all other tasks' blockedBy lists."""
    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)

（agents/s07_task_system.py）

这个"完成即清理边"的设计意味着：判断一个任务是否可执行，只需要看它自己的 blockedBy 是否为空，不需要每次都全图遍历做拓扑计算。

2.3 状态迁移 + 依赖接线：update

update 一个方法同时承担两类职责：改状态、改依赖边。仓库实际实现比文档摘要多了状态校验——传入非法状态会直接抛出 ValueError：

def update(self, task_id: int, status: str = None,
           add_blocked_by: list = None, remove_blocked_by: list = None) -> str:
    task = self._load(task_id)
    if status:
        if status not in ("pending", "in_progress", "completed"):
            raise ValueError(f"Invalid status: {status}")
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)      # 完成 -> 自动解锁下游
    if add_blocked_by:
        task["blockedBy"] = list(set(task["blockedBy"] + add_blocked_by))
    if remove_blocked_by:
        task["blockedBy"] = [x for x in task["blockedBy"] if x not in remove_blocked_by]
    self._save(task)
    return json.dumps(task, indent=2, ensure_ascii=False)

（agents/s07_task_system.py）

由此形成一个极简的三态状态机：

pending ──update──> in_progress ──update(completed)──> completed

add_blocked_by / remove_blocked_by 则让 Agent 可以在规划过程中动态增删依赖边（例如把"写测试"改成同时依赖"写 API"和"写文档"）。

2.4 列表输出：一眼看清阻塞关系

list_all 按 ID 排序输出，用 Markdown 风格的勾选标记 + 阻塞提示呈现整个任务板（agents/s07_task_system.py）：

[x] #1: Setup project
[ ] #2: Write code (blocked by: [1])
[ ] #3: Write tests (blocked by: [2])

标记映射为 pending -> [ ]、in_progress -> [>]、completed -> [x]。

三、四个任务工具接入分发映射

s07 在 s06 的 5 个基础工具（bash、read_file、write_file、edit_file、glob 类工具集）之上，向 TOOL_HANDLERS 注册 4 个任务工具，工具总数变为 8 个（agents/s07_task_system.py）：

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

对应暴露给模型的 JSON Schema（agents/s07_task_system.py）要点如下：

工具	关键参数	说明
task_create	subject（必填）、description	创建任务，初始状态 pending
task_update	task_id（必填）、status（枚举 pending/in_progress/completed）、addBlockedBy / removeBlockedBy（整数数组）	改状态或改依赖边
task_list	无	输出带状态与阻塞关系的任务板
task_get	task_id（必填）	返回单个任务的完整 JSON

注意一个实现细节：工具参数对外是驼峰命名 addBlockedBy / removeBlockedBy，与 TaskManager.update 内部的蛇形参数 add_blocked_by / remove_blocked_by 通过 lambda 适配——模型侧命名与 Python 侧命名解耦。工具异常不会中断 Agent Loop：agent_loop（agents/s07_task_system.py）中 try/except 会把异常转成 "Error: ..." 字符串回填为 tool_result，让模型自行纠错。

s07 之后，任务图成为多步工作的默认规划载体；s03 的 Todo 保留为轻量级单会话检查清单。这一分工也体现在课程最终版 agents/s_full.py 的系统提示中："Prefer task_create/task_update/task_list for multi-step work. Use TodoWrite for short checklists."

四、相对 s06 的变化

组件	Before（s06）	After（s07）
工具数	5	8（新增 task_create / task_update / task_list / task_get）
规划模型	扁平检查清单（内存）	带依赖关系的任务图（磁盘）
关系	无	blockedBy 依赖边
状态追踪	完成/未完成	pending -> in_progress -> completed
持久性	上下文压缩即丢失	压缩、重启后仍然保留

核心洞察正如 agents/s07_task_system.py 文件头注释所写："State that survives compression -- because it's outside the conversation."（状态能存活过压缩，因为它在对话之外。）

五、动手运行 s07

前置条件：已安装 requirements.txt 中的依赖（anthropic、python-dotenv 等），并在环境变量中提供 MODEL_ID（ANTHROPIC_BASE_URL 可选，用于指向兼容 Anthropic 协议的网关）。

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

进入交互 REPL 后（提示符为 s07 >>），依次尝试课程给出的四个提示词：

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

观察要点：

• 工作区下的 .tasks/ 目录中是否生成了 task_1.json … task_N.json，每个文件包含 id、subject、description、status、blockedBy、owner 六个字段；
• 第 3 步完成后，用 cat .tasks/task_2.json 检查：blockedBy 里的 [1] 是否已被 _clear_dependency 移除——这就是"完成即解锁"的直接证据；
• 终端中每次工具调用会打印 > task_create: 等调用预览（截断到 200 字符），便于确认分发链路。

输入 q 或空行即可退出；退出后再次运行，_max_id 会从磁盘恢复已用 ID，任务不会丢——这正是"任务比任何单次会话活得久"的体现。

六、延伸阅读：任务图在仓库中的后续演进

s07 的整数 ID + 自增方案是课程的"最小可运行版本"。仓库中 s10_task_system/ 章节给出了同一思想的强化版 s10_task_system/code.py：改用 task_ 前缀加 8 位随机十六进制的 ID（以 secrets.token_hex(4) 生成，创建时通过独占文件写入避免 ID 冲突），新增 owner 字段与 claim_task / complete_task 两个显式动作，使"任务归属"成为多 Agent 协作的一等概念；其依赖门控与所有者校验逻辑有对应测试覆盖（tests/test_task_system.py）。而 s07 中"完成时改写其他任务文件"的 _clear_dependency，在 s10 中演化为 incomplete_dependencies 惰性查询 + complete_task 返回"刚刚解锁了哪些任务"的提示，信息更丰富。

此外，agents/s_full.py 把 TodoManager 与 TaskManager 同框集成，任务更新逻辑（含完成后的依赖清理与失效任务文件清理）与 s07 一脉相承，可作为理解最终完整 harness 中任务系统形态的对照。

小结：s07 只做了三件事——把清单写到磁盘、给条目加上 blockedBy 依赖边、用"完成即清边"实现自动解锁——却为后续后台任务、Agent 团队与 worktree 隔离提供了共享的协调基础。理解这一章，也就掌握了整个 learn-claude-code 多步工作协调机制的地基。