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.