平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“一文详细解析Claude Code的Task System的实现方法原理”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。

目录

• 1. 前言：TodoWrite 的致命局限
• 2. 核心架构
• 3. 数据模型：Task 的字段设计
• 4. TaskStore 核心实现逐层剖析
• 4.1 文件锁 FileLock：极简的跨进程互斥
• 4.2 线（High Water Mark）：防止 ID 复用
• 4.3 CRUD 操作详解
• create：带锁的原子新建
• get / list_all：无锁只读
• update：粒度文件级锁
• delete：级联清理引用
• 4.4 依赖管理：block 的双向维护
• 4.5 认领机制：claim 的原子性保证
• 4.6 统计接口：stats 的快照视图
• 5. Agent Loop：LLM 如何驱动 Task System
• 5.1 工具注册与分发
• 5.2 Agent 核心循环
• 5.3 System Prompt 设计
• 5.4 交互式入口
• 6. 调用链路全景图
• 7. 与 TodoWrite 的对比总结
• 8. 总结

1. 前言：TodoWrite 的致命局限

在这个场景下，我们在学习了Claude Code 的 TodoWrite 模式，但它存在以下缺点：

1. 纯内存、零持久化：数据仅存于进程堆内存，一旦 Agent 进程重启，所有 Todo 全部丢失，无法跨会话保留。
2. 原子替换、无增量操作：只兼容传入完整列表整体替换，无法单独更新、认领或删除某个任务。多个 Agent 同时发更新时，后写覆盖先写（Last-Write-Wins），数据不一致风险极高。
3. 数据模型过于简陋：仅包含 content、status、activeForm 三个字段，没有任务 ID、没有 owner、没有依赖关系。所以无法分配任务、无法建立任务间的阻塞/依赖、无法稳定引用具体任务。
4. 执行模型只兼容单 Agent：按 agentId 或 sessionId 隔离不同的 Todo 列表，Agent 之间互相不可见，从根本上无法实现 Multi-Agent 协作（共享看板、认领、依赖排序等）。

从实现思路看，为便于解决上述痛点，Claude Code 引入了 Task System（任务系统）。它不再是单个 Agent 内存中的临时记事本，而是一个跨进程、跨会话、多 Agent 共享理解这一步时，的持久化任务看板。借助 JSON 文件存储、文件锁同时发控制、显式的任务 ID 与双向依赖管理，Task System 让 Agent 团队能够真正协同工作：认领任务、更新状态、建立阻塞关系，且数据在进程重启后完整保留。

在这个场景下，本文基于从 Claude Code 源码的 Task System 部分提取最小的 Python 实现，同时深入剖析其实现原理。

2. 核心架构

Claude Code 的 Task System 本质是一个基于文件系统的分布式任务看板，核心思想极为轻松：

实际处理时，每个任务 = 一个 JSON 文件，每个操作 = 一次文件 I/O，同时发控制 = 文件锁。

架构图如下所示：

┌───────────────────────────────────────────────┐
│                 Agent Loop                     │
│    LLM 生成 tool_calls → 路由到 TaskStore     │
└───────────────────┬───────────────────────────┘
                    │
         ┌──────────┴──────────┐
         │     TaskStore        │
         │  ┌────────────────┐  │
         │  │ FileLock       │  │  ← 并发控制
         │  │ (O_EXCL|O_CREAT)│  │
         │  └───────┬────────┘  │
         │  ┌───────┴────────┐  │
         │  │ JSON 文件 I/O  │  │  ← 持久化
         │  │ (.json 文件)   │  │
         │  └───────┬────────┘  │
         │  ┌───────┴────────┐  │
         │  │ .highwatermark │  │  ← ID 防复用
         │  └────────────────┘  │
         └──────────────────────┘
                    │
         ┌──────────┴──────────┐
         │  文件系统            │
         │  tasks/<id>/         │
         │  ├── .highwatermark  │
         │  ├── 1.json          │
         │  ├── 2.json          │
         │  └── ...             │
         └─────────────────────┘

实际处理时，这张架构图清晰地展示了从 Agent 调用到最后文件存储的完整链路：上层是 Agent Loop 将 LLM 生成的 tool_calls 路由到 TaskStore；TaskStore 内部依靠文件锁保证同时发安全，借助 JSON 文件实现持久化，并借助 .highwatermark 文件防止 ID 复用；最底层是文件系统中的具体目录结构。正是这种极简的"文件即任务"设计，解决了 TodoWrite 的所有痛点。

3. 数据模型：Task 的字段设计

结合项目来看，Task System 的数据模型是一个 Python dataclass，定义在 task_store.py 第 37-48 行：

# task_store.py  L37-48
@dataclass
class Task:
    id: str                          # 自增数字 ID，全局唯一引用
    subject: str                     # 任务标题（祈使句）
    description: str                 # 任务详细描述
    status: TaskStatus = TaskStatus.PENDING   # pending / in_progress / completed
    owner: Optional[str] = None      # 认领者标识（Agent 名称）
    blocks: list[str] = []           # 本任务阻塞了哪些其他任务（出边）
    blocked_by: list[str] = []       # 本任务被哪些任务阻塞（入边）
    active_form: Optional[str] = None # 任务的展示形式
    metadata: Optional[dict] = None  # 扩展元数据

与 TodoWrite 的 content/status/activeForm 三字段相比，Task 模型新增了以下关键字段：

TaskStatus 是一个三态枚举（task_store.py L31-34）：

PENDING → IN_PROGRESS → COMPLETED

这是任务状态机的全部合法转换路径。

4. TaskStore 核心实现逐层剖析

4.1 文件锁 FileLock：极简的跨进程互斥

文件路径：task_store.py L54-96

落到代码里，FileLock 是整个 Task System 同时发安全的基石。它选用了操作系统原语 O_CREAT | O_EXCL——这是所有 Unix/Linux/Windows 都兼容的原子操作，语义是"当且仅当文件不存在时才新建它"。

# task_store.py  L61-80
class FileLock:
    def __init__(self, target_path: Path, retries: int = 30,
                 min_wait_ms: int = 5, max_wait_ms: int = 100):
        self._lock_path = Path(str(target_path) + ".lock")  # <path>.lock 伴生锁文件
        self._retries = retries
        self._min_wait = min_wait_ms / 1000   # 5ms
        self._max_wait = max_wait_ms / 1000   # 100ms
    def acquire(self) -> bool:
        for i in range(self._retries):
            try:
                fd = os.open(str(self._lock_path), os.O_CREAT | os.O_EXCL)
                os.close(fd)
                self._held = True
                return True
            except FileExistsError:
                wait = min(self._min_wait * (2 ** min(i, 10)), self._max_wait)
                time.sleep(wait)
        return False

设计要点：

1. 伴生锁文件（Companion Lock File）：对文件 1.json 加锁时，实际新建 1.json.lock。这把锁文件和被保护文件是独立的，不存在"文件已存在就无法加锁"的问题——锁文件和业务文件是两条完全独立的路径。
2. 指数退避重试：wait = min(5ms × 2^i, 100ms)，最多重试 30 次。最短等待 5ms，以 2 的指数增长，上限 100ms。30 次重试的总等待时间上限约 3 秒（实际由指数曲线决定），超过后得到 False。
3. 上下文管理器协议：实现了 __enter__ 和 __exit__（L90-96），能够用 with FileLock(path): 语法，确保锁在离开作用域时自动释放——即使发生异常也不会泄漏锁。
4. 释放是幂等的：release() 中用 try/except FileNotFoundError 包裹 os.unlink()（L82-88）。如果锁已经被其他进程释放或手动删除，不会报错。

为什么不用 flock() 或 fcntl？

• flock()/fcntl 是 Unix 特有的，Windows 不完全兼容
• O_EXCL|O_CREAT 是 POSIX 标准，跨平台兼容
• 在这个场景下，锁的生命周期等于文件的存在周期，进程崩溃后操作系统自动清理锁文件（进程退出时文件描述符关闭），不会死锁

4.2 线（High Water Mark）：防止 ID 复用

文件路径：task_store.py L118-142

在这个场景下，Task System 采用自增整数作为任务 ID。但如果任务被删除后，它的 ID 能够复用吗？答案是：不能。 这正是 .highwatermark 的作用。

# task_store.py  L118-142
def _read_hw(self) -> int:
    try:
        return int(self._hw_file.read_text().strip())
    except (FileNotFoundError, ValueError):
        return 0
def _write_hw(self, value: int):
    self._dir.mkdir(parents=True, exist_ok=True)
    self._hw_file.write_text(str(value))
def _max_id_from_files(self) -> int:
    if not self._dir.exists():
        return 0
    max_id = 0
    for f in self._dir.glob("*.json"):
        try:
            tid = int(f.stem)
            if tid > max_id:
                max_id = tid
        except ValueError:
            pass
    return max_id
def _allocate_id(self) -> str:
    return str(max(self._max_id_from_files(), self._read_hw()) + 1)

ID 分配逻辑：

1. _allocate_id() 取"磁盘上实际存在的最大 ID"和"线记录值"两者的较大值，再 +1 作为新 ID
2. 新新建任务时，目录下 1.json、2.json、3.json 依次递增，线在新建时无需更新（因为 _max_id_from_files() 总能找到最大 ID）
3. 关键是删除路径（task_store.py L206-215）：删除任务时，将它的 ID 写入线

# task_store.py  L210-214
try:
    tid = int(task_id)
    if tid > self._read_hw():     # 只有大于当前线的 ID 才更新
        self._write_hw(tid)        # 防止删除操作把线往回拉
except ValueError:
    pass

举个场景说明为什么需线：

在这个场景下，假设有任务 1、2、3、4、5。删除了 5 号任务，此时 _max_id_from_files() 得到 4，所以下一个 ID 会是 5——这就复用了已删除任务的 ID！但如果任务 5 之前被其他任务借助 blocks 或 blocked_by 字段引用过，这些引用就变成了"悬垂引用"（dangling reference）。

有了线：删除任务 5 时，线被设为 5。下一次 _allocate_id() 计算 max(4, 5) + 1 = 6，不会复用 5。虽然实现中没有进一步清理对已删除任务的引用（这是一个已知的简化），但至少新任务不会"冒充"旧任务被误认。

4.3 CRUD 操作详解

create：带锁的原子新建

文件路径：task_store.py L160-172

def create(self, subject: str, description: str,
           active_form: Optional[str] = None) -> Task:
    self._dir.mkdir(parents=True, exist_ok=True)
    with FileLock(self._lock_file):           # 持有全局锁
        task = Task(
            id=self._allocate_id(),           # 在锁内分配 ID
            subject=subject,
            description=description,
            active_form=active_form or subject
        )
        self._save(task)
    return task

关键点：

• 锁的粒度：create 采用全局锁 self._lock_file（即 <data_dir>/tasks/<task_list_id>/.lock），而不是单个任务文件的锁。这是必要的——因为 ID 分配是全局操作，两个同时发的 create 如果采用不同的锁，可能拿到相同的 ID。
• ID 分配在锁内完成：_allocate_id() 在 with FileLock 块内部调用，保证当前进程看到的文件系统状态是"独占快照"。
• _save() 不检查：_save() 直接 write_text() 写入。由于锁保证了同一时刻只有一个进程在执行 create，不会出现写冲突。

get / list_all：无锁只读

# task_store.py  L174-186
def get(self, task_id: str) -> Optional[Task]:
    return self._load(self._task_path(task_id))
def list_all(self) -> list[Task]:
    if not self._dir.exists():
        return []
    tasks = []
    for f in sorted(self._dir.glob("*.json"),
                    key=lambda p: int(p.stem)):
        t = self._load(f)
        if t:
            tasks.append(t)
    return tasks

两个只读方法都不加锁。list_all 按数字 ID 排序得到，每次调用都会重新扫描文件系统——这意味着它总是得到最新的任务列表（最后一致性）。代价是目录扫描，在任务数量不大（Claude Code 的典型场景是几十到几百个任务）时完全能够接受。

update：粒度文件级锁

# task_store.py  L183-197, L199-204
def _update_unsafe(self, task_id: str, **updates) -> Optional[Task]:
    """内部更新（无锁，由调用方保证锁已持有）"""
    task = self._load(self._task_path(task_id))
    if task is None:
        return None
    for k, v in updates.items():
        if hasattr(task, k):
            setattr(task, k, v)
    self._save(task)
    return task
def update(self, task_id: str, **updates) -> Optional[Task]:
    path = self._task_path(task_id)
    if not path.exists():
        return None
    with FileLock(path):                       # 锁在单个 .json 文件上
        return self._update_unsafe(task_id, **updates)

这里展示了两个重要的设计决策：

1. 锁的粒度下沉：create 用全局锁，update 用任务级别的文件锁。这种精细化控制使得：Agent A 在更新任务 1 的同时，Agent B 能够更新任务 2——互不阻塞。只有同时更新同一任务时才会竞争锁。
2. _update_unsafe 内部方法：这是一个约定——以 _unsafe 结尾的方法不持有锁，由调用方保证锁已拿到。这样 claim()、block() 等方法能够在已持有的锁内部复用 _update_unsafe，避免锁嵌套死锁。

delete：级联清理引用

# task_store.py  L206-222
def delete(self, task_id: str) -> bool:
    path = self._task_path(task_id)
    if not path.exists():
        return False
    # 1. 更新线
    try:
        tid = int(task_id)
        if tid > self._read_hw():
            self._write_hw(tid)
    except ValueError:
        pass
    # 2. 删除文件
    path.unlink()
    # 3. 级联清理引用
    for task in self.list_all():
        new_blocks = [b for b in task.blocks if b != task_id]
        new_blocked = [b for b in task.blocked_by if b != task_id]
        if new_blocks != task.blocks or new_blocked != task.blocked_by:
            self.update(task.id, blocks=new_blocks, blocked_by=new_blocked)
    return True

删除操作分三步：

1. 抬高线：确保删除的 ID 不会被复用（见 4.2 节）
2. 物理删除 JSON 文件：path.unlink()
3. 级联清理引用：遍历所有剩余任务，从它们的 blocks 和 blocked_by 字段中移除对已删除任务的引用。这确保依赖图中不会出现悬垂引用

理解这一步时，注意 delete 本身没有加锁——在多 Agent 同时发场景下，这可能导致短暂的"不一致窗口"（任务文件被删除但引用尚未清理完毕）。这是一个有意的简化，因为 Claude Code 中任务删除是低频操作。

4.4 依赖管理：block 的双向维护

文件路径：task_store.py L226-235

def block(self, blocker_id: str, blocked_id: str) -> bool:
    blocker = self.get(blocker_id)
    blocked = self.get(blocked_id)
    if not blocker or not blocked:
        return False
    if blocked_id not in blocker.blocks:
        self._update_unsafe(blocker_id, blocks=blocker.blocks + [blocked_id])
    if blocker_id not in blocked.blocked_by:
        self._update_unsafe(blocked_id, blocked_by=blocked.blocked_by + [blocker_id])
    return True

建立依赖时，同时维护两个方向的引用：

blocker.blocks = [..., blocked_id]       # A 阻塞了 B
blocked.blocked_by = [..., blocker_id]   # B 被 A 阻塞

为什么要双向存储？

从实现思路看，如果只存单向（比如只在 blockee 上存 blocked_by），那么：

• 要列出"A 阻塞了哪些任务"需扫描所有任务的 blocked_by 字段 → O(n)
• 双向存储使得两种查询都是 O(1)：读 blocker.blocks 即可

关于幂等性：注意条件判断 if blocked_id not in blocker.blocks:——重复调用 block(1, 2) 不会重复添加引用。不过这个方法没有加锁，在极端同时发下可能出现竞争。在实际采用中，依赖建立通常由同一个 Agent 在规划阶段完成，并发压力不大。

4.5 认领机制：claim 的原子性保证

文件路径：task_store.py L239-259

def claim(self, task_id: str, agent: str) -> dict:
    path = self._task_path(task_id)
    if not path.exists():
        return {"success": False, "reason": "not_found"}
    with FileLock(path):                        # 持有任务文件锁
        task = self._load(path)                 # 在锁内重新读取（防止 TOCTOU）
        if task is None:
            return {"success": False, "reason": "not_found"}
        if task.owner and task.owner != agent:
            return {"success": False, "reason": "already_claimed",
                    "owner": task.owner}
        if task.status == "completed":
            return {"success": False, "reason": "already_completed"}
        all_tasks = self.list_all()
        unresolved = {t.id for t in all_tasks if t.status != "completed"}
        active_blockers = [b for b in task.blocked_by if b in unresolved]
        if active_blockers:
            return {"success": False, "reason": "blocked",
                    "blocked_by": active_blockers}
        self._update_unsafe(task_id, owner=agent)
        return {"success": True, "task_id": task_id}

实际处理时，claim 是整个 Task System 中同时发安全性要求最高的操作——两个 Agent 同时认领同一任务时，必须保证只有一个成功。实现要点：

1. 先 check 再加锁：函数入口处先更快检查文件是否存在（无锁），减少加锁开销
2. 锁内重新读取（TOCTOU 防护）：持有锁后重新 _load(path) 读取任务数据。这防止了 TOCTOU（Time-of-check Time-of-use）竞态——入口处的检查可能已过时
3. 四重校验：在锁内依次验证：
   • 任务存在
   • 未被他人认领（或已被自己认领，允许重新认领）
   • 任务未完成
   • 依赖检查：遍历所有未完成任务，检查 blocked_by 中的阻塞者是否均已完成。只有所有阻塞者都完成后，认领才成功
4. 得到结构化结果：得到 {"success": bool, "reason": str, ...} 而不是轻松布尔值，让 Agent 能向 LLM 报告为什么理解这一步时，认领失败——这和 TaskStore 本身无关，是给 LLM 提供更好的上下文

得到值设计：

• success=True → 认领成功，Agent 能够开始工作
• reason="blocked" → 附带 blocked_by 列表，LLM 知道需等待哪些任务
• reason="already_claimed" → 附带 owner，LLM 知道谁抢走了任务

4.6 统计接口：stats 的快照视图

# task_store.py  L263-277
def stats(self) -> dict:
    tasks = self.list_all()
    return {
        "total": len(tasks),
        "pending": sum(1 for t in tasks if t.status == TaskStatus.PENDING),
        "in_progress": sum(1 for t in tasks if t.status == TaskStatus.IN_PROGRESS),
        "completed": sum(1 for t in tasks if t.status == TaskStatus.COMPLETED),
        "available": sum(1 for t in tasks
                         if t.status == TaskStatus.PENDING
                         and not t.owner
                         and not t.blocked_by),
    }

available 字段专门为认领场景设计——它是所有"能够立即认领"的任务计数（待处理 + 无人认领 + 无阻塞依赖）。Agent 能够据此更快判断是否有活可干。

5. Agent Loop：LLM 如何驱动 Task System

文件路径：agent_loop.py（完整 286 行）

理解这一步时，Agent Loop 是 Task System 的上层消费者。它将 TaskStore 的五个核心操作包装为 LLM 可调用的 Function Calling 工具，形成完整的"LLM 规划 → 工具执行 → 结果反馈"闭环。

5.1 工具注册与分发

工具定义在 agent_loop.py L53-122，结构如下所示：

工具名称         对应 TaskStore 方法      核心参数
─────────────────────────────────────────────────────
task_create      create()                subject, description
task_list        list_all() + stats()    无
task_update      update()                task_id, status
task_claim       claim()                 task_id
task_block       block()                 blocker_id, blocked_id

实际处理时，每个工具的实现是一个独立的 handler 函数（L128-185），它们负责：

1. 参数提取：从 JSON 反序列化的 args 字典中提取参数
2. 调用 TaskStore：执行底层操作
3. 格式化输出：将 TaskStore 的得到（Task 对象、dict、bool）转为 LLM 能理解的自然语言字符串

以 tool_task_claim 为例（agent_loop.py L161-170）：

def tool_task_claim(args: dict) -> str:
    result = store.claim(args["task_id"], "agent-main")
    if result["success"]:
        return f"✅ 已认领任务 #{args['task_id']}"
    reason = result.get("reason", "unknown")
    if reason == "blocked":
        return f"❌ 任务 #{args['task_id']} 被阻塞: 等待 #{','.join(result.get('blocked_by', []))} 完成"
    if reason == "already_claimed":
        return f"❌ 任务 #{args['task_id']} 已被 {result.get('owner', '?')} 认领"
    return f"❌ 认领失败: {reason}"

设计精髓：handler 不只是得到成功/失败，而是将 TaskStore 的结构化得到翻译为带 emoji 的自然语言。LLM 看到 "❌ 任务 #3 被阻塞: 等待 #1, #2 完成" 后，能直观理解当前状态同时决定下一步行动——这正是 AI Agent 中"工具输出可解释性"的关键实践。

工具分发借助字典 TOOL_HANDLERS（L179-185）完成：

TOOL_HANDLERS = {
    "task_create": tool_task_create,
    "task_list": tool_task_list,
    "task_update": tool_task_update,
    "task_claim": tool_task_claim,
    "task_block": tool_task_block,
}

5.2 Agent 核心循环

文件路径：agent_loop.py L212-249

def agent_loop(messages: list, max_iterations: int = 15) -> str:
    """Agent 循环：LLM ↔ Tool Calls ↔ TaskStore"""
    for iteration in range(max_iterations):
        response = [email protected](
            model=MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = response.choices[0].message
        messages.append(msg)
        if not msg.tool_calls:
            return msg.content or ""
        for tc in msg.tool_calls:
            name = tc.function.name
            args = json.loads(tc.function.arguments)
            handler = TOOL_HANDLERS.get(name)
            if handler:
                result = handler(args)
            else:
                result = f"未知工具: {name}"
            # 打印执行日志
            arg_str = json.dumps(args, ensure_ascii=False)
            print(f"  ? {name}({arg_str})")
            print(f"     → {result}")
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result,
            })
    return "⚠️ Agent 达到最大迭代次数"

流程如下所示：

User Prompt → messages
     │
     ▼
┌──────────────────────┐
│  LLM 推理            │ ← [email protected](messages, tools=TOOLS)
│  返回 tool_calls 或  │
│  纯文本回复          │
└───────┬──────┬───────┘
        │      │
  tool_calls   text → 结束，返回 回复
        │
        ▼
┌──────────────────────┐
│  遍历 tool_calls     │
│  1. 解析函数名+参数  │
│  2. TOOL_HANDLERS 分发│
│  3. 调用 TaskStore   │
│  4. 结果格式化       │
│  5. 追加到 messages  │
└───────┬──────────────┘
        │
        ▼
   下一轮 LLM 推理 ← (带工具结果)

关键设计点：

1. 最大迭代保护：max_iterations=15，防止 LLM 陷入无限的"调工具→看结果→再调工具"循环
2. 完整的对话历史：每次工具调用的结果都追加到 messages 列表，LLM 在下一轮能看到完整的上下文——包括之前新建了哪些任务、哪些已认领、哪些在等待
3. 单轮多工具调用：for tc in msg.tool_calls 循环兼容 LLM 在一次响应中调用多个工具（比如同时新建 3 个任务）
4. 工具结果即文本：没有复杂的结构化得到，每个工具得到一个自然语言字符串，这是最通用、对 LLM 最友好的格式

5.3 System Prompt 设计

文件路径：agent_loop.py L191-208

SYSTEM_PROMPT = """你是一个项目管理助手，拥有一个持久化的任务看板（Task System）。
## 工作规则
1. 收到复杂需求时，先用 task_create 拆分为 3-5 个步骤
2. 拆分后立即用 task_list 确认
3. 执行时先用 task_claim 认领
4. 完成一步立即 task_update(status="completed")
5. 需要先后顺序的用 task_block
请用中文回复，简洁直接。"""

System Prompt 给 LLM 的是一条硬编码的工作流：Create → List → Claim → Update → Complete。这不是一种"建议"，而是 Agent 理解 Task System 采用方式的唯一入口——LLM 根据这些规则决定何时调用哪个工具。

5.4 交互式入口

文件路径：agent_loop.py L256-287

if __name__ == "__main__":
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    while True:
        query = input("n? 用户 >> ").strip()
        if query.lower() == "list":
            print(tool_task_list({}))    # 快捷命令：直接调用工具
            continue
        history.append({"role": "user", "content": query})
        final = agent_loop(history)
        if final:
            print(f"n? 助手: {final}")

交互式入口提供了两个便利：

• list 快捷命令：绕过 LLM 直接调用 task_list，省去一次 API 调用
• 持续对话：history 列表在循环外部维护，兼容多轮对话，Agent 记住之前的上下文

6. 调用链路全景图

从实现思路看，将 agent_loop.py 和 task_store.py 串起来，一条典型的任务执行流程如下所示：

用户输入："帮我实现用户注册功能"
         │
         ▼
agent_loop()                    [agent_loop.py L212]
  │
  ├─► LLM 推理 → tool_calls: [task_create, task_create, task_create]
  │
  ├─► tool_task_create("设计数据模型", "...")
  │     └─► store.create()      [task_store.py L160]
  │           ├─► FileLock(.lock).acquire()   [L70-80]
  │           ├─► _allocate_id() → "1"        [L141-142]
  │           ├─► Task(id="1", ...)
  │           ├─► _save(task)                  [L146-148]
  │           │     └─► 1.json → write_text(json.dumps(...))
  │           └─► FileLock.release()           [L82-88]
  │
  ├─► LLM 推理 → tool_calls: [task_block(1, 2), task_block(2, 3)]
  │
  ├─► tool_task_block("1", "2")
  │     └─► store.block("1", "2")              [task_store.py L226]
  │           ├─► blocker: 1.blocks += ["2"]
  │           └─► blocked: 2.blocked_by += ["1"]
  │
  ├─► LLM 推理 → tool_calls: [task_claim("1")]
  │
  ├─► tool_task_claim("1")
  │     └─► store.claim("1", "agent-main")     [task_store.py L239]
  │           ├─► FileLock(1.json).acquire()
  │           ├─► _load(1.json) → Task
  │           ├─► 检查 owner=None ✓
  │           ├─► 检查 blocked_by=[] ✓
  │           ├─► _update_unsafe("1", owner="agent-main")
  │           └─► 返回 {"success": True}
  │
  ├─► LLM 推理 → tool_calls: [task_update("1", "completed")]
  │
  ├─► tool_task_update("1", "completed")
  │     └─► store.update("1", status=COMPLETED) [task_store.py L199]
  │           ├─► FileLock(1.json).acquire()
  │           ├─► _update_unsafe("1", status=COMPLETED)
  │           └─► 返回 Task
  │
  └─► LLM 推理 → 无 tool_calls → 返回最终文本

实际处理时，这条路径贯穿了两个文件的全部核心组件：LLM 对话管理 → 工具分发 → 文件锁 → JSON 持久化 → 依赖检查 → 状态流转。

7. 与 TodoWrite 的对比总结

Task System 的代价与局限：

1. 没有并发读取的隔离：get() 和 list_all() 不加锁，读到的是即时快照（最后一致性）。在极端情况下，可能读到"正在被修改"的任务文件（写入未完成时文件可能不完整）。这在 Claude Code 的实际采用中不是问题——因为任务文件通常很小（<1KB），写入是原子的（对于大多数文件系统来说，小文件的 write_text 操作在页缓存层面是原子的）。
2. 依赖环检测缺失：block() 方法不会检查是否形成环（A→B→C→A），如果 LLM 建立了循环依赖，所有任务将永久无法认领。这是当前实现的一个已知局限。
3. 无任务优先级/排序：任务只按 ID 排序，没有显式的优先级字段。LLM 需借助 System Prompt 中的"工作规则"隐式理解执行顺序。

8. 总结

Claude Code 的 Task System 用极简的工程手段解决了 TodoWrite 的四大痛点：

• 文件系统替代内存 → 解决持久化
• FileLock 替代无锁 → 解决并发安全
• Task dataclass 替代三字段 → 解决数据模型简陋
• 共享文件系统替代 session 隔离 → 解决多 Agent 协作

它的设计哲学是：不要引入数据库，不要引入消息队列，不要引入分布式协调服务结合项目来看，。就用文件系统 + 文件锁就能构建一个实用的、兼容多 Agent 协作的持久化任务看板。这种"工程上的最小主义"正是 Claude Code 整体架构的缩影——用最少的依赖解决最核心的问题。

理解这一步时，当然，当任务量增长到数千级别、对查询性能有要求、需事务性保证时，还是要升级到 SQLite 或更重的存储。但 Task System 证明了：在 Agent 协作场景下，文件系统作为存储后端是一个充分且优雅的方案。

实际处理时，以上就是一文详解Claude Code的Task System的实现原理的详细内容，更多关于Claude Code的Task System实现原理的资料请关注脚本之家其它相关文章！