cadrumo.core.observability.store module

Filesystem persistence for run traces and JSONL event logs.

One subdirectory per run_id under core.config.Settings.cadrumo_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.rules.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.

ENVELOPE_FILENAME

Per-run artifact leaf names, owned by this module (see the run_trace / run_events / run_envelope entries in STORAGE_PATH_DEFINITIONS, each declared owner="cadrumo.core.observability" and interpolating these constants into their grammar rather than re-typing the names).

validate_run_id(run_id)[source]

Return run_id if it matches the canonical shape, else raise.

The canonical shape is 16 lowercase hex characters — the form minted by core.observability.context._mint_run_id(). Validating every id reaching this layer prevents path-traversal escapes through runs_dir / run_id.

Parameters:

run_id (str) – Candidate run identifier.

Return type:

str

Returns:

The same run_id when it matches the canonical shape.

Raises:

RunTraceValidationError – When run_id is not a 16-character lowercase hex string.

runs_dir(settings=None)[source]

Return the configured runs directory. Resolution only; never creates it.

Resolved through storage_path() for StorageCategory.RUNS rather than by reading cadrumo_runs_dir directly here. Both answer the same today, because the settings field is what the accessor consults first – but only one of them stays correct if the member’s resolution ever gains a case.

A pure path resolver. Materialisation is a write-side concern: the one production writer (run_dir()) creates the full <runs_dir>/<run_id> chain itself via mkdir(parents=True) when it stages a run for writing, which brings this root into existence as a side effect of that single mkdir call – it does not depend on this function creating anything. A read must never materialise a directory for what it failed to find, so every read-only caller (load_trace(), load_envelope_document(), iter_events(), iter_runs(), prune_run_traces()) treats an absent root exactly like an empty one rather than relying on this resolver to conjure it first.

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. Not guaranteed to exist.

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

Return the per-run directory, creating it (and the runs root) when absent.

The one production write path that materialises anything under the runs root: mkdir(parents=True) creates the per-run subdirectory and, as part of the same call, any missing ancestor – including the runs root itself – in one step. Rejects run_id values that do not match the canonical minted shape so runs_dir / run_id cannot traverse out of the configured runs directory.

Parameters:
Return type:

Path

Returns:

Absolute path to the per-run subdirectory (created if absent).

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

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

Every string leaf passes through core.redaction.rules.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: does not create the per-run directory, nor the runs root itself. 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, when its contents fail strict validation, or when its embedded identity names another run.

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.capture_envelopes() during the run. It is stored key-sorted so the on-disk artifact is byte-stable, and it is durable evidence for the recorded run. 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, nor the runs root itself. 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, nor the runs root itself. A missing file yields no records.

Parameters:
Return type:

Iterator[RunEvent]

Returns:

An iterator of RunEvent records in append order.

iter_runs(*, settings=None)[source]

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

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

Read-only: does not create the runs directory. No run has ever been saved on a fresh install or an isolated test root, so an absent runs directory is treated exactly like an empty one – yielding nothing – rather than raising.

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]]

prune_run_traces(*, retention_days=None, max_total_bytes=None, settings=None)[source]

Delete run directories beyond the retention window or the size ceiling.

Gives the run-trace store a declared retention lifecycle instead of one subdirectory accumulating per run forever. Two independent bounds apply in order, each delegating its survivor decision to the shared select_filesystem_retention_survivors() selector while this function keeps its own rmtree side effect and its own newest-directory-never-size-pruned rule:

  1. Age — run directories whose modification time (last write) is older than retention_days are removed, so crashed runs that never produced a valid trace.json are pruned too rather than accumulating unreadable.

  2. Total size — if the surviving run directories still exceed max_total_bytes on disk, the oldest directories are removed until the store fits under the ceiling. The newest run directory is always kept (protect_newest=1), so the trace whose save triggered the prune survives even when it alone exceeds the ceiling.

retention_days defaults to the centralized cadrumo_runs_retention_days and max_total_bytes to cadrumo_runs_max_total_bytes. Entirely best-effort: a directory that cannot be enumerated, stat’d, or removed is logged and skipped, never raised – pruning must not crash the caller.

Parameters:

  • retention_days (int | None) – Age cutoff in days; run directories whose mtime is strictly older are removed. Defaults to the centralized setting.

  • max_total_bytes (int | None) – Total on-disk size ceiling for the run-trace store. Defaults to the centralized setting.

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

Return type:

int

Returns:

Number of run directories removed.