s11 的 agent 是无状态的：每次对话结束后，完成了哪些工作、还有哪些待做，全部丢失。
s12 引入了 文件持久化的任务图：每个任务存为 .tasks/{id}.json，支持 blockedBy 依赖链，agent 可以创建、认领、完成任务，跨会话保持进度。
from dataclasses import dataclass, asdictTASKS_DIR = .tasks/ 持久化目录@dataclass Task — 6 字段数据类create_task / save_task / load_task / list_tasks / get_task — CRUD 层can_start() — 依赖检查claim_task() / complete_task() — 状态转移agent_loop 简化（移除了 s11 的完整错误恢复，聚焦任务系统）| 代码行 | 语法注解 |
|---|---|
| from dataclasses import dataclass, asdict | dataclass：装饰器，自动生成 __init__、__repr__、__eq__；asdict：将实例转换为 dict（用于 JSON 序列化） |
| @dataclass | 装饰器语法；等价于在类定义后执行 Task = dataclass(Task)；不需要手写 __init__ |
| class Task: | 普通类定义；装饰器负责生成样板代码 |
| id: str | 字段注解（PEP 526）；dataclass 用注解顺序生成 __init__ 参数；类型为 str |
| subject: str | 任务标题（简短描述）；用于 list_tasks() 的显示 |
| description: str | 任务详细描述；可为空字符串 |
| status: str # pending | in_progress | completed | 行内注释说明允许的值；Python 未强制 Literal 类型，这里用字符串约定 |
| owner: str | None # Agent name (multi-agent scenarios) | str | None：联合类型（Python 3.10+）；认领前为 None，认领后设为 agent 名 |
| blockedBy: list[str] # Dependency task IDs | list[str]：泛型类型注解（Python 3.9+）；存储前置任务 ID 列表；空列表 = 无依赖 |
手写 __init__ 相当于：
def __init__(self, id, subject, description,
status, owner, blockedBy):
self.id = id
self.subject = subject
...
@dataclass 自动生成上述代码，减少重复。
| 代码行 | 语法注解 |
|---|---|
| TASKS_DIR = WORKDIR / ".tasks" | Path / str：pathlib 运算符重载，拼接路径；等价于 Path(WORKDIR, ".tasks") |
| TASKS_DIR.mkdir(exist_ok=True) | 模块级副作用：导入时立即创建目录；exist_ok=True = 目录已存在不报错；注意无 parents=True（WORKDIR 必须已存在） |
| def _task_path(task_id: str) -> Path: | 私有辅助函数（_ 前缀约定）；集中管理文件路径格式，便于统一修改 |
| return TASKS_DIR / f"{task_id}.json" | f-string 内嵌变量构造文件名；如 .tasks/task_1234_0001.json |
| 代码行 | 语法注解 |
|---|---|
| def create_task(subject: str, description: str = "", | 默认参数 description=""：调用时可省略；Python 默认参数在函数定义时求值 |
| blockedBy: list[str] | None = None) -> Task: | blockedBy=None（非 []）：避免 Python 可变默认参数陷阱（多次调用共享同一列表） |
| task = Task( | dataclass __init__ 调用；所有字段必须提供（无默认值的字段） |
| id=f"task_{int(time.time())}_{random.randint(0, 9999):04d}", | int(time.time())：Unix 时间戳（秒）；random.randint(0,9999)：4 位随机数；:04d：格式化为至少 4 位，不足补零；组合避免 ID 碰撞 |
| status="pending", | 所有新任务初始状态为 "pending"；由状态机控制后续转移 |
| owner=None, | 未认领；claim_task() 调用后赋值 |
| blockedBy=blockedBy or [], | or []：若 blockedBy 为 None（或其他 falsy 值），使用新建空列表；正确处理可变默认参数 |
| ) | Task 实例构造完毕 |
| save_task(task) | 立即持久化到文件；create 操作原子性（Python 进程内） |
| return task | 返回 Task 实例，让调用方获取 id |
| 代码行 | 语法注解 |
|---|---|
| def save_task(task: Task): | 无返回值（隐式 return None）；副作用：写文件 |
| _task_path(task.id).write_text(json.dumps(asdict(task), indent=2)) | asdict(task)：dataclass → dict；json.dumps(..., indent=2)：格式化 JSON；Path.write_text()：原子写入（在大多数 OS 上） |
| def load_task(task_id: str) -> Task: | 返回 Task 实例；若文件不存在抛 FileNotFoundError（调用方处理） |
| return Task(**json.loads(_task_path(task_id).read_text())) | 链式调用：read_text() → json.loads() → **dict 解包 → Task() 构造；一行完成反序列化 |
| 代码行 | 语法注解 |
|---|---|
| def list_tasks() -> list[Task]: | 返回所有任务列表，按文件名排序（时间戳前缀保证创建时间顺序） |
| return [Task(**json.loads(p.read_text())) | 列表推导式（list comprehension）：对每个文件 p 执行加载 |
| for p in sorted(TASKS_DIR.glob("task_*.json"))] | Path.glob("task_*.json")：仅匹配 task_ 前缀的 JSON；sorted()：按 Path 的字典序排序（等价于时间序） |
| 代码行 | 语法注解 |
|---|---|
| def can_start(task_id: str) -> bool: | 返回布尔值；用于 claim_task() 调用前的守卫检查 |
| """Check if all blockedBy dependencies are completed. | docstring 第一行：单句总结 |
| Missing dependencies are treated as blocked.""" | 关键设计决策文档：依赖 ID 不存在时视为阻塞（保守策略） |
| task = load_task(task_id) | 从文件加载最新状态（不使用缓存） |
| for dep_id in task.blockedBy: | 遍历依赖 ID 列表；若列表为空，循环不执行，直接 return True |
| if not _task_path(dep_id).exists(): | 先检查文件存在性：依赖文件不存在 = 依赖 ID 无效 = 视为阻塞 |
| return False | 早返回（fail-fast）：发现任一阻塞即返回 False，无需检查后续依赖 |
| if load_task(dep_id).status != "completed": | 加载依赖任务检查状态；"pending" 或 "in_progress" 均视为阻塞 |
| return False | 依赖未完成 → 当前任务不可开始 |
| return True | 所有依赖均已完成（或无依赖）→ 可以开始 |
| 代码行 | 语法注解 |
|---|---|
| def claim_task(task_id: str, owner: str = "agent") -> str: | 返回字符串（成功消息或错误描述）；owner 默认 "agent"，多代理时传入唯一名称 |
| task = load_task(task_id) | 从文件读取最新状态（防止竞态条件：另一代理已认领） |
| if task.status != "pending": | 状态守卫：只有 pending 状态可被认领；已 in_progress 或 completed 的任务不可重复认领 |
| return f"Task {task_id} is {task.status}, cannot claim" | 返回描述性错误字符串（不抛异常）；调用方（tool runner）将此作为工具结果返回给 LLM |
| if not can_start(task_id): | 依赖守卫：调用 can_start() 检查所有 blockedBy |
| deps = [d for d in task.blockedBy | 列表推导式：筛选出尚未完成的依赖 ID |
| if not _task_path(d).exists() or load_task(d).status != "completed"] | 条件过滤：文件不存在或状态非 completed |
| return f"Blocked by: {deps}" | 告知 LLM 被哪些任务阻塞，LLM 可据此决定先完成前置任务 |
| task.owner = owner | 直接修改 dataclass 实例的属性（dataclass 默认可变） |
| task.status = "in_progress" | 状态转移：pending → in_progress |
| save_task(task) | 持久化修改后的状态到文件 |
| print(f" \033[36m[claim] {task.subject} → in_progress (owner: {owner})\033[0m") | \033[36m = 青色；调试输出，不影响返回值 |
| return f"Claimed {task.id} ({task.subject})" | 成功消息；LLM 收到后知道任务已认领，可开始执行 |
| 代码行 | 语法注解 |
|---|---|
| def complete_task(task_id: str) -> str: | 返回字符串；包含完成消息和解锁的下游任务列表 |
| task = load_task(task_id) | 从文件加载最新状态 |
| if task.status != "in_progress": | 只有 in_progress 的任务可完成；防止直接从 pending 跳到 completed |
| return f"Task {task_id} is {task.status}, cannot complete" | 错误消息；LLM 需先 claim_task 再 complete_task |
| task.status = "completed" | 状态转移：in_progress → completed |
| save_task(task) | 先保存完成状态，再查询下游（确保下游查询到的是最新状态） |
| unblocked = [t.subject for t in list_tasks() | 列表推导：遍历所有任务，筛选出刚被解锁的任务 |
| if t.status == "pending" and t.blockedBy and can_start(t.id)] | 条件：仍是 pending（未被认领）且 有依赖（t.blockedBy 非空）且 现在可以开始 |
| print(f" \033[32m[complete] {task.subject} ✓\033[0m") | \033[32m = 绿色；✓ Unicode 对勾 |
| msg = f"Completed {task.id} ({task.subject})" | 基础完成消息 |
| if unblocked: | 有任务被解锁时追加信息；LLM 可据此主动认领下游任务 |
| msg += f"\nUnblocked: {', '.join(unblocked)}" | +=：字符串原地拼接（字符串不可变，实际创建新字符串） |
| print(f" \033[33m[unblocked] {', '.join(unblocked)}\033[0m") | \033[33m = 黄色；提示有任务解锁 |
| return msg | 返回完整消息字符串（包含解锁信息） |
save_task(task) 必须在 list_tasks() 之前执行，因为 can_start() 通过读文件检查依赖状态。若先查后保存，当前任务仍显示为 in_progress，下游任务不会被标记为可启动。
| 代码行 | 语法注解 |
|---|---|
| def run_create_task(subject: str, description: str = "", | 包装 create_task()；参数签名与工具定义中的 input_schema 对应 |
| blockedBy: list[str] | None = None) -> str: | 返回字符串（tool result）；LLM 通过字符串了解操作结果 |
| task = create_task(subject, description, blockedBy) | 调用业务逻辑层；分离 tool 协议层和业务层 |
| deps = f" (blockedBy: {', '.join(blockedBy)})" if blockedBy else "" | 三元条件表达式：value_if_true if condition else value_if_false |
| return f"Created {task.id}: {task.subject}{deps}" | 返回包含新任务 ID 的字符串；LLM 后续操作需要此 ID |
| def run_list_tasks() -> str: | 无参数；对应工具定义 input_schema: {properties: {}} |
| icon = {"pending": "○", "in_progress": "●", | 字典查找替代 if-elif 链；Unicode 圆圈表示状态 |
| "completed": "✓"}.get(t.status, "?") | .get(key, default)：未知状态显示 "?"（防御性编程） |
| deps = f" (blockedBy: {', '.join(t.blockedBy)})" if t.blockedBy else "" | 条件格式化：有依赖才显示 blockedBy 信息 |
| owner = f" [{t.owner}]" if t.owner else "" | 条件格式化：有 owner 才显示 |
| def run_get_task(task_id: str) -> str: | 包装 get_task()；增加 try/except 处理文件不存在 |
| except FileNotFoundError: | 捕获特定异常类型；比 except Exception 更精确 |
| def run_claim_task(task_id: str) -> str: | 包装 claim_task()，固定 owner="agent"；工具层不暴露 owner 参数 |
| def run_complete_task(task_id: str) -> str: | 最薄包装层；直接转发给业务逻辑 |
业务函数（claim_task）返回字符串，而 TOOL_HANDLERS 的 value 必须是接受 **block.input 的函数。
Runner 层将业务函数的签名适配为 agent_loop 期望的接口，同时可以在此添加：格式化输出、错误包装、参数验证、调试打印。
| 代码行 | 语法注解 |
|---|---|
| {"name": "create_task", | 工具名 = TOOL_HANDLERS 字典的键 = LLM 在 tool_use 中引用的名称 |
| "description": "Create a new task with optional blockedBy dependencies.", | 描述供 LLM 理解工具功能；应说明可选参数行为 |
| "input_schema": {"type": "object", | JSON Schema 格式；Anthropic API 用此描述参数结构 |
| "properties": { | 参数字段定义 |
| "subject": {"type": "string"}, | 必填参数（在 required 列表中） |
| "description": {"type": "string"}, | 可选参数（不在 required 中）；LLM 可省略 |
| "blockedBy": {"type": "array", | JSON Schema 数组类型 |
| "items": {"type": "string"}}}, | items：数组元素类型约束；每个元素必须是字符串（task ID） |
| "required": ["subject"]}}, | 只有 subject 必填；description 和 blockedBy 可选 |
| {"name": "list_tasks", | 无参数工具 |
| "input_schema": {"type": "object", "properties": {}, | properties: {}：空属性字典 = 无参数 |
| "required": []}}, | required: []：空必填列表；LLM 调用时不需传入任何参数 |