Skill: Resume Pipeline

Purpose

Resume an interrupted analysis pipeline by reading

working/pipeline_state.json

When to Use

Invoke as

/resume-pipeline

• A previous analysis session was interrupted (context limit, user break, connection issue)
• The user wants to continue an analysis started in a prior conversation
• Pipeline state file exists from a partially completed run
• A pipeline failed and the underlying issue has been fixed

Instructions

Step 1: Locate pipeline state (per-run directory aware)

Search for the most recent pipeline state in this order:

1. Per-run directory (preferred): Check

   working/latest/pipeline_state.json

   (symlink to latest run). If found, set

   RUN_DIR

   from the symlink target and proceed to Step 2.
2. Specific run: If the user passed a run ID (e.g.,

   /resume-pipeline 2026-02-23_acme-analytics_why-revenue-dropped

   ), look in

   working/runs/{id}/pipeline_state.json

   . Set

   RUN_DIR

   accordingly.
3. Legacy location: Check

   working/pipeline_state.json

   (pre-run-directory pipelines). If found, read it and proceed to Step 2 without a

   RUN_DIR

   .
4. No state found: Fall back to artifact scanning (Step 1b).

Pipeline state fields to extract (V2):

• run_id

  -- identifies this run

• run_dir

  -- per-run directory path (may be absent for legacy runs)

• dataset

  -- active dataset

• question

  -- the business question

• status

  --

  running

  ,

  paused

  , or

  failed

• agents

  -- map of agent-name to agent state (status, output_file, timestamps)

Step 1a: V1-to-V2 state migration

After loading the state file and before any processing, check whether the state uses the V1 (step-number keyed) format and migrate it to V2 if needed.

from helpers.pipeline_state import detect_schema_version, migrate_v1_to_v2

if detect_schema_version(state) < 2:
    # Resolve dataset from active.yaml or fall back to "unknown"
    dataset = state.get("dataset") or resolve_active_dataset() or "unknown"
    state = migrate_v1_to_v2(state, dataset=dataset)
    # Write migrated state back to disk (same location it was read from)
    write_pipeline_state(state_path, state)
    print("Migrated pipeline state from V1 -> V2 format")

Migration details (handled by

helpers/pipeline_state.py

• pipeline_id

  (ISO timestamp) ->

  started_at

  ; generate

  run_id

  from date + dataset + question slug

• steps.{n}.agent

  keys ->

  agents.{agent_name}

  keys

• steps.{n}.output_files[0]

  ->

  agents.{name}.output_file

  (take first)
• Status values are preserved as-is (compatible between V1 and V2)
• Adds

  schema_version: 2

  and

  updated_at

  set to current time
• If any V1 step had

  status: running

  , it becomes

  paused

  at the pipeline level (was interrupted)

After migration, continue with the V2 fields listed above.

Step 1b: Artifact-based fallback (no pipeline_state.json)

If no state file exists, scan

working/

outputs/

question_brief_*.md

outputs/

hypothesis_doc_*.md

outputs/

data_inventory_*.md

outputs/

tieout_*.md

working/

analysis_report_*.md

outputs/

investigation_*.md

working/

validation_*.md

outputs/

sizing_*.md

working/

storyboard_*.md

working/

coherence_review_*.md

working/

charts/*.png

outputs/

design_review_*.md

working/

narrative_*.md

outputs/

deck_*.md

outputs/

Walk the list top to bottom. If an artifact exists and looks complete (not empty, no "NEEDS REVISION" markers), mark that agent as completed. Reconstruct a pipeline_state.json from this scan.

Step 2: Compute READY set from DAG

1. Read

   agents/registry.yaml

   to build the dependency graph
2. For each agent in the registry, check

   state["agents"][agent_name]["status"]

   :
   • If status is

     complete

     ,

     skipped

     , or

     degraded

     → leave it
   • If status is

     failed

     → reset to

     pending

     (will be retried)
   • If status is

     in_progress

     or

     running

     → reset to

     pending

     (was interrupted)
3. Compute READY agents: those with

   status: pending

   whose every dependency is

   complete

Step 3: Build context summary

Read each completed agent's output files and extract a brief summary:

• From question brief: the framed question and decision context
• From analysis report: key findings (top 3)
• From storyboard: narrative beats and visual plan
• From validation: confidence grade

Compile into a context block for the resumed session.

Step 4: Present resume plan

Display:

Resuming pipeline {run_id}

Completed agents: {count}
  - {agent_name}: {one-line summary from outputs}
  - ...

Failed/interrupted agents (will retry): {count}
  - {agent_name}: {error or "interrupted"}

Next READY agents: {list}

Resume execution?

Step 5: Resume via DAG walker

On confirmation:

1. Update pipeline_state.json: set

   status: running

   , reset failed/running to pending
2. Hand off to the DAG walker in run-pipeline skill (Phase 2)
3. The walker will pick up from the READY set and continue tier-by-tier
4. All existing completed outputs are preserved — only pending agents execute

Special Cases

• Storyboard with "NEEDS ADDITIONS": Mark story-architect as

  pending

  , not completed
• Partial chart generation: Count generated charts vs storyboard beats. If incomplete, mark chart-maker as

  pending

• Source tie-out FAIL: Mark as

  failed

  . User must investigate before resuming
• Stale data (>24h gap): Warn that underlying data may have changed since the original run

Limitations

• Context gap: Resuming restores artifacts but not conversational reasoning. The resumed analysis may be slightly less coherent than a single-session run.
• No partial step recovery: If an agent was interrupted mid-execution, the entire agent must re-run.
• Pipeline state is authoritative: If pipeline_state.json and artifacts disagree, trust pipeline_state.json.