2026/9/7 5:11:33
FEATURED · 精选文章 
learn-claude-code s10 任务系统基于文件持久化的 Task System——从执行检查表到可协调的阻塞图【免费下载链接】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 课程的 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 / completedpending / 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 TaskStoreID 校验、防逃逸与读写TaskStore封装了全部 JSON 读写并在 code.py 处以TASKS TaskStore(TASKS_DIR)实例化为全局任务存储。几个值得注意的实现细节1ID 校验与路径逃逸防护。_path()先用TASK_ID_PATTERN.fullmatch校验 ID再对最终路径做resolve()is_relative_to检查确保.tasks/是符号链接时也无法把任务文件写到工作区之外见 code.pydef _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(fInvalid task ID: {task_id!r}) root self._root(createcreate_root) path (root / f{task_id}.json).resolve() if not path.is_relative_to(root): raise ValueError(fInvalid 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 emptyblockedBy用来声明依赖例如写 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_dependenciescode.py逐个加载前置任务任何前置任务不是completed或者其文件已经不存在FileNotFoundError都被计入未完成任务列表——这意味着被删除的依赖会被保守地视为未完成任务保持阻塞而不是被错误放行。4.2 claim_task认领与状态迁移Agent 开始处理某任务时调用claim_task校验状态与依赖后设置owner并把状态从pending推进到in_progresscode.pydef claim_task(task_id: str, owner: str agent) - str: task load_task(task_id) if task.status ! pending: return fTask {task_id} is {task.status}, cannot claim dependencies incomplete_dependencies(task) if dependencies: return fBlocked by: {dependencies} task.owner owner task.status in_progress TASKS.save(task) return fClaimed {task_id} ({task.subject})两个拒绝分支都返回字符串消息而非异常因此模型能读懂为什么不能开始并自行调整策略。测试 test_dependencies_gate_claim_and_completion_checks_owner 完整走了一遍这条门控链assert lesson.claim_task(api.id) fBlocked 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, ownerother) # 非 owner 不能完成 assert Completed in lesson.complete_task(api.id)4.3 complete_task完成并解锁下游完成一个任务时代码先做完成前快照哪些带依赖的 pending 任务当时已可开始把当前任务置为completed并落盘再重新扫描找出新变可开始的下游任务并写进返回消息code.pydef complete_task(task_id: str, owner: str agent) - str: task load_task(task_id) if task.status ! in_progress: return fTask {task_id} is {task.status}, cannot complete if task.owner ! owner: return fTask {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 fCompleted {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), indent2)会话跨断恢复时Agent 需要读完整描述才能继续工作而不是只看到一行 subject。4.5 状态机两个动作、三个状态pending ──claim──→ in_progress ──complete──→ completedclaim_taskpending → in_progress写入owner开始工作complete_taskin_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) # ✓ Claimedschema 已完成 complete_task(endpoints.id) # ✓ Completed → 解锁 tests claim_task(docs.id) # ✓ Claimedschema 已完成 complete_task(docs.id) # ✓ Completed claim_task(tests.id) # ✓ Claimedendpoints 已完成 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.examplegit 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按章节建议依次输入以下提示词Create tasks: setup database schema, create API endpoints (depends on schema), write tests (depends on endpoints), write docs (depends on schema)List all tasks and their statusesClaim the first unblocked task and complete itList 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_idID 冲突时重新生成绝不覆盖已有任务test_create_rejects_unknown_dependencies创建时拒绝不存在的依赖 IDtest_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完成了三件事把任务从会话内检查表升级为磁盘上的可恢复记录用blockedBycan_start给出确定性的依赖门控用owner 两个动作claim/complete 三个状态定义了最小可用的多 Agent 协作契约。其全部行为均有 tests/test_task_system.py 的断言背书运行python s10_task_system/code.py即可在.tasks/目录中亲手验证文件级持久化与解锁过程。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
learn-claude-code s10 任务系统：基于文件持久化的 Task System——从执行检查表到可协调的阻塞图
learn-claude-code s10 任务系统基于文件持久化的 Task System——从执行检查表到可协调的阻塞图【免费下载链接】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 课程的 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 / completedpending / 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 TaskStoreID 校验、防逃逸与读写TaskStore封装了全部 JSON 读写并在 code.py 处以TASKS TaskStore(TASKS_DIR)实例化为全局任务存储。几个值得注意的实现细节1ID 校验与路径逃逸防护。_path()先用TASK_ID_PATTERN.fullmatch校验 ID再对最终路径做resolve()is_relative_to检查确保.tasks/是符号链接时也无法把任务文件写到工作区之外见 code.pydef _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(fInvalid task ID: {task_id!r}) root self._root(createcreate_root) path (root / f{task_id}.json).resolve() if not path.is_relative_to(root): raise ValueError(fInvalid 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 emptyblockedBy用来声明依赖例如写 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_dependenciescode.py逐个加载前置任务任何前置任务不是completed或者其文件已经不存在FileNotFoundError都被计入未完成任务列表——这意味着被删除的依赖会被保守地视为未完成任务保持阻塞而不是被错误放行。4.2 claim_task认领与状态迁移Agent 开始处理某任务时调用claim_task校验状态与依赖后设置owner并把状态从pending推进到in_progresscode.pydef claim_task(task_id: str, owner: str agent) - str: task load_task(task_id) if task.status ! pending: return fTask {task_id} is {task.status}, cannot claim dependencies incomplete_dependencies(task) if dependencies: return fBlocked by: {dependencies} task.owner owner task.status in_progress TASKS.save(task) return fClaimed {task_id} ({task.subject})两个拒绝分支都返回字符串消息而非异常因此模型能读懂为什么不能开始并自行调整策略。测试 test_dependencies_gate_claim_and_completion_checks_owner 完整走了一遍这条门控链assert lesson.claim_task(api.id) fBlocked 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, ownerother) # 非 owner 不能完成 assert Completed in lesson.complete_task(api.id)4.3 complete_task完成并解锁下游完成一个任务时代码先做完成前快照哪些带依赖的 pending 任务当时已可开始把当前任务置为completed并落盘再重新扫描找出新变可开始的下游任务并写进返回消息code.pydef complete_task(task_id: str, owner: str agent) - str: task load_task(task_id) if task.status ! in_progress: return fTask {task_id} is {task.status}, cannot complete if task.owner ! owner: return fTask {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 fCompleted {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), indent2)会话跨断恢复时Agent 需要读完整描述才能继续工作而不是只看到一行 subject。4.5 状态机两个动作、三个状态pending ──claim──→ in_progress ──complete──→ completedclaim_taskpending → in_progress写入owner开始工作complete_taskin_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) # ✓ Claimedschema 已完成 complete_task(endpoints.id) # ✓ Completed → 解锁 tests claim_task(docs.id) # ✓ Claimedschema 已完成 complete_task(docs.id) # ✓ Completed claim_task(tests.id) # ✓ Claimedendpoints 已完成 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.examplegit 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按章节建议依次输入以下提示词Create tasks: setup database schema, create API endpoints (depends on schema), write tests (depends on endpoints), write docs (depends on schema)List all tasks and their statusesClaim the first unblocked task and complete itList 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_idID 冲突时重新生成绝不覆盖已有任务test_create_rejects_unknown_dependencies创建时拒绝不存在的依赖 IDtest_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完成了三件事把任务从会话内检查表升级为磁盘上的可恢复记录用blockedBycan_start给出确定性的依赖门控用owner 两个动作claim/complete 三个状态定义了最小可用的多 Agent 协作契约。其全部行为均有 tests/test_task_system.py 的断言背书运行python s10_task_system/code.py即可在.tasks/目录中亲手验证文件级持久化与解锁过程。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
