learn-claude-code 任务系统：用磁盘上的 JSON 文件构建带依赖关系的 Agent 任务图

learn-claude-code 任务系统：用磁盘上的 JSON 文件构建带依赖关系的 Agent 任务图

【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code

本篇围绕 learn-claude-code 教程中的 "Task System"（任务系统）章节展开：讲解如何将内存里的扁平待办清单升级为持久化到磁盘的任务图（Task Graph），实现"任务间依赖（blockedBy）+ 三态状态机 + 自动解除阻塞"。读完后你将掌握文件型任务图的完整设计（TaskManager 的 CRUD、依赖解析、工具注册），并能在 agents/s07_task_system.py 中逐行对应实现，为后续后台任务、多智能体协作打下基础。

1. 问题：扁平待办清单为什么不够

在前序章节中，TodoManager（TodoWrite）提供的只是一份内存中的扁平清单：没有顺序、没有依赖、状态只有"完成/未完成"。真实目标是带结构的——任务 B 依赖任务 A，任务 C 和 D 可以并行，任务 E 要等 C 和 D 都完成。

文档指出了两个致命缺陷：

1. 关系缺失：没有显式的依赖关系时，Agent 无法判断哪些任务"已就绪"、哪些"被阻塞"、哪些"可以并发"；
2. 生命周期过短：清单只存在于内存中，一旦触发上下文压缩（context compression），所有计划信息会被清空。

2. 方案：把清单提升为持久化任务图

核心思路：将清单升级为落盘的任务图（task graph）——每个任务是一个 JSON 文件，携带状态与依赖（blockedBy）。这个图随时能回答三个问题：

• 什么是就绪的（What's ready?）——状态为pending且blockedBy为空的任务；
• 什么是被阻塞的（What's blocked?）——还在等待未完成依赖的任务；
• 什么是已完成的（What's done?）——completed任务；其完成会自动解除下游依赖。

文档给出的目录结构与 DAG 示例：

.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"} Task graph (DAG): +----------+ +--> | task 2 | --+ | | pending | | +----------+ +----------+ +--> +----------+ | task 1 | | task 4 | | completed| --> +----------+ +--> | blocked | +----------+ | task 3 | --+ +----------+ | pending | +----------+ Ordering: task 1 must finish before 2 and 3 Parallelism: tasks 2 and 3 can run at the same time Dependencies: task 4 waits for both 2 and 3 Status: pending -> in_progress -> completed

该图同时表达了四层语义：顺序（task 1 必须先于 2、3）、并行（2 和 3 可同时执行）、依赖汇聚（4 等待 2、3 两者）、状态流转（pending → in_progress → completed）。文档还强调：从本章起，任务图成为后续所有章节（后台执行、多智能体团队、worktree 隔离）的"协调骨架"——它们读写的是同一份磁盘结构。

3. TaskManager：文件即任务记录的 CRUD 实现

agents/s07_task_system.py 中，任务系统的全部逻辑集中在TaskManager类（L47-L118）。

3.1 存储与自增 ID

每个任务对应.tasks/下的一个 JSON 文件，文件名即task_{id}.json：

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

• tasks_dir在模块级被固定为WORKDIR / ".tasks"（L41），即工作目录下的隐藏目录；
• 新实例启动时通过_max_id()扫描已有文件计算最大 ID，因此任务编号在重启后也能无缝续接，不会与旧文件冲突；
• _save()使用json.dumps(task, indent=2, ensure_ascii=False)写盘，保证非 ASCII 字符（如中文 subject）可原样落盘。

3.2 create：任务记录的默认值约定

def create(self, subject, description=""): 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)

任务 JSON 的字段约定：

3.3 依赖解析：完成即自动解除下游阻塞

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)

这是整个任务图的"状态传播"机制：当任务 X 被标记为completed时，系统扫描所有任务文件，把 X 的 ID 从每个依赖方的blockedBy中移除并回写磁盘。依赖边因此是"物理删除"而非"逻辑忽略"——任何新会话重新加载文件后看到的都是最新的就绪状态。

3.4 update：状态机与依赖边的统一入口

def update(self, task_id, status=None, add_blocked_by=None, remove_blocked_by=None): 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)

源码中比文档片段多了一层状态合法性校验：非法状态直接抛ValueError，异常在 agent loop 中被捕获并作为Error: ...文本回传给模型（见 L217-L220），形成"工具失败 → 模型自我纠正"的闭环。add_blocked_by通过set去重，避免重复依赖边。

3.5 list_all：一屏看清任务图

marker = {"pending": "[ ]", "in_progress": "[>]", "completed": "[x]"}.get(t["status"], "[?]") blocked = f" (blocked by: {t['blockedBy']})" if t.get("blockedBy") else "" lines.append(f"{marker} #{t['id']}: {t['subject']}{blocked}")

输出按任务 ID 排序，用[ ]/[>]" /[x]三种勾选标记对应三态，并附当前仍存在的依赖边。Agent 只需调用一次task_list` 就能回答"什么是就绪的、什么是被阻塞的"。

4. 四个任务工具进入调度表

任务能力以 4 个工具的形式注册进TOOL_HANDLERS分发映射（L173-L182），Agent 从 5 个基础工具变为 8 个：

TOOL_HANDLERS = { # ...base tools: bash, read_file, write_file, edit_file... "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（L193-L200）定义了模型可见的参数契约：

注意task_update中status的enum约束与源码中的ValueError校验互为呼应：前者在 API 层引导模型生成合法值，后者在运行时兜底。

5. 相对 s06 的变化

文档给出的定位总结：从 s07 起，任务图成为多步骤工作的默认规划方式；s03 的 Todo 只保留给"单会话快速清单"场景。关键洞察如 agents/s07_task_system.py 头部注释所言——"State that survives compression -- because it's outside the conversation"（能活过压缩的状态，因为它在对话之外）。

6. 仓库内的后续演进：从"图"到"协作协议"

教程的后续课程目录 s10_task_system/code.py 保留了同一套.tasks/JSON 文件约定，并在此骨架上叠加了多智能体协作所需的机制，可以作为本章节方案的"下一步参考实现"：

• 任务 ID 改为随机十六进制：task_{8位hex}（secrets.token_hex(4)生成，L112-L129），创建时以"x"独占模式写文件，ID 冲突则重试，最多 100 次；
• owner+ claim/complete 动作：claim_task把pending → in_progress并写入负责人；complete_task校验 owner 一致才允许in_progress → completed，并在完成时扫描出"刚刚被解除阻塞"的下游任务，把Unblocked: ...直接回传给模型（L185-L228）；
• 就绪判定函数can_start：所有blockedBy前置任务必须为completed，缺失文件视为未完成依赖（L174-L186）；
• 路径安全：TaskStore校验任务存储目录必须位于工作区内，拒绝符号链接逃逸（L80-L95）。

测试用例 tests/test_task_system.py 验证了这些契约：依赖未满足时claim_task返回Blocked by: [...]（test_dependencies_gate_claim_and_completion_checks_owner）；他人 owner 无法完成任务；../outside之类的非法 ID 与不存在的任务都会变成工具错误文本而非崩溃；创建时重复 ID 走重试而非覆盖；符号链接指向工作区外时create_task返回Error: Task store escapes the workspace且不产生任何外部文件。

7. 动手实践（Try It）

运行环境依赖见 requirements.txt（anthropic、python-dotenv、pyyaml），并需要设置MODEL_ID环境变量（可选ANTHROPIC_BASE_URL指向兼容网关）。

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

交互提示符为s07 >>，输入q退出。按文档建议依次尝试这四条指令：

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_2.json等文件，内容与task_get返回一致；
• 完成 task 1 后，task 2 的blockedBy是否被自动清空并落盘（重新运行程序后依赖关系依然保留——这正是"磁盘状态活过会话"的验证）；
• 第 4 条指令会构造出parse → {transform, emit} → test的并行结构，task_list中 transform 与 emit 应同时显示为无阻塞的[ ]任务。

8. 小结

learn-claude-code 的任务系统用最朴素的手段——"每个任务一个 JSON 文件 +blockedBy数组"——实现了完整的任务图语义：顺序、并行、汇聚依赖、三态状态机、完成即解阻、跨会话持久化。它不需要数据库、不需要图引擎，Agent 通过四个原子工具（create/update/list/get）即可规划任意复杂度的多步骤工作，并天然把"什么是就绪的"这一判断暴露给模型。从源码结构看，这套.tasks/约定也是后续后台任务、多 Agent 团队与 worktree 隔离章节的公共存储层——文件即状态，状态即协作协议。

【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code