learn-claude-code 任务系统实战：磁盘持久化任务图、blockedBy 依赖与多 Agent 协调骨架

learn-claude-code 任务系统实战：磁盘持久化任务图、blockedBy 依赖与多 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 教程的 s07 章（Task System，任务系统）为主线，讲清楚如何把 Agent 的"执行清单"升级为落盘到.tasks/目录的带依赖关系的任务图（DAG）：每个任务一个 JSON 文件，用blockedBy记录前置依赖，完成任务时自动解锁下游。读完本文，你可以理解这套任务系统的完整数据结构、TaskManager的源码级实现、四个任务工具的注册方式，以及它在后续章节（后台任务、多 Agent 团队、worktree 隔离）中如何演化为整个 Harness 的协调骨架。

一、问题：扁平清单为什么不够

s03 章引入的TodoManager（见 agents/s03_todo_write.py）只是一个内存中的扁平清单：没有顺序、没有依赖、状态只有"做完没做完"。

而真实目标是有结构的：任务 B 依赖任务 A；任务 C 和 D 可以并行；任务 E 要等 C 和 D 都完成。没有显式关系，Agent 分不清什么能做、什么被卡住、什么能同时跑。更致命的是——清单只活在内存里，s06 的上下文压缩（context compact）一跑就全没了。

s07 给出的答案是：把扁平清单升级为持久化到磁盘的任务图，让目标比任何一次对话都更长寿。

二、解决方案：随时回答三个问题的任务图

任务图持久化在.tasks/目录，每个任务是一个 JSON 文件，带状态和前置依赖blockedBy。它可以随时回答三个问题（引自 docs/zh/s07-task-system.md）：

• 什么可以做？—— 状态为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）都读写这同一个结构。

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

核心实现位于 agents/s07_task_system.py，TaskManager类（L47-L118）负责 CRUD 和依赖图维护。

3.1 任务数据结构

每个任务就是一个 JSON 文件task_{id}.json，create方法生成的初始结构（L67-L74）：

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	int	自增 ID，构造时通过_max_id() + 1扫描目录中已有的task_*.json得出（L53-L55），因此重启进程后 ID 不会冲突
subject	str	任务标题，task_create的唯一必填参数
description	str	详细描述，默认空串
status	str	状态机三值：pending/in_progress/completed
blockedBy	list[int]	前置依赖任务 ID 列表；列表为空表示可开始
owner	str	负责 Agent，默认空串，为多 Agent 分工预留

读写细节：_load直接读task_{id}.json，不存在时抛ValueError；_save用json.dumps(..., indent=2, ensure_ascii=False)落盘，保证人类可读且中文不被转义（L57-L65）。

3.2 状态变更与依赖边：update 方法

update是任务系统里唯一会同时触碰"状态"和"依赖边"的方法（L79-L93）：

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)

三个关键行为：

1. 状态校验：源码里对status做了白名单校验，非法状态（如done）会直接抛ValueError，由 agent loop 捕获后作为Error: ...返回给模型——状态机是pending → in_progress → completed的三态，不允许随意取值。
2. 完成即解锁：状态置为completed时立即调用_clear_dependency(task_id)，把该 ID 从所有其他任务的blockedBy中移除。
3. 依赖边可增可删：add_blocked_by用集合去重后追加（list(set(...))），remove_blocked_by做差集，允许事后调整任务图。

3.3 依赖解除：_clear_dependency

解锁逻辑非常朴素但足够健壮——直接扫描磁盘上的全部任务文件（L95-L101）：

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)

因为每个任务的完整状态就在磁盘文件里，"完成任务 1 → task_2 和 task_3 的blockedBy里移除 1 → task 4 仍需等待"整个过程不需要任何内存索引。这正是文档强调的"State that survives compression —— because it's outside the conversation"：状态活在对话之外，压缩和重启都动不了它。

3.4 列表输出：list_all 的状态标记

list_all把全部任务按 ID 排序后渲染成带状态标记的一行式摘要（L103-L118）：

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

标记映射为pending → [ ]、in_progress → [>]、completed → [x]，未知状态兜底为[?]；blockedBy非空时追加(blocked by: [...])。这让模型每轮都能用极低 token 成本掌握全局进度。

四、四个任务工具：加入 dispatch map

s07 在原 4 个基础工具（bash/read_file/write_file/edit_file）上注册了 4 个任务工具，工具总数从 5（含上下文）扩充到 8。工具注册表与参数 schema 见 agents/s07_task_system.py：

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"]), }

各工具的完整参数约束（摘自源码中的TOOLS定义，L193-L200）：

工具	必填参数	可选参数	说明
task_create	subject(string)	description(string)	创建pending任务，返回任务 JSON
task_update	task_id(integer)	status(enum:pending/in_progress/completed)、addBlockedBy(int 数组)、removeBlockedBy(int 数组)	改状态或改依赖边；置completed时自动解锁下游
task_list	无	无	返回全量状态摘要
task_get	task_id(integer)	无	返回单个任务的完整 JSON

注意task_update的工具入参是 camelCase（addBlockedBy/removeBlockedBy），与 Python 方法参数（add_blocked_by）在 lambda 里做了转换——这是模型接口与内部 API 的边界。

在 agent loop（L204-L224）中，tool_use块按block.name查TOOL_HANDLERS，执行结果以tool_result回传给模型；handler 抛出的任何异常（包括"任务不存在"、"非法状态"）都会被捕获并转成Error: ...字符串返回，让模型自行纠错而不是进程崩溃。

系统提示词也随之更新为You are a coding agent at {WORKDIR}. Use task tools to plan and track work.（L43），引导模型在多步工作中优先使用任务图。

五、相对 s06 的变更

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

使用边界也由此划清：从 s07 起，任务图是多步工作的默认选择；s03 的 Todo 仍可用于单次会话内的快速清单。

六、源码级纵深：s10 的演进验证与完整集成

s07 的设计在后续章节中被不断加深，读这三处代码可以印证任务图确实成为整个 Harness 的协调骨架：

6.1 s10 版本：字符串 ID、ID 校验与认领/完成动作

章节化目录 s10_task_system/code.py 把任务记录进一步结构化：Task是 dataclass（L66-L73），TaskStore负责 ID 校验（正则^task_[0-9a-f]{8}$，ID 为task_+ 8 位随机十六进制）、工作区逃逸检查（L76-L95）；依赖判断从"清空列表"演化为can_start——blockedBy中全部前置任务 completed才放行，且前置文件缺失同样视为未就绪。生命周期变成显式动作：

pending --claim_task--> in_progress --complete_task--> completed

claim_task负责认领（写owner、校验依赖），complete_task校验 owner 后才允许置为 completed，并回报"刚刚被解锁"的下游任务。测试用例 tests/test_task_system.py 覆盖了这些契约：依赖未满足时拒绝认领（Blocked by: [...]）、owner 不符时拒绝完成（owned by agent, not other）、创建任务时随机 ID 冲突会自动重试而非覆盖、blockedBy指向不存在的任务会报错、.tasks被符号链接到工作区外时直接拒绝写入（Task store escapes the workspace）。这些行为在 s07 的简洁实现里尚未出现，属于后续版本的加固——引用时请注意适用版本。

6.2 s_full.py：任务图如何驱动多 Agent

完整集成版 agents/s_full.py 中保留了=== SECTION: file_tasks (s07) ===（L261-L324），可以推断出两点扩展：

• update额外支持deleted状态：置为deleted时直接unlink对应 JSON 文件，任务图可以真正"删"任务；
• 新增claim(tid, owner)方法（L319-L324），把owner字段填上并置in_progress——这正是 s09 多 Agent 团队里"谁来做"的落地机制。系统提示词同时写明分工原则：Prefer task_create/task_update/task_list for multi-step work. Use TodoWrite for short checklists.（L554）。

6.3 TodoWrite 与 Task System 的定位对比

配套章节文档 s10_task_system/README.zh.md 给出了一张更完整的对照表，可作为选型参考：

TodoWrite (s05)	Task System (s07/s10)
定位	当前任务的执行清单	可恢复的任务系统
存储	进程内 / 会话状态	.tasks/{id}.json
依赖	无	blockedBy依赖图
生命周期	当前会话 / 当前任务	跨会话保留
分工	不负责任务认领	owner/ claim
更新契约	整表替换	对单条记录执行创建、读取、更新、列举

七、试一试：运行 s07

7.1 环境要求

脚本通过load_dotenv加载.env，并从环境读取模型配置（agents/s07_task_system.py）：

• 环境变量MODEL_ID（必填，os.environ["MODEL_ID"]直接取值）；
• 可选ANTHROPIC_BASE_URL（自定义网关；设置后会自动移除ANTHROPIC_AUTH_TOKEN）；
• Python 依赖为anthropicSDK 与python-dotenv，见 requirements.txt。

7.2 运行与推荐 prompt

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

交互提示符为s07 >>，输入q/exit退出。推荐尝试这些 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

观察重点：工作目录下是否生成了.tasks/及其中的task_N.json文件？完成任务 1 后，task_2.json的blockedBy是否变为空数组？重启脚本后task_list是否能恢复完整进度？——这三点正是"任务图比对话长命"的直接证据。

八、小结

s07 用不到两百行代码完成了三个关键动作：把计划从内存搬到磁盘（每任务一 JSON）、把顺序隐式约定变成显式的blockedBy依赖边、把"完成"变成自动解锁下游的原子动作。这套结构没有引入任何数据库或额外服务，纯文件即可表达 DAG、状态机与所有权，也因此成为后续后台任务、Agent 团队和 worktree 隔离机制共同读写的协调骨架。如果你在设计自己的 Agent Harness，"状态放在对话之外"这条原则值得直接借鉴。

【免费下载链接】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