Observe production runs

Choose a channel

Cuprum reports a run through several channels, and each answers a different question. Pick by the question, and keep every callback bounded:

• Lifecycle events. observe delivers an ExecEvent as a command plans, starts, writes, exits, or times out, and one settled event with its terminal outcome. Correlate by exec_id, not pid, which the operating system recycles.
• Lines. on_line or SafeCmd.lines, in Control output, for progress as it happens.
• A heartbeat. idle_after says that no monitored output arrived; it says nothing about whether the child is stuck.
• Control operations. observe_control reports what signal(), terminate(), or kill() on a RunningCommand did to the child, which the exit code alone cannot say.
• Echo fallbacks. observe_echo reports a truncated line, or a sink that could not encode or had closed.
• Stream operations. One aggregate event per completed drain or pipeline transfer, for throughput.
• The Rust pump. Whether each pipeline hop used the native pump or fell back to Python; a decline is a routing decision, not a failure. See the Rust extension.

Only observe hooks can fail a run. Every other channel is fail-open: a hook that raises is logged at WARNING and the operation it was observing carries on. Their events carry only bounded values, never output, argument vectors, or exception text, so the ready-made metrics hooks can label series without unbounded cardinality.

Read the one terminal outcome

Every execution whose plan event began observation ends with exactly one settled event, including a spawn failure and a cancellation, which have no exit event to speak of. A timeout keeps its exit event: the timeout event comes first, then exit, then settled, and only then does TimeoutExpired reach the caller. Its TerminalOutcome is one of exit_zero, exit_nonzero, timeout, cancelled, or error; it carries a pid and exit_code only when those are known and never any exception text. Phases will be added again, so give a hook that matches on phase a default arm:

import sys

from cuprum import ExecEvent, Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="phases")
python = sh.make(Program(sys.executable), catalogue=catalogue)
outcomes = []


def hook(event: ExecEvent) -> None:
    match event.phase:
        case "settled":
            outcomes.append(event.terminal_outcome)
        case _:
            pass  # Phases this hook does not track, including future ones.


with sh.observe(hook):
    python("-c", "pass").run_sync()
assert [outcome.value for outcome in outcomes] == ["exit_zero"]

Every event also carries the execution's env_mode and capture_mode, known before the child spawns and describing the whole run, so a byte-exact run under a replacement environment is distinguishable from a text run under an overlay in every log record, span, and counter. A binding adds resolved_path, the executable that ran, to the event and the result. Catalogue lookup and allowlist refusals happen before an exec_id is minted and emit no events at all.

Log, measure, and trace runs

The adapters turn lifecycle events into structured log records (logging_adapter), metrics (metrics_adapter), and trace spans (tracing_adapter). Each is an observe hook. InMemoryMetrics and InMemoryTracer are reference backends; production code implements the MetricsCollector protocol, or the Tracer and Span protocols, over its own telemetry library.

import logging
import sys

from cuprum import Program, ProgramCatalogue, sh
from cuprum.adapters.logging_adapter import structured_logging_hook
from cuprum.adapters.metrics_adapter import InMemoryMetrics, MetricsHook
from cuprum.adapters.tracing_adapter import InMemoryTracer, TracingHook

catalogue = ProgramCatalogue.from_programs(sys.executable, name="telemetry")
python = sh.make(Program(sys.executable), catalogue=catalogue)
metrics = InMemoryMetrics()
tracer = InMemoryTracer()
logger = logging.getLogger("myapp.telemetry")

with (
    sh.observe(structured_logging_hook(logger=logger)),
    sh.observe(MetricsHook(metrics)),
    sh.observe(TracingHook(tracer)),
):
    python("-c", "print('traced')").run_sync()

assert metrics.counters["cuprum_executions_total"] == 1
assert [span.name for span in tracer.spans] and tracer.spans[0].ended

MetricsHook counts cuprum_failures_total and observes cuprum_duration_seconds from settled, so the failure counter covers every outcome except exit_zero, and cuprum_terminal_outcomes_total is labelled by the outcome. cuprum_executions_total and cuprum_failures_total carry env_mode and capture_mode labels, bounded by construction; resolved_path deliberately never becomes a label. TracingHook closes its span on settled. The structured logging adapter and the tracing adapter record argv in full, so a secret passed as an argument reaches the log unless a redaction policy masks it. Never turn untrusted arguments into metric labels.

Keep secrets out of telemetry

The adapters export what a run looked like, and that can include secrets: structured logging and tracing record the full argument vector, and events can carry environment values, output lines, notes, and tag values. Passing a secret in a file, rather than as an argument or environment value, keeps it out of these records. Where that is not possible, a RedactionPolicy masks the text it is told about before it is exported:

import logging
import re
import sys

from cuprum import Program, ProgramCatalogue, sh
from cuprum.adapters.logging_adapter import structured_logging_hook
from cuprum.adapters.tracing_adapter import InMemoryTracer, tracing_hook
from cuprum.redaction import RedactionPolicy, redacting_hook

catalogue = ProgramCatalogue.from_programs(sys.executable, name="redaction")
python = sh.make(Program(sys.executable), catalogue=catalogue)
policy = RedactionPolicy(
    literals=("token-value",),
    patterns=(re.compile(r"session-[0-9]+"),),
)
tracer = InMemoryTracer()
exports = []
logger = logging.getLogger("myapp.telemetry")
with (
    sh.observe(structured_logging_hook(logger=logger, redaction=policy)),
    sh.observe(tracing_hook(tracer, redaction=policy)),
    sh.observe(redacting_hook(exports.append, policy)),
):
    result = python(
        "-c", "import sys; print(sys.argv[1])", "token-value session-314"
    ).run_sync()

assert result.stdout == "token-value session-314\n"
assert exports and tracer.spans
assert "token-value" not in repr(exports) + repr(tracer.spans)
assert "session-314" not in repr(exports) + repr(tracer.spans)

• It masks only what it is given. Rules are literal strings (literals=) and compiled regular expressions (patterns=); the policy does not guess at secrets. Each match becomes the mask, [REDACTED] by default or the mask= text, and overlapping matches become a single mask.
• It is opt-in at each destination. Pass redaction=policy to structured_logging_hook(), tracing_hook() or TracingHook, logging_hook, and GitHubActionsSink, and wrap any other observer as redacting_hook(observer, policy). A destination without a policy exports text unchanged, so configure every destination that leaves the process.
• It changes exports, not the run. The command's arguments and environment, the captured result, and events seen by other observers are untouched.
• It covers free text. Argument items, environment values, output lines, notes, and top-level string tag values are redacted. The program name, project, paths, tag keys, and numeric and enum fields are not.
• It fails closed. If a rule fails, RedactionError is raised and nothing is exported in raw form. Like any observe-hook failure, it fails the run.

Metrics need no policy: their labels never carry arguments, environment values, output, or tags. The GitHubActionsSink accepts the same redaction= argument for its group and annotation titles; the group and annotate_failure flags have no redaction setting, and redaction never alters the mirrored child output.

Watch what a stop request did

An exit code of -15 says the child ended on SIGTERM, and -9 that it needed SIGKILL after its grace ran out, but neither records which control request was made, and a stop refused because the child had already gone leaves no trace at all. observe_control registers a ControlHook on a context-local channel; every completed signal(), terminate(), and kill() on a RunningCommand then reports one ControlEvent with the closed ControlOperation and ControlOutcome, the measured duration, and the run's exec_id:

import asyncio
import sys

from cuprum import ExecutionContext, Program, ProgramCatalogue, observe_control, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="control")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python("-c", "import time; time.sleep(30)")
events = []


async def stop_a_child() -> None:
    async with await command.start(
        context=ExecutionContext(cancel_grace=0.2),
    ) as handle:
        await handle.terminate()


with observe_control(events.append):
    asyncio.run(stop_a_child())
assert [(event.operation, event.outcome) for event in events] == [
    ("terminate", "delivered"),
]

An outcome of escalated means the child ignored SIGTERM and had to be killed; refused_exited and refused_unsupported mirror the two signal() refusals. Emission is fail-open, so an observer that raises cannot replace the refusal the caller is owed or interrupt the cleanup under way, and a caller who registers nothing pays no clock read. control_metrics_hook projects the events to bounded counters and a duration histogram whose only labels are the operation and the outcome; the PID reaches the event and the span but never a label, because a recycled PID would fold two children into one series.

Measure stream throughput

observe_stream_operation registers a hook for the pure-Python stream paths. It receives one event per completed drain or transfer, never one per read, with the bytes moved, the number of reads, and the duration. stream_operation_metrics_hook turns those into counters and a histogram whose only labels are the closed operation and outcome values. Observer failures are logged and suppressed, so observing never changes a run.

import sys

from cuprum import Program, ProgramCatalogue, sh
from cuprum.adapters.metrics_adapter import InMemoryMetrics
from cuprum.adapters.stream_metrics import stream_operation_metrics_hook
from cuprum.stream_observation import observe_stream_operation

catalogue = ProgramCatalogue.from_programs(sys.executable, name="stream-metrics")
python = sh.make(Program(sys.executable), catalogue=catalogue)
metrics = InMemoryMetrics()
with observe_stream_operation(stream_operation_metrics_hook(metrics)):
    python("-c", "print('hello')").run_sync()
assert metrics.counters["cuprum_stream_operation_bytes_total"] > 0

Present output in GitHub Actions

On GitHub Actions, Cuprum can frame a run's echoed output in a collapsible log group and turn a failed run into an ::error:: annotation. Both are presentation only: capture, exit codes, and the result are unchanged. The framing activates only when the parent process runs on GitHub Actions (GITHUB_ACTIONS is true); elsewhere a run behaves as if it were absent.

Group and annotate flags

group and annotate_failure on RunOutputOptions are the shorthand for the common case. They build a GitHubActionsSink and store it as the options' sink, so shared options can carry them safely:

import dataclasses
import io
import sys

from cuprum import Program, ProgramCatalogue, RunOutputOptions, sh
from cuprum.sinks import GitHubActionsSink

catalogue = ProgramCatalogue.from_programs(sys.executable, name="ci-flags")
python = sh.make(Program(sys.executable), catalogue=catalogue)

shared = RunOutputOptions(echo=True, group=True, annotate_failure=True)
assert isinstance(shared.sink, GitHubActionsSink)
result = python("-c", "print('building')").run_sync(output=shared)
assert result.ok and result.stdout == "building\n"

# An explicit sink wins, so shared flagged options never displace it.
explicit = GitHubActionsSink(io.StringIO(), force=True)
assert dataclasses.replace(shared, sink=explicit).sink is explicit

try:
    RunOutputOptions(group="yes")
except ValueError:
    pass
else:
    raise AssertionError("a non-bool flag should be rejected")

The flags are independent, and both default to False. group=True frames the run in a ::group:: and ::endgroup:: pair and protects echoed output with a stop-commands lease, so child output cannot inject workflow commands; a pipeline receives one group for the whole run. annotate_failure=True writes one ::error:: annotation when the run ends in a non-zero exit, a timeout, or an error; on its own it frames no group and takes no lease.

Workflow commands go to the parent's stderr. An explicit sink= takes precedence and makes both flags no-ops, which is what makes them safe in shared options; pass it on a new RunOutputOptions or through dataclasses.replace(shared, sink=...). The flags carry no force, and anything but a bool raises ValueError. Run grouped commands one after another when they share the parent's stderr, or their frames can interleave.

Configure the sink directly

Use GitHubActionsSink for anything the flags do not cover: a custom title, another destination, force=True to reproduce the framing locally, or emit_group=False and emit_annotation=False to switch off either half of the frame.

import io
import sys

from cuprum import Program, ProgramCatalogue, RunOutputOptions, sh
from cuprum.sinks import GitHubActionsSink

catalogue = ProgramCatalogue.from_programs(sys.executable, name="ci")
python = sh.make(Program(sys.executable), catalogue=catalogue)
workflow_log = io.StringIO()
sink = GitHubActionsSink(workflow_log, title="Build", force=True)
result = python("-c", "print('building')").run_sync(
    output=RunOutputOptions(echo=True, sink=sink),
)
assert result.ok and result.stdout == "building\n"
assert workflow_log.getvalue().startswith("::group::Build")