pipeline

Failure policy for a pipeline phase.

Controls what happens when a phase's handler raises an exception. Default behavior (on_fail="stop") matches the original fail-fast semantics.

A single captured output line from a phase handler.

A named pipeline phase.

Runtime status of a single phase.

Observable state for a running pipeline.

UI state for the pipeline detail TUI (wraps PipelineState).

This is a view-layer wrapper — the pipeline reducer remains pure. The TUI reducer wraps it and handles @@KEY actions for navigation.

Raised when a pipeline dependency graph contains a cycle.

Declarative build pipeline that executes through the Store.

Usage::

pipeline = Pipeline(
    "build",
    Phase("discover", handler=discover),
    Phase("parse", handler=parse, depends_on=("discover",)),
    Phase("render", handler=render, depends_on=("parse",), parallel=True),
    Phase("assets", handler=assets, depends_on=("parse",), parallel=True),
    Phase("write", handler=write, depends_on=("render", "assets")),
)

# Extend with >>
pipeline = pipeline >> Phase("health", handler=check)

The pipeline generates a saga and reducer that work with the Store for observable, testable execution.

RaisesCycleErrorat construction if the dependency graph contains a cycle.

Proxy that routes writes to a per-phase buffer when capture is active.

When_phase_buffercontextvar holds a list, writes are buffered there. Otherwise writes pass through to the original stream. Thread-safe because contextvars are per-thread.

Publish a PipelineState for the milo://pipeline/timeline resource.

Call this from a Store listener to keep the resource up to date::

store.subscribe(lambda: set_active_pipeline(store.state))

Serialize a PipelineState to a timeline dict for MCP.

Returns structured JSON matching the milo://pipeline/timeline schema::

{
    "pipeline": "build",
    "status": "completed",
    "elapsed": 2.34,
    "progress": 1.0,
    "phases": [
        {"name": "discover", "status": "completed", "elapsed": 0.52, "attempt": 1, "log_count": 42},
        ...
    ]
}

Create a wrapping reducer for the interactive pipeline detail view.

Handles @@KEY actions for cursor navigation, expansion/collapse, log scrolling, and auto-follow. Delegates all other actions to the inner pipeline reducer.

Usage::

reducer = make_detail_reducer(pipeline.build_reducer())

Call a handler with stdout/stderr capture. Returns (result, log_entries).

On exception, the captured logs are attached to the exception as __captured_logs__so the caller can flush them before emitting PHASE_FAILED / PHASE_RETRY / PHASE_SKIPPED.

Run a sequential phase inline in the main saga. Returns True if pipeline should stop.

Create a saga for a single phase (used by All for parallel phases).

When fail_fast is True and the phase fails withon_fail="stop" (the default), the saga re-raises after dispatchingPHASE_FAILED so thatAllcan cancel sibling sagas.