Skip to content

Spec Engine

Reference

The spec engine exposes its public API through two packages: attune.pipeline (execution) and attune.spec (state and presentation). execute_with_approval lives in attune.spec.runner.

Pipeline execution

Symbol Purpose
PipelineOrchestrator(spec_path, *, skip_gates=False, skip_tests=False, skip_simplify=False) Load a plan file and prepare tasks for execution with optional gate overrides.
PipelineOrchestrator.run_all(*, on_task_complete=None, skip_task_ids=None) Async. Execute all tasks, awaiting an optional async callback after each; returns a PipelineResult. The callback receives (task, result) and may return "redo" / "auto" / "stop" (or None/"approve" to continue).
PipelineOrchestrator.run_gates_for_task(task) Run quality gates for a single DecomposedTask and return a TaskResult.
read_spec(plan_path) Parse a plan file and extract its XML task blocks into DecomposedTask objects. Raises FileNotFoundError (missing file) or ValueError (empty path).
execute_with_approval(spec_path, on_task_complete, *, skip_gates=False, skip_tests=False, skip_simplify=False) Async. Execute a spec with an interactive per-task approval loop. Import from attune.spec.runner.

State management

Symbol Purpose
load_state(plan_path) Read a SpecState from the HTML comment embedded in a plan file; returns None if no state exists.
save_state(state) Write or update the spec-state comment in a plan file.
clear_state(plan_path) Remove the spec-state comment from a plan file.
find_resumable_plans(plans_dir='.claude/plans') Return all SpecState objects whose plans have incomplete execution.
get_pending_tasks(tasks, state) Filter a task list to those whose IDs are not in state.completed.

Presentation

Symbol Purpose
present_tasks(tasks, state) Format a task list as a markdown table, optionally annotated with completion state.
present_task_detail(task) Format a single task with its full acceptance criteria and metadata.
present_task_result(task, gate_result) Format execution output including quality-gate status and score.
format_progress_bar(completed, total) Render a visual progress indicator for a running pipeline.

Result fields

TaskResult fields you'll inspect most often:

Field Type Meaning
quality_gate_passed bool \| None Gate outcome; None when gates were skipped.
tests_passed bool \| None Test outcome; None when tests were skipped.
gate_score float \| None Numeric quality score from the gate.
severity str (property) Classified severity of the gate result.
error str \| None Error message if the task failed to execute.
cost float Cost attributed to this task.

PipelineResult top-level members:

Member Type Meaning
success bool (property) True only when all tasks executed and passed gates.
summary str (property) Human-readable run summary.
tasks list[TaskResult] Per-task outcomes.
total_cost float Aggregated cost across all tasks.
duration_ms int Wall-clock time for the full run.

Example output

An approval run prints a progress bar, per-task gate results, and a final summary:

[========--] 4/5 tasks

✔ task-1  add-jwt-config          gate: passed  score: 92.0  cost: $0.003
✔ task-2  token-service           gate: passed  score: 87.5  cost: $0.004
✔ task-3  auth-middleware         gate: passed  score: 81.0  cost: $0.005
✔ task-4  wire-routes             gate: passed  score: 78.3  cost: $0.004
  task-5  integration-tests       pending

Pipeline complete: 4/5 tasks executed  total_cost: $0.016  duration: 18402ms