my-pi-agent--todolist与background

发表于 2026-08-31 更新于 2026-09-07 分类于 agent实战， my-pi-agent 阅读次数：

三大组成部分

┌─────────────────────────────────────────────────────────────────────────────┐
│                       统一任务系统的三大组成部分                            │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  【第一部分：数据与 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 的数据结构

在 packages/my-agent-core/src/my_agent_core/task_store.py 中，每一个任务卡片都是一个 TaskItem 实例：

@dataclass
class TaskItem:
    id: str                                  # 服务端分配的唯一标识符，如 "task_1", "task_2"
    subject: str                             # 任务短标题（如 "设计数据库表结构"）
    description: str = ""                    # 任务长描述与具体要求
    status: TaskStatus = "pending"           # 状态机：pending | in_progress | completed | deleted
    owner: str | None = None                 # 负责人（如 "agent" 或 "subagent:reviewer"）
    active_form: str | None = None           # 进行时的动态文案（如 "writing schema.sql"）
    blocked_by: list[str] = field(default_factory=list)  # 该任务所依赖的前置任务 ID 列表
    metadata: dict[str, Any] = field(default_factory=dict) # 扩展元数据

### 1. status（工单生命周期状态机）

它记录一个任务“当前处于哪个人生阶段”。一共有 4 种合法状态：

┌──────────┐      认领开工      ┌─────────────┐      完工打勾      ┌───────────┐
│ pending  │ ────────────────► │ in_progress │ ────────────────► │ completed │
└──────────┘  (必须依赖已清空)   └─────────────┘                   └───────────┘
     │                                                                   ▲
     │                              软删除                               │
     └───────────────────────────────────────────────────────────────────┘
                                      │
                                      ▼
                                ┌───────────┐
                                │  deleted  │ (归档墓碑，不影响拓扑)
                                └───────────┘

pending（排队待办）：刚创建出来的初始状态。它可能正被别人卡住（blocked_by 非空），也可能已经就绪随时可做；
in_progress（正在干活）：已经被 Agent 或 Subagent 认领开工。铁律：只有当它的 blocked_by == [] 时，才允许进入此状态！
completed（打勾完工）：任务已经搞定。一旦进入此状态，就会像多米诺骨牌一样，去下游任务的 blocked_by 列表里把自己划掉；
deleted（软删除/墓碑）：作废的任务，不再参与依赖计算。

────────────────────────────────────────────────────────────────────────────────

### 2. blocked_by（前置依赖卡点名单）

它是一个字符串数组（list[str]），记录着“有哪几个上游任务挡在我的面前”： - blocked_by = []（空列表）：说明当前没有任何卡点！它是一条绿色通道，随时可以被认领变成 in_progress； - blocked_by = [“task_1”, “task_2”]（非空列表）：说明这是一个被红灯锁定的任务。即便大模型想认领它，系统也会说：“不行！你必须先把 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_agent_core/tasks.json，断电永不损坏历史；
写操作工具（task_create, task_update, todo_write）诚实声明为 is_parallel_safe=False，由 ToolRegistry 按模型原序串行调度，TaskStore 内部无需加冗余锁，保持架构纯粹极简。

大模型操作面：从离散 CRUD 到单一统一 todo 工具

在设计底层数据操作面时，系统涵盖了 5 种核心操作能力：

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"] 统一分发），既保持了概念清晰，又极大精简了大模型的工具调用认知负担。

内部结构解剖图

┌─────────────────────────────────────────────────────────────────────────────────────────┐
│                                 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_agent_core/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），填入标题和要求，生成一张全新的待办工单，存入内存并落盘。”

端到端全链路

【人类用户】: "帮我重构认证模块并编写测试"
       │
       ▼
【大模型 (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，我下一步可以去给它绑定依赖或者认领它了！”

内部流水线

             [ 调用者发起: 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：为什么任务写操作要严格串行？ - 场景：大模型在同一轮并发调用 task_create(subject="建表") 和 task_create(subject="写API")； - 方案：写操作工具诚实声明为 is_parallel_safe=False，由 ToolRegistry 保序串行执行。任务 A 先进拿到 task_1 并累加计数器；任务 B 随后进入拿到 task_2。保证每个任务绝对拥有唯一的身份证号，且 TaskStore 无需内部加锁，代码极简。

#### 疑问 2：为什么任务 ID 不能让大模型自己填？ - 如果让大模型自己传 ID，大模型第一轮可能起名叫 "1"，第二轮叫 "step_one"，第三轮叫 "task_A"；后续绑定依赖时极易引用错乱； - 由系统统一派发递增 ID（task_1, task_2），规范统一。

update() 方法（增量流转与自动解锁）

端到端全链路图

【大模型 (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 步严格处理：

             [ 调用发起: 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 打勾，准备直接输出纯文本说“我干完了”溜走。对标 Pi 官方扩展事件哲学，核心 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")

为什么必须检查？

大模型常见幻觉：模型在更新 task_1 时，误把参数写成了 add_blocked_by=[“task_1”]；
严重后果：这叫“逻辑自噬”。task_1 必须等 task_1 完成后才能开始，但它不开始就永远无法完成！
防御：只要发现依赖的目标编号和自己一模一样，当场拒绝！

二、检查 2：前置存在性检查（禁止依赖幽灵任务）

源码：

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

什么是“传递性闭环”？

成环往往不是直接的，而是跨越了多个层级：

已有的依赖链条:
task_3 ──依赖──► task_2 ──依赖──► task_1

此时模型试图让:
task_1 ──依赖──► task_3 ？？

_depends_on 是怎么查出这个死锁的？

def _depends_on(self, task_id: str, target_id: str) -> bool:
    visited = set()
    queue = [task_id]          # 初始把 dep 放进队列
    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

核心逻辑：在把 task_1 ➔ 依赖 ➔ task_3 写入前，先从 task_3 开始往上溯源。结果系统发现：task_3 的上游是 task_2，task_2 的上游正是 task_1！系统立刻得出结论：“如果我现在允许 task_1 依赖 task_3，就会形成 1 ➔ 3 ➔ 2 ➔ 1 的闭环死锁！”
防御：直接掐死，并在报错信息中把完整的回路打印出来引导大模型。

render_board() 方法（看板与上下文投影）

作用与使用位置

调用时机与架构演进：
- 早期探索：曾考虑在 Agent.run() 的模型视图构建阶段（BeforeModelCall 决策点前）将看板注入系统提示词；
- 生产架构定型：为了 100% 捍卫大模型供应商的 Prompt Prefix Cache（避免动态改动前缀导致每轮 KV Cache 全量击穿），系统最终将看板投影全面升级为“在 todo 工具的 ToolResult.data['board'] 中随路回显（In-Band Echo）”！

执行效果：若存在活跃任务，自动渲染紧凑 Markdown 块随工具执行结果回显：

1
2
3

[x] task_1: 设计数据库表结构 (completed)
[>] task_2: 编写 API 接口 (in_progress - writing endpoints)
[ ] task_3: 编写单元测试 (pending, blocked by: ['task_2'])

核心定位：这不是面向人类的 UI 可视化（人类 UI 属于产品层/TUI/Web 的职责），而是面向大模型的上下文工程（Prompt Projection）。大模型每轮执行完修改后即可在结果中看到当前焦点与阻塞关系，0 工具额外往返消耗，前缀缓存 100% 稳定，且 Session 磁盘历史绝对零污染。

_save_to_disk

真实磁盘文件内容长这样（完整样例）

打开你的 .my_agent_core/tasks.json，它的真实内容如下：

{
  "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：大模型思考并调用 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 里）

实现方式：每轮大模型说话前，框架在 messages[0]（System Prompt）尾部动态追加当前的 …。
致命缺陷：
1. 击穿前缀缓存（Cache Busting）：Anthropic、OpenAI、DeepSeek 的 KV Cache 是按前缀从前往后匹配的。如果 messages[0] 每一轮因为任务状态变化而改变，整段前缀缓存每轮全量失效！首字延迟（TTFT）从 200ms 飙升到 2~3 秒，Token 账单直接翻倍；
2. 会话历史污染：写入 Session 文件的 System 提示词每一轮都在变，导致回溯历史极其混乱。

### 做法 B：我们当前的做法（利用 ToolResult 随路回传）

实现方式：在 packages/my-agent-core/src/my_agent_core/tools/builtin/task_tools.py 中：

# 当模型调 todo 工具修改状态时：
return ToolResult(
    ok=True,
    data={
        "action": "update",
        "task": {...},
        "unblocked": unblocked,
        "board": store.render_board(), # 看板随路返回！
    },
)

核心优势：
1. 100% 保护前缀缓存：开头的 System Prompt 永久冻结、一字不改；看板作为最新工具调用的客观返回结果自然追加在消息队列最末端，前序的所有对话前缀完美命中缓存；
2. 0 额外查询往返：大模型执行了更新后，不需要再傻傻地调一次 todo(action=“list”)，在当前轮次就当场看清了全盘最新格局；
3. 数据流绝对真实纯净：Session JSONL 磁盘文件只记录正常的 ToolCall 和 ToolResult，没有任何人为伪造的系统幽灵消息。

底层三大模块是如何协同工作的？

这套机制由 3 个模块像钟表齿轮一样紧密咬合：

### 1. 排版中枢：TaskStore.render_board()（task_store.py）

它负责把内存里的任务字典“脱水”压缩成大模型最容易解析的紧凑 Markdown： - [x]：已完工； - [>]：当前唯一在跑的焦点（带 active_form 如 writing schema）； - [ ]：待办（若有依赖，带 blocked by: [‘task_1’]）。（刻意丢掉几十上百字的长篇描述，保证看板只有几十个 Token，极度节省上下文空间）。

### 2. 通信载体：todo 工具（task_tools.py）

单一标准入口，支持 create、update、list、get、clear、write；
所有可能改变任务状态的操作，返回时统统附带 store.render_board()；
声明 is_parallel_safe=False，由底层 ToolRegistry 保证按顺序串行执行，防止大模型同一轮派发多个修改导致看板数据错乱。

### 3. 纪律委员：TaskGuardHook（task_tools.py）

守在 TurnEnd 事件出口；
如果模型把代码写完了，觉得自己做完了想直接输出文本交差，但忘记调工具去把任务状态从 in_progress 改为 completed；
钩子会立刻从外部调用 agent.steer(…)，在安全点把模型拦截拉回：“你还有一个任务在 in_progress，请先更新状态再交差！”，实现管理闭环。

task_tools.py

它扮演着极其关键的角色：它是连接底层数据层（TaskStore）与大模型 ReAct 认知循环的“桥梁与执行终端”。

它在工程上主要交付了两件核心产物： 1. make_todo_tool(store)：为大模型提供一个单一、多功能、防并发竞态的 todo 工具； 2. TaskGuardHook：为框架装配一个基于生命周期事件的后台守卫，防止模型早退。

在哪进行注册

一、Agent.init 中的装配流水线

在 Agent.__init__ 中（约第 135~145 行），系统按清晰的顺序执行初始化：

# 1. 解析 task_store（检查工作区或用户传入配置）
self.task_store = self._init_task_store(task_store)

# 2. ① 注册工具（把 todo 工具注入 ToolRegistry）
self._register_tools(tools)

# 3. ② 注册 Hooks（把 TaskGuardHook 注入 HookRegistry）
self._register_hooks(hooks)

只要 self.task_store 处于启用状态，这两个组件就会分别注册进工具注册表与事件钩子注册表。

### 二、1. todo 工具是在哪注册的？

在 agent.py 的 _register_tools() 方法中（第 195~215 行）：

def _register_tools(self, tools: list[Tool]) -> None:
    """注册用户工具 + 内置 task 工具 + 内置 memory 工具 + 内置 todo 工具。"""
    for t in tools:
        self.registry.register(t)

    ...

    # 关键：如果启用了 task_store，自动把 make_task_tools 产出的 todo 工具注册进注册表！
    if self.task_store:
        for task_tool in make_task_tools(self.task_store):
            if self.registry.get(task_tool.name) is not None:
                raise ValueError(
                    f"Tool name '{task_tool.name}' conflicts with built-in task tool"
                )
            self.registry.register(task_tool) # 注册到 self.registry！

注册到哪：self.registry（即 ToolRegistry，框架的统一工具库）；
注册效果：大模型调用 tools = self.registry.get_schemas() 时，就能看到 todo 工具的定义，从而可以在 ReAct 循环中主动调用它。

### 三、2. TaskGuardHook 守卫是在哪注册的？

在 agent.py 的 _register_hooks() 方法中（第 250~260 行）：

def _register_hooks(self, hooks) -> None:
    """构造时批量注册 hooks（对称 _register_tools）。"""
    # 关键：如果启用了 task_store，自动实例化守卫并挂载两个生命周期事件！
    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)

    # 注册用户显式传入的其他自定义 hooks
    for event_cls, callback in hooks or []:
        self.hooks.register(event_cls, callback)

注册到哪：self.hooks（即 HookRegistry，框架的事件总线）；
注册效果：
1. 会话启动时触发 AgentStart → 执行 guard.on_agent_start，清空防死循环集合；
2. 每轮模型说完话触发 TurnEnd → 执行 guard.on_turn_end，检查是否有在跑任务，若有则调用 self.steer 抓回大模型！

统一工具工厂：make_todo_tool(store: TaskStore)

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 行开始）：

  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 工单
          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)  # 防死循环：单次会话只提醒一次

                  # 关键：调用 steer_fn 注入纠偏指令，在下一轮安全点打断大模型！
                  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），底层消息队列在内层循环安全点自动拉起下一轮：“别走！你还没打勾更新状态，先调 todo 更新！”；
- 防死循环（nudged_ids）：每个任务只催一次，如果模型执意不听劝，第二次不再重复催促，保证系统绝不陷入死循环。

后台异步执行 (Background Tasks)

后台任务跑完了，何时通知大模型？

┌───────────────────────────┐
│ 1. 生产者 (任务完成时)    │  BackgroundRunner 发现子进程结束，
│                           │  直接调用：message_queue.add_followup(通知)
└─────────────┬─────────────┘
              │
              ▼
┌───────────────────────────┐
│ 2. 邮箱缓冲区 (暂存排队)  │  通知安静躺在 followup 队列里，
│                           │  绝不粗暴打断大模型正在说的话或正在调的工具
└─────────────┬─────────────┘
              │
              ▼
┌───────────────────────────┐
│ 3. 消费者 (安全点自唤醒)  │  大模型把当前手头的活干完、准备退出的那一瞬间，
│                           │  外层循环检查 if has_followup()，
│                           │  自动把通知取出来，拉起新一轮让大模型总结！
└───────────────────────────┘

并发调度抉择：用多线程还是原生异步？

真正的耗时命令（如 pytest），是在操作系统里作为一个独立的「子进程（Subprocess）」在跑；而我们在 Python 内部，专门为它创建了一个极其轻量的「监工协程（_worker）」去默默盯着它！

┌─────────────────────────────────────────────────────────────┐
│ Python 虚拟机 (单线程内部)                                   │
│                                                             │
│  【主 Agent 协程】                                          │
│   • 负责和大模型聊天、写文件                                │
│   • 调完后台命令后，立刻脱身！继续推进主线！                 │
│                                                             │
│       │ 派发 (asyncio.create_task)                          │
│       ▼                                                     │
│  【监工协程 _worker】 (只占几百字节内存)                    │
│   • 负责盯着后台子进程                                      │
│   • 在 await proc.communicate() 处闭眼挂起，0% CPU          │
│   • 等子进程跑完了，醒来把结果投递进 Follow-up 邮箱！       │
└──────────────────────────────┬──────────────────────────────┘
                               │ 异步监听管道 (操作系统内核 IOCP / epoll)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 操作系统层面 (Python 外部)                                  │
│                                                             │
│  【真实的操作系统子进程】 (PID: 36372)                      │
│   • 真正跑 pytest / npm install / 编译的地方                │
└─────────────────────────────────────────────────────────────┘

我们当前实现（原生异步）的硬核原理

在 packages/my-agent-core/src/my_agent_core/background.py 中，我们彻底抛弃了多线程，采用了纯粹的 Native Asyncio 协程调度：

async def run_process(self, command: str, cwd: Path | str, description: str = "") -> str:
    ...
    # 【定义一个后台协程】：它是一个可以随时暂停、恢复的任务
    async def _worker() -> None:
        # 1. 异步启动子进程，让操作系统内核接管
        proc = await asyncio.create_subprocess_shell(...)

        # 2. 关键点：【让出 CPU！】
        # 子进程在跑慢测试时，这个 _worker 协程在这里原地“暂停挂起”，
        # 整个 Python 线程瞬间腾出空来，去继续服务大模型和用户！
        stdout, stderr = await proc.communicate()

        # 3. 任务彻底跑完了，内核唤醒此协程，继续往下走：
        self.message_queue.add_followup(notification)

    # 【关键调度动作】：把 _worker 协程交给事件循环，立即放飞！
    asyncio.create_task(_worker())

    # 毫秒级瞬间返回任务 ID 给大模型！大模型完全感受不到任何卡顿！
    return job_id

如何管理后台task

      【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）

所有后台任务在启动那一刻，就会被登记在一张唯一的“任务花名册”中。

在我们的 packages/my-agent-core/src/my_agent_core/background.py 中：

@dataclass
class BackgroundJob:
    """单个后台作业的完整档案。"""
    id: str                                  # 唯一任务编号，如 bg_000001
    description: str                         # 命令描述或命令本身
    status: Literal["running", "completed", "failed", "cancelled"] = "running"
    result: str | None = None                # 执行结果/报错截断
    exit_code: int | None = None             # 操作系统退出码（0为正常，非0为异常）
    process: asyncio.subprocess.Process | None = None # 绑定的系统子进程对象（含 PID）
    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() 时，管理调度引擎会立刻执行全量清理：

# background.py 中的 cancel_all 实现：
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)。无论用户是直接关掉终端窗口，还是 Python 发生严重崩溃，解释器在退出前都会强制把存活的子进程全部扫地出门，保证操作系统的干净。

场景全流程演示

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")，项目圆满完成！

 初始状态
    │
    ▼ 调 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"])
└──────────────┘