你好我是程序员无隅 这是我的个人主页无隅的主页技术永无止境希望我的内容能帮到你 热门专栏Claude Code 源码解读 | LeetCode 算法笔记 | JVM 面试八股文s03 的 TodoManager 是内存里的扁平清单不能表达依赖也不能在重启后恢复。Task System 就是来解决这个问题的。 一句话说清楚Task System 本质就是把内存里的扁平 todo 升级成磁盘上的持久化任务图。每个任务是独立的 JSON 文件带状态、带依赖、带负责人。Agent 不怕上下文压缩了因为任务图一直在磁盘上。 s03 的 TodoManager 有什么问题回顾一下 s03 的方案TodoManager 内存里的 dict记录 {task_id: {subject, status}}看着挺好但有几个硬伤问题1不支持依赖。A 做完了才能做 B对不起没这个字段。 问题2上下文压缩后丢了。auto_compact 一执行内存里的 todo 没了。 问题3会话重启就没了。Agent 挂了再起todo 清单从零开始。真实复杂任务需要知道三件事什么能做没有前置依赖的 pending 任务 什么被卡住blockedBy 不为空的任务 做完后会解锁什么依赖解除机制s03 管不了。所以 Claude Code 加了 Task System。 核心数据结构每个任务存成独立的 JSON 文件.tasks/ task_1.json task_2.json task_3.json一个任务长这样{id:1,subject:Setup project,status:pending,blockedBy:[],owner:}五个字段每个都有用id 任务编号全局唯一 subject 任务标题一句话说清楚做什么 status pending / in_progress / completed三态流转 blockedBy 前置依赖的任务 ID 列表 owner 负责人多 Agent 协作时分配用注意 blockedBy 的语义它不是我阻塞谁而是**“谁阻塞我”**。这个字段容易搞反。task_1: Setup projectblockedBy: []✅ 可执行task_2: Write codeblockedBy: [1]⛔ 等待中task_3: Write testsblockedBy: [2]⛔ 等待中task_4: Generate summaryblockedBy: [3]⛔ 等待中 TaskManager.create — 创建任务create()负责新建任务并写入磁盘classTaskManager:def__init__(self,tasks_dir:Path):self.dirtasks_dir self.dir.mkdir(exist_okTrue)self._next_idself._max_id()1defcreate(self,subject,description):task{id:self._next_id,subject:subject,status:pending,blockedBy:[],owner:,}self._save(task)self._next_id1returnjson.dumps(task,indent2)拆解一下mkdir(exist_okTrue) .tasks 目录不存在就创建 _max_id() 1 找到下一个可用 ID _save(task) 写成 .tasks/task_x.json 文件 _next_id 1 ID 自增保证不重复说白了就是新建任务对象 写成 JSON 文件没什么花活。 blockedBy 表达依赖这是整个 Task System 最核心的设计。举个例子代码审查 Agent 要做四件事task 1: 扫描代码 task 2: 定位问题 task 3: 生成修复 task 4: 运行测试依赖关系是扫描 → 定位 → 修复 → 测试创建后的初始状态task 1: blockedBy[] ← 没人阻塞可以做 task 2: blockedBy[1] ← 被 task 1 阻塞 task 3: blockedBy[2] ← 被 task 2 阻塞 task 4: blockedBy[3] ← 被 task 3 阻塞判断一个任务能不能做只看两个条件status pending 并且 blockedBy []两个都满足才是当前可执行的任务。完成后解锁完成后解锁完成后解锁扫描代码✅ 可执行定位问题⏳ 等待生成修复⏳ 等待运行测试⏳ 等待 _clear_dependency — 自动解锁完成一个任务后要把它的 ID 从其他任务的blockedBy里删掉def_clear_dependency(self,completed_id):forfinself.dir.glob(task_*.json):taskjson.loads(f.read_text())ifcompleted_idintask.get(blockedBy,[]):task[blockedBy].remove(completed_id)self._save(task)逻辑很简单遍历所有任务文件找到 blockedBy 包含已完成 ID 的任务删掉它。举个例子执行前 task 2: blockedBy[1] task 3: blockedBy[2] task 1 completed 后调用 _clear_dependency(1) task 2: blockedBy[] ← 被解锁了 task 3: blockedBy[2] ← 没变化 task 2 completed 后调用 _clear_dependency(2) task 3: blockedBy[] ← 被解锁了解锁是级联的——完成一个下游可能连续解锁多个。 TaskManager.update — 统一状态管理update()是一个万能方法处理状态变更和依赖修改defupdate(self,task_id,statusNone,add_blocked_byNone,remove_blocked_byNone):taskself._load(task_id)ifstatus:task[status]statusifstatuscompleted:self._clear_dependency(task_id)ifadd_blocked_by:task[blockedBy]list(set(task[blockedBy]add_blocked_by))ifremove_blocked_by:task[blockedBy][xforxintask[blockedBy]ifxnotinremove_blocked_by]self._save(task)六个步骤一气呵成1. _load(task_id) → 读取任务文件 2. 改 status → 更新状态 3. completed 时 → 自动调 _clear_dependency 解锁下游 4. add_blocked_by → 添加依赖用 set 去重 5. remove_blocked_by → 移除依赖 6. _save(task) → 覆盖写回 JSON 文件踩坑提醒add_blocked_by用的是list(set(...))去重防止同一个依赖被加两次。 四个工具模型怎么用 TaskManagerTaskManager 是内部实现模型通过工具来操作它TOOL_HANDLERS{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]),}四个工具各司其职LLM 模型task_create创建新任务task_update更新状态/依赖task_list列出所有任务task_get查询单个任务TaskManager读写 .tasks/ 目录task_create 创建任务写 .tasks/task_x.json task_update 更新状态或依赖触发自动解锁 task_list 列出所有任务看全局进度 task_get 查询单个任务详情模型的工作流先 task_list 看有什么可做的挑一个执行完事后 task_update 标记完成再 task_list 看解锁了什么。 完整执行流程把上面的步骤串起来走一遍完整例子用户创建 4 个任务扫描代码、定位问题、生成修复、运行测试 按依赖关系串起来然后开始执行 Agent 操作 task_create(扫描代码) → task 1 task_create(定位问题) → task 2 task_create(生成修复) → task 3 task_create(运行测试) → task 4 task_update(2, add_blocked_by[1]) task_update(3, add_blocked_by[2]) task_update(4, add_blocked_by[3]) task_list() → 看到只有 task 1 可执行 执行 task 1扫描代码... task_update(1, statuscompleted) task_list() → task 1 完成task 2 解锁 执行 task 2定位问题... task_update(2, statuscompleted) task_list() → task 2 完成task 3 解锁 ...依此类推直到全部完成 和 s03 TodoWrite 的区别维度s03 TodoWrites07 Task System存储位置内存磁盘 JSON 文件数据结构扁平 dict任务图DAG依赖支持无blockedBy 字段持久化否重启丢失是文件在磁盘自动解锁无_clear_dependency适用场景单次会话短任务长任务、多 Agent 协作一句话总结s03 管当前会话进度。 s07 管长期任务结构。 踩坑记录坑1blockedBy 语义搞反blockedBy 是谁阻塞我不是我阻塞谁。写代码时如果搞反了整个依赖链就乱了。记住blockedBy[1]意思是task 1 没完成之前我不能做。坑2_clear_dependency 没处理级联如果 task 1 完成后解锁了 task 2task 2 完成后又解锁了 task 3。但 _clear_dependency 只处理当前这一层不会自动递归。需要在 Agent loop 里循环调用 task_list 来检查是否有新解锁的任务。坑3并发写入文件冲突多个 Agent 同时操作 .tasks/ 目录时可能同时写同一个 JSON 文件。实际工程中需要加文件锁fcntl.flock或者用 SQLite 替代 JSON 文件。 怎么验证用这个 prompt 测一下Create 3 tasks: Setup project, Write code, Write tests. Make them depend on each other in order.期望结果.tasks/ 下生成 3 个 JSON 文件 task 2 blockedBy[1] task 3 blockedBy[2] 当前只有 task 1 可执行再执行Complete task 1 and list tasks.期望结果task 1 statuscompleted task 2 blockedBy[]被解锁 task 2 成为可执行任务 面试话术【问题】Task System 如何支持复杂 Agent 任务【回答】Task System 将复杂目标拆成持久化任务图每个任务有独立状态和blockedBy依赖Agent 通过查询任务图选择当前可执行任务并在任务完成后自动解锁后续任务。任务存储在磁盘上即使上下文压缩或会话重启也能恢复。【关键词】持久化任务图、JSON task、blockedBy、依赖解锁、task_create、task_update、task_list、DAG。【项目里怎么做】在代码审查 Agent 中我会把一次审查拆成扫描代码、定位问题、生成修复、运行测试、生成总结等任务。比如运行测试依赖修复完成总结依赖测试完成。任务状态写入磁盘即使上下文压缩或会话重启Agent 也能通过 task_list 恢复当前进度。【面试追问】Task System 和 TodoWrite 有什么区别TodoWrite 更像单次会话里的临时进度表只记录当前任务状态。Task System 是持久化任务图每个任务独立存储并通过 blockedBy 表达依赖关系。TodoWrite 适合短任务Task System 适合长任务、多 Agent 协作和可恢复执行。制作不易觉得有帮助的话点个关注吧Claude Code 源码解读系列持续更新中。有问题欢迎评论区交流
【 Learn Claude7 】Task System：让 Agent 管理持久化任务图
你好我是程序员无隅 这是我的个人主页无隅的主页技术永无止境希望我的内容能帮到你 热门专栏Claude Code 源码解读 | LeetCode 算法笔记 | JVM 面试八股文s03 的 TodoManager 是内存里的扁平清单不能表达依赖也不能在重启后恢复。Task System 就是来解决这个问题的。 一句话说清楚Task System 本质就是把内存里的扁平 todo 升级成磁盘上的持久化任务图。每个任务是独立的 JSON 文件带状态、带依赖、带负责人。Agent 不怕上下文压缩了因为任务图一直在磁盘上。 s03 的 TodoManager 有什么问题回顾一下 s03 的方案TodoManager 内存里的 dict记录 {task_id: {subject, status}}看着挺好但有几个硬伤问题1不支持依赖。A 做完了才能做 B对不起没这个字段。 问题2上下文压缩后丢了。auto_compact 一执行内存里的 todo 没了。 问题3会话重启就没了。Agent 挂了再起todo 清单从零开始。真实复杂任务需要知道三件事什么能做没有前置依赖的 pending 任务 什么被卡住blockedBy 不为空的任务 做完后会解锁什么依赖解除机制s03 管不了。所以 Claude Code 加了 Task System。 核心数据结构每个任务存成独立的 JSON 文件.tasks/ task_1.json task_2.json task_3.json一个任务长这样{id:1,subject:Setup project,status:pending,blockedBy:[],owner:}五个字段每个都有用id 任务编号全局唯一 subject 任务标题一句话说清楚做什么 status pending / in_progress / completed三态流转 blockedBy 前置依赖的任务 ID 列表 owner 负责人多 Agent 协作时分配用注意 blockedBy 的语义它不是我阻塞谁而是**“谁阻塞我”**。这个字段容易搞反。task_1: Setup projectblockedBy: []✅ 可执行task_2: Write codeblockedBy: [1]⛔ 等待中task_3: Write testsblockedBy: [2]⛔ 等待中task_4: Generate summaryblockedBy: [3]⛔ 等待中 TaskManager.create — 创建任务create()负责新建任务并写入磁盘classTaskManager:def__init__(self,tasks_dir:Path):self.dirtasks_dir self.dir.mkdir(exist_okTrue)self._next_idself._max_id()1defcreate(self,subject,description):task{id:self._next_id,subject:subject,status:pending,blockedBy:[],owner:,}self._save(task)self._next_id1returnjson.dumps(task,indent2)拆解一下mkdir(exist_okTrue) .tasks 目录不存在就创建 _max_id() 1 找到下一个可用 ID _save(task) 写成 .tasks/task_x.json 文件 _next_id 1 ID 自增保证不重复说白了就是新建任务对象 写成 JSON 文件没什么花活。 blockedBy 表达依赖这是整个 Task System 最核心的设计。举个例子代码审查 Agent 要做四件事task 1: 扫描代码 task 2: 定位问题 task 3: 生成修复 task 4: 运行测试依赖关系是扫描 → 定位 → 修复 → 测试创建后的初始状态task 1: blockedBy[] ← 没人阻塞可以做 task 2: blockedBy[1] ← 被 task 1 阻塞 task 3: blockedBy[2] ← 被 task 2 阻塞 task 4: blockedBy[3] ← 被 task 3 阻塞判断一个任务能不能做只看两个条件status pending 并且 blockedBy []两个都满足才是当前可执行的任务。完成后解锁完成后解锁完成后解锁扫描代码✅ 可执行定位问题⏳ 等待生成修复⏳ 等待运行测试⏳ 等待 _clear_dependency — 自动解锁完成一个任务后要把它的 ID 从其他任务的blockedBy里删掉def_clear_dependency(self,completed_id):forfinself.dir.glob(task_*.json):taskjson.loads(f.read_text())ifcompleted_idintask.get(blockedBy,[]):task[blockedBy].remove(completed_id)self._save(task)逻辑很简单遍历所有任务文件找到 blockedBy 包含已完成 ID 的任务删掉它。举个例子执行前 task 2: blockedBy[1] task 3: blockedBy[2] task 1 completed 后调用 _clear_dependency(1) task 2: blockedBy[] ← 被解锁了 task 3: blockedBy[2] ← 没变化 task 2 completed 后调用 _clear_dependency(2) task 3: blockedBy[] ← 被解锁了解锁是级联的——完成一个下游可能连续解锁多个。 TaskManager.update — 统一状态管理update()是一个万能方法处理状态变更和依赖修改defupdate(self,task_id,statusNone,add_blocked_byNone,remove_blocked_byNone):taskself._load(task_id)ifstatus:task[status]statusifstatuscompleted:self._clear_dependency(task_id)ifadd_blocked_by:task[blockedBy]list(set(task[blockedBy]add_blocked_by))ifremove_blocked_by:task[blockedBy][xforxintask[blockedBy]ifxnotinremove_blocked_by]self._save(task)六个步骤一气呵成1. _load(task_id) → 读取任务文件 2. 改 status → 更新状态 3. completed 时 → 自动调 _clear_dependency 解锁下游 4. add_blocked_by → 添加依赖用 set 去重 5. remove_blocked_by → 移除依赖 6. _save(task) → 覆盖写回 JSON 文件踩坑提醒add_blocked_by用的是list(set(...))去重防止同一个依赖被加两次。 四个工具模型怎么用 TaskManagerTaskManager 是内部实现模型通过工具来操作它TOOL_HANDLERS{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]),}四个工具各司其职LLM 模型task_create创建新任务task_update更新状态/依赖task_list列出所有任务task_get查询单个任务TaskManager读写 .tasks/ 目录task_create 创建任务写 .tasks/task_x.json task_update 更新状态或依赖触发自动解锁 task_list 列出所有任务看全局进度 task_get 查询单个任务详情模型的工作流先 task_list 看有什么可做的挑一个执行完事后 task_update 标记完成再 task_list 看解锁了什么。 完整执行流程把上面的步骤串起来走一遍完整例子用户创建 4 个任务扫描代码、定位问题、生成修复、运行测试 按依赖关系串起来然后开始执行 Agent 操作 task_create(扫描代码) → task 1 task_create(定位问题) → task 2 task_create(生成修复) → task 3 task_create(运行测试) → task 4 task_update(2, add_blocked_by[1]) task_update(3, add_blocked_by[2]) task_update(4, add_blocked_by[3]) task_list() → 看到只有 task 1 可执行 执行 task 1扫描代码... task_update(1, statuscompleted) task_list() → task 1 完成task 2 解锁 执行 task 2定位问题... task_update(2, statuscompleted) task_list() → task 2 完成task 3 解锁 ...依此类推直到全部完成 和 s03 TodoWrite 的区别维度s03 TodoWrites07 Task System存储位置内存磁盘 JSON 文件数据结构扁平 dict任务图DAG依赖支持无blockedBy 字段持久化否重启丢失是文件在磁盘自动解锁无_clear_dependency适用场景单次会话短任务长任务、多 Agent 协作一句话总结s03 管当前会话进度。 s07 管长期任务结构。 踩坑记录坑1blockedBy 语义搞反blockedBy 是谁阻塞我不是我阻塞谁。写代码时如果搞反了整个依赖链就乱了。记住blockedBy[1]意思是task 1 没完成之前我不能做。坑2_clear_dependency 没处理级联如果 task 1 完成后解锁了 task 2task 2 完成后又解锁了 task 3。但 _clear_dependency 只处理当前这一层不会自动递归。需要在 Agent loop 里循环调用 task_list 来检查是否有新解锁的任务。坑3并发写入文件冲突多个 Agent 同时操作 .tasks/ 目录时可能同时写同一个 JSON 文件。实际工程中需要加文件锁fcntl.flock或者用 SQLite 替代 JSON 文件。 怎么验证用这个 prompt 测一下Create 3 tasks: Setup project, Write code, Write tests. Make them depend on each other in order.期望结果.tasks/ 下生成 3 个 JSON 文件 task 2 blockedBy[1] task 3 blockedBy[2] 当前只有 task 1 可执行再执行Complete task 1 and list tasks.期望结果task 1 statuscompleted task 2 blockedBy[]被解锁 task 2 成为可执行任务 面试话术【问题】Task System 如何支持复杂 Agent 任务【回答】Task System 将复杂目标拆成持久化任务图每个任务有独立状态和blockedBy依赖Agent 通过查询任务图选择当前可执行任务并在任务完成后自动解锁后续任务。任务存储在磁盘上即使上下文压缩或会话重启也能恢复。【关键词】持久化任务图、JSON task、blockedBy、依赖解锁、task_create、task_update、task_list、DAG。【项目里怎么做】在代码审查 Agent 中我会把一次审查拆成扫描代码、定位问题、生成修复、运行测试、生成总结等任务。比如运行测试依赖修复完成总结依赖测试完成。任务状态写入磁盘即使上下文压缩或会话重启Agent 也能通过 task_list 恢复当前进度。【面试追问】Task System 和 TodoWrite 有什么区别TodoWrite 更像单次会话里的临时进度表只记录当前任务状态。Task System 是持久化任务图每个任务独立存储并通过 blockedBy 表达依赖关系。TodoWrite 适合短任务Task System 适合长任务、多 Agent 协作和可恢复执行。制作不易觉得有帮助的话点个关注吧Claude Code 源码解读系列持续更新中。有问题欢迎评论区交流
