Control output

Capture and echo

Capture is on and echo is off by default. With capture on, stdout and stderr on the result are decoded strings; with it off, they are None. Echo mirrors output to the parent's sinks and can be set per stream. The two are independent: a stream can be captured silently, or echoed without being kept.

Echo limits each mirrored line to 64 KiB by default, counting its truncation marker and terminator; captured output is always complete. Set max_echo_line_bytes=None only where unbounded mirroring is safe. CommandResult.relay_fallbacks records any echo that fell back, by stream and closed category and without the output itself; see when an echo destination fails.

Observe each line

RunOutputOptions.on_line receives every decoded line as a LineEvent, even when capture and echo are both off. The callback runs synchronously and should return promptly. Lines keep their order within a stream, not across streams.

import sys

from cuprum import Program, ProgramCatalogue, RunOutputOptions, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="output")
python = sh.make(Program(sys.executable), catalogue=catalogue)
lines = []
result = python("-c", "print('one'); print('two')").run_sync(
    output=RunOutputOptions(capture=False, on_line=lines.append),
)
assert result.ok and result.stdout is None
assert [(line.stream, line.text) for line in lines] == [
    ("stdout", "one"),
    ("stdout", "two"),
]

Stream lines as they arrive

When a caller must react before the command finishes, iterate SafeCmd.lines, an async iterator with backpressure. Use it as an async context manager whenever the loop can stop early: leaving the block closes the stream and tears the child down, as a cancelled run() would. Merely breaking out of the loop does not.

import asyncio
import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="follow")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python(
    "-c", "import time\nfor n in range(100):\n print(n, flush=True); time.sleep(0.01)"
)


async def first_line() -> str:
    async with command.lines() as stream:
        async for event in stream:
            # Leaving the block closes the stream and stops the child.
            return event.text
    raise AssertionError("the child should print at least one line")


assert asyncio.run(first_line()) == "0"

After a loop that runs to the end, stream.result holds the same CommandResult that run() would have returned. Pipelines have no lines(); on_line on a pipeline run observes the final stage's stdout and every stage's stderr, since interior stdout feeds the next stage instead.

When an echo destination fails

A sink can fail while echoing: a text-only console may be unable to encode the child's output, or the reader at the other end of a pipe may go away. A sink that cannot encode is always handled: echo stops for that stream while capture completes. A sink whose destination has closed is handled only when the run opts in, with BrokenPipePolicy.BEST_EFFORT on RunOutputOptions:

import io
import sys

from cuprum import (
    BrokenPipePolicy,
    ExecutionContext,
    Program,
    ProgramCatalogue,
    RunOutputOptions,
    sh,
)


class ClosedReader(io.TextIOBase):
    """An echo destination whose reader has gone away."""

    def write(self, text: str) -> int:
        raise BrokenPipeError


catalogue = ProgramCatalogue.from_programs(sys.executable, name="broken-pipe")
command = sh.make(Program(sys.executable), catalogue=catalogue)("-c", "print('hello')")
result = command.run_sync(
    output=RunOutputOptions(echo=True, broken_pipe_policy=BrokenPipePolicy.BEST_EFFORT),
    context=ExecutionContext(stdout_sink=ClosedReader()),
)
assert result.ok and result.stdout == "hello\n"
assert [(f.stream, f.error_category) for f in result.relay_fallbacks] == [
    ("stdout", "broken_pipe")
]

Under best-effort, the BrokenPipeError from the sink disables echo for the affected stream only; capture, line observation, and child reaping continue, and the result's relay_fallbacks records one entry naming the stream and the broken_pipe category, never the output. The default, STRICT, propagates the error and aborts the run. Only BrokenPipeError is affected; any other sink OSError propagates under both policies.

A heartbeat for a quiet child

RunOutputOptions(idle_after=30.0) writes a notice to the stderr sink after 30 seconds without output, then once per further silent interval; the default is off. The notice is never mixed into captured output. on_idle(total, idle) replaces it. A heartbeat reports silence, not a deadlock, and changes neither the deadline nor the exit status. Keep the callback and any sink it writes to prompt: both run on the command's event loop.

import sys

from cuprum import Program, ProgramCatalogue, RunOutputOptions, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="idle")
command = sh.make(Program(sys.executable), catalogue=catalogue)("-c", "print('done')")
result = command.run_sync(output=RunOutputOptions(idle_after=30.0))
assert result.ok and result.stdout == "done\n"

Send a stream to a file or the parent

Capture keeps output in memory for the caller to read. When output should land somewhere else, such as a log file or the parent's own terminal, redirect the stream instead. RunOutputOptions.stdout and .stderr accept a StdioTarget, and the four kinds differ in who owns the destination.

When a stream is left unset, Cuprum uses a pipe if something needs to read it and discards the output otherwise, so inherit() is the way to let an unobserved stream reach the parent's terminal. Any target other than pipe() counts as a redirect, and capture is on by default, so pass capture=False when redirecting stdout:

import sys
import tempfile
from pathlib import Path

from cuprum import Program, ProgramCatalogue, RunOutputOptions, StdioTarget, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="redirect")
python = sh.make(Program(sys.executable), catalogue=catalogue)
log = Path(tempfile.mkdtemp()) / "out.log"
result = python("-c", "print('redirected')").run_sync(
    output=RunOutputOptions(capture=False, stdout=StdioTarget.path(log)),
)
assert result.stdout is None, "a redirected stream is not captured"
assert log.read_text(encoding="utf-8") == "redirected\n"

A few combinations are contradictory, and Cuprum rejects them with ValueError before the child starts rather than silently choosing one side:

• Capture or echo with a redirected stream, including inherit(). Both read from a pipe, and a redirected stream has none. Line observation and the heartbeat are not rejected, but they see nothing from a redirected stream.
• One path for both stdout and stderr. Each open starts at offset 0, so the two streams would overwrite each other. Use two paths, or one fd() target the caller manages.
• A redirected stdout with lines(), or either stream with start(). Both read the child's output in the parent. Use run() or run_sync() instead.
• An inherited stdin alongside a stdin source. RunOutputOptions(stdin=StdioTarget.inherit()) with stdin=StdinInput(...) or stdin=StdinStream(...) gives two answers to where the input comes from; this is checked when the run starts. A pipe() stdin target with a source is accepted, because the source is written through that pipe.
• Targets on a Pipeline. Redirect a single command instead.
• path() or fd() on Windows. File and descriptor redirection needs POSIX descriptors; pipe() and inherit() work everywhere.

Reusing one fd() target across several runs shares its file offset, so each run continues where the previous one stopped. RunOutputOptions.stdin accepts only pipe() and inherit(): the input itself is passed to the run call's stdin= argument, as Stream input from a producer describes.

Binary output

Capture decodes by default, so stdout and stderr are strings, and a byte the child wrote that is not valid UTF-8 arrives as U+FFFD. That is right for text and wrong for an archive or an image, because a decoded capture cannot be turned back into the bytes the child wrote. SafeCmd.run_bytes and Pipeline.run_bytes(), with run_bytes_sync() counterparts, are the same runs with capture left undecoded:

import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="binary-output")
python = sh.make(Program(sys.executable), catalogue=catalogue)
payload = bytes(range(256)) + b"\xff\x00\xfe\x80"
result = python(
    "-c", f"import sys; sys.stdout.buffer.write({payload!r})"
).run_bytes_sync()
assert result.ok
assert result.stdout == payload

The byte-mode entry points take the same output, timeout, context, and, for a single command, stdin arguments, and return BytesCommandResult or BytesPipelineResult, which declare the same fields as their text counterparts with stdout and stderr as bytes | None. The mode is explicit: run() never returns bytes and run_bytes() never returns text. A timeout still carries its partial output in the same field, capture=False still yields None, and an external cancellation still re-raises asyncio.CancelledError.

Binary mode covers capture. A caller-supplied on_line is rejected with ValueError before the child is spawned, since a callback carrying decoded text would be a second, contradictory contract for the same stream. Echo, sinks, and the heartbeat are unaffected, and hooks registered with observe still receive decoded lines: undecodable bytes are replaced in the event while the captured payload keeps the child's own bytes. That replacement is scoped to byte mode; a text-mode run under errors="strict" still raises UnicodeDecodeError as before.