aeat.core.observability._context module

Contextvars-backed run context with nesting support and JSONL sink wiring.

Entering run_context() at the outermost CLI entry point mints a fresh run_id, fingerprints the corpus / db / cert state, attaches a aeat.core.observability._sink.JsonlRunSink to the root logger for the duration of the block, emits a aeat.core.observability._models.RunEventKind.STEP_START event, and persists the final aeat.core.observability._models.RunTrace on exit. Nesting is idempotent: an inner enter reuses the outer run_id and only pushes a new step identifier.

class RunContextInfo(**data)[source]

Bases: BaseModel

Immutable bag of run-level metadata exposed to call sites.

Yielded by run_context() so callers can stamp recorded events with the active run_id or surface it in user-facing output.

Variables:

  • run_id – 16-char lowercase hex identifier for this run.

  • entrypoint – Stable string identifying the CLI entry point (e.g. "aeat workflow run").

  • started_at – Wall-clock UTC timestamp captured at run-context enter.

  • arguments – Tuple of ArgumentRecord capturing the CLI flags / positional values for replay.

  • corpus_sha256 – Fingerprint of .vault/ plus Settings plus env/.env at enter time.

  • db_sha256 – Fingerprint of the local var/ state tree at enter time, excluding cache subdirectories.

  • cert_fingerprint – SHA-256 of the configured PKCS#12 certificate, or "" when no cert path is configured.

  • initial_step_id – Step identifier emitted with the first STEP_START boundary event.

Parameters:
run_id: str
entrypoint: str
started_at: datetime
arguments: tuple[ArgumentRecord, ...]
corpus_sha256: str
db_sha256: str
cert_fingerprint: str
initial_step_id: str
RUN_CONTEXT_VAR: ContextVar

Active RunContextInfo for the current task / thread, or None.

STEP_CONTEXT_VAR: ContextVar

Active step identifier within the current run context, or None.

current_run_context()[source]

Return the RunContextInfo bound to the current task, if any.

Return type:

RunContextInfo | None

run_context(*, entrypoint, arguments=(), run_id=None, step_id=None)[source]

Enter a run context, emitting STEP_START / STEP_END boundary events.

The outermost enter mints a run_id, fingerprints the corpus / db / cert state, attaches a aeat.core.observability._sink.JsonlRunSink to the root logger, emits a STEP_START event, and on exit emits a STEP_END plus persists the finalised aeat.core.observability._models.RunTrace (even on exception, with RunOutcome.FAILED).

Inner enters reuse the outer run_id and only push a new step_id, so callers can wrap higher-level commands without every callee knowing whether a run is already active.

Parameters:

  • entrypoint (str) – Stable string identifying the CLI entry point (e.g. "aeat workflow run").

  • arguments (Sequence[ArgumentRecord]) – Sequence of ArgumentRecord capturing the CLI flags / values for replay.

  • run_id (str | None) – Optional caller-supplied run_id (used by aeat.core.observability.replay_run()).

  • step_id (str | None) – Optional initial step identifier; defaults to "step-0" for the outermost enter and a derived nested id for inner enters.

Yields:

The active RunContextInfo for the block.

Raises:

Exception – The error captured during the save_trace call, re-raised when trace persistence fails and the yielded body completed successfully (outcome OK).

Return type:

Iterator[RunContextInfo]