aeat.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 can be replayed and asserted byte-identical after masking (ADR 2026-06-30-deterministic-output-replay-substrate). 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 core.observability._golden.

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 replay 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:

Iterator[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

capture_is_armed()

Return whether an envelope-capture scope is active for the current context.

Return type:

bool