from cuprum.adapters import logging_adapter
Structured logging adapter for Cuprum execution events.
This module provides an observe hook that emits structured log records for each execution event phase. Unlike the simpler logging_hook() which logs only start/exit with before/after hooks, this adapter leverages the full ExecEvent stream for fine-grained observability.
The adapter demonstrates how to transform ExecEvent values into structured log records suitable for log aggregation systems (ELK, Splunk, etc.).
Example::
import logging from cuprum import ScopeConfig, scoped, sh from cuprum.adapters.logging_adapter import structured_logging_hook
logging.basicConfig(level=logging.DEBUG)
with scoped( ScopeConfig(allowlist=my_allowlist) ), sh.observe(structured_logging_hook()): sh.make(ECHO)("hello").run_sync()
3 exports
class cuprum.adapters.logging_adapter.LogLevels
class LogLevels(
plan_level: int = logging.DEBUG,
start_level: int = logging.INFO,
output_level: int = logging.DEBUG,
exit_level: int = logging.INFO,
fail_fast_level: int = logging.WARNING,
group_teardown_level: int = logging.WARNING,
stdio_error_level: int = logging.WARNING,
)
Configuration for logging levels per execution event phase.
Attributes
plan_level- Log level for
plan events (intent to execute). Default DEBUG.
start_level- Log level for
start events (process spawned). Default INFO.
output_level- Log level for
stdout/stderr events. Default DEBUG.
exit_level- Log level for
exit events (process completed). Default INFO.
fail_fast_level- Log level for
pipeline_fail_fast events (a pipeline is being torn down because a non-final stage failed first). Default WARNING: unlike the other phases this reports an abnormal outcome, and it would otherwise fall through to the unhandled-phase default of DEBUG and be invisible in a normal configuration.
group_teardown_level- Log level for
group_teardown events (a teardown could not establish settlement). Default WARNING for the same reason as fail_fast_level: unlike the other phases this reports an abnormal outcome, and it would otherwise fall through to the unhandled-phase default of DEBUG and be invisible in a normal configuration.
stdio_error_level- Log level for
stdio_error events (a standard-stream boundary failed). Default WARNING, for the same reason as fail_fast_level: it reports a failure, and a record nobody retains is not a diagnostic.
function cuprum.adapters.logging_adapter.structured_logging_hook
def structured_logging_hook(
*,
logger: logging.Logger | None = None,
levels: LogLevels | None = None,
redaction: RedactionPolicy | None = None,
) -> ExecHook
Create an observe hook that logs execution events with structured data.
Parameters
logger- Logger instance for event emission. Defaults to
logging.getLogger("cuprum.exec").
levels- Log level configuration for different event phases. Defaults to
LogLevels() with standard levels.
redaction- Optional projection policy applied before rendering messages and extras. Defaults to
None; execution inputs and raw events remain unchanged.
Returns
- ExecHook
- A hook suitable for use with
sh.observe().
Notes
This hook is synchronous and non-blocking. Log emission happens inline with event processing. For high-throughput scenarios, consider using an async handler or buffered logging configuration.
The hook attaches structured extra data to log records including:
- cuprum_phase: Event phase (plan, start, stdout, stderr, stdin, stdin_error, timeout, teardown_error, capture_eof_grace_expired, exit, settled, pipeline_fail_fast) - cuprum_program: Program being executed - cuprum_argv: Full argument vector - cuprum_pid: Process ID (when available) - cuprum_exit_code: Exit code (for exit and pipeline_fail_fast events) - cuprum_terminal_outcome: Bounded category on settled events - cuprum_duration_s: Duration in seconds (for exit and pipeline_fail_fast events) - cuprum_stage_index / cuprum_stage_count: Position of the failing stage and the pipeline width (for pipeline_fail_fast events) - cuprum_eof_grace_s / cuprum_pending_readers: Fixed grace duration and pending-reader count (for capture_eof_grace_expired events) - cuprum_max_rss_bytes / cuprum_user_cpu_seconds / cuprum_system_cpu_seconds / cuprum_resource_usage_mode: Child resource measurements and the mode naming their source. The mode is carried on every terminal exit event — wait4_child, aggregate_cpu_delta, or unavailable — while the three figures are present only where a source produced them
The adapter projects selected execution fields into log extras; it does not emit the full tags mapping.
from cuprum.adapters import metrics_adapter
Prometheus-style metrics adapter for Cuprum execution events.
This module provides an observe hook that collects metrics about command execution in a format compatible with Prometheus client libraries. The adapter demonstrates how to collect:
- **Counters**: Total executions, failures, output lines - **Byte counters**: Successful stdin bytes written - **Histograms**: Execution duration distribution
The implementation uses protocol classes to remain decoupled from specific metrics libraries. Projects can implement the protocols with their preferred backend (prometheus_client, statsd, OpenTelemetry metrics, etc.).
Example with the in-memory reference implementation::
from cuprum import ScopeConfig, scoped, sh from cuprum.adapters.metrics_adapter import MetricsHook, InMemoryMetrics
metrics = InMemoryMetrics()
with scoped( ScopeConfig(allowlist=my_allowlist) ), sh.observe(MetricsHook(metrics)): sh.make(ECHO)("hello").run_sync()
print(metrics.counters) # {'cuprum_executions_total': 1, ...} print(metrics.histograms) # {'cuprum_duration_seconds': [...]}
Example with prometheus_client::
from prometheus_client import Counter, Histogram from cuprum.adapters.metrics_adapter import MetricsCollector, MetricsHook
class PrometheusMetrics: def __init__(self): self._exec_total = Counter( "cuprum_executions_total", "Total command executions", ["program", "project"], ) self._duration = Histogram( "cuprum_duration_seconds", "Execution duration", ["program", "project"], )
def inc_counter(self, name, value, labels): if name == "cuprum_executions_total": self._exec_total.labels(**labels).inc(value)
def observe_histogram(self, name, value, labels): if name == "cuprum_duration_seconds": self._duration.labels(**labels).observe(value)
hook = MetricsHook(PrometheusMetrics())
4 exports
class cuprum.adapters.metrics_adapter.InMemoryMetrics
class InMemoryMetrics(
counters: dict[str, float] = dc.field(default_factory=dict),
histograms: dict[str, list[float]] = dc.field(default_factory=dict),
)
Reference in-memory metrics collector for testing and examples.
Storage and locking follow the shared _LockedStore contract: every mutator holds the lock, and reset() clears the store under it.
Attributes
counters- Dict mapping metric names to accumulated counter values.
histograms- Dict mapping metric names to lists of observed values.
class cuprum.adapters.metrics_adapter.MetricsCollector
Protocol for metrics collection backends.
Implementations must be thread-safe; hooks may be invoked from multiple threads or async tasks concurrently.
class cuprum.adapters.metrics_adapter.MetricsHook
class MetricsHook(collector: MetricsCollector) -> None
Observe hook that collects Prometheus-style metrics.
The hook emits the following metrics:
- cuprum_executions_total: Counter incremented on each start event - cuprum_terminal_outcomes_total: Counter of terminal outcomes, labelled by the bounded terminal_outcome category - cuprum_failures_total: Counter incremented for every category except exit_zero - cuprum_duration_seconds: Histogram of measured terminal durations - cuprum_stdout_lines_total: Counter of stdout lines emitted - cuprum_stderr_lines_total: Counter of stderr lines emitted - cuprum_stdin_bytes_total: Counter of successful stdin bytes written - cuprum_stdin_errors_total: Counter of stdin writer failures - cuprum_timeouts_total: Counter of subprocess and pipeline expiries - cuprum_teardown_errors_total: Counter of consumer drain failures - cuprum_capture_eof_grace_expired_total: Counter of capture EOF grace expiries with readers still pending - cuprum_pipeline_fail_fast_total: Counter of pipelines terminated early after their first non-final stage failure - cuprum_resource_usage_measurements_total: Counter of terminal events that reported how their child's resource figures were obtained - cuprum_child_max_rss_bytes: Histogram of per-child maximum RSS, from the attributable wait4 path only - cuprum_child_user_cpu_seconds: Histogram of child user CPU time - cuprum_child_system_cpu_seconds: Histogram of child system CPU time
All metrics include program and project labels. The three resource histograms and their counter additionally carry a low-cardinality resource_usage_mode label naming their source — wait4_child, aggregate_cpu_delta, or unavailable — so a run that measures nothing is distinguishable from one that measured a small value. The histograms are observed only where the figure was actually measured; the counter is emitted for every recorded mode, including unavailable.
Parameters
collector- A
MetricsCollector implementation for the target backend.
function cuprum.adapters.metrics_adapter.metrics_hook
def metrics_hook(collector: MetricsCollector) -> ExecHook
Create a metrics observe hook for the given collector.
This is a convenience factory that returns a MetricsHook instance cast to the ExecHook type.
Parameters
collector- A
MetricsCollector implementation.
Returns
- ExecHook
- A hook suitable for use with
sh.observe().
from cuprum.adapters import tracing_adapter
OpenTelemetry-style tracing adapter for Cuprum execution events.
This module provides an observe hook that creates distributed traces for command execution. The adapter demonstrates how to:
- Create spans for command execution lifecycle - Attach structured attributes to spans - Handle parent-child span relationships for pipelines - Record span events for output lines
The implementation uses protocol classes to remain decoupled from specific tracing libraries. Projects can implement the protocols with their preferred backend (OpenTelemetry, Jaeger, Zipkin, etc.).
Example with the in-memory reference implementation::
from cuprum import ScopeConfig, scoped, sh from cuprum.adapters.tracing_adapter import TracingHook, InMemoryTracer
tracer = InMemoryTracer()
with scoped( ScopeConfig(allowlist=my_allowlist) ), sh.observe(TracingHook(tracer)): sh.make(ECHO)("hello").run_sync()
print(tracer.spans) # [Span(name='cuprum.exec echo', ...)]
Example with OpenTelemetry::
from opentelemetry import trace from cuprum.adapters.tracing_adapter import Tracer, Span, TracingHook
class OTelSpan: def __init__(self, otel_span): self._span = otel_span
def set_attribute(self, key, value): self._span.set_attribute(key, value)
def add_event(self, name, attributes=None): self._span.add_event(name, attributes=attributes or {})
def set_status(self, *, ok): from opentelemetry.trace import StatusCode code = StatusCode.OK if ok else StatusCode.ERROR self._span.set_status(code)
def end(self): self._span.end()
class OTelTracer: def __init__(self, tracer): self._tracer = tracer def start_span(self, name, attributes=None): span = self._tracer.start_span(name, attributes=attributes) return OTelSpan(span)
otel_tracer = trace.get_tracer("cuprum") hook = TracingHook(OTelTracer(otel_tracer))
7 exports
class cuprum.adapters.tracing_adapter.InMemorySpan
class InMemorySpan(
name: str,
attributes: dict[str, object] = dc.field(default_factory=dict),
events: list[tuple[str, dict[str, object]]] = dc.field(default_factory=list),
status_ok: bool | None = None,
ended: bool = False,
)
Reference in-memory span for testing and examples.
A mutable test double implementing the Span protocol from tracing_adapter. It records everything set on it so that tests can inspect the result. It performs no synchronization of its own; a span is expected to be mutated by a single thread at a time.
Attributes
name str- Name of the span, as supplied to
InMemoryTracer.start_span.
attributes dict[str, object]- Span attributes keyed by name. Seeded from the attributes passed at creation and extended in place by
set_attribute.
events list[tuple[str, dict[str, object]]]- Recorded events in insertion order, each a
(name, attributes) pair appended by add_event.
status_ok bool | None- Completion status:
None until set_status is called, then True for success or False for failure.
ended bool- Whether
end has been called; False until then.
class cuprum.adapters.tracing_adapter.InMemoryTracer
class InMemoryTracer(
spans: list[InMemorySpan] = dc.field(default_factory=list),
)
Reference in-memory tracer for testing and examples.
Storage and locking follow the shared _LockedStore contract: every mutator holds the lock, and reset() clears the store under it.
Attributes
spans- List of all spans created by this tracer.
class cuprum.adapters.tracing_adapter.Span
Protocol for a tracing span.
Spans represent a unit of work and can be enriched with attributes and events.
class cuprum.adapters.tracing_adapter.Tracer
Protocol for a tracing backend.
Implementations must be thread-safe; hooks may be invoked from multiple threads or async tasks concurrently.
class cuprum.adapters.tracing_adapter.TracingHook
class TracingHook(
tracer: Tracer,
*,
record_output: bool = True,
redaction: RedactionPolicy | None = None,
) -> None
Project correlated execution events onto backend spans.
Events without exec_id are ignored rather than correlated by a PID. Attributes include cuprum.program, cuprum.argv, cuprum.pid, cuprum.cwd, cuprum.exit_code, cuprum.duration_s, cuprum.project, cuprum.pipeline_stage_index, and cuprum.pipeline_stages. Every terminal exit event also carries cuprum.resource_usage_mode, naming how its child's resource figures were obtained — wait4_child, aggregate_cpu_delta, or unavailable. A terminal exit event that measured its child's resource usage additionally carries cuprum.max_rss_bytes, cuprum.user_cpu_seconds, and cuprum.system_cpu_seconds; those three figures alone are absent rather than null when no measurement applies.
Parameters
tracer- A
Tracer implementation for the target backend.
record_output- If True, record stdout/stderr lines as span events. Default True.
redaction- Optional policy applied to event copies before span projection. Defaults to
None; raw observations and execution remain unchanged.
function cuprum.adapters.tracing_adapter.record_control_events
def record_control_events(hook: TracingHook) -> ControlHookRegistration
Record control operations as events on the execution span they act on.
Parameters
hook- The tracing hook whose spans the control events are recorded on. It must already be registered as an observe hook, because this function reports events onto spans that hook has opened.
Returns
- ControlHookRegistration
- A detachable registration. It fails open: an observer that raises is logged and skipped so it can never replace a control operation's own error or interrupt the child cleanup the operation is performing.
function cuprum.adapters.tracing_adapter.tracing_hook
def tracing_hook(
tracer: Tracer,
*,
record_output: bool = True,
redaction: RedactionPolicy | None = None,
) -> ExecHook
Create a tracing observe hook for the given tracer.
This is a convenience factory that returns a TracingHook instance cast to the ExecHook type.
Parameters
tracer- A
Tracer implementation.
record_output- If True, record stdout/stderr lines as span events. Default True.
redaction- Optional policy applied before span projection. Defaults to
None.
Returns
- ExecHook
- A hook suitable for use with
sh.observe().
from cuprum.adapters import stream_metrics
Bounded aggregate metrics for completed pure-Python stream operations.
5 exports
constant cuprum.adapters.stream_metrics.STREAM_OPERATION_BYTES_TOTAL
STREAM_OPERATION_BYTES_TOTAL = 'cuprum_stream_operation_bytes_total'
constant cuprum.adapters.stream_metrics.STREAM_OPERATION_DURATION_SECONDS
STREAM_OPERATION_DURATION_SECONDS = 'cuprum_stream_operation_duration_seconds'
constant cuprum.adapters.stream_metrics.STREAM_OPERATION_READ_OPERATIONS_TOTAL
STREAM_OPERATION_READ_OPERATIONS_TOTAL = 'cuprum_stream_operation_read_operations_total'
class cuprum.adapters.stream_metrics.StreamOperationMetricsHook
class StreamOperationMetricsHook(collector: MetricsCollector) -> None
Project aggregate stream-operation events into bounded metrics.
The hook emits one byte counter, one reader-operation counter, and one duration histogram for every received event. Every metric uses exactly the closed operation and outcome labels from the event enums.
function cuprum.adapters.stream_metrics.stream_operation_metrics_hook
def stream_operation_metrics_hook(
collector: MetricsCollector,
) -> StreamOperationHook
Create a bounded metrics hook for stream-operation observation.
from cuprum.adapters import line_stream_metrics
Bounded metrics projection for line-stream lifecycle events.
3 exports
constant cuprum.adapters.line_stream_metrics.LINE_STREAM_EVENTS_TOTAL
LINE_STREAM_EVENTS_TOTAL = 'cuprum_line_stream_events_total'
Counter of SafeCmd.lines() lifecycle events by closed categories.
class cuprum.adapters.line_stream_metrics.LineStreamMetricsHook
class LineStreamMetricsHook(collector: MetricsCollector) -> None
Count line-stream lifecycle events with bounded categorical labels.
Parameters
collector- The metrics collector that receives one counter increment per event.
function cuprum.adapters.line_stream_metrics.line_stream_metrics_hook
def line_stream_metrics_hook(
collector: MetricsCollector,
) -> LineStreamHook
Return a lifecycle hook that projects events into collector.
Parameters
collector- The metrics collector that receives line-stream event counters.
Returns
- LineStreamHook
- A synchronous hook that records bounded lifecycle labels.
from cuprum.adapters import echo_metrics
Metrics adapter for stream-echo failure and truncation events.
Counts how often a text-only echo sink rejected the subprocess output, per output stream, and how often a bounded echoed line was successfully truncated. Those facts are otherwise visible only through a WARNING log record or the echo observation hook, which dashboards cannot aggregate.
The hook consumes EchoEvent values from observe_echo. It reuses the MetricsCollector protocol, so one collector can back the execution, pump, and echo hooks and no new telemetry dependency is introduced.
5 exports
constant cuprum.adapters.echo_metrics.ECHO_BROKEN_PIPE_TOTAL
ECHO_BROKEN_PIPE_TOTAL = 'cuprum_echo_broken_pipe_total'
Echo writes abandoned after BrokenPipeError, labelled by stream.
A distinct series from ECHO_ENCODING_FAILURES_TOTAL: both count a handled echo disablement, but they report different faults with different remedies. An encoding failure is inherent to the sink's configuration and recurs for every run that reaches it; a broken pipe means a destination that was expected to stay open did not, which is exactly the signal an operator wants to alert on separately. Folding them together would make a permanently misencoded sink indistinguishable from a downstream reader that keeps disconnecting.
Only recorded when the caller opted in with BEST_EFFORT; under the default policy the error propagates and no event is emitted.
constant cuprum.adapters.echo_metrics.ECHO_ENCODING_FAILURES_TOTAL
ECHO_ENCODING_FAILURES_TOTAL = 'cuprum_echo_encoding_failures_total'
Echo writes rejected with UnicodeEncodeError, labelled by stream.
The only other label is error_category, whose single value (UNICODE_ENCODE) names the closed category the counter counts. The label exists so a future failure category extends the metric rather than silently mixing into this series — which is what ECHO_BROKEN_PIPE_TOTAL then did.
constant cuprum.adapters.echo_metrics.ECHO_TRUNCATIONS_TOTAL
ECHO_TRUNCATIONS_TOTAL = 'cuprum_echo_truncations_total'
Successfully written bounded echo lines, labelled by stream.
class cuprum.adapters.echo_metrics.EchoMetricsHook
class EchoMetricsHook(collector: MetricsCollector) -> None
Echo observation hook that records bounded echo outcomes.
The hook emits cuprum_echo_encoding_failures_total: incremented once per first UnicodeEncodeError on a drain's echo path, labelled stream with stdout or stderr and error_category with unicode_encode. It also emits cuprum_echo_truncations_total once per successfully written truncation, labelled only by stream.
A BrokenPipeError the caller chose to tolerate under BEST_EFFORT increments cuprum_echo_broken_pipe_total instead, with the same label set and error_category of broken_pipe. It is a separate series, so a sink that cannot encode is never confused with a destination that keeps closing.
A drain that has already disabled echo emits nothing further, so repeated chunks after the first failure do not compound the count.
Parameters
collector- A
MetricsCollector implementation for the target backend. It must be thread-safe: a failure can be recorded from any task draining a stream.
function cuprum.adapters.echo_metrics.echo_metrics_hook
def echo_metrics_hook(collector: MetricsCollector) -> EchoHook
Create an echo observation hook for the given collector.
Parameters
collector- A
MetricsCollector implementation.
from cuprum.adapters import pump_metrics
Metrics adapter for Rust-pump routing events.
Counts the pump lifecycle facts that are otherwise visible only as DEBUG log records: how often an inter-stage hop declined the Rust pump, broken down by the seam that refused; how often a cancelled hop's worker failed; and how long native cleanup took after cancellation.
The hook consumes PumpEvent values from observe_pump, not ExecEvent values from sh.observe. It reuses the MetricsCollector protocol, so one collector can back both hooks and no new telemetry dependency is introduced.
10 exports
constant cuprum.adapters.pump_metrics.RUST_PUMP_CLEANUP_DEFERRED_TOTAL
RUST_PUMP_CLEANUP_DEFERRED_TOTAL = 'cuprum_rust_pump_cleanup_deferred_total'
Deferred native-pump cleanups completed after the caller had returned.
constant cuprum.adapters.pump_metrics.RUST_PUMP_CLEANUP_DURATION_SECONDS
RUST_PUMP_CLEANUP_DURATION_SECONDS = 'cuprum_rust_pump_cleanup_duration_seconds'
Monotonic wait durations for completed native-pump cleanup.
constant cuprum.adapters.pump_metrics.RUST_PUMP_CLEANUP_GRACE_EXPIRED_TOTAL
RUST_PUMP_CLEANUP_GRACE_EXPIRED_TOTAL = 'cuprum_rust_pump_cleanup_grace_expired_total'
Caller-facing cleanup waits that exhausted their configured grace.
constant cuprum.adapters.pump_metrics.RUST_PUMP_CLEANUP_TOTAL
RUST_PUMP_CLEANUP_TOTAL = 'cuprum_rust_pump_cleanup_total'
Completed native-pump cleanups after a pipeline cancellation.
constant cuprum.adapters.pump_metrics.RUST_PUMP_DECLINED_TOTAL
RUST_PUMP_DECLINED_TOTAL = 'cuprum_rust_pump_declined_total'
Declined Rust-pump hand-offs, labelled by the reason that refused.
constant cuprum.adapters.pump_metrics.RUST_PUMP_FAILED_AFTER_CANCEL_TOTAL
RUST_PUMP_FAILED_AFTER_CANCEL_TOTAL = 'cuprum_rust_pump_failed_after_cancel_total'
Rust-pump worker failures consumed after their hop was cancelled.
constant cuprum.adapters.pump_metrics.RUST_PUMP_HANDOFF_TOTAL
RUST_PUMP_HANDOFF_TOTAL = 'cuprum_rust_pump_handoff_total'
Rust writer-resource hand-offs, labelled by their bounded outcome.
constant cuprum.adapters.pump_metrics.UNKNOWN_DECLINE_REASON
UNKNOWN_DECLINE_REASON = 'unknown'
The reason label a decline carrying no recognized reason degrades to.
Public because it is operator-visible: it is the one value the reason label can take that is not a RustPumpDeclineReason member, so a dashboard filtering on that enum alone would silently miss these. Declared here rather than as an enum member because no seam reports it — it exists only so a malformed event cannot introduce an unbounded label.
class cuprum.adapters.pump_metrics.PumpMetricsHook
class PumpMetricsHook(collector: MetricsCollector) -> None
Collect bounded metrics from Rust-pump routing events.
The hook emits the following metrics:
- cuprum_rust_pump_declined_total: a counter labelled reason by the closed RustPumpDeclineReason vocabulary, plus unknown for an unrecognized hand-built event. - cuprum_rust_pump_failed_after_cancel_total: an unlabelled counter for worker failures observed after cancellation. - cuprum_rust_pump_cleanup_total: an unlabelled counter for completed native-pump cleanup. - cuprum_rust_pump_cleanup_duration_seconds: an unlabelled histogram of normal cleanup durations. - cuprum_rust_pump_cleanup_grace_expired_total: an unlabelled counter for cleanup waits that reach their configured grace limit. - cuprum_rust_pump_cleanup_deferred_total: an unlabelled counter for deferred cleanup completed by the worker's callback. - cuprum_rust_pump_handoff_total: a counter labelled outcome by the closed RustPumpHandoffOutcome vocabulary, including the Windows raw-handle rejection.
All labels are closed and bounded; descriptor values, exception details, and other event data are never labels. Unknown phases are ignored so a newer producer cannot make an existing observer fail. The collector must be thread-safe because pump events may be emitted from worker threads.
Parameters
collector- Thread-safe metrics backend receiving counters and histograms.
function cuprum.adapters.pump_metrics.pump_metrics_hook
def pump_metrics_hook(collector: MetricsCollector) -> PumpHook
Create a pump observation hook for the given collector.
Parameters
collector- A
MetricsCollector implementation.
from cuprum.adapters import control_metrics
Bounded metrics for the control operations an owned handle performs.
5 exports
constant cuprum.adapters.control_metrics.CONTROL_OPERATIONS_TOTAL
CONTROL_OPERATIONS_TOTAL = 'cuprum_control_operations_total'
constant cuprum.adapters.control_metrics.CONTROL_OPERATION_DURATION_SECONDS
CONTROL_OPERATION_DURATION_SECONDS = 'cuprum_control_operation_duration_seconds'
constant cuprum.adapters.control_metrics.CONTROL_REFUSALS_TOTAL
CONTROL_REFUSALS_TOTAL = 'cuprum_control_refusals_total'
class cuprum.adapters.control_metrics.ControlMetricsHook
class ControlMetricsHook(collector: MetricsCollector) -> None
Project control events into bounded metrics.
The hook emits one counter and one duration histogram for every received event. Both metrics carry exactly the closed operation and outcome labels from the event enums, so the series count is fixed by the two enums rather than by how many children a process happens to control. No PID, execution identifier, or command argument reaches a label.
function cuprum.adapters.control_metrics.control_metrics_hook
def control_metrics_hook(collector: MetricsCollector) -> ControlHook
Create a bounded metrics hook for control observation.
from cuprum import redaction
Opt-in redaction for execution-event projections.
Redaction happens on copies at export boundaries. It never changes the arguments, environment, or captured output used by an execution.
3 exports
exception cuprum.redaction.RedactionError
class RedactionError(
*,
rule_kind: typ.Literal['literal', 'pattern'],
rule_index: int,
error_type: str,
) -> None
A configured redaction rule failed before an export could be built.
Attributes
rule_kind- Whether the failing rule was a literal or regular expression.
rule_index- Zero-based position of the rule in its configured tuple.
error_type- Name of the underlying exception class. The exception text and source value are omitted so diagnostics cannot echo the value being redacted.
class cuprum.redaction.RedactionPolicy
class RedactionPolicy(
literals: str | cabc.Iterable[str] = (),
patterns: re.Pattern[str] | cabc.Iterable[re.Pattern[str]] = (),
mask: str = _DEFAULT_MASK,
) -> None
Apply configured literal and regular-expression masks to exports.
The default policy is inert. Rules are supplied by the caller; this class does not attempt to identify secrets automatically.
Parameters
literals- Literal strings to replace. A single string or any iterable of strings is coerced to an immutable tuple, ordered longest first. Empty strings are rejected because they would match every position in the input.
patterns- Compiled text regular expressions to replace. A single pattern or any iterable of patterns is coerced to an immutable tuple.
mask- Opaque text substituted for each merged match. Defaults to
"[REDACTED]".
function cuprum.redaction.redacting_hook
def redacting_hook(
consumer: ExecHook,
policy: RedactionPolicy,
) -> ExecHook
Wrap an observation consumer so it receives sanitized event copies.
Parameters
consumer- The hook that exports or stores execution observations.
policy- The explicit policy applied before the consumer receives an event.
Returns
- ExecHook
- A hook suitable for registration with
observe(). The wrapper returns the consumer's result unchanged, including an awaitable result.
Presentation sink adapters for parent-facing run output.
A sink adapter reshapes how a run's parent-facing output appears — for example, framing it in a GitHub Actions group — without changing capture, success semantics, or the returned result. Sinks are opt-in: a caller passes one explicitly via sink, and runs without a sink are byte-for-byte unchanged.
The protocol lives in base; concrete adapters (such as GitHubActionsSink) live alongside it in this package.
7 exports
class cuprum.sinks.GitHubActionsSession
class GitHubActionsSession(
log: typ.IO[str],
label: str,
*,
annotation_label: str | _Annotation,
) -> None
One run's framed presentation session on GitHub Actions.
Framing is complete the moment the session exists: the group opening and the stop-commands lease are written before the sink returns it, so child output can never appear above the group opening. The session holds that lease for the run's entire lifetime and closes the group with an error annotation when the run's terminal outcome is a failure.
Both halves of that frame are separately switchable. A session opened with emit_group=False writes no group command and takes no lease — it is a plain pass-through destination that may still annotate. A session opened with emit_annotation=False frames normally but never annotates.
class cuprum.sinks.GitHubActionsSink
class GitHubActionsSink(
destination: typ.IO[str] | None = None,
**options: typ.Unpack[_GitHubActionsSinkOptionsKwargs],
) -> None
Opt-in presentation sink for GitHub Actions CI callers.
The sink holds only immutable configuration; every run gets a fresh GitHubActionsSession from open_session, so one sink instance can serve many sequential or concurrent runs.
emit_group and emit_annotation independently control whether the session writes the collapsible group frame and failure annotation.
Parameters
destination- Writable text stream receiving the workflow commands and framed output. Defaults to the parent's stderr, matching where GitHub Actions reads workflow commands from.
title- Optional explicit label for both the group and failure annotation. By default, argv titles the group and the bounded run label titles the annotation.
force- Activate the sink even when the parent process does not run on GitHub Actions. Intended for local reproduction of CI framing and non-standard runners; when
False (the default) the sink defers to the environment check.
emit_group- Whether a session writes the group commands and their stop-commands lease.
False leaves the run's log destination unframed, which is what RunOutputOptions(annotate_failure=True) asks for: an annotation without collapsible logs.
emit_annotation- Whether a failed outcome emits its
::error:: annotation. False keeps the framing and drops the annotation, which is what RunOutputOptions(group=True) asks for.
redaction- Optional policy for every emitted group and annotation title, including the bounded fallback label, applied before workflow-command escaping. Defaults to
None. Mirrored child output is unchanged.
Raises
TypeError- If
emit_group or emit_annotation is not a bool. They gate workflow commands, so a merely truthy value would frame a run on the strength of something the caller never documented as a flag.
class cuprum.sinks.OutputSession
The write side of one active adapter run.
Implementations keep their framing state here and remain safe for the sequential single-writer use the execution layer guarantees: child-output writes and control writes are issued from the run's own task context, not from unrelated tasks.
An adapter emits its opening framing as the session is constructed, so it is already written when OutputSink.open_session returns; the execution layer offers no framing hook and never reaches into an adapter's internals. log is therefore a pure accessor. A session may expose an optional redirects_echo boolean attribute; when it is False, the shared lifecycle leaves echoed output on the caller's original destinations while the session still handles its terminal annotation. Sessions without this hint continue to route echoed output through log.
class cuprum.sinks.OutputSink
A factory for per-run presentation sessions.
Implementations hold only immutable configuration. All mutable state lives on the OutputSession returned by open_session, which the execution layer creates fresh for every invocation.
class cuprum.sinks.SessionOutcome
class SessionOutcome(
outcome: TerminalOutcome,
exit_code: int | None = None,
detail: str | None = None,
)
Immutable terminal report for one run.
Attributes
outcome- The closed-set terminal category.
exit_code- The actual exit code when the run produced one, otherwise
None. Adapters must not synthesize shell-style codes.
detail- Optional bounded, categorical detail (for example
"timeout"). The execution layer never passes exception text or argv here.
class cuprum.sinks.SessionStart
class SessionStart(label: str, argv: tuple[str, ...])
Immutable inputs an adapter needs to frame one run.
Attributes
label- Bounded, already-safe display label for the group (for example
"project: program"). The execution layer builds it from catalogue metadata, not from argv, so secrets never reach it.
argv- The full command vector (program followed by arguments) the run will execute. Adapters that frame a run per command — for example a GitHub Actions group titled with the program args — use this directly. It is caller-controlled input: an adapter that surfaces it verbatim opts the caller into displaying it (GitHub Actions groups are visible in the workflow log regardless, because echoed child output is shown there).
class cuprum.sinks.TerminalOutcome
Why an observed execution reached its definitive terminal event.
The values are shared with presentation sinks, so execution telemetry and output adapters classify the same outcome with the same bounded labels.