Pipeline Cancellation

What you’ll learn¶

Cancel a running pipeline programmatically with pipeline.cancel()
Inspect cancelled step results and their metadata
Re-run a cancelled pipeline — completed steps are cached, cancelled steps re-execute
How Ctrl+C / SIGTERM signal handling works automatically

Prerequisites: Run vs Submit, Resume and Caching, Error Handling in Practice
Estimated time: 10 minutes
GPU required: No.

In production pipelines, you sometimes need to stop early — bad upstream data, wrong parameters, or resource limits. The framework supports cooperative cancellation: call pipeline.cancel() (or press Ctrl+C), and remaining steps skip immediately with zero execution overhead.

Section	What you’ll see
Programmatic cancellation	`pipeline.cancel()` stops remaining steps
Inspecting cancelled results	Step metadata distinguishes cancelled from skipped
Re-running after cancellation	Completed steps cached, cancelled steps re-execute
Signal handling and scripts	Ctrl+C and SLURM cancellation demos

from __future__ import annotations

from artisan.operations.examples import Wait
from artisan.orchestration import PipelineManager
from artisan.utils import tutorial_setup
from artisan.visualization import inspect_pipeline

env = tutorial_setup("pipeline_cancellation")

Programmatic cancellation¶

Build a 4-step pipeline using Wait (a lightweight operation that sleeps for a configurable duration). Run the first two steps, call pipeline.cancel(), then run the remaining two. The cancelled steps skip immediately with zero execution overhead.

pipeline = PipelineManager.create(
    name="cancellation_demo",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)

step_0 = pipeline.run(
    operation=Wait,
    name="wait_0",
    params={"duration": 0.1},
)
step_1 = pipeline.run(
    operation=Wait,
    name="wait_1",
    params={"duration": 0.1},
)
print(f"Step 0: {step_0.succeeded_count} succeeded")
print(f"Step 1: {step_1.succeeded_count} succeeded")

Steps 0–1 completed normally. Now cancel the pipeline and run two more steps:

pipeline.cancel()

step_2 = pipeline.run(
    operation=Wait,
    name="wait_2",
    params={"duration": 0.1},
)
step_3 = pipeline.run(
    operation=Wait,
    name="wait_3",
    params={"duration": 0.1},
)

result = pipeline.finalize()
print(f"Pipeline: {result['total_steps']} steps, success={result['overall_success']}")

Pipeline finished without error. Steps 2–3 were skipped instantly — no execution, no SLURM jobs, no wasted compute.

Inspecting cancelled results¶

The Error Handling tutorial introduced skip_reason values for empty filter cascades. Cancellation adds a third reason:

`skip_reason`	Meaning
`"empty_inputs"`	First step to receive zero artifacts
`"pipeline_stopped"`	Skipped because an earlier step had empty inputs
`"cancelled"`	Skipped because `cancel()` was called

for step in [step_0, step_1, step_2, step_3]:
    skipped = step.metadata.get("skipped", False)
    if skipped:
        reason = step.metadata["skip_reason"]
        print(f"Step {step.step_number} ({step.step_name}): skipped — {reason}")
    else:
        print(
            f"Step {step.step_number} ({step.step_name}): "
            f"{step.succeeded_count}/{step.total_count} succeeded"
        )

Cancelled steps have success=True with zero counts — cancellation is not a failure. The pipeline’s overall_success stays True, so you can distinguish “cancelled cleanly” from “something broke.”

inspect_pipeline(env.delta_root)

inspect_pipeline shows completed steps from the delta table. Steps cancelled at submit-time (steps 2–3 above) never wrote to delta, so they don’t appear here. Steps cancelled mid-execution do appear with status = "cancelled".

Re-running after cancellation¶

Cancelled steps are not cached. Only steps with status="completed" qualify as cache hits. Re-run the same pipeline with the same parameters: steps 0–1 load from cache, steps 2–3 execute fresh.

env = tutorial_setup("pipeline_cancellation", clean=False)

pipeline2 = PipelineManager.create(
    name="cancellation_rerun",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)

step_0r = pipeline2.run(
    operation=Wait,
    name="wait_0",
    params={"duration": 0.1},
)
step_1r = pipeline2.run(
    operation=Wait,
    name="wait_1",
    params={"duration": 0.1},
)
step_2r = pipeline2.run(
    operation=Wait,
    name="wait_2",
    params={"duration": 0.1},
)
step_3r = pipeline2.run(
    operation=Wait,
    name="wait_3",
    params={"duration": 0.1},
)

result2 = pipeline2.finalize()
print(f"Pipeline: {result2['total_steps']} steps, success={result2['overall_success']}")

Log output shows “CACHED” for steps 0–1 and normal execution for steps 2–3. See Resume and Caching for more on caching.

for step in [step_0r, step_1r, step_2r, step_3r]:
    print(
        f"Step {step.step_number} ({step.step_name}): "
        f"{step.succeeded_count}/{step.total_count} succeeded"
    )

inspect_pipeline(env.delta_root)

Signal handling and interactive cancellation¶

In production, you cancel pipelines with Ctrl+C (SIGINT) or kill (SIGTERM), not by calling cancel() in code. The framework installs signal handlers automatically:

Behavior	Detail
Installation	Signal handlers installed when the first step starts
Restoration	Previous handlers restored in `finalize()`
Jupyter	No-op on non-main threads; use `cancel()` programmatically

Three-press escalation: The signal handler uses an escalating response model. The first Ctrl+C triggers graceful cancellation — the current step drains and remaining steps are skipped. The second Ctrl+C restores Python’s default signal handlers. The third Ctrl+C raises KeyboardInterrupt and force-kills the process. If graceful shutdown feels slow, you can spam Ctrl+C to force exit.

To try interactive cancellation, run the companion script from a terminal:

pixi run python docs/tutorials/execution/cancel_demo.py

The script submits 10 Wait steps that each sleep for 30 seconds. Press Ctrl+C while it’s running — you’ll see completed steps, the cancelled step, and skipped steps in the summary. Press Ctrl+C multiple times to force exit if the graceful shutdown takes too long.

SLURM cancellation¶

On SLURM, Ctrl+C cancels the pipeline — remaining steps are skipped and finalize() returns a clean summary. However, SLURM jobs that are already running on the cluster continue running. They must be cancelled manually:

squeue -u $USER          # find your running jobs
scancel <job_id>         # cancel specific jobs

Auto-scancel is planned but not yet implemented.

Run the SLURM demo script on a cluster:

pixi run python docs/tutorials/execution/cancel_demo_slurm.py

The script submits Wait steps to SLURM. Press Ctrl+C to cancel the pipeline, then check squeue and scancel any remaining jobs.

Summary¶

Concept	API
Programmatic cancel	`pipeline.cancel()`
Signal handling	Automatic — SIGINT/SIGTERM call `cancel()`
Escalation	1st Ctrl+C: graceful, 2nd: restore defaults, 3rd: force kill
Cancelled step metadata	`step.metadata["skip_reason"] == "cancelled"`
Cache behavior	Cancelled steps are not cached — they re-execute
SLURM	Running cluster jobs must be cancelled manually (`scancel`)

Key points:

Cancellation is cooperative — current phase completes, no partial writes
Cancelled steps have success=True — cancellation is not failure
Spam Ctrl+C to force exit if graceful shutdown is too slow
Use inspect_pipeline for persisted state, iterate step results for in-memory state

Next steps¶

Resume and Caching — how caching works and why cancelled steps re-execute
Error Handling in Practice — runtime failures, failure logs, and FailurePolicy
Error Handling — design rationale for failure containment