aeat.core.observability._store module

Filesystem persistence for run traces and JSONL event logs.

One subdirectory per run_id under core.config.Settings.aeat_runs_dir, containing trace.json and events.jsonl. Both files round-trip through the strict pydantic models in core.observability._models.

Run traces are DIAGNOSTIC class. The redaction rule set returned by core.redaction.default_rules_for_class() for DIAGNOSTIC walks every string leaf — NIFs SHA-256-prefixed, URLs reduced to host-only, bearer-shaped tokens fingerprinted, opaque bearers fingerprinted — before serialisation. The core redaction helper is imported lazily so commands that never persist traces avoid resolving the rule registry on import.

runs_dir(settings=None)[source]

Return the configured runs directory, creating it when absent.

Parameters:

settings (Settings | None) – Optional core.config.Settings override (used by tests). When None, the active settings are loaded via core.config.load_settings().

Return type:

Path

Returns:

Absolute path to the per-process runs root.

save_trace(trace, *, settings=None)[source]

Persist a RunTrace to <runs_dir>/<run_id>/trace.json.

Every string leaf passes through core.redaction.redact_structured() at DIAGNOSTIC class before serialisation so the on-disk record never carries a plaintext NIF, bearer token, or sensitive URL path even if a caller fed one into arguments.

Parameters:
Return type:

Path

Returns:

Absolute path of the written trace.json file.

load_trace(run_id, *, settings=None)[source]

Load and strictly validate a persisted RunTrace.

Read-only lookups do not create the per-run directory — a missing trace.json raises RunTraceValidationError without polluting the runs directory with an empty entry.

Parameters:
Return type:

RunTrace

Returns:

The validated RunTrace.

Raises:

RunTraceValidationError – When run_id has an invalid shape, when the file is missing, or when its contents fail strict validation.

save_envelope(run_id, document, *, settings=None)[source]

Persist an emitted envelope document to <runs_dir>/<run_id>/envelope.json.

The document is the verbatim, already-CLI-redacted SchemaEnvelope mapping captured by core.observability.capture_envelopes() during the run. It is stored key-sorted so the on-disk artifact is byte-stable, and it is the golden expectation a later replay_run() asserts against. Re-validation into a typed envelope happens on load via core.observability.validate_captured_envelope(); this writer stays free of any JSON-contract dependency.

Parameters:
Return type:

Path

Returns:

Absolute path of the written envelope.json file.

load_envelope_document(run_id, *, settings=None)[source]

Load the persisted emitted-envelope document for a run.

Read-only: does not create the per-run directory. Returns the raw mapping; type it with core.observability.validate_captured_envelope().

Parameters:
Return type:

dict[str, object]

Returns:

The persisted envelope mapping.

Raises:

RunTraceValidationError – When run_id has an invalid shape, the file is missing, or its contents are not a JSON object.

save_events_append(run_id, event, *, settings=None)[source]

Append a single RunEvent line to the per-run events.jsonl.

newline="" pins the on-disk line terminator to \\n on every platform — mirroring core.observability._sink.JsonlRunSink — so events.jsonl is byte-stable across Windows and POSIX writers. Every string leaf in the event is redacted at DIAGNOSTIC class before serialisation so the on-disk record stays free of plaintext NIFs / tokens / sensitive URLs.

Parameters:
Return type:

Path

Returns:

Absolute path of the appended events.jsonl file.

iter_events(run_id, *, settings=None)[source]

Return an iterator of RunEvent records from the per-run events.jsonl.

Streams records so callers processing a long-running run’s event log can avoid holding the entire file in memory. The run_id is validated eagerly — before the iterator starts — so a bad id surfaces at the call site instead of on first iteration.

Read-only: does not create a run directory when absent. A missing file yields no records.

Parameters:
Return type:

Iterator[RunEvent]

Returns:

An iterator of RunEvent records in append order.

load_events(run_id, *, settings=None)[source]

Load and strictly validate every JSONL event for a run.

Thin wrapper over iter_events() that drains the iterator into a tuple. Prefer iter_events() for long-running traces where the whole log may exceed available memory.

Read-only: does not create a run directory when absent.

Parameters:
Return type:

tuple[RunEvent, ...]

Returns:

Tuple of every recorded RunEvent in append order.

iter_runs(*, settings=None)[source]

Yield (run_id, RunTrace) pairs sorted by started_at descending.

Directories without a valid trace.json — or whose name does not match the canonical run_id shape — are skipped silently. This lets crashed runs (no on-exit finaliser call) coexist with healthy ones rather than poisoning aeat run list, and blocks any non-run artefacts that may have been dropped into the runs directory by hand.

Parameters:

settings (Settings | None) – Optional core.config.Settings override.

Yields:

(run_id, trace) pairs in newest-first order, where each trace is a RunTrace loaded from the run directory.

Return type:

Iterator[tuple[str, RunTrace]]