learn-claude-code s10 任务系统：基于文件持久化的 Task System——从执行检查表到可协调的阻塞图

本篇技术文章围绕 learn-claude-code 课程的 s10 章节（Task System）展开：讲解如何用 .tasks/ 目录下的 JSON 文件为 Agent 构建一个带 blockedBy 依赖关系、owner 认领机制和三态生命周期（pending → in_progress → completed）的任务系统。读完本篇，你将掌握任务持久化、依赖解锁（unblock）、认领/完成状态机的完整实现方式，并能结合 s10_task_system/code.py 与 tests/test_task_system.py 复现并验证每一个行为细节。

一、为什么 TodoWrite 不够：问题定义

在 learn-claude-code 的渐进式课程中，s05_todo_write/ 章节引入的 TodoWrite 工具让 Agent 把当前任务的执行步骤记成检查表——每个条目有内容和状态，帮助 Agent 确认接下来该做什么。但检查表解决不了另一类问题：

• 跨任务依赖：一个项目被拆成"建数据库表 → 写 API → 补测试"三个任务时，API 必须等数据库表就绪，测试必须等 API 稳定。TodoWrite 只能显示"实现 API 未完成"，却无法让 Harness 判断"这个任务现在能不能开始"。
• 任务归属：多 Agent 协作场景下，需要记录谁在负责哪个任务，避免两个 Agent 同时认领同一工作。
• 跨会话恢复：检查表只活在当前进程/会话状态里，会话结束即丢失；而真实项目中，任务进度需要比一次对话活得更久。

s10 的回答是引入 Task System：每个任务有独立 ID 与状态，blockedBy 记录前置任务，owner 记录负责的 Agent，并以 .tasks/{id}.json 文件持久化到磁盘。章节口号即是对这一层的概括："把大目标拆成小任务，排序并持久化"（Big goals break into small tasks, ordered, persisted to disk）。

二、整体方案：保留 s04 内核，叠加 5 个任务工具

s10 的代码建立在 s04 已有的基础设施之上：保留 5 个基础工具（bash、read_file、write_file、edit_file、glob）、Permission 权限钩子、Hooks 事件机制和共享的 execute_tool 分发路径，然后追加：

• 5 个任务工具：create_task、list_tasks、get_task、claim_task、complete_task；
• .tasks/ 目录下的 JSON 文件持久化；
• 基于 blockedBy 的依赖检查与解锁逻辑。

这一点有测试直接佐证——tests/test_task_system.py 的 test_s10_keeps_the_s04_kernel_and_adds_task_tools 断言 TOOLS 列表的顺序恰好是 5 个基础工具加 5 个任务工具，且 permission_hook 仍注册在 PreToolUse 钩子上、execute_tool 存在：

assert [tool["name"] for tool in lesson.TOOLS] == [
    "bash", "read_file", "write_file", "edit_file", "glob",
    "create_task", "list_tasks", "get_task", "claim_task", "complete_task",
]
assert lesson.permission_hook in lesson.HOOKS["PreToolUse"]

TodoWrite 与 Task System 的定位对比（完整继承自章节文档）：

维度	TodoWrite (s05)	Task System (s10)
定位	当前任务的执行检查表	可恢复的任务系统
存储	进程内 / 会话状态	.tasks/{id}.json
依赖关系	无	blockedBy 依赖图
生命周期	当前会话 / 当前任务	跨会话
分工协调	无任务认领	owner / claim
状态	pending / in_progress / completed	pending / in_progress / completed
粒度	Agent 自己的步骤	可认领、可追踪、可解锁的任务
更新契约	整体替换清单	对单条记录做创建/读取/更新/列表

三、核心机制：数据结构与持久化

3.1 Task 数据类与 ID 规则

每个任务是一个 JSON 文件，存储在 .tasks/ 目录下（相对当前工作目录）。code.py 中的定义：

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str          # pending | in_progress | completed
    owner: str | None    # 负责该任务的 Agent
    blockedBy: list[str] # 依赖任务 ID 列表

ID 由 task_ 前缀加 8 位随机十六进制字符组成，对应 code.py 中的两条常量：

TASKS_DIR = WORKDIR / ".tasks"
TASK_ID_PATTERN = re.compile(r"^task_[0-9a-f]{8}$")

文件以"排他创建"方式写入（open(..., "x")），若生成的 ID 已存在则重新生成，最多重试 100 次。secrets.token_hex(4) 恰好产生 8 个十六进制字符。该重试行为有专门测试：test_create_retries_instead_of_overwriting_an_existing_id 通过 monkeypatch 让 token_hex 连续两次返回 deadbeef，断言第二个任务没有覆盖第一个，而是拿到了重新生成的 task_cafebabe。

3.2 TaskStore：ID 校验、防逃逸与读写

TaskStore 封装了全部 JSON 读写，并在 code.py 处以 TASKS = TaskStore(TASKS_DIR) 实例化为全局任务存储。几个值得注意的实现细节：

（1）ID 校验与路径逃逸防护。 _path() 先用 TASK_ID_PATTERN.fullmatch 校验 ID，再对最终路径做 resolve() + is_relative_to 检查，确保 .tasks/ 是符号链接时也无法把任务文件写到工作区之外（见 code.py）：

def _path(self, task_id: str, create_root: bool = False) -> Path:
    if not isinstance(task_id, str) or not TASK_ID_PATTERN.fullmatch(task_id):
        raise ValueError(f"Invalid task ID: {task_id!r}")
    root = self._root(create=create_root)
    path = (root / f"{task_id}.json").resolve()
    if not path.is_relative_to(root):
        raise ValueError(f"Invalid task ID: {task_id!r}")
    return path

对应测试 test_task_store_rejects_a_symlink_outside_the_workspace 把 .tasks 做成指向工作区外临时目录的符号链接，调用 create_task 得到 Error: Task store escapes the workspace，并断言外部目录保持为空。而 test_invalid_and_missing_task_ids_become_tool_results 验证了非法 ID（如 ../outside）会抛出 Invalid task ID，且异常最终被 execute_tool 捕获为 Error: ... 形式的工具结果，而不是让进程崩溃。

（2）依赖必须在创建时存在。 TaskStore.create 会去重（dict.fromkeys）并逐个检查 blockedBy 中每个依赖的 JSON 文件是否已存在，否则抛出 Dependency not found（见 code.py）。测试 test_create_rejects_unknown_dependencies 确认传入不存在的 task_00000000 时，工具输出恰为 Error: Dependency not found: task_00000000——即依赖图不允许出现"悬空边"。

（3）加载时二次校验。 load() 读取 JSON 后会核对文件内 id 字段与文件名一致，且 status 只能是三态之一（见 code.py）。list() 则用 sorted(root.glob("task_*.json")) 按文件名排序返回全部任务，因此 code.py 的列表顺序是确定性的。

3.3 create_task：创建带依赖的任务

对模型暴露的工具入口：

def create_task(subject: str, description: str = "",
                blockedBy: list[str] | None = None) -> Task:
    return TASKS.create(subject, description, blockedBy)

subject 为空会报 Task subject cannot be empty；blockedBy 用来声明依赖，例如"写 API"任务引用数据库任务的 ID。工具包装层 run_create_task 会把依赖回显在返回值里（Created task_xxxx: xxx (blockedBy: task_yyyy)），方便模型在下一轮决策时看到刚建立的边。

在 code.py 的 TOOLS 定义中，各任务工具的 JSON Schema 如下（create_task 与 claim_task/complete_task）：

{"name": "create_task",
 "description": "Create a task with optional dependencies.",
 "input_schema": {"type": "object", "properties": {
     "subject": {"type": "string"},
     "description": {"type": "string"},
     "blockedBy": {"type": "array", "items": {"type": "string"}}},
     "required": ["subject"]}},
{"name": "claim_task",
 "description": "Claim a pending task whose dependencies are complete.",
 "input_schema": {"type": "object", "properties": {"task_id": {"type": "string"}},
     "required": ["task_id"]}},
{"name": "complete_task",
 "description": "Complete the task claimed by this agent.",
 "input_schema": {"type": "object", "properties": {"task_id": {"type": "string"}},
     "required": ["task_id"]}},

四、依赖检查与状态机

4.1 can_start 与 incomplete_dependencies：阻塞判定

任务只有在 blockedBy 中的所有前置任务都 completed 之后才能开始：

def can_start(task_id: str) -> bool:
    return not incomplete_dependencies(load_task(task_id))

其中 incomplete_dependencies（code.py）逐个加载前置任务：任何前置任务不是 completed，或者其文件已经不存在（FileNotFoundError），都被计入未完成任务列表——这意味着被删除的依赖会被保守地视为"未完成"，任务保持阻塞而不是被错误放行。

4.2 claim_task：认领与状态迁移

Agent 开始处理某任务时调用 claim_task：校验状态与依赖后，设置 owner 并把状态从 pending 推进到 in_progress（code.py）：

def claim_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)
    if task.status != "pending":
        return f"Task {task_id} is {task.status}, cannot claim"
    dependencies = incomplete_dependencies(task)
    if dependencies:
        return f"Blocked by: {dependencies}"
    task.owner = owner
    task.status = "in_progress"
    TASKS.save(task)
    return f"Claimed {task_id} ({task.subject})"

两个拒绝分支都返回字符串消息而非异常，因此模型能读懂"为什么不能开始"并自行调整策略。测试 test_dependencies_gate_claim_and_completion_checks_owner 完整走了一遍这条门控链：

assert lesson.claim_task(api.id) == f"Blocked by: ['{schema.id}']"  # 依赖未完成 → 拒绝认领
assert "Claimed" in lesson.claim_task(schema.id)                     # 无依赖任务 → 认领成功
assert "Unblocked: write API" in lesson.complete_task(schema.id)     # 完成 → 解锁下游
assert "owned by agent, not other" in lesson.complete_task(
    api.id, owner="other")                                           # 非 owner 不能完成
assert "Completed" in lesson.complete_task(api.id)

4.3 complete_task：完成并解锁下游

完成一个任务时，代码先做"完成前快照"（哪些带依赖的 pending 任务当时已可开始），把当前任务置为 completed 并落盘，再重新扫描，找出"新变可开始"的下游任务并写进返回消息（code.py）：

def complete_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)
    if task.status != "in_progress":
        return f"Task {task_id} is {task.status}, cannot complete"
    if task.owner != owner:
        return f"Task {task_id} is owned by {task.owner}, not {owner}"
    ready_before = {t.id for t in list_tasks()
                    if t.status == "pending" and t.blockedBy
                    and can_start(t.id)}
    task.status = "completed"
    TASKS.save(task)
    unblocked = [t.subject for t in list_tasks()
                 if t.status == "pending" and t.blockedBy
                 and t.id not in ready_before
                 and can_start(t.id)]
    msg = f"Completed {task_id} ({task.subject})"
    if unblocked:
        msg += f"\nUnblocked: {', '.join(unblocked)}"
    return msg

两个校验点：只有 in_progress 状态可完成；且完成者必须是认领时的 owner（默认 "agent"）。快照-差集（ready_before 对比 unblocked）的设计保证消息里只报告本次完成动作新解锁的任务，而不是把所有恰好可开始的任务都列一遍。owner 校验正是多 Agent 协调的最小原语：谁认领、谁完成，防止另一个 Agent 半途截胡。

4.4 get_task：跨会话恢复所需的全量详情

list_tasks 只给一行摘要（run_list_tasks 渲染为 [ ] / [>] / [x] 标记加状态、owner、依赖），而 get_task 返回完整任务 JSON（含 description 与依赖明细）：

def get_task(task_id: str) -> str:
    task = load_task(task_id)
    return json.dumps(asdict(task), indent=2)

会话跨断恢复时，Agent 需要读完整描述才能继续工作，而不是只看到一行 subject。

4.5 状态机：两个动作、三个状态

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

• claim_task：pending → in_progress，写入 owner，开始工作；
• complete_task：in_progress → completed，落盘并解锁下游。

从源码结构看，这里没有 cancel / reassign 之类的回退动作，状态迁移是单向的——这是 s10 保持最小化的体现；owner 冲突只能靠"非 owner 无法 complete"这条规则兜住。

五、端到端演练：依赖图上的完整执行

章节文档给出的组合示例（在 code.py 模块 docstring 中也有对应的依赖图示意）：

# 创建带依赖的任务
schema = create_task("setup database schema")
endpoints = create_task("create API endpoints", blockedBy=[schema.id])
tests = create_task("write tests", blockedBy=[endpoints.id])
docs = create_task("write docs", blockedBy=[schema.id])

# Agent 认领第一个可执行任务
claim_task(schema.id)       # ✓ Claimed（无依赖）
complete_task(schema.id)    # ✓ Completed → 解锁 endpoints、docs

claim_task(endpoints.id)    # ✓ Claimed（schema 已完成）
complete_task(endpoints.id) # ✓ Completed → 解锁 tests

claim_task(docs.id)         # ✓ Claimed（schema 已完成）
complete_task(docs.id)      # ✓ Completed

claim_task(tests.id)        # ✓ Claimed（endpoints 已完成）
complete_task(tests.id)     # ✓ Completed

每一次 create_task 都会写出一个 JSON 文件，每一次 claim_task / complete_task 都会更新对应文件。会话结束后 .tasks/ 目录仍然存在，Agent 重新读取这些文件即可恢复进度——这就是"文件持久化任务图"作为多 Agent 协调基座的原因：任何进程都可以通过同一份磁盘事实达成一致。

工具调用路径与基础工具完全一致：模型发出 tool_use 块 → execute_tool 先触发 PreToolUse 钩子（权限检查）→ 按 TOOL_HANDLERS 分发到 run_* 处理器 → 异常统一转成 Error: ... 文本回填到 tool_result。也就是说任务系统没有为 Harness 打开任何"后门"，它只是又一批普通工具。

六、动手运行

6.1 环境准备

s10 与课程其他章节共用同一套依赖与环境变量，见 requirements.txt 与 .env.example：

git clone <仓库地址> learn-claude-code && cd learn-claude-code
pip install -r requirements.txt   # anthropic / python-dotenv / pyyaml
cp .env.example .env              # 填入 ANTHROPIC_API_KEY、MODEL_ID（必填）

注意 code.py 中 MODEL = os.environ["MODEL_ID"] 是硬依赖：缺少 MODEL_ID 会直接抛 KeyError；如配置了 ANTHROPIC_BASE_URL，脚本会自动清除 ANTHROPIC_AUTH_TOKEN 以适配 Anthropic 兼容端点。

6.2 启动与观察要点

cd learn-claude-code
python s10_task_system/code.py

按章节建议依次输入以下提示词：

1. Create tasks: setup database schema, create API endpoints (depends on schema), write tests (depends on endpoints), write docs (depends on schema)
2. List all tasks and their statuses
3. Claim the first unblocked task and complete it
4. List tasks again — which ones are now unblocked?

观察点：.tasks/ 目录下是否生成了 task_XXXXXXXX.json 文件？完成一个任务后，被它阻塞的下游任务是否出现在 Unblocked: 消息中？另外，由于终端会打印调试行（如 [create]、[claim]、[complete]、[unblocked]，见 code.py），你可以直接对照文件内容与日志验证状态迁移。

七、测试矩阵：行为契约速览

tests/test_task_system.py 用 tempfile.TemporaryDirectory 作为隔离工作目录、以假 anthropic/dotenv 模块加载 lesson 代码，覆盖了 s10 的全部关键契约：

测试	验证的行为
test_s10_keeps_the_s04_kernel_and_adds_task_tools	工具表 = s04 内核 + 5 个任务工具；权限钩子仍在 PreToolUse；无副作用目录
test_dependencies_gate_claim_and_completion_checks_owner	依赖门控认领、完成解锁下游、非 owner 不能完成
test_invalid_and_missing_task_ids_become_tool_results	非法/缺失 ID 转成 Error: 工具结果而非异常
test_create_retries_instead_of_overwriting_an_existing_id	ID 冲突时重新生成，绝不覆盖已有任务
test_create_rejects_unknown_dependencies	创建时拒绝不存在的依赖 ID
test_task_store_rejects_a_symlink_outside_the_workspace	符号链接逃逸防护：.tasks 指向外部目录时拒绝写入

八、定位与局限：从源码结构看

• 顺序更新，无并发保护：claim/complete 是"读文件 → 改内存 → 整体写回"，s10 阶段按章节文档的说法是顺序更新任务状态，没有文件锁。若多个进程同时写入同一任务文件，后写者会覆盖先写者；这是留给后续章节（Agent Teams）讨论的协作问题。
• 单向状态机：没有取消、重开或改派动作，任务一旦 completed 即终态；需要"回滚"只能删除 JSON 文件或新建任务。
• 依赖图无环检测：从源码结构看，create 只校验依赖存在，不校验是否会形成环；循环依赖会使两个任务永久互锁（can_start 恒为 False）。这是教学实现有意保留的最简边界。
• 与 s11 的衔接：任务图解决"做什么、按什么顺序做"，但跑全量测试、安装依赖、部署这类慢命令在同步执行时会阻塞整个 Agent Loop。s11_background_tasks/ 章节的 Background Tasks 正是为此而来：慢操作转入后台线程，Agent Loop 继续处理其他任务，后台完成后再以通知形式注入结果。

九、小结

s10 用不到两百行新增代码（Task/TaskStore 加 5 个工具函数，见 s10_task_system/code.py）完成了三件事：把任务从"会话内检查表"升级为"磁盘上的可恢复记录"，用 blockedBy + can_start 给出确定性的依赖门控，用 owner + 两个动作（claim/complete）+ 三个状态定义了最小可用的多 Agent 协作契约。其全部行为均有 tests/test_task_system.py 的断言背书，运行 python s10_task_system/code.py 即可在 .tasks/ 目录中亲手验证文件级持久化与解锁过程。