Claude Code 07：Task System——磁盘持久化任务图

在第 3 篇中，我们通过内存版的 TodoManager 让模型具备了基本的工作清单意识。但随着任务规模从“修改两个函数”升级到“迁移大型模块与微服务”，内存快照的脆弱性暴露无遗：

  1. 进程崩溃即前功尽弃：只要终端意外中断或电源故障，整个会话中的任务进度全部归零；
  2. 缺乏前置依赖拓扑（DAG）：无法清晰表达“任务 C（部署）必须严格等待任务 A（构建）和任务 B（单测通过）完成后才能开始”；
  3. 无法支撑多智能体协同：多个并发的 Agent 无法安全地共享和认领同一个内存对象。

Claude Code 在其进阶架构中建立了一套以磁盘文件为中枢的持久化任务框架（Task System）。所有任务被建模为一个有向无环图（DAG），每个任务对应物理磁盘上的独立实体。

本文将深入 Claude Code 真实的统一任务骨架，并从零实现一个支持依赖阻塞与崩溃恢复的持久化任务引擎。

生产源码探秘：Claude Code 的统一任务体系

在 Claude Code 源码 src/Task.ts 中，任务绝不是普通待办那么简单，它承载着整个系统的异步调度与多智能体分工。

1. 七大任务类型与带前缀的 ID 系统

系统为不同应用场景定义了 7 种专有 TaskType，每种类型分配了一个唯一的英文字符前缀：

1
2
3
4
5
6
7
8
9
// src/Task.ts 生产类型定义
export type TaskType =
  | 'local_bash'          // 前缀 'b' —— 本地 Shell 命令执行
  | 'local_agent'         // 前缀 'a' —— 本地派生的子 Agent
  | 'remote_agent'        // 前缀 'r' —— 远端云环境沙箱 Agent
  | 'in_process_teammate' // 前缀 't' —— 同进程内的协作队友 Agent
  | 'local_workflow'      // 前缀 'w' —— 本地自动化工作流脚本
  | 'monitor_mcp'         // 前缀 'm' —— 长期存活的 MCP 事件监听
  | 'dream'               // 前缀 'd' —— 记忆系统后台夜间整理任务

系统通过 generateTaskId() 生成由“类型前缀 + 8 位 36 进制随机串”构成的 ID（例如 b3k7f9x2a 或 ahj4m8n1p）。这使得任何组件只需看一眼 ID，就能立即推断出它的执行模型与生命周期。

2. 严格的五态生命周期守卫

所有任务无论类型，统一遵循严谨的 5 态状态机：

当任务进入 completed、failed 或 killed 终态后，系统实施严格的终态守卫（Terminal Guard）：禁止任何针对已死任务的状态覆写，防止异步回调晚到引发状态倒退。

3. TaskStateBase 物理底座

1
2
3
4
5
6
7
8
9
10
11
export type TaskStateBase = {
  id: string              // 带前缀的全局唯一 ID
  type: TaskType          // 任务类型
  status: TaskStatus      // 当前运行状态
  description: string     // 人类可读的任务说明
  startTime: number       // 启动毫秒时间戳
  endTime?: number        // 结束时间戳
  outputFile: string      // 专属的磁盘输出日志文件路径
  outputOffset: number    // 游标已读取的字节偏移量
  notified: boolean       // 状态流转是否已向父级发出通知
}

值得注意的是 outputFile 与 outputOffset 机制：任务产生的持续日志不保存在内存，而是直接追加写入磁盘流。外部观察者只需要记录游标 offset，就能增量读取新增日志，彻底避免把大文件重复加载进内存。

磁盘持久化 DAG 引擎实现

我们用 Python 实现一个具备依赖拓扑阻塞与崩溃恢复能力的持久化任务引擎：

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
67
68
69
import os
import json
import uuid
from typing import List, Dict, Optional, Literal

Status = Literal["pending", "running", "completed", "failed"]

class PersistentTaskDAG:
    def __init__(self, storage_dir: str = "./.claude_tasks"):
        self.storage_dir = storage_dir
        os.makedirs(self.storage_dir, exist_ok=True)

    def _task_path(self, task_id: str) -> str:
        return os.path.join(self.storage_dir, f"{task_id}.json")

    def create_task(self, description: str, blocked_by: Optional[List[str]] = None) -> str:
        """创建一个持久化任务并建立前置依赖"""
        task_id = f"t_{uuid.uuid4().hex[:8]}"
        task_data = {
            "id": task_id,
            "description": description,
            "status": "pending",
            "blocked_by": blocked_by or []
        }
        with open(self._task_path(task_id), "w", encoding="utf-8") as f:
            json.dump(task_data, f, indent=2, ensure_ascii=False)
        return task_id

    def load_all_tasks(self) -> Dict[str, dict]:
        tasks = {}
        for fname in os.listdir(self.storage_dir):
            if fname.endswith(".json"):
                with open(os.path.join(self.storage_dir, fname), "r", encoding="utf-8") as f:
                    data = json.load(f)
                    tasks[data["id"]] = data
        return tasks

    def update_status(self, task_id: str, new_status: Status):
        path = self._task_path(task_id)
        if not os.path.exists(path):
            raise FileNotFoundError(f"任务 {task_id} 不存在")
        
        with open(path, "r", encoding="utf-8") as f:
            task = json.load(f)
        
        task["status"] = new_status
        with open(path, "w", encoding="utf-8") as f:
            json.dump(task, f, indent=2, ensure_ascii=False)

    def get_executable_tasks(self) -> List[dict]:
        """
        依赖决策核心算法：
        只有当一个 pending 任务的所有前置依赖都达到 completed 终态时，它才处于可执行就绪状态
        """
        all_tasks = self.load_all_tasks()
        ready = []
        for tid, t in all_tasks.items():
            if t["status"] == "pending":
                # 检查所有前置依赖项的状态
                all_dependencies_met = True
                for dep_id in t["blocked_by"]:
                    dep_task = all_tasks.get(dep_id)
                    if not dep_task or dep_task["status"] != "completed":
                        all_dependencies_met = False
                        break
                
                if all_dependencies_met:
                    ready.append(t)
        return ready

依赖拓扑执行实战展示

设想一个标准的工程重构任务链条：

  • 任务 1：重构鉴权模块代码（t_1）；
  • 任务 2：重写配套单元测试（t_2，依赖 t_1）；
  • 任务 3：编写使用文档（t_3，依赖 t_1）；
  • 任务 4：打包发布镜像（t_4，依赖 t_2 和 t_3）。

在启动执行时：

  1. get_executable_tasks() 仅会筛选出 t_1。其余任务由于前置未就绪，处于安全阻塞状态；
  2. 任务 1 完毕并标记为 completed 后，t_2 与 t_3 同时满足解锁条件，成为并行就绪任务；
  3. 即使系统在此刻意外掉电重启，重新加载目录后，引擎依然能精准辨识出未完成的依赖拓扑，直接实现零损失断点续跑。

总结

持久化任务框架将 Agent 从单会话的内存束缚中彻底解救出来：

  1. 状态物理落地：以独立 JSON 文件为实体，具备天然的抗崩溃韧性；
  2. DAG 依赖闭环：通过 blocked_by 构建执行图，杜绝任务前后顺序颠倒的工程灾难；
  3. 为多智能体铺路：磁盘上的独立任务文件为后续多个 Agent 协同认领任务（Task Claiming）奠定了物理基础。

有了清晰的依赖调度体系后，另一个瓶颈浮出水面：某些任务（如 npm run build 或大型测试套件）耗时长达数分钟。如果主 Agent 一直同步等待，用户界面与推理循环就会卡死。

下一篇我们将探讨非阻塞调度的核心机制：Background Tasks 后台任务与异步轮询通知。