基于 s07_task_system.py 源码逐行分析，配合 s07-task-system.md 设计思路。

---

一、问题：s03 的 Todo 在 s06 面前不堪一击

回顾一下到目前为止的演进：

• s03 引入了 TodoManager——在内存中维护一个带状态的待办列表
• s06 引入了三层压缩——当上下文超阈值时，LLM 做摘要后丢弃所有旧消息

这两者在 s06 出现了矛盾：TodoManager 存储在 Python 对象的内存中，而 auto_compact 会把整个 messages 替换为摘要。压缩后，todo 列表的细节会丢失——模型知道"有一个任务叫重构 utils.py"，但不记得它完成了 3/5 还是 4/5。

更根本的问题是：s03 的 TodoManager 是扁平的。任务之间没有顺序、没有依赖关系。"重构 utils.py"在 "添加类型注解"之前还是之后？"写测试"能并行做吗？还是必须等"实现功能"完成？扁平的 [ ] [>] [x] 回答不了这些问题。

二、解决方案：持久化到磁盘的任务图

s07 的核心思路：状态只有放在对话外部，才能在压缩后存活。

对话（messages）          ← 会被压缩，会丢失细节
磁盘（.tasks/*.json）     ← 压缩对它没影响，重启程序后还在

每个任务是一个独立的 JSON 文件，存在 .tasks/ 目录下：

.tasks/
  task_1.json  {"id":1, "subject":"设计数据模型", "status":"completed", "blockedBy":[]}
  task_2.json  {"id":2, "subject":"实现 API 接口",  "status":"pending",    "blockedBy":[1]}
  task_3.json  {"id":3, "subject":"写单元测试",    "status":"pending",    "blockedBy":[2]}
  task_4.json  {"id":4, "subject":"写文档",        "status":"pending",    "blockedBy":[2]}

这不再是一个扁平列表，而是一个有向无环图（DAG）：

           +----------+
      +--> | task 2   | --+--> +----------+
      |    | pending  |   |    | task 3   |
+----------+  +----------+   +--> | pending  |
| task 1   |                          +----------+
| completed| --> +----------+
+----------+     | task 4   |
                 | pending  |
                 +----------+

这个图能回答三个问题：

• 什么可以做？ — pending 且 blockedBy 全空的 task 2 和 task 4
• 什么被卡住？ — blockedBy 不为空的 task 3（等 task 2）
• 什么做完了？ — completed 的 task 1

完成 task 2 后，_clear_dependency 自动把它从 task 3 的 blockedBy 中移除——task 3 自动解锁。

---

三、和 s06 相比，多了什么？

组件	s06	s07
规划模型	无（s03 的 Todo 被压缩挂掉）	TaskManager：磁盘持久化 + 依赖图
新工具	compact	task_create/update/list/get（4 个）
持久化	.transcripts/（存档）	.tasks/（任务状态）
关系建模	无	blockedBy：DAG 依赖边
agent_loop	+ 压缩层	+ 任务 CRUD

从 s07 开始，任务图成为后续所有高级机制的骨架——后台执行（s08）、多 agent 团队（s09+）、worktree 隔离（s12）都构建在这个持久化任务系统之上。

---

四、TaskManager 类：CRUD + 依赖图

4.1 初始化和 ID 分配

class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)     # 目录不存在就创建
        self._next_id = self._max_id() + 1  # 找到最大 ID 后 +1

    def _max_id(self) -> int:
        ids = [int(f.stem.split("_")[1]) for f in self.dir.glob("task_*.json")]
        return max(ids) if ids else 0

自增 ID 的实现方式：扫描 .tasks/ 下所有 task_*.json 文件，从文件名提取数字，取最大值 +1。这不是线程安全的，但在单 agent 场景下没问题。

初学者注意：f.stem 是 Path 对象的属性，返回不带后缀的文件名。如 task_3.json → task_3。split("_")[1] 按下划线分割取第二部分，得到 "3"，再 int() 转数字。

4.2 _load() 和 _save() — 文件系统的读写

def _load(self, task_id: int) -> dict:
    path = self.dir / f"task_{task_id}.json"
    if not path.exists():
        raise ValueError(f"Task {task_id} not found")
    return json.loads(path.read_text())

def _save(self, task: dict):
    path = self.dir / f"task_{task['id']}.json"
    path.write_text(json.dumps(task, indent=2, ensure_ascii=False))

_save 中的 indent=2 让 JSON 文件可读（人可以 cat 看），ensure_ascii=False 保留中文等非 ASCII 字符不转义。

4.3 create() — 创建新任务

def create(self, subject: str, description: str = "") -> str:
    task = {
        "id": self._next_id,
        "subject": subject,
        "description": description,
        "status": "pending",          # 新建任务默认未开始
        "blockedBy": [],              # 默认无前置依赖
        "owner": "",                  # 为 s09 多 agent 预留的字段
    }
    self._save(task)
    self._next_id += 1
    return json.dumps(task, indent=2, ensure_ascii=False)

注意 owner 字段——当前是空字符串，但在 s09 引入多 agent 后，这个字段会标识"哪个 agent 负责这个任务"。s07 在设计时就在为未来铺路。

4.4 update() — 状态变更和依赖操作

def update(self, task_id: int, status: str = None,
           add_blocked_by: list = None,
           remove_blocked_by: list = None) -> str:
    task = self._load(task_id)

    if status:
        if status not in ("pending", "in_progress", "completed"):
            raise ValueError(f"Invalid status: {status}")
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)     # 完成后自动解锁后续任务

    if add_blocked_by:
        task["blockedBy"] = list(set(task["blockedBy"] + add_blocked_by))
        # set 去重 — 防止同一个 task_id 被重复加入

    if remove_blocked_by:
        task["blockedBy"] = [x for x in task["blockedBy"]
                             if x not in remove_blocked_by]

    self._save(task)
    return json.dumps(task, indent=2, ensure_ascii=False)

update 可以同时改变状态和依赖关系：

• status="completed" → 自动调用 _clear_dependency 解锁后续任务
• add_blocked_by=[3, 4] → 增加前置依赖（这些任务完成后本任务才能做）
• remove_blocked_by=[3] → 移除前置依赖（手动解锁，不一定等完成）

4.5 _clear_dependency() — 自动解锁后续任务

def _clear_dependency(self, completed_id: int):
    """Remove completed_id from all other tasks' blockedBy lists."""
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

当一个任务完成时（标记 completed），这个方法遍历所有其他任务，把这个 ID 从它们的 blockedBy 中移除。这是一个级联操作：task 1 完成 → task 2 和 task 4 的 blockedBy 被清空 → 它们现在可以开始了。

这个设计意味着模型不需要手动管理依赖——只说"这个任务完成了"，系统自动处理好后续。减少模型的操作负担，降低出错概率。

4.6 list_all() — 渲染任务状态

def list_all(self) -> str:
    tasks = []
    files = sorted(
        self.dir.glob("task_*.json"),
        key=lambda f: int(f.stem.split("_")[1])  # 按 ID 排序
    )
    for f in files:
        tasks.append(json.loads(f.read_text()))

    if not tasks:
        return "No tasks."

    lines = []
    for t in tasks:
        marker = {"pending": "[ ]", "in_progress": "[>]",
                  "completed": "[x]"}.get(t["status"], "[?]")
        blocked = f" (blocked by: {t['blockedBy']})" \
                  if t.get("blockedBy") else ""
        lines.append(f"{marker} #{t['id']}: {t['subject']}{blocked}")
    return "\n".join(lines)

输出效果：

[x] #1: 设计数据模型
[ ] #2: 实现 API 接口 (blocked by: [1])
[ ] #3: 写单元测试 (blocked by: [2])
[ ] #4: 写文档

模型看到这个，一目了然：task 1 做完了，task 4 可以做（无依赖），task 2 被 task 1 卡住（但 task 1 已完成，所以实际可做），task 3 要等 task 2。

---

五、四个任务工具

s07 新增了 4 个工具，和 s02 的基础工具一起组成 8 工具集：

工具	作用	关键参数
task_create	创建新任务	subject（必填）、description（可选）
task_update	更新状态/依赖	task_id（必填）、status、addBlockedBy、removeBlockedBy
task_list	列出所有任务	无参数
task_get	查看单个任务详情	task_id（必填）

分发：

TOOL_HANDLERS = {
    # ...s06 的 4 个基础工具...
    "task_create": lambda **kw: TASKS.create(kw["subject"],
                                              kw.get("description", "")),
    "task_update": lambda **kw: TASKS.update(kw["task_id"],
                                              kw.get("status"),
                                              kw.get("addBlockedBy"),
                                              kw.get("removeBlockedBy")),
    "task_list":   lambda **kw: TASKS.list_all(),
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),
}

agent_loop 完全不变——还是 while 循环 + dispatch map 查找。加任务系统不影响循环结构。

---

六、完整流程走读

假设用户说：“规划一个重构项目：先设计数据模型，再做 API 实现和写文档（可并行），最后写单元测试（要等 API 实现完成）。”

第 1 轮

模型调用 4 次 task_create：

task_create("设计数据模型")           → task 1
task_create("实现 API 接口")         → task 2
task_create("写单元测试")            → task 3
task_create("写文档")                → task 4

第 2 轮

模型设置依赖关系：

task_update(2, addBlockedBy=[1])    ← task 2 要等 task 1 完成
task_update(3, addBlockedBy=[1, 2]) ← task 3 要等 task 1 和 task 2 都完成

task 4 无依赖，和 task 2 可以并行。

第 3 轮

模型调用 task_list()，看到：

[ ] #1: 设计数据模型
[ ] #2: 实现 API 接口 (blocked by: [1])
[ ] #3: 写单元测试 (blocked by: [1, 2])
[ ] #4: 写文档

模型决定先做 task 1：task_update(1, status="in_progress")。

第 4-10 轮

模型专注做完 task 1（读代码 → 设计模型 → 写文件）。

第 11 轮

模型标记完成：task_update(1, status="completed")。_clear_dependency(1) 自动从 task 2 和 task 3 的 blockedBy 中移除 1。

task_list() 现在显示：

[x] #1: 设计数据模型
[ ] #2: 实现 API 接口              ← 自动解锁了！
[ ] #3: 写单元测试 (blocked by: [2]) ← 还被 task 2 卡着
[ ] #4: 写文档

模型看到 task 2 和 task 4 都可以做（无依赖），task 3 要等 task 2。于是启动 task 2，或者在 s09+ 中可以把 task 4 派给另一个 agent 并行执行。

如果此时发生压缩（s06 的 auto_compact）

messages 被替换为摘要。但 .tasks/ 下的 JSON 文件完好无损。下一轮模型调 task_list()，完整的任务图重新出现在上下文——磁盘是抗压缩的。

---

七、和 s03 TodoManager 的本质区别

	s03 TodoManager	s07 TaskManager
存哪里	内存（Python 对象）	磁盘（JSON 文件）
压缩后	细节丢失	完好无损
重启后	消失	存在
结构	扁平列表	DAG（依赖图）
关系	无	blockedBy 边
多 agent	不支持	有 owner 字段预留

s03 的 Todo 适合单次对话内的快速清单——“我要做这三件事，做一件勾一件”。s07 的 Task 适合跨对话的持久化项目——“这个项目有 20 个任务，依赖关系复杂，可能被压缩多次、被不同 agent 执行”。

它们不是替代关系，是互补关系。实际使用中，模型可以同时维护一个 Todo（短期跟踪）和一个 Task 图（长期规划）。

---

八、设计洞察

8.1 文件系统即数据库

TaskManager 没有引入 SQLite、没有用 ORM。就用 JSON 文件 + 目录遍历。这看起来"简陋"，但在这个项目里有巨大的优势：

• 零依赖：不需要安装数据库驱动
• 可调试：cat .tasks/task_3.json 就能看状态
• 可版本控制：.tasks/ 可以加入 git，任务历史就是 commit 历史
• 可脚本化：jq '.status' .tasks/task_*.json 批量查询

当数据量小时，文件系统比数据库更合适。 简单是最高级的复杂。

8.2 任务图是协调骨架

从 s07 开始，任务图不再只是"模型自己的备忘录"。它成为多个 agent 之间协调的数据结构——s08 的后台 agent 从任务图取任务执行，s09 的 agent 团队通过 owner 字段分配任务，s12 的 worktree 隔离为每个任务提供独立工作空间。一个数据结构，被后续 5 个章节复用——好的抽象有这种辐射力。

8.3 自动解锁 vs 手动管理

_clear_dependency 的设计体现了一个原则：让系统承担自动化操作，让模型专注于决策。 模型只需要说"这个任务完成了"，不需要记得"哦对，task 2 还依赖这个任务，我得手动去更新 task 2 的 blockedBy"。减少模型的操作步骤，就是减少出错。

8.4 “owner” 字段的远见

owner 在 s07 中始终为空，但它出现在 schema 里。这不是"不小心多写的"，是为 s09 多 agent 预留的钩子。写代码时知道未来会往哪个方向走，在数据结构上提前开孔——不做功能，但留接口。