NAME

LLM::Data::Pipeline - Generic pipeline framework for sequential data processing with checkpointing

SYNOPSIS

use LLM::Data::Pipeline;
use LLM::Data::Pipeline::Context;
use LLM::Data::Pipeline::Step;
use LLM::Data::Pipeline::Plan;
use LLM::Data::Pipeline::Runner;

class MyStep does LLM::Data::Pipeline::Step {
    method name(--> Str:D) { 'my-step' }
    method description(--> Str:D) { 'Does something useful' }
    method requires(--> List) { ('input',) }
    method optional(--> List) { ('config',) }
    method provides(--> List) { ('output',) }
    method execute(LLM::Data::Pipeline::Context:D $ctx --> Nil) {
        my $input = $ctx.get('input');
        $ctx.set('output', "processed: $input");
    }
}

my LLM::Data::Pipeline::Plan $plan .= new;
$plan.add-step(MyStep.new);

my LLM::Data::Pipeline::Context $ctx .= new;
$ctx.set('input', 'hello');

my LLM::Data::Pipeline::Runner $runner .= new(
    on-event => -> LLM::Data::Pipeline::Event $e {
        say "{$e.seq}\t{$e.kind}";
    }
);
$runner.run($plan, $ctx, :checkpoint-path('checkpoint.json'.IO));

say $ctx.get('output');  # "processed: hello"

# Resume from checkpoint (completed steps arrive as step-skipped events)
my $ctx2 = $runner.resume($plan, 'checkpoint.json'.IO);

DESCRIPTION

LLM::Data::Pipeline provides a domain-agnostic framework for building sequential data processing pipelines with automatic checkpointing and resume capability.

LLM::Data::Pipeline::Step (Role)

Any class can be a pipeline step by doing this role.

method name(--> Str:D) { ... }           # Unique step identifier
method description(--> Str:D) { ... }    # Human-readable description
method requires(--> List) { () }         # Context keys this step must have
method optional(--> List) { () }         # Context keys this step can use
method provides(--> List) { ... }        # Context keys this step writes
method execute(Context:D $ctx --> Nil) { ... }  # Do the work

LLM::Data::Pipeline::Context

String-keyed data bag flowing through the pipeline. Values must be JSON-serializable.

my LLM::Data::Pipeline::Context $ctx .= new;
$ctx.set('key', 'value');
$ctx.get('key');              # 'value'
$ctx.has('key');              # True
$ctx.keys;                   # Sorted list of all keys

# Serialization (for checkpointing)
my %snap = $ctx.snapshot;    # Deep-copy Hash via JSON round-trip
my $restored = LLM::Data::Pipeline::Context.from-snapshot(%snap);

LLM::Data::Pipeline::Plan

Ordered list of steps with dependency validation.

my LLM::Data::Pipeline::Plan $plan .= new;
$plan.add-step(StepA.new);
$plan.add-step(StepB.new);
$plan.steps;                          # List of steps in order
$plan.validate($ctx);                 # Dies with diagnostic if deps unsatisfied

Validation walks steps in order, tracking available keys from each step's provides. Reports all missing keys and duplicate step names in one error.

LLM::Data::Pipeline::Runner

Executes a Plan against a Context with checkpointing, surfacing everything observable as a typed event stream.

my LLM::Data::Pipeline::Runner $runner .= new(
    on-event => -> LLM::Data::Pipeline::Event $e {
        given $e {
            when LLM::Data::Pipeline::Event::StepStarted {
                say "→ {$e.step} ({$e.ordinal}/{$e.total})";
            }
            when LLM::Data::Pipeline::Event::StepCompleted {
                say "✓ {$e.step} in {$e.duration.round(0.01)}s";
            }
            when LLM::Data::Pipeline::Event::RunCompleted {
                say "done: {$e.steps-run} run, {$e.steps-skipped} skipped";
            }
        }
    }
);

# Run from beginning (validates first)
my $ctx = $runner.run($plan, $ctx, :checkpoint-path($path));

# Resume from checkpoint (completed steps arrive as step-skipped events)
my $ctx = $runner.resume($plan, $checkpoint-path);

Checkpoints are JSON files written after each step, containing completed step names and the full context snapshot. Writes use temp file + rename for atomicity.

Observing a run

&.on-event is the primary hook: a synchronous callback invoked once per event on the run thread, in strict seq order. Handler exceptions are shielded (logged via note) so a broken observer can neither abort the run nor cause later events to be dropped.

A secondary method events(-- Supply)> mirrors the same stream for reactive consumers. Taps run synchronously on the run thread, so tap before calling run/resume; the Supply receives done when the run finishes, on normal completion and on failure (run-failed is emitted first, then the exception is rethrown, then done fires). Bridge with .Channel for thread isolation.

$runner.events.tap(
    -> LLM::Data::Pipeline::Event $e { say to-json($e.to-hash) },
    done => { say 'stream closed' },
);
$runner.run($plan, $ctx);   # tap BEFORE running

The legacy &.on-step string callback ('skip'/'start'/'complete') is retained as a thin shim over the event stream, firing identically to prior releases. It is deprecated and will be removed at 1.0 — migrate to &.on-event.

Ordering guarantees

• Events form a per-run total order by seq (contiguous, starting at 1, no gaps).

• step-started precedes its item events, which precede that step's step-completed/step-failed (item events arrive from 0.4.0).

• Cross-item ordering is defined only by seq.

run-id is stable within a single run/resume call and differs between calls. A validation failure throws before run-started is emitted.

Step retries

Each plain step runs under the &.step-retry RetryPolicy. The default is a single attempt, which preserves the pre-0.3.0 contract exactly — a failing step emits run-failed and rethrows its original exception (so domain types like X::LLM::Data::Inference::Exhausted keep flowing out).

# Opt into retries for every plain step:
my $runner = LLM::Data::Pipeline::Runner.new(
    step-retry => LLM::Data::Pipeline::RetryPolicy.new(:max-attempts(3)),
);

With max-attempts above 1, each failed attempt emits step-failed (will-retry True, retry-delay = the backoff), the Runner waits (via the injectable &.schedule-after), then emits step-retry. When the budget is spent it emits a final step-failed (will-retry False), then run-failed, then throws X::LLM::Data::Pipeline::StepExhausted carrying the step name, attempt count, and the last error.

Retry implies idempotency: a failed attempt's Context mutations are not rolled back. A step that opts into retries must tolerate its own partial writes; one that cannot must keep max-attempts at 1.

&.now and &.schedule-after are injectable so tests can drive a virtual clock with zero real sleeps.

Runner internals

You do not need this section to use the Runner, but if you are debugging a failure, supervising a long pipeline, or implementing Step::Items, these are the rules the engine actually enforces.

Step drive loop

run validates the Plan, then walks the steps. resume checks the checkpoint file exists, loads it (v2, or legacy v1 with no version key; a v2 checkpoint's run-id is adopted so DLQ/checkpoint lineage stays stable across resumes), validates the plan against the rehydrated Context, optionally applies retry-dead to reopen dead items, and walks.

For each step: steps already recorded in completed-steps emit step-skipped and are not re-run. Every other step emits step-started, executes, is pushed onto the completed list, and — if a checkpoint path is configured — writes a step-boundary checkpoint (trigger 'step') before step-completed is emitted. Only after a Step::Items step completes does the fail-on-dead check run: a step with dead items and fail-on-dead set throws X::LLM::Data::Pipeline::ItemsDead (after step-completed, after the checkpoint). Any exception propagates out of the loop as run-failed — except Cancelled, which rethrows without a second terminal event because it is the caller's intent, not a failure.

Plain-step retry contract

With the default budget of one attempt a failing step's original exception propagates unchanged. With more, each failed attempt emits step-failed (will-retry True), the Runner backs off via the injectable &.schedule-after (blocking the run thread — correct, plain steps have no worker threads), emits step-retry, and re-runs. The spent budget emits a final step-failed (will-retry False) and throws X::LLM::Data::Pipeline::StepExhausted. A failed attempt's Context mutations are never rolled back, so retries require idempotency or partial-write tolerance.

Item coordinator model

• items($ctx) is called once per activation and materialized. Each item gets a key via item-key; duplicates abort the run. A fingerprint (count, SHA-256 of the sorted keys' JSON, SHA-256 of the items' JSON) is stored in the checkpoint; on resume it is recomputed and any mismatch throws X::LLM::Data::Pipeline::CheckpointDrift.

• The coordinator (the run thread) owns all mutable state: done results, attempts, dead keys, attempt history, telemetry, in-flight set, partials. Workers are start blocks pulling from a work Channel; they run process-item under try and message results back through an inbox Channel, touching no state, files, or events. Every state transition — and therefore event seq, checkpoint, and DLQ writes — happens in one place: the inbox loop. Single-writer by construction, no locks.

• The Context is frozen and snapshotted once before the first worker starts; every per-item, dead-letter, and cancel checkpoint reuses that snapshot.

• Dispatch is throttled to at most degree items in flight, so the work Channel never buffers more than degree jobs and cancellation only has to drain the in-flight ones.

A worker can never strand its key: the whole job body is wrapped so a result is always sent, even if the engine itself throws. A failure's item-retryable advice is probed worker-side (the exception object never crosses the Channel; a plain Bool is shipped), duck-typed so an exception that never heard of the contract reads as retryable, and an accessor that throws reads as retryable — advice must never dead-letter an item by malfunctioning.

Checkpoint v2 schema

Written as v2 always, atomically (temp file + rename), with sorted keys:

{
  "version": 2,
  "run-id": "…stable across resumes…",
  "completed-steps": ["step-a", "step-b"],
  "context": { "…full Context snapshot…": null },
  "step-state": {
    "step-name": {
      "fingerprint": { "count": 100, "keys-sha256": "…", "items-sha256": "…" },
      "started-at": "2026-08-21T12:00:00Z",
      "done":     { "item-key": true },
      "results":  { "item-key": "…item result value…" },
      "attempts": { "item-key": 3 },
      "dead":     { "item-key": true },
      "partials": { "item-key": { "…saved sub-unit state…": null } }
    }
  },
  "updated-at": "2026-08-21T12:34:56Z"
}

Once a step completes, results (and partials) are pruned from its state — a completed step's checkpoint record shrinks to keys and counters, so a finished stage never bloats the checkpoint.

DLQ schema

The dead-letter journal sits beside the checkpoint as JSONL: foo.checkpoint.json → foo.dlq.jsonl; any other checkpoint name keeps its basename minus the extension plus .dlq.jsonl. No checkpoint path ⇒ no DLQ. Each record:

The write order on death is DLQ first, then checkpoint (forced — dead-letter writes are never coalesced away). A DLQ append failure is deliberately fatal: the journal is the one artifact this feature exists to produce.

resume(:retry-dead) requeues every dead item: the digest is re-verified against the journal's dead record (CheckpointDrift on mismatch; the checkpoint, not the journal, is authoritative for control flow), the key is cleared from dead/attempts/partials, a requeued record is appended, and item-requeued is emitted. Reopening a completed step also invalidates every later completed step (they consumed its provides), while rehydrating the reopened step's successes from its reserved Context key — only the formerly-dead items re-run.

Failure modes

LLM::Data::Pipeline::RetryPolicy

Exponential-backoff retry budget: delay-for(attempt) = min(max-delay, base-delay * 2 ** (attempt - 1)) + rand * jitter (1-based attempt). exhausted($attempts) reports when the budget is spent. Defaults: max-attempts 3, base-delay 5s, max-delay 120s, jitter 0.5s — deliberately slower than the inference layer's per-call backoff, since it is the outer retry ring above Task's own fast fallback chain.

LLM::Data::Pipeline::Exceptions

X::LLM::Data::Pipeline::StepExhausted (step, attempts, last-error) is thrown when a step's retry budget is spent. X::LLM::Data::Pipeline::Cancelled, ItemsDead, and CheckpointDrift are declared now and thrown from 0.4.0.

LLM::Data::Pipeline::Event

Every observable moment in a run is an immutable object composing the LLM::Data::Pipeline::Event role, which supplies three envelope fields — UInt $.seq (per-run, from 1), Instant $.at, Str $.run-id — plus kind (a kebab-case discriminator) and to-hash (a JSON-safe Hash; at serializes to a POSIX epoch-seconds number).

use LLM::Data::Pipeline::Event;
use JSON::Fast;

# Pattern-match on class, or switch on the kind string:
given $e {
    when LLM::Data::Pipeline::Event::RunStarted   { ... }
    when LLM::Data::Pipeline::Event::StepCompleted { ... }
}

say to-json($e.to-hash);   # JSON-safe log line, round-trips via from-json

The full taxonomy is declared now; the Stage-2 subset is emitted by this release. Reserved kinds let consumers pattern-match today and start receiving events in the noted release without a breaking change.

An item step emits one progress at activation (from 0.5.1), before its first item-started, so a progress bar can show 0/N for a fresh stage and done/N for a resumed one immediately instead of staying blank until the stage's first item finishes.

LLM::Data::Pipeline::Step::Items

A step whose body is a set of items processed in parallel by the Runner. Instead of execute, implement items (deterministic, JSON-safe), process-item (worker thread; no Context mutation; at-least-once), and finalize (coordinator thread; assembles provides from the reserved "{name}/items" key). Per-item retries dead-letter to a crash-safe JSONL journal beside the checkpoint; a run can be resumed, cancelled, or resume(:retry-dead)'d. See the Runner internals section for the coordinator model, checkpoint v2, DLQ schema, and failure-mode table.

use LLM::Data::Pipeline::Step::Items;

class TagChunks does LLM::Data::Pipeline::Step::Items {
    method name(--> Str:D) { 'tag-chunks' }
    method description(--> Str:D) { 'Tag every chunk' }
    method provides(--> List) { ('tag-chunks/items', 'tags') }
    method items($ctx --> Iterable) { $ctx.get('chunks').list }
    method item-key(Int:D $i, $chunk --> Str:D) { $chunk<id> }
    method process-item($ctx, $chunk, Str:D $key --> Any) { tag-one($chunk) }
    method finalize($ctx --> Nil) {
        $ctx.set('tags', $ctx.get('tag-chunks/items').values.flat.unique.sort.list);
    }
}

my $runner = LLM::Data::Pipeline::Runner.new(:degree(8));
$runner.run($plan, $ctx, :checkpoint-path('run.checkpoint.json'.IO));
# Dead items are journaled to run.dlq.jsonl; retry them once fixed:
$runner.resume($plan, 'run.checkpoint.json'.IO, :retry-dead);

Resumable items (0.6.0)

An item that is a run of sequential sub-units — twenty model calls to rewrite twenty turns of one chunk — can persist its progress as it goes and be handed it back on the next attempt, so a failure on sub-unit 19 no longer throws eighteen finished ones away. Opt in with two lines: mint $runner.partial-sink(:$step, :$key) and call it with a JSON-safe Hash after each completed sub-unit; read :%partial (%() = start fresh) in process-item.

method process-item($ctx, $chunk, Str:D $key, :%partial --> Any) {
    my &save   = $runner.partial-sink(:step(self.name), :$key);
    my @turns  = (%partial<turns> // []).list.Array;
    my Int $at = (%partial<next>  // 0).Int;
    for @($chunk<beats>)[$at ..^ *] -> $beat {
        @turns.push(write-turn($beat));
        save({ turns => @turns, next => ++$at });
    }
    @turns;
}

Partials are state, not events: they ride the checkpoint (step-state.partials, trigger 'partial', coalesced by checkpoint-every) and never the event stream or the DLQ. They are cleared when the item goes terminal and kept across a cancellation, which is what makes a cancelled run resume mid-item.

Operational recipes

Three entry points, three failure lifetimes:

• run — one shot. Use it when the pipeline is short and you want failures in your face immediately. Exceptions propagate unchanged.

• resume — restart from a checkpoint. Use it after an interrupt, a process death, or a deliberate shutdown. Completed steps replay as step-skipped; an in-progress item step picks up from the checkpoint, mid-item if partials were saved.

• run-until-done — stay alive across transient failures. Use it for long or unattended pipelines. Attempt 1 resumes an existing checkpoint (so it composes with an external re-invoker after a crash), else runs; transient failures back off (run-retry event) and resume, while validation / cancellation / drift rethrow immediately. Dead items with :retry-dead are requeued on each attempt; without it ItemsDead is unretryable — a rethrow would recur identically, since nothing about the dead items changed.

In-process retries do not replace an external supervisor; they complement one. run-until-done retries failures inside a live process — a transient 502 from the API, an exhausted step budget. The supervisor (cron, systemd, CI, or just re-running the command) handles the process itself dying: because a checkpoint is written after every step and every item terminal, the next invocation resumes from the latest state with no special-mode flag. If the failure mode is "the machine restarted", rely on the supervisor; if it is "the wire glitched for thirty seconds", rely on run-until-done.

DLQ triage

Dead items are journaled, not thrown away: each one lands in the .dlq.jsonl sibling as a dead record with its full attempt history and (when the step provided it) an inference summary. Read it with JSONL::Reader, group by error.exception / error.message, and the decision tree is short:

• Transient (rate limits, DNS, timeouts, spot kills): fix the environment, then resume(:retry-dead). Only the dead items re-run.

• Poison item (one malformed chunk, a schema rejection): fix the item at its source before retrying — but note the digest check compares the item re-materialized from items() against the journal's record, so edited content will CheckpointDrift (deliberately: a re-run must be against the item that died, or nothing was verified). retry-dead re-runs items exactly as they were dead-lettered; fix the code that handles them or accept the dead keys (they are listed in "{name}/dead") and handle them downstream.

• Systematic (the whole step failed one way): do not retry-dead — the same items die again. Fix the step code; the next run or resume re-executes it (a completed step re-runs after retry-dead invalidation, or start a fresh checkpoint).

:retry-dead is an explicit, deliberate act — nudge it per attempt via run-until-done(:retry-dead) if the deaths are infra-shaped, or withhold it while triaging. The digest verification turns "the inputs changed upstream" into a loud CheckpointDrift instead of a silent replay against new data.

The retry-dead decision guide

Note the invalidation cascade: re-opening a completed step removes it and every later step from completed-steps, because they consumed its provides. Downstream work re-runs too — that is the honest cost of re-admitting a finished stage's dead items; budget for it.

Checkpoint sizing

• checkpoint-every / checkpoint-interval coalesce per-item checkpoints (defaults: every item, no time floor). Dead-letter, step-boundary, and cancel writes are never coalesced away.

• results are pruned once a step completes, so finished stages shrink to counters; the long pole is always the current step's accumulated results and partials.

• Partials are whole Hashes per item. Saving a running transcript of a 20-turn chunk 20 times grows the checkpoint quadratically-ish; save the prefix structure that lets you re-derive the rest, not a duplicate of everything already produced.

run-until-done

Supervise a pipeline to completion across whole-run retries — the outer ring above per-item and per-step retries. Attempt 1 resumes an existing checkpoint (so it composes with an external re-invoker after a crash), else runs; transient failures back off (run-retry event) and resume, while validation / cancellation / drift rethrow immediately. See the Operational recipes section for the full contract and the operational recipes (in-process retries vs external supervisor, DLQ triage, the retry-dead decision guide).

my $ctx = $runner.run-until-done(
    $plan, $seed-ctx,
    checkpoint-path => 'run.checkpoint.json'.IO,
    run-retry       => LLM::Data::Pipeline::RetryPolicy.new(:max-attempts(5), :base-delay(30)),
    retry-dead      => True,   # requeue transiently-dead items on each attempt
);

AUTHOR

Matt Doughty <[email protected]>

COPYRIGHT AND LICENSE

Copyright 2026 Matt Doughty

This library is free software; you can redistribute it and/or modify it under the Artistic License 2.0.