s12：任务系统 — 大目标拆成有依赖的小任务
s12：任务系统 — 大目标拆成有依赖的小任务
核心改进点
s12 的任务系统相对 s05 的 TodoWrite 有几个关键升级：
-
任务持久化为文件，而非内存变量
s05 的
TodoWrite存在进程内存里，会话结束就没了。s12 每个任务是一个.tasks/{id}.json文件，重启 Agent、换机器都能恢复进度。 -
任务之间通过
blockedBy形成依赖图can_start检查所有上游依赖是否全部completed，缺一个就阻塞。 -
两动作、三状态的状态机
创建 → pending ──(claim)──→ in_progress ──(complete)──→ completed ↑ │ │ │ blockedBy 未满足 │ 依赖满足后 │ 解锁下游 └────────────────────────┘ ←────────────────────┘
- `claim_task`：锁任务 + 设置 `owner`，防止多 Agent 重复认领
- `complete_task`：标记完成 + 扫描下游，自动报告"哪些任务刚被解锁"
4. **五个新工具覆盖完整生命周期**
`create_task`(建)、`list_tasks`(览)、`get_task`(读详情)、`claim_task`(认领)、`complete_task`(完成)——从创建到完结形成闭环。
5. **与 `TodoWrite` 是两类系统**
`TodoWrite` 是"当前会话的临时清单"，`Task` 是"可恢复的持久化任务图"。CC 源码中两者同时存在，按场景切换。
> **`blockedBy` 的本质是 DAG（有向无环图）**：任务依赖必须无环，否则会出现"互相锁死"的死锁。Harness 没有显式检测环，而是依靠"上游不完成就阻塞"的自然约束。
| 维度 | `TodoWrite`（s05） | `Task` 系统（s12） |
|---|---|---|
| 存储 | 进程内存 | `.tasks/{id}.json` 文件 |
| 生命周期 | **会话结束**即消失 | 重启/换机可恢复 |
| 依赖 | 无 | `blockedBy` 依赖图 |
| 多 Agent | 不适用 | `owner` 防重复认领 |
| 适用场景 | 临时清单 | 持久化任务规划 |
---
## Task 数据结构
```python
# Python 的轻量数据类装饰器
@dataclass
class Task:
id: str # 唯一标识，task_<timestamp>_<random>
subject: str # 简短标题
description: str # 自由格式描述
status: str # pending | in_progress | completed
owner: str | None # 谁在做（多 agent 场景防重复认领）
blockedBy: list[str] # 上游依赖的任务 ID 列表
持久化读写
- 存储路径：
.tasks/task_<timestamp>_<random>.json - 三个函数构成最小 CRUD
asdict+json.dumps和json.loads+**展开的两步走，实现了"Python 对象 ⇄ JSON 文件"的零摩擦互转。
TASKS_DIR = WORKDIR / ".tasks"
TASKS_DIR.mkdir(exist_ok=True)
# 将逻辑 ID 映射到文件系统路径
def _task_path(task_id: str) -> Path:
return TASKS_DIR / f"{task_id}.json"
def save_task(task: Task):
# asdict(task) 将 dataclass 转为 dict（如 {"id":"task_1","status":"pending",...}）
# json.dumps(..., indent=2) 格式化为可读 JSON
_task_path(task.id).write_text(json.dumps(asdict(task), indent=2))
def load_task(task_id: str) -> Task:
# **json.loads(...) 将 dict 解包为 Task 构造函数的参数
return Task(**json.loads(_task_path(task_id).read_text()))
# 按文件名排序，返回完整的 Task 对象列表
def list_tasks() -> list[Task]:
# glob("task_*.json")：只扫描任务文件，排除其他 JSON
return [Task(**json.loads(p.read_text()))
for p in sorted(TASKS_DIR.glob("task_*.json"))]
任务创建
每个任务是一个 JSON 文件，存于 .tasks/ 目录。
def create_task(subject: str, description: str = "",
blockedBy: list[str] | None = None) -> Task:
task = Task(
# ID 生成：timestamp + random hex
id=f"task_{int(time.time())}_{random.randint(0, 9999):04d}",
subject=subject,
description=description,
status="pending",
owner=None,
blockedBy=blockedBy or [],
)
# 创建后立即写入磁盘，避免内存丢失
save_task(task)
return task
依赖检查 can_start
一个任务能否开始，取决于它的所有上游依赖是否都已完成。
- 依赖任务的文件不存在 →
False（保守策略：未知=阻塞） - 如果依赖状态不是
completed→False
def can_start(task_id: str) -> bool:
task = load_task(task_id)
# 遍历依赖 blockedBy 列表中的依赖 ID
for dep_id in task.blockedBy:
# 依赖任务的文件不存在
if not _task_path(dep_id).exists():
return False
# 依赖存在但状态不是 completed
if load_task(dep_id).status != "completed":
return False
# 所有依赖都 completed 或 blockedBy 为空 → 可以开始
return True
claim_task — 认领
状态守卫（Guard Clause）
- 只有
pending的任务可以认领 blockedBy必须全部completed
认领后
pending → in_progress- 设置
owner标记所有权（多 Agent 场景下防止重复认领）
def claim_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
# 状态守卫：只有 pending 的任务可认领
if task.status != "pending":
return f"Task {task_id} is {task.status}, cannot claim"
# 依赖守卫：blockedBy 必须全部 completed
if not can_start(task_id):
# 列出具体哪些依赖还没完成，方便 agent 知道该等什么
deps = [d for d in task.blockedBy
if not _task_path(d).exists() or load_task(d).status != "completed"]
return f"Blocked by: {deps}"
# 通过两道守卫 → 认领
task.owner = owner
# 状态转换
task.status = "in_progress"
save_task(task)
return f"Claimed {task.id} ({task.subject})"
complete_task — 任务完成
任务做完后，设为 completed。同时扫描所有其他任务，找出刚刚被解锁的下游任务，需满足以下条件：
- 状态为
pending - 有
blockedBy（有依赖） - 现在
can_start()
def complete_task(task_id: str) -> str:
task = load_task(task_id)
# 只有 `in_progress` 的任务可以完成
if task.status != "in_progress":
return f"Task {task_id} is {task.status}, cannot complete"
# 修改任务状态
task.status = "completed"
save_task(task)
# 找出被解锁的下游任务
unblocked = [t.subject for t in list_tasks()
if t.status == "pending" and t.blockedBy and can_start(t.id)]
msg = f"Completed {task.id} ({task.subject})"
if unblocked:
msg += f"\nUnblocked: {', '.join(unblocked)}"
return msg
工具注册
| 工具 | 动作 | 说明 |
|---|---|---|
create_task |
建 | 让 LLM 分解复杂请求为子任务，可带 blockedBy |
list_tasks |
览 | 让 LLM 了解全局进度，只显示一行摘要 |
get_task |
读 | 查看特定任务的完整信息，返回完整的任务 JSON，包括 description 和依赖细节 |
claim_task |
认领 | 声明任务所有权 |
complete_task |
完成 | 推进任务图，获得下游提示 |
TOOLS = [
# ... 基础工具(bash, read_file, write_file) ...
{"name": "create_task",
"description": "Create a new task with optional blockedBy dependencies.",
"input_schema": {"type": "object",
"properties": {
"subject": {"type": "string"},
"description": {"type": "string"},
"blockedBy": {"type": "array",
"items": {"type": "string"}}},
"required": ["subject"]}},
...
]
Agent Loop 中任务流程
用户: "实现一个 Web 服务器，支持路由和中间件"
LLM 分析:
1. create_task("设计路由系统") → task_A
2. create_task("实现中间件管道") → task_B
3. create_task("编写服务器入口") → task_C (blockedBy: [task_A, task_B])
4. create_task("编写测试") → task_D (blockedBy: [task_C])
LLM 执行:
→ claim_task(task_A) # 无依赖，可开始
→ [bash: 创建路由代码]
→ complete_task(task_A) # 返回: "Unblocked: (无)" (task_C 仍需 task_B)
→ claim_task(task_B) # 无依赖
→ [bash: 创建中间件代码]
→ complete_task(task_B) # 返回: "Unblocked: 编写服务器入口" ← task_C 解锁!
→ claim_task(task_C) # 依赖已满足
→ ...
本模块知识点
- 任务持久化为
.tasks/{id}.json文件，重启 / 换机可恢复（对比 s05 内存TodoWrite）。 - 依赖图：
blockedBy形成 DAG，can_start检查上游全completed。 - 状态机：
pending → in_progress → completed，两动作claim/complete。 - 多 Agent 防重复认领：认领时设
owner。
| 工具 | 动作 | 说明 |
|---|---|---|
create_task |
建 | 分解复杂请求为子任务，可带 blockedBy |
list_tasks |
览 | 全局进度一行摘要 |
get_task |
读 | 单任务完整 JSON（含依赖） |
claim_task |
认领 | pending→in_progress，设 owner 防重复认领 |
complete_task |
完成 | 标记完成并扫描解锁下游 |
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 茯茶养生人的博客！
