三大组成部分 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 ┌─────────────────────────────────────────────────────────────────────────────┐ │ 统一任务系统的三大组成部分 │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ 【第一部分：数据与 DAG 依赖层】（对标 s10 Task System / Claude Code） │ │ • 解决：“任务是什么、怎么存、前后依赖关系是什么？” │ │ • 核心组件：TaskItem + TaskStore + 单一标准 todo 工具（整合增量与批量） │ │ • 关键能力：DAG 依赖图、传递性成环检测、单 in_progress 约束、自动解锁、 │ │ Crash-Safe 原子文件持久化、串行写操作因果调度。 │ │ │ │ ───────────────────────────────────────────────────────────────────────── │ │ │ │ 【第二部分：看板与视图投影层】（对标 s05 TodoWrite / Pi Todo） │ │ • 解决：“大模型每轮如何感知全局进度，做到不迷路且 0 Token 额外开销？” │ │ • 核心组件：ToolResult 随路回显投影 + action="write" 批量便签支持 │ │ • 关键能力：工具返回时自动携带最新 <TASK_BOARD>，100% 捍卫 Prompt 前缀 │ │ 缓存（Prefix Cache），0 额外轮次开销，Session 历史零污染。 │ │ │ │ ───────────────────────────────────────────────────────────────────────── │ │ │ │ 【第三部分：后台异步执行与通知收割层】（对标 s11 Background / Pi 通知流） │ │ • 解决：“慢操作（跑测试、编译构建）如何不卡死主 Agent 循环？” │ │ • 核心组件：BackgroundRunner + MessageQueue Follow-up 队列 │ │ • 关键能力：bash(run_in_background=True) 立即返回占位符、后台非阻塞运行、│ │ 跨平台进程树递归强杀（杜绝孤儿进程）、完成后通知自动收割闭环。│ │ │ └─────────────────────────────────────────────────────────────────────────────┘
领域模型分工：Subagent 委派句柄 vs 项目工单卡片 为避免概念混淆，彻底解耦子代理委派与看板工单：
subagent_tasks.py（SubagentTask / SubagentTaskManager）运行时执行容器/句柄 ，存活于一次 task() 工具调用期间，状态为 RUNNING / COMPLETED / ERROR；task_store.py（TaskItem / TaskStore）持久化待办卡片 ，存在于磁盘 tasks.json 中，状态为 pending / in_progress / completed / deleted。
Task System 在简单的单步任务中，Agent 可以凭记忆搞定；但当面对大型工程开发（如“重构认证模块、新增 OAuth 接口并补全单元测试”）时，
前后顺序不可乱：你不能在“数据库表结构还没建好”之前就去“写 API 接口代码”，也不能在“接口没写完”之前就去“跑集成测试”；
死锁与混乱防范：如果大模型逻辑混乱，让任务 A 等待任务 B，又让任务 B 等待任务 A，就会导致整个 Agent 陷入死锁；
Token 浪费严重：如果像早期系统那样每次修改都重传整张大表，随着任务增多，每轮都会浪费成千上万的 Token。
Task System 的使命：提供一个带 DAG（有向无环图）依赖拓扑、增量修改、自动解锁、崩溃安全的项目工单引擎！
TaskItem 的数据结构 在 src/my_agent_core/task_store.py 中，每一个任务卡片都是一个 TaskItem 实例：
1 2 3 4 5 6 7 8 9 10 @dataclass class TaskItem : id : str subject: str description: str = "" status: TaskStatus = "pending" owner: str | None = None active_form: str | None = None blocked_by: list [str ] = field(default_factory=list ) metadata: dict [str , Any ] = field(default_factory=dict )
1. status（工单生命周期状态机） 它记录一个任务“当前处于哪个人生阶段” 。一共有 4 种合法状态：
1 2 3 4 5 6 7 8 9 10 11 ┌──────────┐ 认领开工 ┌─────────────┐ 完工打勾 ┌───────────┐ │ pending │ ────────────────► │ in_progress │ ────────────────► │ completed │ └──────────┘ (必须依赖已清空) └─────────────┘ └───────────┘ │ ▲ │ 软删除 │ └───────────────────────────────────────────────────────────────────┘ │ ▼ ┌───────────┐ │ deleted │ (归档墓碑，不影响拓扑) └───────────┘
pending（排队待办）：
in_progress（正在干活）：
completed（打勾完工）：
deleted（软删除/墓碑）：
────────────────────────────────────────────────────────────────────────────────
2. blocked_by（前置依赖卡点名单） 它是一个字符串数组（list[str]），记录着“有哪几个上游任务挡在我的面前” ：
blocked_by = []（空列表）：
blocked_by = [“task_1”, “task_2”]（非空列表）：
TaskStore 核心算法与机制 ① 两阶段 DAG 建图（Two-Phase DAG Construction） 大模型在同一轮中发出多个 task_create 调用时，无法提前预知系统分配的 ID。
阶段一（节点创建）：模型调 task_create 创建 task_1 和 task_2；
阶段二（依赖绑定）：模型拿到 ID 后，调用 task_update(task_id="task_2", add_blocked_by=["task_1"]) 建立依赖边。
② 传递性成环检测（Cycle Detection） 在执行 add_blocked_by 时，TaskStore 沿着依赖链进行深度遍历回溯（_depends_on 广度优先算法）：
如果尝试添加 task_1 ➔ task_2 ➔ task_1，系统立即拦截并报错抛出 Cycle detected，从根本上防止任务图死锁。
③ 单 in_progress 聚焦原则（Single In-Progress Invariant） 默认开启 enforce_single_in_progress=True；
当 Agent 尝试把 task_2 设为 in_progress 时，如果 task_1 还在进行中，系统会拒绝并提示必须先完成或暂停前一个任务，强制大模型保持注意力聚焦。
在多 Agent 或后台任务场景下，可设为 False 放开限制，允许互不依赖的分支并发进行。
④ 下游任务自动解锁（Auto-Unblocking Feedback） 当一个任务调用 task_update(status="completed") 时，系统自动遍历所有 pending 任务；
只要某个任务的前置依赖因本次完成而全部清空，系统在本次工具返回值中显式附带：
1 { "task" : { "id" : "task_1" , "status" : "completed" } , "unblocked" : [ "task_2" ] }
大模型收到回显，下一轮立刻知道 task_2 已就绪，实现自动交接！
⑤ 崩溃安全原子落盘与串行调度 持久化写入使用 tempfile + fsync + os.replace 写入 <workspace>/.my-pi-agent/tasks.json，断电永不损坏历史；
写操作工具（task_create, task_update, todo_write）诚实声明为 is_parallel_safe=False，由 ToolRegistry 按模型原序串行调度，TaskStore 内部无需加冗余锁，保持架构纯粹极简。
大模型操作面：从离散 CRUD 到单一统一 todo 工具 在设计底层数据操作面时，系统涵盖了 5 种核心操作能力：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 1. create(subject, description="", active_form=None) ➔ 创建工单，返回分配的 ID 与初始看板 2. update(task_id, status=None, add_blocked_by=None, remove_blocked_by=None, ...) ➔ 增量修改字段与依赖，返回最新状态、unblocked 解锁列表与最新看板 3. get(task_id) ➔ 查阅单条工单的完整信息（包含长文本 description） 4. list(include_deleted=False) ➔ 获取看板摘要列表（默认不吐出长 description，极省 Token） 5. write(todos: list[dict]) ➔ 批量便签覆盖写入快捷能力（对标 s05 TodoWrite 草稿纸模式）
架构演进注记 ：在初始探索期，这 5 种操作曾作为独立的 5 个工具（task_create、task_update 等）导出；在最终架构收敛时，为了防止工具 Schema 数量膨胀并严格对标 Pi 与 Hermes-Agent，我们将其统一封装为了单一的标准 todo 工具 （通过 action: Literal["create", "update", "list", "get", "clear", "write"] 统一分发），既保持了概念清晰，又极大精简了大模型的工具调用认知负担。
内部结构解剖图 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 ┌─────────────────────────────────────────────────────────────────────────────────────────┐ │ TaskStore 内部全景 │ ├─────────────────────────────────────────────────────────────────────────────────────────┤ │ │ │ 【1. 串行调度保护】: 工具层声明 is_parallel_safe=False，ToolRegistry 原序串行执行 │ │ │ │ 【2. 自增序列号】: self._next_id = 3 (自动派发 task_1, task_2, task_3...) │ │ │ │ 【3. 内存工单字典】: self.tasks: dict[str, TaskItem] │ │ │ │ │ ├── "task_1" ──► TaskItem(id="task_1", subject="设计表结构", status="completed", │ │ │ blocked_by=[]) │ │ │ │ │ └── "task_2" ──► TaskItem(id="task_2", subject="编写 API", status="in_progress", │ │ blocked_by=[], owner="agent") │ │ │ │ 【4. 物理文件】: <workspace>/.my-pi-agent/tasks.json (Crash-Safe 原子持久化) │ │ │ └─────────────────────────────────────────────────────────────────────────────────────────┘
内部方法
方法名 大白话作用 现实类比
create(...)创建新工单：服务端自动分配递增编号（如 task_1），填入标题，初始化为待办（pending）并落盘。
挂号取号机吐出一张新排队票。
update(...) (最核心方法)增量修改与流转： 1. 改状态（设为进行中或已完成）； 2. 绑依赖（add_blocked_by，自带成环死锁拦截）； 3. 自动解锁（若完成上游，主动返回哪些下游任务被解锁了）。
工单推进：认领开工、绑定先后顺序、打勾并通知下一位接力。
get(task_id)查单个任务详情：按编号精确调阅一张工单的全部内容（包含长篇描述和具体要求）。
调阅某一份病历或工单的完整档案。
list()看所有活跃任务：列出当前所有没被删除的任务清单。
查看当前看板上的全部工单卡片。
batch_write(todos)批量便签覆盖：支持大模型一次性传一个列表进行批量新建或覆盖更新（用于兼容便签式 todo_write）。
在便签纸上一口气写下一组 3~5 步小计划。
render_board()渲染排版看板：把所有任务格式化为打勾图标的 Markdown 文本（[x] 已完成、[>] 进行中、[ ] 待办），每轮自动投影给大模型看。
把整个工单进度整理成大屏幕实时滚动看板。
create() 方法（工单创建） create() 只做一件事：
“分配一个全局唯一的递增 ID（如 task_1），填入标题和要求，生成一张全新的待办工单，存入内存并落盘。”
端到端全链路 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 【人类用户】: "帮我重构认证模块并编写测试" │ ▼ 【大模型 (LLM)】: 思考后认为需要拆分任务，决定发起工具调用 │ ▼ 输出 JSON: {"name": "todo", "arguments": {"action": "create", "subject": "设计数据库表"}} 【Agent 调度器】: 拦截到工具调用，路由给 todo 工具 │ ▼ 传入实参调用: todo(action="create", subject="设计数据库表") 【task_tools.py】: 调用底层仓库 await store.create(...) │ ▼ ═══════════════ 进入 TaskStore.create() 核心执行 ═══════════════ │ │ 1. 校验参数 ➔ 2. 派发 ID (task_1) ➔ 3. 构造工单 ➔ │ 4. 存内存字典 ➔ 5. 刷盘到 tasks.json │ ▼ ═══════════════════════════════════════════════════════════════ 【TaskStore】: 返回刚刚建好的 TaskItem 对象 │ ▼ 【task_tools.py】: 包装成标准工具观察结果 ToolResult: │ {"task": {"id": "task_1", "subject": "设计数据库表", "status": "pending"}, │ "message": "Created task_1"} ▼ 【大模型 (LLM)】: 收到观察结果，心里有了底： “好的，系统分配的编号是 task_1，我下一步可以去给它绑定依赖或者认领它了！”
内部流水线 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 [ 调用者发起: await store.create(subject="设计表结构") ] │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 1 步：参数清洗与非空校验】 │ │ • sub = subject.strip() 剔除前后空格； │ │ • 检查 sub 是否为空？ │ │ ├── 为空 ──► 抛出 ValueError("Task subject cannot be empty") │ │ └── 合法 ──► 进入下一步。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 2 步：自增 ID 签发与计数器推进】 │ │ • 读取当前计数器: task_id = f"task_{self._next_id}" （如 "task_1"）； │ │ • 计数器立即累加: self._next_id += 1 （变成 2，留给下一个人）。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 3 步：实例化 TaskItem 结构体】 │ │ • id = "task_1" │ │ • subject = "设计表结构" │ │ • status = "pending" (初始状态永远是待办，不能一步登天) │ │ • owner = None (初始无人认领) │ │ • blocked_by = [] (初始依赖为空，等待第二阶段用 update 绑定) │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 4 步：更新内存注册表】 │ │ • self.tasks["task_1"] = task │ │ • 此时内存字典已完成登记，后续的 get/list 方法立刻能查到。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 5 步：触发原子安全落盘】 self._save_to_disk() │ │ • 内存全量数据序列化为 JSON 字符串； │ │ • 写入临时文件 tasks_xxx.tmp ➔ 强制 os.fsync 刷盘 ➔ os.replace 覆盖； │ │ • 确保此时即使操作系统崩溃，磁盘上的任务记录也 100% 完整无损。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 6 步：返回工单实体】 │ │ • 将构建好的 TaskItem 实体对象 return 返回给调用方。 │ └───────────────────────────────────────────────────────────────────────────┘
疑问 1：为什么任务写操作要严格串行？
update() 方法（增量流转与自动解锁） 端到端全链路图 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 【大模型 (LLM)】: 任务 1 完成了，调用工具 todo(action="update", task_id="task_1", status="completed") │ ▼ 【ToolRegistry 调度器】: │ • 识别出 todo (action="update") 是 is_parallel_safe=False（写操作）； │ • 按照因果顺序严格串行执行，调用底层 task_tools.py。 ▼ 【task_tools.py】: 调用底层仓库 await store.update(task_id="task_1", status="completed") │ ▼ ═══════════════ 进入 TaskStore.update() 核心流水线 ═══════════════ │ │ 1. 查工单是否存在 │ 2. 若改状态为 in_progress，检查是否已有在跑任务 │ 3. 若增删依赖，做图遍历成环检测（Cycle Detection） │ 4. 覆盖新字段（status, subject, owner...） │ 5. 【自动解锁流水线】：若是 completed，自动帮下游清除依赖并挑出就绪任务 │ 6. 触发原子刷盘 (_save_to_disk) │ ▼ ═════════════════════════════════════════════════════════════════ 【TaskStore】: 返回元组 (task_1, unblocked=["task_2"]) │ ▼ 【task_tools.py】: 包装成标准 ToolResult 给大模型回显： │ { │ "task": {"id": "task_1", "status": "completed"}, │ "unblocked": ["task_2"], │ "message": "Updated task_1" │ } ▼ 【大模型 (LLM)】: 拿到回显：“task_1 已完成，下游 task_2 刚刚解锁！我下一轮去认领 task_2！”
内部 6 步微观流水线详解 内部经过 6 步严格处理：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 [ 调用发起: await store.update(task_id, status, add_blocked_by...) ] │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 1 步：存在性防御检查】 │ │ • if task_id not in self.tasks: │ │ raise KeyError(f"Task '{task_id}' not found") │ │ • 拦截大模型凭空捏造不存在的 ID，保证只操作真实已登记的工单。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 2 步：单进行中（in_progress）聚焦校验】 │ │ • if status == "in_progress" and self.enforce_single_in_progress: │ │ 扫描其它任务，若发现已有任务处于 in_progress，直接抛出 ValueError！ │ │ • 作用：防止单 Agent 一心二用、多线开花导致全部烂尾。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 3 步：依赖图增删边与死锁成环拦截】 │ │ • 如果传入了 add_blocked_by=[dep, ...]: │ │ 1. 检查 dep == task_id ➔ 严禁自己依赖自己； │ │ 2. 检查 dep in self.tasks ➔ 依赖的目标必须真实存在； │ │ 3. 调用 self._depends_on(dep, task_id) 广度优先回溯检查； │ │ 若发现 dep 已经在等 task_id，成环！➔ 抛出 Cycle detected 拦截； │ │ 4. 安全 ➔ 追加到 task.blocked_by（自动去重）。 │ │ • 如果传入了 remove_blocked_by ➔ 从依赖列表中剔除。 │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 4 步：更新标量字段】 │ │ • 哪个参数传了就更新哪个： │ │ status, subject, description, active_form, owner, metadata │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 5 步：下游多米诺骨牌自动解锁（最核心逻辑）】 │ │ • if status == "completed": │ │ 遍历内存中所有其它处于 pending 的任务: │ │ 1. 检查该任务的 blocked_by 是否包含当前完成的 task_id？ │ │ 2. 如果包含，自动把 task_id 从依赖中移出； │ │ 3. 移出后检查: len(blocked_by) == 0？ │ │ ➔ 如果前置障碍已全清空，追加进 unblocked 列表！ │ └─────────────────────────────────────┬─────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────────────────────────────────────┐ │ 【第 6 步：原子安全落盘并返回双结果】 │ │ • 调用 self._save_to_disk() (临时文件 ➔ fsync 刷盘 ➔ replace 覆盖) │ │ • 返回元组: (更新后的 TaskItem, unblocked 列表) │ └───────────────────────────────────────────────────────────────────────────┘
任务收尾守卫（Early-Exit Guard 自动提醒） 大模型做完任务经常容易忘记调用 todo 打勾，准备直接输出纯文本说“我干完了”溜走。核心 Agent 循环 (Agent.run) 保持 100% 纯净通用，绝不硬编码任何任务状态检查 。系统通过挂载在 TurnEnd 上的独立 TaskGuardHook 守卫进行优雅拦截：
当大模型未发起任何工具调用（not event.tool_results）准备退出本轮时，钩子检查 TaskStore；
若发现看板上仍有处于 in_progress 的任务未结清，钩子调用 agent.steer(nudge) 注入一条强制提醒：Task '{task_id}' is still marked as 'in_progress'. If you have completed it, please call todo(action='update', task_id='{task_id}', status='completed') before concluding.
Agent.run 在安全点检测到 Steering 消息，自动继续推进下一轮，大模型被精准拉回乖乖调 todo 打勾结清，保证任务流 100% 闭环！钩子在单次运行中对同一任务最多敲打一次（nudged_ids 集合，在 AgentStart 时重置），杜绝死循环。
添加前置依赖时的检查 检查 1：自依赖防范（禁止自己等自己）
源码：
1 2 if dep == task_id: raise ValueError("Task cannot depend on itself" )
为什么必须检查？
源码：
1 2 if dep not in self .tasks: raise KeyError(f"Dependency task '{dep} ' not found" )
为什么必须检查？
大模型常见幻觉：模型随手编造了一个不存在的前置编号，比如 add_blocked_by=[“task_888”]；
严重后果：系统里根本没有 task_888，也就永远没有人能把 task_888 标记为 completed；
结果就是该任务的 blocked_by 永远无法清空，变成一张永远被冻结的死卡片！
防御：只有当前 tasks.json 中真实存在且已分配的工单，才允许被绑定为前置依赖。
────────────────────────────────────────────────────────────────────────────────
三、检查 3：传递性成环死锁检测（最硬核的图论防御）
这是最关键、技术含量最高的一道关卡！
源码：
1 2 if self ._depends_on(dep, task_id): raise ValueError(f"Cycle detected: {task_id} -> {dep} -> {task_id} " )
什么是“传递性闭环”？
成环往往不是直接的，而是跨越了多个层级：
1 2 3 4 5 已有的依赖链条: task_3 ──依赖──► task_2 ──依赖──► task_1 此时模型试图让: task_1 ──依赖──► task_3 ？？
_depends_on 是怎么查出这个死锁的？
1 2 3 4 5 6 7 8 9 10 11 12 13 def _depends_on (self, task_id: str , target_id: str ) -> bool : visited = set () queue = [task_id] while queue: curr = queue.pop(0 ) if curr == target_id: return True if curr in visited: continue visited.add(curr) if curr in self .tasks: queue.extend(self .tasks[curr].blocked_by) return False
核心逻辑：
防御：直接掐死，并在报错信息中把完整的回路打印出来引导大模型。
render_board() 方法（看板与上下文投影） 作用与使用位置
调用时机与架构演进 ：
早期探索 ：曾考虑在 Agent.run() 的模型视图构建阶段（BeforeModelCall 决策点前）将看板注入系统提示词；生产架构定型 ：为了 100% 捍卫大模型供应商的 Prompt Prefix Cache（避免动态改动前缀导致每轮 KV Cache 全量击穿） ，系统最终将看板投影全面升级为“在 todo 工具的 ToolResult.data['board'] 中随路回显（In-Band Echo）” ！
执行效果 ：1 2 3 [x] task_1: 设计数据库表结构 (completed) [>] task_2: 编写 API 接口 (in_progress - writing endpoints) [ ] task_3: 编写单元测试 (pending, blocked by: ['task_2'])
核心定位 ：面向大模型的上下文工程（Prompt Projection） 。大模型每轮执行完修改后即可在结果中看到当前焦点与阻塞关系，0 工具额外往返消耗，前缀缓存 100% 稳定，且 Session 磁盘历史绝对零污染。
_save_to_disk 真实磁盘文件内容长这样（完整样例） 打开你的 .my-pi-agent/tasks.json，它的真实内容如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 { "next_id" : 4 , "tasks" : [ { "id" : "task_1" , "subject" : "设计数据库表结构" , "description" : "定义 users 表与 orders 表，使用 PostgreSQL 语法" , "status" : "completed" , "owner" : "agent" , "active_form" : "writing schema.sql" , "blocked_by" : [ ] , "metadata" : { "priority" : "high" } } , { "id" : "task_2" , "subject" : "编写 API 接口" , "description" : "基于 FastAPI 实现用户注册与查询接口" , "status" : "in_progress" , "owner" : "agent" , "active_form" : "implementing endpoints" , "blocked_by" : [ ] , "metadata" : { } } , { "id" : "task_3" , "subject" : "编写单元测试" , "description" : "覆盖 100% 的边界测试用例" , "status" : "pending" , "owner" : null , "active_form" : null , "blocked_by" : [ "task_2" ] , "metadata" : { } } ] }
1. 顶层自增序号：”next_id”（极其重要！） 1 "next_id" : self ._next_id
持久化内容：一个递增整数（比如 4）；
为什么必须持久化它？
这是业界标准的 高水位线（Highwatermark）机制；
如果程序中途退出或者机器重启，重新加载时如果丢失了这个数字，计数器就会重置为 1；
计数器一旦重置为 1，下次创建新任务又会分配出重复的 “task_1”，导致新旧任务编号严重撞车冲突！
存下 “next_id”，保证重启后分配的一定是全新的 task_4。
todolist 整体流转流程图 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 1：用户下发多步复杂需求 │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ 用户指令: "帮我实现登录接口并补全单测" ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 2：大模型思考并调用 todo 工具规划工单 │ │ • 模型发起工具调用: │ │ 1. todo(action="create", subject="写数据库模型") ──► 生成 task_1 │ │ 2. todo(action="create", subject="写登录接口") ──► 生成 task_2 │ │ 3. todo(action="update", task_id="task_2", add_blocked_by=["task_1"]) │ │ 4. todo(action="update", task_id="task_1", status="in_progress", active_form="...") │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 3：TaskStore 引擎执行状态流转与持久化 │ │ • DAG 成环检测（防死锁）、单 in_progress 校验 │ │ • Crash-Safe 原子落盘（tasks.json） │ │ • 触发 self.render_board() 生成紧凑 Markdown 看板 │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 4：【核心机制】通过 ToolResult 随路回显最新看板！ │ │ • 工具返回给模型的 JSON: │ │ { │ │ "action": "update", │ │ "board": "<TASK_BOARD>\n" │ │ "[>] task_1: 写数据库模型 (in_progress - writing schema)\n" │ │ "[ ] task_2: 写登录接口 (pending, blocked by: ['task_1'])\n" │ │ "</TASK_BOARD>" │ │ } │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 5：大模型阅读 ToolResult，胸有成竹写代码 │ │ • 模型看到当前焦点是 [>] task_1 │ │ • 调用 write("models.py", ...) 写入代码 │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 6：模型完工打勾，系统自动解锁下游 │ │ • 模型调用: todo(action="update", task_id="task_1", status="completed") │ │ • TaskStore 剥离下游 task_2 的 blocked_by │ │ • 再次在 ToolResult 中回传: │ │ { │ │ "unblocked": ["task_2"], │ │ "board": "[x] task_1: 写数据库模型 (completed)\n" │ │ "[ ] task_2: 写登录接口 (pending)" │ │ } │ └───────────────────────────────────────────┬────────────────────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────────────────────────────────────────┐ │ 阶段 7：【守卫把关】TaskGuardHook（防模型溜走） │ │ │ │ 分支 A：模型想溜（不调工具准备交差，但看板上还有 in_progress 任务） │ │ └─► 触发 TurnEnd ──► TaskGuardHook 捕获 ──► 调用 agent.steer(催促指令) │ │ ──► Agent 安全点拉起下一轮，把模型抓回来打勾！ │ │ │ │ 分支 B：所有任务正常结清（或无在跑任务） │ │ └─► 顺利收工，Agent 结束本轮输出总结。 │ └────────────────────────────────────────────────────────────────────────────────────────┘
为什么一定要用“工具返回值回传看板”？ 我们对比一下两种不同做法，您就会立刻感受到当前实现的优越性：
做法 A：早期粗暴做法（把看板动态拼在 System Prompt 里）
底层三大模块是如何协同工作的？ 这套机制由 3 个模块像钟表齿轮一样紧密咬合：
1. 排版中枢：TaskStore.render_board()（task_store.py） 它负责把内存里的任务字典“脱水”压缩成大模型最容易解析的紧凑 Markdown：
[x]：已完工；
[>]：当前唯一在跑的焦点（带 active_form 如 writing schema）；
[ ]：待办（若有依赖，带 blocked by: [‘task_1’]）。
单一标准入口，支持 create、update、list、get、clear、write；
所有可能改变任务状态的操作，返回时统统附带 store.render_board()；
声明 is_parallel_safe=False，由底层 ToolRegistry 保证按顺序串行执行，防止大模型同一轮派发多个修改导致看板数据错乱。
守在 TurnEnd 事件出口；
如果模型把代码写完了，觉得自己做完了想直接输出文本交差，但忘记调工具去把任务状态从 in_progress 改为 completed；
钩子会立刻从外部调用 agent.steer(…)，在安全点把模型拦截拉回：“你还有一个任务在 in_progress，请先更新状态再交差
它扮演着极其关键的角色：它是连接底层数据层（TaskStore）与大模型 ReAct 认知循环的“桥梁与执行终端”。
它在工程上主要交付了两件核心产物：
make_todo_tool(store)：为大模型提供一个单一、多功能、防并发竞态的 todo 工具；
TaskGuardHook：为框架装配一个基于生命周期事件的后台守卫，防止模型早退。
在哪进行注册 一、Agent.init 中的装配流水线 在 Agent.init 中（约第 135~145 行），系统按清晰的顺序执行初始化：
1 2 3 4 5 6 7 8 self .task_store = self ._init_task_store(task_store)self ._register_tools(tools)self ._register_hooks(hooks)
只要 self.task_store 处于启用状态，这两个组件就会分别注册进工具注册表与事件钩子注册表。
────────────────────────────────────────────────────────────────────────────────
二、1. todo 工具是在哪注册的？ 在 agent.py 的 _register_tools() 方法中（第 195~215 行）：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 def _register_tools (self, tools: list [Tool] ) -> None : """注册用户工具 + 内置 task 工具 + 内置 memory 工具 + 内置 todo 工具。""" for t in tools: self .registry.register(t) ... if self .task_store: todo_tool = make_todo_tool(self .task_store) if self .registry.get(todo_tool.name) is not None : raise ValueError( f"Tool name '{todo_tool.name} ' conflicts with built-in task tool" ) self .registry.register(todo_tool)
注册到哪：self.registry（即 ToolRegistry，框架的统一工具库）；
注册效果：大模型调用 tools = self.registry.get_schemas() 时，就能看到 todo 工具的定义，从而可以在 ReAct 循环中主动调
────────────────────────────────────────────────────────────────────────────────
三、2. TaskGuardHook 守卫是在哪注册的？ 在 agent.py 的 _register_hooks() 方法中（第 250~260 行）：
1 2 3 4 5 6 7 8 9 10 11 def _register_hooks (self, hooks ) -> None : """构造时批量注册 hooks（对称 _register_tools）。""" if self .task_store: guard = TaskGuardHook(self .task_store, self .steer) self .hooks.register(AgentStart, guard.on_agent_start) self .hooks.register(TurnEnd, guard.on_turn_end) for event_cls, callback in hooks or []: self .hooks.register(event_cls, callback)
注册到哪：self.hooks（即 HookRegistry，框架的事件总线）；
注册效果：
会话启动时触发 AgentStart → 执行 guard.on_agent_start，清空防死循环集合；
每轮模型说完话触发 TurnEnd → 执行 guard.on_turn_end，检查是否有在跑任务，若有则调用 self.steer 抓回大模型！
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 def make_todo_tool (store: TaskStore ) -> Tool: """生成单一统一的标准 todo 工具（对标 Pi & Hermes-Agent）。""" @tool( name="todo" , description=TODO_GUIDELINES, is_parallel_safe=False , ) async def todo ( action: Literal ["create" , "update" , "list" , "get" , "clear" , "write" ], subject: str | None = None , task_id: str | None = None , status: Literal ["pending" , "in_progress" , "completed" , "deleted" ] | None = None , description: str | None = None , active_form: str | None = None , owner: str | None = None , add_blocked_by: list [str ] | None = None , remove_blocked_by: list [str ] | None = None , todos: list [dict [str , Any ]] | None = None , include_deleted: bool = False , ) -> ToolResult:
6 个 Action todo 工具的 6 个 Action 分点精简说明：
1. create（单项新建）
作用 ：传入 subject 新建任务，系统分配自增编号（如 task_1），初始为 pending。返回 ：新任务基本信息 + 最新随路看板（board）。
2. update（状态流转与自动解锁）
作用 ：凭 task_id 推进任务（开工标为 in_progress、完工标为 completed）或调整依赖。返回 ：任务状态 + 自动解锁的下游列表（unblocked） + 最新随路看板（board）。规则 ：强制同一时间只准有 1 个任务在跑；完工打勾时自动解除下游的前置阻塞。
3. write（批量草稿覆写）
作用 ：一次性传入任务数组，像草稿纸一样一键批量初始化全部待办（对标 s05 TodoWrite）。返回 ：批量任务清单 + 最新随路看板（board）。
4. list（轻量全局查）
作用 ：主动查阅全局进展。返回 ：极简摘要列表 + 紧凑看板（board）。刻意滤掉长篇大论的描述，极省 Token。
5. get（深度单卡查）
作用 ：凭 task_id 查看某项任务的详细要求。返回 ：该任务的完整字段（包含长文本 description 与 metadata），专为深入阅读具体要求设计（唯一不随带看板的分支）。
6. clear（一键清盘）
作用 ：清空所有工单，并将自增计数器重置为 1。返回 ：清空确认文案 + 空看板 (No active tasks)。
事件守卫类：TaskGuardHook（防止模型早退） 这是刚才重构的核心亮点（位于文件尾部第 193 行开始）：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 class TaskGuardHook : """任务收尾早退守卫钩子（对标 Pi 扩展架构）：在 TurnEnd 时检查未结清工单，通过 steer 提醒大模型。""" def __init__ (self, task_store: TaskStore, steer_fn: Callable [[str ], None ] ) -> None : self .task_store = task_store self .steer_fn = steer_fn self .nudged_ids: set [str ] = set () def on_agent_start (self, event: AgentStart ) -> None : """会话开始时重置已提醒集合。""" self .nudged_ids.clear() def on_turn_end (self, event: TurnEnd ) -> None : """Turn 结束时检查：若无工具调用且仍有 in_progress 任务，发起 steer 提醒。""" if event.tool_results: return in_progress = [t for t in self .task_store.list () if t.status == "in_progress" ] for t in in_progress: if t.id not in self .nudged_ids: self .nudged_ids.add(t.id ) self .steer_fn( f"Task '{t.id } ' ({t.subject} ) is still marked as 'in_progress'. " f"If you have completed it, please call todo(action='update', task_id='{t.id } ', status='completed') " f"to update your progress before concluding." ) break
它的运作逻辑：
on_agent_start：每次用户发起新的任务运行，清空 nudged_ids，保证新一轮有完整的提醒机会；
on_turn_end：
过滤工具轮：if event.tool_results: return，模型在写文件、跑命令时，不触发；
捕获早退：当大模型没有调用任何工具，准备说“我做完了”直接交差退出时；
状态核实：去 task_store 查一眼——“咦？你还有 in_progress 的任务挂着呢！”；
自动干预：调用 steer_fn（直通底层 agent.steer），底层消息队列在内层循环安全点自动拉起下一轮：“别走！你还没打勾更
防死循环（nudged_ids）：每个任务只催一次，如果模型执意不听劝，第二次不再重复催促，保证系统绝不陷入死循环。
后台异步执行 (Background Tasks) 后台任务跑完了，何时通知大模型？ 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 ┌───────────────────────────┐ │ 1. 生产者 (任务完成时) │ BackgroundRunner 发现子进程结束， │ │ 直接调用：message_queue.add_followup(通知) └─────────────┬─────────────┘ │ ▼ ┌───────────────────────────┐ │ 2. 邮箱缓冲区 (暂存排队) │ 通知安静躺在 followup 队列里， │ │ 绝不粗暴打断大模型正在说的话或正在调的工具 └─────────────┬─────────────┘ │ ▼ ┌───────────────────────────┐ │ 3. 消费者 (安全点自唤醒) │ 大模型把当前手头的活干完、准备退出的那一瞬间， │ │ 外层循环检查 if has_followup()， │ │ 自动把通知取出来，拉起新一轮让大模型总结！ └───────────────────────────┘
并发调度抉择：用多线程还是原生异步？
真正的耗时命令（如 pytest），是在操作系统里作为一个独立的「子进程（Subprocess）」在跑；
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 ┌─────────────────────────────────────────────────────────────┐ │ Python 虚拟机 (单线程内部) │ │ │ │ 【主 Agent 协程】 │ │ • 负责和大模型聊天、写文件 │ │ • 调完后台命令后，立刻脱身！继续推进主线！ │ │ │ │ │ 派发 (asyncio.create_task) │ │ ▼ │ │ 【监工协程 _worker】 (只占几百字节内存) │ │ • 负责盯着后台子进程 │ │ • 在 await proc.communicate() 处闭眼挂起，0% CPU │ │ • 等子进程跑完了，醒来把结果投递进 Follow-up 邮箱！ │ └──────────────────────────────┬──────────────────────────────┘ │ 异步监听管道 (操作系统内核 IOCP / epoll) ▼ ┌─────────────────────────────────────────────────────────────┐ │ 操作系统层面 (Python 外部) │ │ │ │ 【真实的操作系统子进程】 (PID: 36372) │ │ • 真正跑 pytest / npm install / 编译的地方 │ └─────────────────────────────────────────────────────────────┘
我们当前实现（原生异步）的硬核原理 在 src/my_agent_core/background.py 中，我们彻底抛弃了多线程，采用了纯粹的 Native Asyncio 协程调度：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 async def run_process (self, command: str , cwd: Path | str , description: str = "" ) -> str : ... async def _worker () -> None : proc = await asyncio.create_subprocess_shell(...) stdout, stderr = await proc.communicate() self .message_queue.add_followup(notification) asyncio.create_task(_worker()) return job_id
如何管理后台task 1 2 3 4 5 6 7 【1. 诞生】 【2. 运行中】 【3. 终结】 bash(background=True) self.jobs 花名册实时监控 │ │ │ ▼ ▼ ▼ 登记进 BackgroundJob ───► 可通过 bg_status 探活 正常完成 ──► 投递 follow_up 邮件 分配唯一 ID (bg_000001) 可通过 bg_logs 查尾部日志 用户按 Ctrl+C ──► taskkill 整树强杀 毫秒级返回凭证给模型 可通过 bg_kill 手工强杀 程序崩溃退出 ──► atexit 自动兜底收尸
内存状态花名册（BackgroundJob） 所有后台任务在启动那一刻，就会被登记在一张唯一的“任务花名册” 中。
在我们的 src/my_agent_core/background.py 中：
1 2 3 4 5 6 7 8 9 10 @dataclass class BackgroundJob : """单个后台作业的完整档案。""" id : str description: str status: Literal ["running" , "completed" , "failed" , "cancelled" ] = "running" result: str | None = None exit_code: int | None = None process: asyncio.subprocess.Process | None = None started_at: float = field(default_factory=time.time)
BackgroundRunner 内部维护了 self.jobs: dict[str, BackgroundJob] 字典。
当前一共有多少个任务在跑？
每一个任务已经跑了多少秒（time.time() - job.started_at）？
对应的操作系统物理 PID 是多少？
生命周期治理：如何主动取消与强杀（Cancellation） 当大模型或用户发现一个任务卡死、陷入死循环、或者不再需要时，如何管理它的退出？
1. 框架级联动强杀（agent.abort()） 当用户按 Ctrl+C 中止会话，或者调用 agent.abort() 时，管理调度引擎会立刻执行全量清理：
1 2 3 4 5 6 7 async def cancel_all (self ) -> None : """取消所有正在运行的后台子进程并递归杀死进程树。""" for job in self .jobs.values(): if job.status == "running" : job.status = "cancelled" _kill_process_tree(job.process)
2. 操作系统整树根除（杜绝孤儿） 调用 _kill_process_tree：
Windows：taskkill /F /T /PID （/T 沿着句柄树向下遍历，杀绝父进程、子进程与孙进程）；
Unix：os.killpg(os.getpgid(proc.pid), signal.SIGKILL)（对进程组一网打尽）。
3. 解释器退出兜底（atexit） BackgroundRunner.init 中注册了 atexit.register(self._sync_cleanup)。
场景全流程演示 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 Turn 1: 大模型规划 ➔ 调 todo(action="create", subject="设计表结构") ➔ 获得 "task_1" ➔ 调 todo(action="create", subject="编写API") ➔ 获得 "task_2" ➔ 调 todo(action="update", task_id="task_2", add_blocked_by=["task_1"]) ➔ 调 todo(action="update", task_id="task_1", status="in_progress", active_form="writing schema") Turn 2: 大模型干活 ➔ 调 write("schema.sql", ...) 写好了表结构 Turn 3: 大模型打勾并自动解锁 ➔ 调 todo(action="update", task_id="task_1", status="completed") ➔ 工具返回: {"unblocked": ["task_2"]} ➔ 大模型收到反馈，立即调 todo(action="update", task_id="task_2", status="in_progress") 开始写 API！ Turn 4: 启动后台测试 ➔ 调 bash("pytest tests/ -q", run_in_background=True) ➔ 立即返回: "[Background task bg_000001 started]" ➔ 主 Agent 释放控制权，等待后台跑完 Turn 5: 两层循环自动收割通知并总结 ➔ 后台测试完成，MessageQueue 自动注入 <task_notification> ➔ 大模型自动获得测试通过日志，调用 todo(action="update", task_id="task_3", status="completed")，项目圆满完成！
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 初始状态 │ ▼ 调 todo(action="create") ┌──────────────┐ │ task_1 (建表)│ ──状态: pending (blocked_by: []) ───► 【可立即认领】 └──────┬───────┘ │ 绑定依赖 add_blocked_by=["task_1"] ▼ ┌──────────────┐ │ task_2 (API) │ ──状态: pending (⚠️ blocked_by: ["task_1"]) ───► 【不可认领，必须等 task_1】 └──────┬───────┘ │ 绑定依赖 add_blocked_by=["task_2"] ▼ ┌──────────────┐ │ task_3 (测试)│ ──状态: pending (⚠️ blocked_by: ["task_2"]) ───► 【不可认领，必须等 task_2】 └──────────────┘ ══════════════════════════════════════════════════════════════════════════════════════════ 状态机推进：task_1 完成 (调用 todo(action="update", task_id="task_1", status="completed")) ══════════════════════════════════════════════════════════════════════════════════════════ ┌──────────────┐ │ task_1 (建表)│ ──状态: [x] COMPLETED └──────┬───────┘ │ TaskStore 自动从下游移除依赖 (task_2.blocked_by 从 ["task_1"] 变为 []) ▼ ┌──────────────┐ │ task_2 (API) │ ──状态: [>] IN_PROGRESS ◄─── 【自动解锁！返回 unblocked: ["task_2"]】 └──────┬───────┘ │ (task_3 依然被 task_2 阻塞) ▼ ┌──────────────┐ │ task_3 (测试)│ ──状态: [ ] PENDING (⚠️ blocked_by: ["task_2"]) └──────────────┘