task-coordinator

What this skill does

# Task Coordinator

Patterns for integrating Claude Code's native Task system into workflow orchestration.

## When to Use

- Workflow has 3+ sequential steps needing progress visibility
- Parallel execution with dependency tracking
- Long-running workflows needing crash recovery
- Cross-session work requiring persistent state

## Config Check (Guard Pattern)

```
TASK_TRACKING = config_read("task_tracking.enabled", "false")
LEDGER_ENABLED = config_read("task_tracking.ledger", "false")
LEDGER_PATH = config_read("task_tracking.ledger_path", ".agents/workflow-ledger.yml")
AUTO_CLEANUP = config_read("task_tracking.auto_cleanup", "true")

If NOT TASK_TRACKING: skip all Task operations, run workflow normally
```

## Task Lifecycle

### 1. Create

```
TASK = TaskCreate(
  subject: "Imperative verb phrase",     # "Run security review"
  description: "What and why",
  activeForm: "Present continuous form",  # "Running security review"
  metadata: {
    workflow: WORKFLOW_ID,
    step: STEP_NUMBER,
    milestone: true|false,
    phase: "phase-name",
    parallel_group: "group-id"
  }
)
```

### 2. Full Step Pattern

```
If TASK_TRACKING:
  STEP_TASK = TaskCreate(subject: "Step N: Name", activeForm: "Doing name", metadata: {...})
  TaskUpdate(STEP_TASK.id, status: "in_progress")

... existing step logic (unchanged) ...

If TASK_TRACKING:
  TaskUpdate(STEP_TASK.id, status: "completed")

If LEDGER_ENABLED AND step.milestone:
  Write checkpoint -> LEDGER_PATH
```

## Metadata Conventions

| Key | Type | Purpose | Example |
|-----|------|---------|---------|
| `workflow` | string | Groups tasks to one workflow run | `"build-task-abc123"` |
| `step` | integer | Step number within workflow | `4` |
| `milestone` | boolean | Whether step triggers ledger checkpoint | `true` |
| `phase` | string | Named phase for multi-phase workflows | `"discovery"` |
| `parallel_group` | string | Groups tasks launched in parallel | `"reviewers"` |

Workflow ID format: `"{workflow-name}-{timestamp}"` (e.g., `"build-task-20260213T1430"`)

## Dependency Mapping

```
TASK_A = TaskCreate(subject: "Foundation step", ...)
TASK_B = TaskCreate(subject: "Depends on A", ...)
TaskUpdate(TASK_B.id, addBlockedBy: [TASK_A.id])
```

**Rules:**
- Create ALL tasks first, then set dependencies
- Blocked tasks cannot be claimed until blockers complete

**Multi-phase example:**

```
PHASES = [
  {name: "Discovery", deps: []},
  {name: "Architecture", deps: ["Discovery"]},
  {name: "Implementation", deps: ["Architecture"]},
  {name: "Review", deps: ["Implementation"]}
]

PHASE_TASKS = {}
For each P in PHASES:
  PHASE_TASKS[P.name] = TaskCreate(
    subject: "Phase: {P.name}",
    activeForm: "Running {P.name}",
    metadata: {workflow: WORKFLOW_ID, phase: P.name}
  )

For each P in PHASES:
  If P.deps is not empty:
    BLOCKER_IDS = [PHASE_TASKS[d].id for d in P.deps]
    TaskUpdate(PHASE_TASKS[P.name].id, addBlockedBy: BLOCKER_IDS)
```

## Parallel Execution

Create all tasks first, then launch Task agents in a single message for parallelism.

```
GROUP_ID = "reviewers-{WORKFLOW_ID}"

REVIEWER_TASKS = []
For each REVIEWER in REVIEWER_LIST:
  T = TaskCreate(
    subject: "Run {REVIEWER}",
    activeForm: "Running {REVIEWER}",
    metadata: {workflow: WORKFLOW_ID, parallel_group: GROUP_ID, reviewer: REVIEWER}
  )
  REVIEWER_TASKS.append(T)

# Launch ALL agents in a SINGLE message (enables parallelism)
For each T in REVIEWER_TASKS:  # all in ONE message
  Task(subagent_type: T.metadata.reviewer, prompt: "...", run_in_background: true)

# Collect results
For each T in REVIEWER_TASKS:
  RESULT = wait for agent completion
  TaskUpdate(T.id, status: "completed")
```

**Key rule:** All `Task()` calls MUST be in the same response message. Splitting across messages forces sequential execution.

### Result Aggregation

```
ALL_TASKS = TaskList()
GROUP_TASKS = [T for T in ALL_TASKS where T.metadata.parallel_group == GROUP_ID]
COMPLETED = [T for T in GROUP_TASKS where T.status == "completed"]
FAILED = [T for T in GROUP_TASKS where T.status != "completed"]

If FAILED is not empty: Handle failures per error table
Else: Aggregate results from COMPLETED tasks
```

## Ledger Checkpoint Pattern

YAML-based receipts for crash recovery at milestone steps.

### Schema

```yaml
version: 1
workflow_id: "build-task-20260213T1430"
workflow_type: "build-task"
started_at: "2026-02-13T14:30:00Z"
last_checkpoint: "2026-02-13T14:35:00Z"
current_step: 4
status: "in_progress"  # in_progress | completed | failed
checkpoints:
  - step: 1
    name: "Workspace setup"
    status: "completed"
    completed_at: "2026-02-13T14:31:00Z"
    data: {}
```

### Writing Checkpoints

```
If LEDGER_ENABLED AND step.milestone:
  CHECKPOINT = {step: STEP_NUMBER, name: STEP_NAME, status: "completed", completed_at: NOW(), data: {...}}
  Read existing ledger from LEDGER_PATH (or create new)
  Append CHECKPOINT to checkpoints array
  Update last_checkpoint, current_step
  Write ledger to LEDGER_PATH
```

### Milestone Steps by Workflow

| Workflow | Milestone Steps |
|----------|----------------|
| build-task | Workspace setup (1), Build (4), Verify (5), Quality (7), Ship (9) |
| blueprint | Discovery (1), Architecture (4), Plan written (6), Execution (9) |
| run-blueprint | Task creation (2), Each task completion |
| quality-gate | Each reviewer completion, Final verdict |

## Resume from Crash

```
If LEDGER_ENABLED:
  LEDGER = Read LEDGER_PATH
  If LEDGER exists AND LEDGER.status == "in_progress":
    LAST_STEP = LEDGER.current_step
    COMPLETED_STEPS = [C.step for C in LEDGER.checkpoints where C.status == "completed"]
    For each STEP in WORKFLOW_STEPS:
      If STEP.number in COMPLETED_STEPS: skip
      Else: Execute STEP
```

**Recovery decision table:**

| Ledger State | Action |
|-------------|--------|
| No ledger file | Fresh start |
| `status: completed` | Fresh start (previous run finished) |
| `status: in_progress` | Resume from last checkpoint |
| `status: failed` | Resume from failed step |
| Ledger parse error | Log warning, fresh start |

## Cross-Session Persistence

```
# At workflow start: export task list ID
TASK_LIST_ID = current task list identifier
Set env: CLAUDE_CODE_TASK_LIST_ID = TASK_LIST_ID

# In new session: restore task list
If env CLAUDE_CODE_TASK_LIST_ID is set:
  Restore task list from CLAUDE_CODE_TASK_LIST_ID
  ALL_TASKS = TaskList()
  Resume from incomplete tasks
```

**Note:** Cross-session persistence requires the task list ID to be stored externally (env var, file, or ledger).

## Cleanup

```
If TASK_TRACKING AND workflow completed successfully:
  If AUTO_CLEANUP:
    ALL_TASKS = TaskList()
    WORKFLOW_TASKS = [T for T in ALL_TASKS where T.metadata.workflow == WORKFLOW_ID]
    For each T in WORKFLOW_TASKS:
      TaskUpdate(T.id, status: "completed")

  If LEDGER_ENABLED:
    Update ledger: status = "completed", last_checkpoint = NOW()
```

## Error Handling

| Error | Action |
|-------|--------|
| TaskCreate fails | Log warning, continue without tracking |
| TaskUpdate fails | Retry once, then log and continue |
| TaskList timeout | Fall back to ledger if available |
| TaskGet returns stale data | Re-fetch before update |
| Ledger write fails | Log warning, continue without checkpoint |
| Ledger parse error | Log warning, treat as fresh start |
| Dependency cycle detected | Log error, remove cycle, continue |

**Principle:** Task tracking failures MUST NEVER block workflow execution. Always degrade gracefully.

## State Relationship

```
+---------------------------+     +---------------------------+
|     Native Tasks          |     |     YAML Ledger           |
|  (TaskCreate/Update/List) |     |  (.agents/workflow-ledger) |
+---------------------------+     +---------------------------+
|  Source of truth for:     |     |  Supplementary for:       |
|  - Current step status    |     |  - Crash recovery         |
|  - Dependency tracking    |     |  - Cross-session resume   |
|  - Progress visibility    |     |