写在前面

  还记得我们在第六章subagent末尾的提到的有向无环图吗？那时我们幻想让多个agent去分解一个大任务，那本章就基于这一问题的解决方案task_system。

  在本章，你会了解到：任务如何做持久化操作、A 做完了 B 才能开始，这层关系怎么表达，谁来检查、如何构建这种依赖关系、如何维护生命周期等等这些操作。原项目完整代码如下：
https://github.com/shareAI-lab/learn-claude-code/blob/main/s12_task_system/code.py

  本章，我们的任务是：
  1，从0开始构建整个任务分发系统。
  2，跑几个测试。

---

一、从0开始构建任务分发系统。

  在AI浪潮来之前，那时传统的前后段系统也有甚至可以说更加成熟的任务分发系统。用过ruoyi这个框架的小伙伴可能都注意到过，在他的后台管理平台中，我们可以通过可视化方法的对模块轻松管理。比如当我们下达一个任务时候，Quarts框架会根据表达式规则通过调用目标字符串来找到具体要执行的代码。比如说：你在页面上配置时输入了 ryTask.ryNoParams，系统就知道要去调用名为 ryTask 的 Spring Bean 里的 ryNoParams 方法。

  这是一种非常硬绑定的方法，好处是不会出错。随着现在AI agent浪潮到来，一个AI系统交给Function calling去匹配接口，本专栏第一章就出现的大模型的工具函数，就是这种操作。然而AI的输出是概率，导致这种方法是一种软绑定，坏处是有可能会出错（不过现在模型参数量上来后，出错的可能性也很小了），这也让AI系统显得比传统系统更加“智能”。

  本章讲的是给AI系统引入的任务分发的能力，如果这种逻辑放在上面提到的传统系统中去实现，那编码量可想有多么巨大，并且效果也许甚微。而正是因为AI系统的软绑定，才让这种能力显得更加智慧。比方说：当用户交付给agent一个任务时，这个用户不可能每次对话都会说要分解任务（或者其他能力），而这个决策（要不要在这个任务上用这个能力）是交给agent自己去处理的。当用户交给agent一个短任务，即便有任务分发能力，它也可能不会使用，而是很轻松的就完成了，不会绕弯路。

  这样说大概很抽象，我们直接进入代码去看。

  首先，整个任务系统的最小运行单元一定是任务对吧，我们来定义一个这样的task类。先用 @dataclass 把任务的结构固定下来，字段类型写死，模型乱填会报错。

@dataclass
class Task:
    id: str                 # 系统生成，唯一标识
    subject: str            # 任务标题
    description: str        # 任务详情
    status: str             # pending / in_progress / completed
    owner: str | None       # 谁认领的（agent 名称）
    blockedBy: list[str]    # 依赖的任务 id 列表

  这个@dataclass是新版（大概是3.7）python引入的装饰器，也就是说对于一个类，只要加上这个装饰器，就相当于下面这种写法。

class Task:
    def __init__(self, id, subject, description, status, owner, blockedBy):
        self.id = id
        self.subject = subject
        self.description = description
        self.status = status
        self.owner = owner
        self.blockedBy = blockedBy
    
    def __repr__(self):
        return f"Task(id={self.id!r}, subject={self.subject!r}, ...)"
    
    def __eq__(self, other):
        if not isinstance(other, Task):
            return False
        return (self.id == other.id and 
                self.subject == other.subject and
                self.description == other.description and
                self.status == other.status and
                self.owner == other.owner and
                self.blockedBy == other.blockedBy)
    
    def __hash__(self):
        return hash((self.id, self.subject, self.description, 
                     self.status, self.owner, tuple(self.blockedBy)))

。。。。。。等等

  点进去可以看到，我们还可以对这些生成方法进行自定义。

  每个task存成一个 JSON 文件，放在 .tasks/ 目录下：

def _task_path(task_id: str) -> Path:
    """传入任务id拿到对应路径"""
    return TASKS_DIR / f"{task_id}/.json"

def save_task(task: Task):
    """传入任务，写入指定位置。也可以传入已存在的任务，写入更新"""
    # asdict将类转化为字典，jumps将字典转化为json，indent代表格式化，=2说明开头缩紧两个空格
    _task_path(task.id).write_text(json.dumps(asdict(task), indent=2))


def load_task(task_id: str) -> Task:
    """传入task_id，返回其内容（对应的Task对象）"""
    return Task(**json.loads(_task_path(task_id).read_text()))


def list_tasks() -> list[Task]:
    """返回全部的task"""
    return [Task(**json.loads(p.read_text()))
            for p in sorted(TASKS_DIR.glob("task_*.json"))]

  ID是交由系统生成。这里思考一个问题，为什么那些MEMORY.md和SKILL.md以及本章的JSON全是一个个单独的文件？因为多个 Agent 可能并发写：A 完成 task_1、B 同时完成 task_2，写各自文件互不干扰；写一个大 list 就得加锁，否则后写覆盖先写。代价是列全部任务要扫目录（glob），但任务量不大时这不是问题。

def create_task(subject: str, description: str = "", blockedBy: list[str] | None = None) -> Task:
    """传入信息，返回并保存对应的Task对象"""
    task = Task(
        id = f"task_{int(time.time())}_{random.randint(0, 9999):04d}", # 04d意为四个随机数补零
        subject=subject,
        description=description,
        status="pending",
        owner=None,
        blockedBy=blockedBy or []
    )
    save_task(task)
    return task

  然后是任务的 blockedBy 是它依赖的前置任务 id 列表。can_start 检查所有前置是否都已完成。这就构成了一个 DAG（有向无环图）：B blockedBy A，意思是 A→B 有一条边，A 必须先完成。Agent 不用自己在 prompt 里记"先 A 再 B"，结构本身就编码了顺序，上下文怎么压缩都丢不了。

def get_task(task_id: str) -> str:
    """传入taskID 返回json版的内容，与load不同的是。这里返回的内容，而不是对象"""
    task = load_task(task_id)
    return json.dumps(asdict(task), indent=2)

def can_start(task_id: str) -> bool:
    """检查传入的task的阻塞是否都已经完成，返回是否允许启动"""
    task = load_task(task_id)
    for dep_id in task.blockedBy:
        if not _task_path(dep_id).exists(): # 如果阻塞明明记录了，但对应路径却不存在。
            return False
        if load_task(dep_id).status != "completed":
            return False
    return True

  所有的任务都有三个状态，转换被严格执行：

pending ──（由claim_task转换）──▶ in_progress ──（由complete_task转换）──▶ completed

  claim_task 把 pending → in_progress，同时记下 owner；非 pending 不能认领，被阻塞的也不能认领：

def claim_task(task_id: str, owner: str = "agent") -> str:
    """传入task_id和owner，判断能不能认领，并更新被认领的任务"""
    task = load_task(task_id)
    if task.status != "pending": # 如果task.status不是待定状态，就不能被认领
        return f"Task {task_id} is {task.status}, cannot claim"

    if not can_start(task_id): # 如果还有其他task阻塞它
        des = [d for d in task.blockedBy
               # 如果这些task不存在或者它们的状态不是已完成
               if not _task_path(d).exists() or load_task(d).status != "completed"]
        return f"Blocked by: {des}"

    task.owner = owner
    task.status = "in_progress"
    save_task(task) # 更新一下
    print(f"  \033[36m[claim] {task.subject} → in_progress (owner: {owner})\033[0m")
    return f"Claimed {task.id} ({task.subject})"

  complete_task 把 in_progress → completed，然后扫描所有任务，报告哪些被解锁了：

def complete_task(task_id: str) -> str:
    """传入task_id，用于完成该任务，并自动解锁被他阻塞的任务，返回一段显示完成的字符串"""

    # 完成task
    task = load_task(task_id)
    if task.status != "in_progress": # 完成前的状态必须是in_progress
        return f"Task {task_id} is {task.status}, cannot complete"
    task.status = "completed"
    save_task(task)


    # 解锁被他阻塞的task
    unblocked = [t.subject for t in list_tasks()
                 # 记录它们的subject，前提是这些任务状态待定 且 它们的被阻塞存在 且 它们的blocked都已经完成可以启动
                 # 本质上并不是通过它们直接的锁关系，而是直接遍历所有task去解锁，暴力但通俗易懂
                 if t.status == "pending" and t.blockedBy and can_start(t.id)]
    print(f"  \033[32m[complete] {task.subject} ✓\033[0m")
    msg = f"Completed {task.id} ({task.subject})"
    if unblocked:
        mag += f"\nUnblocked: {', '.join(unblocked)}"
        print(f"  \033[33m[unblocked] {', '.join(unblocked)}\033[0m")
    return msg # 一段string，记录了完成了什么task，进而又解锁了什么task

  到这里基本就结束了，剩下的就是一些“复制”类的操作，因为我们写完了这些工具，必须像tool_use那样给LLM也提供一套接口。这里我直接贴在下面。

# 新增的工具函数，功能与上述一样，只不过是封装为LLM交互的接口

def run_create_task(subject: str, description: str = "",
                    blockedBy: list[str] | None = None) -> str:
    """
    传入task的信息，创建task（生成id等），返回一段字符串表示创造成功。
    这个接口应该是交给LLM用的。
    """
    task = create_task(subject, description, blockedBy)
    deps = f" (blockedBy: {', '.join(blockedBy)})" if blockedBy else ""
    print(f"  \033[34m[create] {task.subject}{deps}\033[0m")
    return f"Created {task.id}: {task.subject}{deps}"

def run_list_tasks() -> str:
    """
    展示所有的tasks
    用于和LLM用的工具
    """
    tasks = list_tasks()
    if not tasks:
        return "No tasks. Use create_task to add some."
    lines = []
    for t in tasks:
        icon = {"pending": "○", "in_progress": "●",
                "completed": "✓"}.get(t.status, "?")
        deps = f" (blockedBy: {', '.join(t.blockedBy)})" if t.blockedBy else ""
        owner = f" [{t.owner}]" if t.owner else ""
        lines.append(f"  {icon} {t.id}: {t.subject} "
                     f"[{t.status}]{owner}{deps}")
    return "\n".join(lines)

def run_get_task(task_id: str) -> str:
    try:
        return get_task(task_id)
    except FileExistsError:
        return f"Error: Task {task_id} not found"


def run_claim_task(task_id: str) -> str:
    return claim_task(task_id, owner="agent") # 为什么要传入默认值？？所有小任务都是交给同一个agent执行的，所以该系统本质上是串行。这样写也是在为后面并发处理提供好接口

def run_complete_task(task_id: str) -> str: 
    return complete_task(task_id)

  最后在TOOLS和分发表里注册一下，agentloop里也保持精简状态即可。这个main函数又改回去了，不过无需在意，逻辑都是一样的（找到messages中的最后一段text并打印）。

# 新增
TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object",
                      "properties": {"command": {"type": "string"}},
                      "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"},
                                     "limit": {"type": "integer"}},
                      "required": ["path"]}},
    {"name": "write_file", "description": "Write content to a file.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"},
                                     "content": {"type": "string"}},
                      "required": ["path", "content"]}},
    {"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"]}},
    {"name": "list_tasks",
     "description": "List all tasks with status, owner, and dependencies.",
     "input_schema": {"type": "object", "properties": {},
                      "required": []}},
    {"name": "get_task",
     "description": "Get full details of a specific task by ID.",
     "input_schema": {"type": "object",
                      "properties": {"task_id": {"type": "string"}},
                      "required": ["task_id"]}},
    {"name": "claim_task",
     "description": "Claim a pending task. Sets owner, changes status to in_progress.",
     "input_schema": {"type": "object",
                      "properties": {"task_id": {"type": "string"}},
                      "required": ["task_id"]}},
    {"name": "complete_task",
     "description": "Complete an in-progress task. Reports unblocked downstream tasks.",
     "input_schema": {"type": "object",
                      "properties": {"task_id": {"type": "string"}},
                      "required": ["task_id"]}},
]

# 新增
TOOL_HANDLERS = {
    "bash": run_bash, "read_file": run_read, "write_file": run_write,
    "create_task": run_create_task, "list_tasks": run_list_tasks,
    "get_task": run_get_task, "claim_task": run_claim_task,
    "complete_task": run_complete_task,
}


# 最精简的版本。
def agent_loop(messages: list, context: dict):
    system = get_system_prompt(context)
    while True:
        try:
            response = client.messages.create(
                model=MODEL, system=system, messages=messages,
                tools=TOOLS, max_tokens=8000)
        except Exception as e:
            messages.append({"role": "assistant", "content": [
                {"type": "text",
                 "text": f"[Error] {type(e).__name__}: {e}"}]})
            return

        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            print(f"\033[36m> {block.name}\033[0m")
            handler = TOOL_HANDLERS.get(block.name)
            output = handler(**block.input) if handler else f"Unknown: {block.name}"
            print(str(output)[:300])
            results.append({"type": "tool_result",
                            "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
        context = update_context(context, messages)
        system = get_system_prompt(context)





# 又改了回来，不知为何
if __name__ == "__main__":
    print("s12: task system")
    print("Enter a question, press Enter to send. Type q to quit.\n")
    history = []
    context = update_context({}, [])
    while True:
        try:
            query = input("\033[36ms12 >> \033[0m")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        history.append({"role": "user", "content": query})
        agent_loop(history, context)
        context = update_context(context, history)
        for block in history[-1]["content"]:
            if getattr(block, "type", None) == "text":
                print(block.text)
            elif isinstance(block, dict) and block.get("type") == "text":
                print(block.get("text", ""))
        print()
    

  来跑几个测试捋一下。

---

二、跑几个测试

  在写本章博客的时候，原项目作者对项目进行了一些改动，有点难受。。。不过还是先看看哪里变动了吧。首先是与本章节无关的，对于s10和s11作者直接将其删去了。本章就成为了第10章，不过其他章节的代码部分是没有变动的。而本章内容相当于直接解耦，将所有关于task相关的函数封装成类，删去了不太相关的context、memory和prompt_system，转而引入原来的hooks和permission去为本章task的执行增加保障。

  首先来看作者是如何封装的。相当于把原先_take_path()、save_task()等工具函数直接集成到了类里面，成为TaskStore的属性。

# 封装成了类
class TaskStore:
    def __init__(self, directory: Path):
        self.directory = directory

    # 获得根目录
    def _root(self, create: bool = False) -> Path:

        # 如果需要，就创建目录
        if create:
            self.directory.mkdir(parents=True, exist_ok=True)

        # 解析为绝对路径
        root = self.directory.resolve()

        # 安全检查，防止逃逸到路径外面
        if not root.is_relative_to(WORKDIR.resolve()):
            raise ValueError("Task store escapes the workspace")
        return root

    # 获得任务文件路径
    def _path(self, task_id: str, create_root: bool = False) -> Path:

        # 验证task_id的格式
        if not isinstance(task_id, str) or not TASK_ID_PATTERN.fullmatch(task_id):
            raise ValueError(f"Invalid task ID: {task_id!r}")

        # 获得根目录
        root = self._root(create=create_root)
        # 构造文件路径
        path = (root / f"{task_id}.json").resolve()
        # 安全检查，防止逃逸到外面
        if not path.is_relative_to(root):
            raise ValueError(f"Invalid task ID: {task_id!r}")
        return path


    # 检查传入的task_id对应文件路径时候存在
    def exists(self, task_id: str) -> bool:
        return self._path(task_id).is_file()


    # 创建任务，返回task对象
    def create(self, subject: str, description: str = "",
               blocked_by: list[str] | None = None) -> Task:
        subject = subject.strip()

        # 验证标题，不能为空
        if not subject:
            raise ValueError("Task subject cannot be empty")

        # 依赖列表去重，然后验证所有对应文件都存在
        dependencies = list(dict.fromkeys(blocked_by or []))
        for dependency in dependencies:
            if not self.exists(dependency):
                raise ValueError(f"Dependency not found: {dependency}")

        # 确保根目录存在
        self._root(create=True)

        # 尝试100次去创建，以避免命名冲突
        for _ in range(100):
            task = Task(
                id=f"task_{secrets.token_hex(4)}", # 随机生成八位16进制字符串
                subject=subject,
                description=description,
                status="pending",
                owner=None,
                blockedBy=dependencies,
            )
            try:
                # 使用x模式，文件不存在才创建
                with self._path(task.id, create_root=True).open(
                    "x", encoding="utf-8"
                ) as handle:
                    json.dump(asdict(task), handle, indent=2)
                return task
            except FileExistsError:
                # id冲突，继续尝试
                continue
        raise RuntimeError("Could not allocate a unique task ID")


    # 保存任务，也可以用于更新
    def save(self, task: Task) -> None:
        self._path(task.id, create_root=True).write_text(
            json.dumps(asdict(task), indent=2),
            encoding="utf-8",
        )


    # 加载任务
    def load(self, task_id: str) -> Task:

        # 读文件
        data = json.loads(self._path(task_id).read_text(encoding="utf-8"))

        # 转为Task对象
        task = Task(**data)

        # 校验id是否一致
        if task.id != task_id:
            raise ValueError(f"Task file ID does not match {task_id}")

        # 检查状态是否合法
        if task.status not in ("pending", "in_progress", "completed"):
            raise ValueError(f"Invalid task status: {task.status}")
        return task


    # 列出所有task
    def list(self) -> list[Task]:
        if not self.directory.exists():
            return []
        root = self._root()
        return [self.load(path.stem)
                for path in sorted(root.glob("task_*.json"))]


TASKS = TaskStore(TASKS_DIR)

  以load_task()为例，只需要调用方法返回即可。这种封装方式使程序的状态清晰可控 。

# def load_task(task_id: str) -> Task:
#     """传入task_id，返回其内容（对应的Task对象）"""
#     return Task(**json.loads(_task_path(task_id).read_text()))
def load_task(task_id: str) -> Task:
    return TASKS.load(task_id)

  此外，原来的方法做了优化，目的是让整个任务拆解，执行的过程更加的安全。

# 观察变化
def claim_task(task_id: str, owner: str = "agent") -> str:
    """传入task_id和owner，判断能不能认领，并更新被认领的任务"""
    task = load_task(task_id)
    if task.status != "pending": # 如果task.status不是待定状态，就不能被认领
        return f"Task {task_id} is {task.status}, cannot claim"

    # if not can_start(task_id): # 如果还有其他task阻塞它
    #     des = [d for d in task.blockedBy
    #            # 如果这些task不存在或者它们的状态不是已完成
    #            if not _task_path(d).exists() or load_task(d).status != "completed"]
    #     return f"Blocked by: {des}"
    dependencies = incomplete_dependencies(task)
    if dependencies:
        return f"Blocked by: {dependencies}"

    task.owner = owner
    task.status = "in_progress"

    # save_task(task) # 更新一下
    TASKS.save(task)

    print(f"  \033[36m[claim] {task.subject} → in_progress (owner: {owner})\033[0m")
    return f"Claimed {task.id} ({task.subject})"


# 观察变化
def complete_task(task_id: str, owner: str = "agent") -> str:
    """传入task_id，用于完成该任务，并自动解锁被他阻塞的任务，返回一段显示完成的字符串"""

    # 完成task
    task = load_task(task_id)
    if task.status != "in_progress": # 完成前的状态必须是in_progress
        return f"Task {task_id} is {task.status}, cannot complete"
    # 新增了所属owner的判断
    if task.owner != owner:
        return f"Task {task_id} is owned by {task.owner}, not {owner}"

    # 记录当前完成之前，可以执行的所有sask。想一下为什么记录？是为了解锁！！配合下面unblocked中增加一个判断，这是最简单的防止死锁的方式
    ready_before = {
        candidate.id
        for candidate in list_tasks()
        if candidate.status == "pending"
        and candidate.blockedBy
        and can_start(candidate.id)
    }

    task.status = "completed"
    TASKS.save(task)
    # 解锁被他阻塞的task
    unblocked = [t.subject for t in list_tasks()
                 # 记录它们的subject，前提是这些任务状态待定 且 它们的被阻塞存在 且 它们的blocked都已经完成可以启动
                 # 本质上并不是通过它们直接的锁关系，而是直接遍历所有task去解锁，暴力但通俗易懂
                 # 为什么要记录subject，而不是id？
                 if t.status == "pending" and t.blockedBy and can_start(t.id) and t.id not in ready_before] # 新增了最后一个
    print(f"  \033[32m[complete] {task.subject} ✓\033[0m")
    msg = f"Completed {task.id} ({task.subject})"
    if unblocked:
        msg += f"\nUnblocked: {', '.join(unblocked)}"
        print(f"  \033[33m[unblocked] {', '.join(unblocked)}\033[0m")
    return msg # 一段string，记录了完成了什么task，进而又解锁了什么task

  最后简单总结一下两者的区别变化：不同点在于做了一些安全类别的优化、其他附加的功能有删减和做了一些封装，相同点都是作为工具函数去给模型赋予的这种能力。那我们来做个测试跑一下看看吧。

这里我给它一个依赖任务：

1. write_file 写一个 demo.py，内容两行:
   print(“hello”)
   print(“world”)
2. 用 edit_file 把 “hello” 换成 “task system works”
3. read_file 读回来确认

输出没问题，非常好。但是没有生成.task文件夹和下面的任务文件。

s12 >> in the /Users/bx/Documents/coding/learn_cladudecode/s12_task_system/test_unit, 1. write_file 写一个 demo.py，内容两行:   print("hello")   print("world")2. 用 edit_file 把 "hello" 换成 "task system works"3. read_file 读回来确认
[HOOK] UserPromptSubmit: working in /Users/bx/Documents/coding/learn_cladudecode
[HOOK] write_file(['print("hello")\nprint("world")\n', '/Users/bx/Documents/co)
[HOOK] edit_file(['print("task system works")', 'print("hello")'])
[HOOK] read_file(['/Users/bx/Documents/coding/learn_cladudecode/s12_task_syst)
[HOOK] Stop: session used 3 tool calls
三步全部完成！✅ 最终确认文件内容为：


print("task system works")
print("world")

| 步骤 | 操作 | 结果 |
|------|------|------|
| 1 | `write_file` demo.py | 写入 `print("hello")` / `print("world")` |
| 2 | `edit_file` 替换 | `"hello"` → `"task system works"` ✅ |
| 3 | `read_file` 确认 | 内容正确：`print("task system works")` + `print("world")` ✅ |

Task system 工作正常！🎉

  原因可能是system_prompt或者是用户的提示词太简单了，改的强制点试试。

s12 >> in the /Users/bx/Documents/coding/learn_cladudecode/s12_task_system/test_unit, 用任务系统管理以下工作：1. write_file 写一个 demo.py，内容两行:   print("hello")   print("world")2. 用 edit_file 把 "hello" 换成 "task system works"3. read_file 读回来确认
[HOOK] UserPromptSubmit: working in /Users/bx/Documents/coding/learn_cladudecode
[HOOK] create_task(['write_file /Users/bx/Documents/coding/learn_cladudecode/s1)
  [create] Write demo.py with hello/world
[HOOK] create_task([['task_74cb46a5'], '用 edit_file 把 demo.py 中的 "hello" 替换为 "t)
  [create] Edit demo.py: hello → task system works (blockedBy: task_74cb46a5)
[HOOK] create_task([['task_74cb46a5'], 'Read demo.py to confirm changes'])
  [create] Read demo.py to confirm changes (blockedBy: task_74cb46a5)
[HOOK] create_task([['task_ba5e757c'], 'read_file 读回 demo.py，确认 "task system wo)
  [create] Read demo.py to confirm final content (blockedBy: task_ba5e757c)
[HOOK] claim_task(['task_74cb46a5'])
  [claim] Write demo.py with hello/world → in_progress (owner: agent)
[HOOK] write_file(['print("hello")\nprint("world")\n', '/Users/bx/Documents/co)
[HOOK] complete_task(['task_74cb46a5'])
  [complete] Write demo.py with hello/world ✓
  [unblocked] Read demo.py to confirm changes, Edit demo.py: hello → task system works
[HOOK] claim_task(['task_ba5e757c'])
  [claim] Edit demo.py: hello → task system works → in_progress (owner: agent)
[HOOK] edit_file(['print("task system works")', 'print("hello")'])
[HOOK] complete_task(['task_ba5e757c'])
  [complete] Edit demo.py: hello → task system works ✓
  [unblocked] Read demo.py to confirm final content
[HOOK] claim_task(['task_a5ec949c'])
  [claim] Read demo.py to confirm final content → in_progress (owner: agent)
[HOOK] read_file(['/Users/bx/Documents/coding/learn_cladudecode/s12_task_syst)
[HOOK] complete_task(['task_a5ec949c'])
  [complete] Read demo.py to confirm final content ✓
[HOOK] list_tasks([])
[HOOK] create_task(['这个任务的依赖关系不对（依赖于 task_74cb46a5 而非 task_ba5e757c），已经有正确的 tas)
  [create] Cleanup: remove stale task_56e08906
[HOOK] claim_task(['task_56e08906'])
  [claim] Read demo.py to confirm changes → in_progress (owner: agent)
[HOOK] complete_task(['task_56e08906'])
  [complete] Read demo.py to confirm changes ✓
[HOOK] claim_task(['task_65d63b75'])
  [claim] Cleanup: remove stale task_56e08906 → in_progress (owner: agent)
[HOOK] complete_task(['task_65d63b75'])
  [complete] Cleanup: remove stale task_56e08906 ✓
[HOOK] complete_task(['task_65d63b75'])
[HOOK] list_tasks([])
[HOOK] Stop: session used 21 tool calls
所有任务均已完成 ✅ 依赖链正确运转：**写文件 → 编辑文件 → 读回确认**，每一步都在前置任务完成后才解锁执行。

  生成的task文件也成功了，非常完美。

---

总结

  整个任务分发有点像操作系统中死锁那一章节。后续我会将准备考研时学习408的这些笔记免费分享给大家。

总结一下测试发生的过程吧。

  从日志里逐条还原了完整的 DAG 结构和执行轨迹。整个过程其实有两个 DAG：模型先建了一个有缺陷的 DAG，发现后补救了。

一、DAG 结构（最终形态）

                                    ┌─────────────────────────┐
                                    │ task_56e08906           │
                                    │ "Read demo.py to        │  ← 建错了依赖
                                    │  confirm changes"       │     （应该依赖 edit，
                            ┌──────▶│ status: completed       │      却依赖了 write）
                            │       │ owner: agent            │
                            │       └─────────────────────────┘
                            │
  ┌──────────────────┐      │       ┌─────────────────────────┐
  │ task_74cb46a5    │      ├──────▶│ task_ba5e757c           │
  │ "Write demo.py"  │──────┤       │ "Edit demo.py"          │
  │ status: completed│      │       │ status: completed       │
  │ owner: agent     │      │       │ owner: agent            │
  └──────────────────┘      │       └───────────┬─────────────┘
                            │                   │
                            │                   ▼
                            │       ┌─────────────────────────┐
                            │       │ task_a5ec949c           │
                            │       │ "Read demo.py to        │
                            │       │  confirm final content" │  ← 正确的读回任务
                            │       │ status: completed       │
                            │       │ owner: agent            │
                            │       └─────────────────────────┘
                            │
                            │       ┌─────────────────────────┐
                            └──────▶│ task_65d63b75           │
                                    │ "Cleanup: remove stale  │  ← 补救任务
                                    │  task_56e08906"          │     （无依赖，独立）
                                    │ status: completed       │
                                    │ owner: agent            │
                                    └─────────────────────────┘

二、DAG 的动态演化（4 个阶段）

阶段 1：模型建图（4 次 create_task）

模型一口气建了 4 个任务，但第 3 个建错了依赖：

创建顺序：
  ① task_74cb46a5  "Write demo.py"        blockedBy: []           ✅ 正确
  ② task_ba5e757c  "Edit demo.py"         blockedBy: [①]          ✅ 正确
  ③ task_56e08906  "Read confirm changes" blockedBy: [①]          ❌ 依赖了 write，应该依赖 edit
  ④ task_a5ec949c  "Read final content"   blockedBy: [②]          ✅ 正确（补建的正确版本）

此时的 DAG：

                    ┌── ③ Read confirm changes (blockedBy ①)  ← 错！
 ① Write demo.py ──┤
                    └── ② Edit demo.py (blockedBy ①) ── ④ Read final (blockedBy ②)

模型犯的错：第 3 个任务"读回确认更改"逻辑上应该在编辑之后才读，但模型把它的 blockedBy 指向了 ①（write）而不是 ②（edit）。这意味着 write 一完成，③ 就解锁了——此时文件还没被 edit，读回来只会看到 “hello” 而非 “task system works”。

阶段 2：执行 ① → 解锁 ②③

claim(①)  → write_file(demo.py, "hello"/"world")
complete(①) → ① 变 completed
              unblocked: ③ Read confirm changes, ② Edit demo.py

complete_task 的 ready_before 机制生效：完成 ① 前，② 和 ③ 都是 pending 且 blockedBy 非空且 can_start=False；完成 ① 后 can_start 变 True，所以两者都报为 unblocked。

阶段 3：模型选择先做 ②（跳过错误的 ③）

claim(②)  → edit_file(demo.py, "hello" → "task system works")
complete(②) → ② 变 completed
              unblocked: ④ Read final content   ← 只有 ④ 新解锁

模型没有先做 ③（虽然 ③ 在阶段 2 就已解锁），而是先做了 ②。这是模型自己的判断——它意识到应该先 edit 再 read。完成 ② 后，只有 ④ 被新解锁（③ 在阶段 2 就已解锁，不算新的，不报）。

阶段 4：执行 ④ → 发现 ③ 是孤儿 → 补救

claim(④)  → read_file(demo.py)  → 确认内容正确
complete(④) → ④ 变 completed

list_tasks() → 模型看到 ③ 还是 pending（从来没被认领过）
              意识到 ③ 是建错的任务

create(⑤ task_65d63b75 "Cleanup: remove stale task_56e08906")  ← 补救任务
claim(③)  → 直接认领（③ 此时 can_start=True，因为 ① 已完成）
complete(③) → ③ 变 completed（废弃掉）

claim(⑤)  → 认领补救任务
complete(⑤) → ⑤ 变 completed

complete(⑤) → 重复完成（返回错误，但已被 try/except 兜住）

list_tasks() → 全部 completed

三、完整执行时间线（21 次工具调用）

轮次  工具              任务        DAG 状态变化
────  ────────────────  ──────────  ──────────────────────────────
 1    create_task       ① 创建      节点①加入图
 2    create_task       ② 创建      节点②加入图, 边①→②
 3    create_task       ③ 创建      节点③加入图, 边①→③ (错误!)
 4    create_task       ④ 创建      节点④加入图, 边②→④
 5    claim_task        ① 认领      ①: pending → in_progress
 6    write_file        —           (执行①的实际工作)
 7    complete_task     ① 完成      ①: in_progress → completed
                                    ②③解锁 (ready_before 机制)
 8    claim_task        ② 认领      ②: pending → in_progress
 9    edit_file         —           (执行②的实际工作)
10    complete_task     ② 完成      ②: in_progress → completed
                                    ④解锁
11    claim_task        ④ 认领      ④: pending → in_progress
12    read_file         —           (执行④的实际工作)
13    complete_task     ④ 完成      ④: in_progress → completed
14    list_tasks        —           (模型检查, 发现③还 pending)
15    create_task       ⑤ 创建      节点⑤加入图 (补救任务)
16    claim_task        ③ 认领      ③: pending → in_progress
17    complete_task     ③ 完成      ③: in_progress → completed (废弃)
18    claim_task        ⑤ 认领      ⑤: pending → in_progress
19    complete_task     ⑤ 完成      ⑤: in_progress → completed
20    complete_task     ⑤ 重复完成   (返回错误, 被兜住)
21    list_tasks        —           (确认全部 completed)

四、模型建错依赖说明了什么

这次运行暴露了 s12 的一个设计特点：DAG 的正确性完全靠 LLM 自觉。模型把 ③ 的 blockedBy 指向 ① 而非 ②，系统不会纠正它——create_task 只检查依赖是否存在（[code.py:115-117](file:///Users/bx/Documents/coding/learn_cladudecode/s12_task_system/code.py#L115-L117)），不检查依赖是否"合理"。

模型最终自己发现了错误（通过 list_tasks 看到 ③ 还是 pending），然后建了补救任务 ⑤ 来清理。这是 LLM 自我纠错，不是系统纠错——如果模型没发现，③ 就会永远 pending。