Learn Claude Code
s07

Tasks

Planning & Coordination

Task Graph + Dependencies

204 LOC8 toolsTaskManager with file-based state + dependency graph
Overview

任务系统把每条任务存成磁盘上的一份文件,跨会话保留。今天关掉 Claude Code 明天回来开新会话,所有任务还在原处——状态、依赖关系、负责人全部记得。

Running example
Research project
耐心资本对企业 ESG 表现的实证项目
What happens in this section

phase 1 跑完 do0–do6 后落地了 main_panel.dta(26441 obs)与 5 张三线表,phase 2 还要做 do7 IV/PSM、do8 机制、do9 异质性、do10 稳健性矩阵 ABC 三个面板共 13 组。任务跨周推进,checkpoint.md 顶层四类——已完成 / 进行中 / 待做 / 待解决——的具体条目按依赖关系拆进 .tasks/,每条任务记 stata-mcp do-file 名、产物表号、依赖前置任务。

s01 > s02 > s03 > s04 > s05 > s06 | [ s07 ] s08 > s09 > s10 > s11 > s12

"Break big goals into small tasks, order them, persist to disk" -- a file-based task graph with dependencies, laying the foundation for multi-agent collaboration.

Harness layer: Persistent tasks -- goals that outlive any single conversation.

Problem

s03's TodoManager is a flat checklist in memory: no ordering, no dependencies, no status beyond done-or-not. Real goals have structure -- task B depends on task A, tasks C and D can run in parallel, task E waits for both C and D.

Without explicit relationships, the agent can't tell what's ready, what's blocked, or what can run concurrently. And because the list lives only in memory, context compression (s06) wipes it clean.

Solution

Promote the checklist into a task graph persisted to disk. Each task is a JSON file with status, dependencies (blockedBy). The graph answers three questions at any moment:

  • What's ready? -- tasks with pending status and empty blockedBy.
  • What's blocked? -- tasks waiting on unfinished dependencies.
  • What's done? -- completed tasks, whose completion automatically unblocks dependents.
.tasks/
  task_1.json  {"id":1, "status":"completed"}
  task_2.json  {"id":2, "blockedBy":[1], "status":"pending"}
  task_3.json  {"id":3, "blockedBy":[1], "status":"pending"}
  task_4.json  {"id":4, "blockedBy":[2,3], "status":"pending"}

Task graph (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

Ordering:     task 1 must finish before 2 and 3
Parallelism:  tasks 2 and 3 can run at the same time
Dependencies: task 4 waits for both 2 and 3
Status:       pending -> in_progress -> completed

This task graph becomes the coordination backbone for everything after s07: background execution (s08), multi-agent teams (s09+), and worktree isolation (s12) all read from and write to this same structure.

How It Works

  1. TaskManager: one JSON file per task, CRUD with dependency graph.
class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self._next_id = self._max_id() + 1

    def create(self, subject, description=""):
        task = {"id": self._next_id, "subject": subject,
                "status": "pending", "blockedBy": [],
                "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)
  1. Dependency resolution: completing a task clears its ID from every other task's blockedBy list, automatically unblocking dependents.
def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)
  1. Status + dependency wiring: update handles transitions and dependency edges.
def update(self, task_id, status=None,
           add_blocked_by=None, remove_blocked_by=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    if add_blocked_by:
        task["blockedBy"] = list(set(task["blockedBy"] + add_blocked_by))
    if remove_blocked_by:
        task["blockedBy"] = [x for x in task["blockedBy"] if x not in remove_blocked_by]
    self._save(task)
  1. Four task tools go into the dispatch map.
TOOL_HANDLERS = {
    # ...base tools...
    "task_create": lambda **kw: TASKS.create(kw["subject"]),
    "task_update": lambda **kw: TASKS.update(kw["task_id"], kw.get("status")),
    "task_list":   lambda **kw: TASKS.list_all(),
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),
}

From s07 onward, the task graph is the default for multi-step work. s03's Todo remains for quick single-session checklists.

What Changed From s06

ComponentBefore (s06)After (s07)
Tools58 (task_create/update/list/get)
Planning modelFlat checklist (in-memory)Task graph with dependencies (on disk)
RelationshipsNoneblockedBy edges
Status trackingDone or notpending -> in_progress -> completed
PersistenceLost on compressionSurvives compression and restarts

Try It

cd claude-code-for-researchers
python agents/s07_task_system.py
  1. Create phase 2 tasks: do7 IV/PSM → do8 mechanism → do9 heterogeneity → do10 robustness, with blocked_by chain
  2. List all tasks and show the dependency graph
  3. Complete the IV/PSM task and list tasks to see mechanism unblocked
  4. Create a task board: "audit protected terms" -> "rerun do6 baseline" + "regen tables 1-5 from data" in parallel -> "merge into 07_论文写作"