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
- Access to the project source code
- A plan file saved under
.claude/plans/with valid XML task blocks
Read a plan and inspect its tasks
-
Import
read_specfrompipelineand call it with the path to your plan file:from pipeline import read_spec tasks = read_spec(".claude/plans/my-plan.md")read_specraisesFileNotFoundErrorif the path does not exist andValueErrorifplan_pathis an empty string. -
Pass the returned task list to
present_tasksto 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_statereturnsNone, no saved state exists yet and all tasks are treated as pending. -
To inspect a single task in full detail, call
present_task_detailwith the task object:from spec import present_task_detail print(present_task_detail(tasks[0]))
Execute a plan with per-task approval
-
Call
execute_with_approvalwith 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, orskip_simplify=Trueto bypass the corresponding quality checks during development. -
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 -
To track progress mid-run, call
format_progress_barwith 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
-
Call
find_resumable_plansto list any plans that have saved state:from spec import find_resumable_plans resumable = find_resumable_plans(".claude/plans") -
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) -
Pass the pending tasks to a
PipelineOrchestrator, skipping the IDs already recorded instate.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
-
Instantiate
PipelineOrchestratorwith your plan path:orchestrator = PipelineOrchestrator(".claude/plans/my-plan.md") -
Call
run_gates_for_taskwith the task you want to evaluate:gate_result = orchestrator.run_gates_for_task(tasks[0]) -
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 -
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 |