cadrumo.core.observability.capture module

In-memory capture sink for emitted CLI success envelopes.

The deterministic-output substrate captures the verbatim emitted SchemaEnvelope document so a recorded run output can be compared byte-identically after masking. The sink is a context variable holding a list; it is unset (None) in production, so record_emitted_envelope() is a no-op unless a capture_envelopes() scope has armed it — the emit path pays only a single ContextVar.get when capture is off.

This module deliberately has NO dependency on core.json_contract, so the emit path (core.json_contract.emit_json_success()) can feed it through a cheap lazy import without an import cycle. Typed re-validation of a captured document against the schema registry lives in cadrumo.tests.golden_comparison.

CAPTURE_SINK: ContextVar

Active capture list for the current context, or None when capture is off.

capture_envelopes()

Arm envelope capture for the current context, yielding the sink list.

Nesting-aware: when a sink is already active (e.g. armed by an outer capture scope), this reuses it rather than shadowing it, so a re-entered command’s emitted envelope lands in the outermost armed sink. The reused case does not reset the outer sink on exit.

Yields:

The list that record_emitted_envelope() appends to; each entry is a shallow copy of an emitted envelope document.

Return type:

Generator[list[dict[str, object]]]

record_emitted_envelope(envelope)

Append envelope to the active capture sink; a no-op when unarmed.

Parameters:

envelope (Mapping[str, object]) – The already-redacted, emitted envelope document.

Return type:

None