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 = /majestic:config task_tracking.enabled false LEDGER_ENABLED = /majestic:config task_tracking.ledger false LEDGER_PATH = /majestic:config task_tracking.ledger_path .agents/workflow-ledger.yml AUTO_CLEANUP = /majestic:config 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

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.