Work with the spec engine

Use the spec engine when you need to execute a structured plan file through an approval loop — running tasks one at a time, applying quality gates after each, and resuming interrupted runs from saved state.

Prerequisites

Read a plan and inspect its tasks

  1. Import read_spec from pipeline and call it with the path to your plan file:

    from pipeline import read_spec
    
    tasks = read_spec(".claude/plans/my-plan.md")
    

    read_spec raises FileNotFoundError if the path does not exist and ValueError if plan_path is an empty string.

  2. Pass the returned task list to present_tasks to see a formatted markdown table of all tasks and their current state:

    from spec import present_tasks, load_state
    
    state = load_state(".claude/plans/my-plan.md")
    print(present_tasks(tasks, state))
    

    If load_state returns None, no saved state exists yet and all tasks are treated as pending.

  3. To inspect a single task in full detail, call present_task_detail with the task object:

    from spec import present_task_detail
    
    print(present_task_detail(tasks[0]))
    

Execute a plan with per-task approval

  1. Call execute_with_approval with your plan path and a callback to handle each completed task:

    from spec import execute_with_approval
    
    result = execute_with_approval(
        ".claude/plans/my-plan.md",
        on_task_complete=my_callback,
    )
    

    Pass skip_gates=True, skip_tests=True, or skip_simplify=True to bypass the corresponding quality checks during development.

  2. After the run completes, check whether all tasks passed:

    print(result.success)   # True if every task executed and passed its gate
    print(result.summary)   # Human-readable summary of the run
    
  3. To track progress mid-run, call format_progress_bar with the count of completed and total tasks:

    from spec import format_progress_bar
    
    bar = format_progress_bar(len(state.completed), len(tasks))
    

Resume an interrupted run

  1. Call find_resumable_plans to list any plans that have saved state:

    from spec import find_resumable_plans
    
    resumable = find_resumable_plans(".claude/plans")
    
  2. Load the saved state for the plan you want to resume:

    from spec import load_state, get_pending_tasks
    
    state = load_state(".claude/plans/my-plan.md")
    pending = get_pending_tasks(tasks, state)
    
  3. Pass the pending tasks to a PipelineOrchestrator, skipping the IDs already recorded in state.completed:

    from pipeline import PipelineOrchestrator
    
    orchestrator = PipelineOrchestrator(".claude/plans/my-plan.md")
    result = orchestrator.run_all(
        skip_task_ids=set(state.completed),
    )
    

Run quality gates for a single task

  1. Instantiate PipelineOrchestrator with your plan path:

    orchestrator = PipelineOrchestrator(".claude/plans/my-plan.md")
    
  2. Call run_gates_for_task with the task you want to evaluate:

    gate_result = orchestrator.run_gates_for_task(tasks[0])
    
  3. Inspect the result fields to determine next steps:

    print(gate_result.severity)             # e.g. "error", "warning", "ok"
    print(gate_result.quality_gate_passed)  # True / False / None
    print(gate_result.gate_score)           # float score if available
    
  4. Display the outcome to the user with present_task_result:

    from spec import present_task_result
    
    print(present_task_result(tasks[0], gate_result))
    

Clear saved state

Call clear_state to remove the state comment from a plan file and reset it for a fresh run:

from spec import clear_state

clear_state(".claude/plans/my-plan.md")

Verify success

A run is successful when PipelineResult.success returns True. This property is True only when every TaskResult in PipelineResult.tasks has executed=True and quality_gate_passed=True. Check PipelineResult.summary for a human-readable breakdown and inspect individual TaskResult.error fields for any tasks that failed.

Unresolved references

Auto-generated by attune-author fact-check. Review and either fix the source code, fix this doc, or add an override.

Location Severity Issue
Line 145 (code fence) error from spec import … — module not importable