Resuming Runs¶

Guide to continuing interrupted pipeline executions and reusing previous results.

Run Resumption Basics¶

The pipeline supports resuming execution from previous runs, avoiding redundant computation.

When Resumption Helps¶

Resumption is beneficial when:

Pipeline Interruptions: Execution stopped due to errors, timeouts, or system issues
Incremental Updates: Adding new documents to an existing graph
Configuration Changes: Modifying settings for specific phases
Resource Management: Pausing long-running processes

How Resumption Works¶

The pipeline automatically detects:

Which phases completed successfully in previous runs
Whether artifacts remain valid for reuse
Where execution should resume

Identifying Previous Runs¶

Locate runs available for resumption:

# List recent runs
from pipeline.runs.persistence import JsonRunManifestStore

manifest_store = JsonRunManifestStore("./.pipeline_runs")
recent_runs = manifest_store.list_recent_manifests(limit=10)

for manifest in recent_runs:
    print(f"Run ID: {manifest.run_id}")
    print(f"Started: {manifest.started_at}")
    print(f"Status: {manifest.status}")
    print("---")

Resuming Execution¶

Continue from a previous run using the same pipeline instance:

# Resume from last completed phase
result = await pipeline.resume_from_run(
    run_id="run-abc123",
    input=PipelineInput(source_paths=["document.tex"])
)

# Resume from a specific phase
result = await pipeline.resume_from_run(
    run_id="run-abc123",
    input=PipelineInput(source_paths=["document.tex"]),
    from_phase=3  # Start from Phase 3
)

Artifact Reuse¶

The system automatically reuses valid artifacts from previous runs.

Reuse Conditions¶

Artifacts are reused when:

Input fingerprints match exactly
Configuration parameters are unchanged
No invalidated dependencies exist
Artifact files remain accessible

Monitoring Reuse¶

Track which artifacts are being reused:

# The resume operation returns information about reuse decisions
result = await pipeline.resume_from_run(run_id, input)

print("Reused phases:", result.manifest.reused_phase_ordinals)
print("Invalidated phases:", result.manifest.invalidated_phase_ordinals)
print("Parent run:", result.manifest.parent_run_id)

Configuration Changes and Invalidation¶

Understand how configuration changes affect resumption.

Compatible Changes¶

These modifications typically allow artifact reuse:

Adding new input documents (extends rather than modifies)
Adjusting confidence thresholds for subsequent filtering
Changing output formatting options
Updating non-functional configuration

Invalidating Changes¶

These modifications require recomputation:

Changing LLM models or prompts
Modifying chunking strategies
Altering entity extraction parameters
Updating relationship extraction logic
Changing graph schema definitions

Smart Invalidation¶

The system uses fingerprinting to detect changes:

# Configuration changes automatically invalidate affected phases
# based on content-addressed fingerprints
cfg_v1 = PipelineConfig(phase2=Phase2Config(ner_confidence_threshold=0.8))
cfg_v2 = PipelineConfig(phase2=Phase2Config(ner_confidence_threshold=0.9))

# Different fingerprints trigger invalidation
print(stable_fingerprint(cfg_v1) != stable_fingerprint(cfg_v2))  # True

Incremental Processing¶

Add new documents to existing runs efficiently.

Extending Input Sets¶

Process additional documents while reusing previous results:

# Original run processed documents A and B
original_input = PipelineInput(source_paths=["docA.pdf", "docB.pdf"])

# New run adds document C while reusing A and B results
extended_input = PipelineInput(source_paths=["docA.pdf", "docB.pdf", "docC.pdf"])

result = await pipeline.resume_from_run(
    run_id="original-run-id",
    input=extended_input
)

Conditional Processing¶

Control which documents trigger reprocessing:

from pipeline.config import ExecutionConfig

cfg = PipelineConfig(
    execution=ExecutionConfig(
        allow_phase_reuse=True,
        reuse_policy="conservative"  # Only reuse when certain safe
    )
)

Advanced Resumption Patterns¶

Handle complex resumption scenarios.

Selective Phase Re-execution¶

Force re-execution of specific phases:

# Even if artifacts exist, force Phase 3 re-execution
result = await pipeline.run_from_phase(
    phase_number=3,
    input=original_input
)

Hybrid Approaches¶

Combine manual and automatic resumption:

# Manually process Phase 1 with new settings
phase1_result = await phase1_runner.run(custom_input)

# Resume automatic processing from Phase 2
# using manually generated Phase 1 artifacts
result = await pipeline.resume_from_run(
    run_id="manual-phase1-run",
    input=original_input,
    from_phase=2
)

Troubleshooting Resumption¶

Address common resumption issues.

Missing Artifacts¶

Handle cases where expected artifacts are unavailable:

# System gracefully falls back to re-execution when artifacts are missing
# Check logs for "artifact_missing" events

Fingerprint Mismatches¶

Diagnose configuration inconsistency issues:

# Look for fingerprint mismatch warnings in logs
# These indicate why artifacts weren't reused

Dependency Invalidation¶

Understand cascading invalidation effects:

# If Phase 2 artifacts are invalidated,
# subsequent phases (3, 4, 5) typically also invalidate
# unless specifically configured otherwise

Performance Considerations¶

Optimize resumption for efficiency.

Cache Management¶

Control artifact caching behavior:

cfg = PipelineConfig(
    execution=ExecutionConfig(
        artifact_cache_size=2000,    # Increase cache size
        cache_eviction_policy="lru"  # Least Recently Used eviction
    )
)

Parallel Resumption¶

Resume multiple independent runs simultaneously:

async def resume_independent_run(run_id, input_data):
    return await pipeline.resume_from_run(run_id, input_data)

# Process multiple runs concurrently
results = await asyncio.gather(*[
    resume_independent_run(run_id, input_data) 
    for run_id in compatible_runs
])

Best Practices¶

Follow recommended patterns for reliable resumption.

Run Identification¶

Use descriptive run identifiers:

# Include meaningful context in run IDs
run_id = f"philosophy-corpus-{datetime.now().strftime('%Y%m%d')}-v1"

State Management¶

Keep runs organized and manageable:

# Clean up old runs periodically
manifest_store.prune_old_manifests(retention_days=30)

Validation¶

Verify resumption success:

# Confirm expected reuse occurred
assert len(result.manifest.reused_phase_ordinals) > 0

# Check result integrity
assert result.manifest.status == "completed"

Monitoring and Logging¶

Track resumption behavior effectively.

Resumption Events¶

Monitor key resumption activities:

@emitter.on("run.resumed")
def on_run_resumed(event):
    print(f"Resumed run {event['run_id']} from phase {event['from_phase']}")

@emitter.on("artifact.reused")
def on_artifact_reused(event):
    print(f"Reused artifact: {event['artifact_id']}")

Performance Metrics¶

Measure resumption efficiency:

# Compare execution time with and without resumption
# Track percentage of work avoided through reuse
reuse_ratio = len(reused_artifacts) / len(total_artifacts)