SuperAPI AI Directory
Partner with us中文AI newsFavorites
Skillstask-coordinator

task-coordinator

Agents & models

View repository

Updated 10-16 00:04

About this skill

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    |     |  - Milestone receipts     |
|  - Parallel coordination  |     |  - Audit trail            |
+---------------------------+     +---------------------------+
         PRIMARY                         SECONDARY

If conflict: Tasks win. Ledger is advisory only.