Pipeline, tee, and stage failure · Worked examples

01 — Problem

The job

A job needs two programs in a row — one produces text, the next transforms it — and you want to watch the result in the terminal while still keeping it for the code that follows. In a shell that is a | b | tee and a prayer about which stage failed. In Cuprum it is a Pipeline built with |, run once, returning a result per stage.

The second stage here is the running Python interpreter itself, so the example needs nothing installed beyond echo. Because sys.executable is an absolute path, the catalogue records that exact path.

02 — Before you start

Before you start

Level: Intermediate
Needs: Python 3.12 or later, and echo on PATH (any POSIX system)
Install: Cuprum 0.2.0b2 from PyPI; see Getting started
Tested: 2026-10-19 against Cuprum 0.2.0b2 on Linux, Python 3.12; the failure run was repeated eight times with identical output
Uses: Pipeline via |, RunOutputOptions(echo=True), PipelineResult.stages, failure_index

03 — Code

The code

pipeline_tee.py Complete

import sys

from cuprum import (
    ECHO,
    Program,
    ProgramCatalogue,
    RunOutputOptions,
    scoped,
    sh,
)

PYTHON = Program(sys.executable)
CATALOGUE = ProgramCatalogue.from_programs(ECHO, PYTHON, name="pipeline-tee")

echo = sh.make(ECHO, catalogue=CATALOGUE)
python = sh.make(PYTHON, catalogue=CATALOGUE)

UPPER = "import sys; sys.stdout.write(sys.stdin.read().upper())"

pipeline = echo("-n", "routed through two stages") | python("-c", UPPER)

with scoped(catalogue=CATALOGUE):
    # echo=True tees the final stdout and every stage's stderr to the
    # terminal while the final stdout is still captured.
    result = pipeline.run_sync(output=RunOutputOptions(echo=True))

print()
print(f"ok={result.ok}")
print(f"captured={result.stdout!r}")
for index, stage in enumerate(result.stages):
    print(index, stage.exit_code, repr(stage.stdout))

Verified 2026-10-19 · Cuprum 0.2.0b2 · Python 3.12 · Linux

Fig. 02

Which stream flows where, and what does the caller get back?

Stage 0 echo -n "routed through two stages" stdout is piped to stage 1; nothing is captured here.

passes bytes (stdout → stdin) to the next stage
Stage 1 (final) python -c UPPER Its stdout is the pipeline’s output.

passes bytes (final stdout) to the next stage
Tee echo=True Copies the bytes to the terminal as they arrive; capture keeps them too.

passes captured stdout to the next stage
Result PipelineResult stdout, ok, and one result per stage with its exit code: stages[], failure, failure_index.

A non-final stage exits non-zero while others still run Fail-fast: Cuprum terminates every other still-running stage, upstream and downstream, and records the first failure.
The final stage exits non-zero No fail-fast termination is needed; the failure is still recorded as failure_index.
Any stage writes to stderr With echo=True every stage’s stderr is teed; only the final stage’s stdout is.

Only the final stage’s stdout is captured and echoed. Intermediate stdout feeds the next stage and is reported as None. Source: cuprum/_pipeline_internals.py and docs/users-guide.md at v0.2.0-beta2

04 — Output

Expected output

Expected outputterminal

ROUTED THROUGH TWO STAGES
ok=True
captured='ROUTED THROUGH TWO STAGES'
0 0 None
1 0 'ROUTED THROUGH TWO STAGES'

The first line is the tee: the final stage's output, echoed as it arrived. The print() after the run starts a new line, because the output had none.

Then the result: the pipeline succeeded, its captured stdout is the final stage's output, and each stage has its own exit code. Stage 0's stdout is None because its bytes went to stage 1, not to the caller.

05 — Failure path

When it fails

Now a three-stage pipeline where the middle stage gives up. The first stage writes forever; the second reads one line and exits with status 3; the third copies whatever it receives.

fail_fast.py Failure path

import sys

from cuprum import (
    ECHO,
    Program,
    ProgramCatalogue,
    scoped,
    sh,
)

PYTHON = Program(sys.executable)
CATALOGUE = ProgramCatalogue.from_programs(ECHO, PYTHON, name="pipeline-tee")

python = sh.make(PYTHON, catalogue=CATALOGUE)

# A producer that never stops, a filter that gives up, and a consumer.
PRODUCE = "import sys\nwhile True: sys.stdout.write('tick\\n'); sys.stdout.flush()"
REJECT = "import sys; sys.stdin.readline(); sys.exit(3)"
CONSUME = "import sys; sys.stdout.write(sys.stdin.read())"

pipeline = python("-c", PRODUCE) | python("-c", REJECT) | python("-c", CONSUME)

with scoped(catalogue=CATALOGUE):
    result = pipeline.run_sync()

print(f"ok={result.ok}")
print(f"failure_index={result.failure_index}")
print(f"failure exit={result.failure.exit_code if result.failure else None}")
for index, stage in enumerate(result.stages):
    print(index, stage.exit_code)

Verified 2026-10-19 · Cuprum 0.2.0b2 · Python 3.12 · Linux

Observed outputstdout, Linux

ok=False
failure_index=1
failure exit=3
0 -15
1 3
2 -15

Stage 1 was the first non-final stage to exit non-zero while others were still running, so Cuprum terminated both of them. On POSIX a negative exit code names the signal that ended the process: -15 is SIGTERM. Without fail-fast the producer would have written until something else stopped it.

failure_index names the stage that failed first, and failure is its CommandResult. Later non-zero exits are usually consequences of the first, which is why only the first counts.

06 — Why it is allowed

Why the policy permits it

The pipeline's catalogue lists two programs, echo and this interpreter's absolute path, and the scope's allowlist is that catalogue's. Cuprum checks every stage against the scope before any stage is spawned, so a pipeline with one forbidden stage runs none of them.

Using sys.executable keeps the example self-contained. In real code, catalogue the tools the pipeline actually needs — the allowlist is the list a reviewer reads.