src.dackar.RCA.orchestrators.artifact_store

artifact_store — Concrete ArtifactStore and SchemaValidator implementations.

Extracted from rca_reasoning_orchestrator.py. The parent module re-exports both classes for backward-compatible imports.

Attributes

JsonDict

Classes

NoOpSchemaValidator	Permissive SchemaValidator that only enforces JSON-object shape.
FileArtifactStore	ArtifactStore that persists artifacts as JSON files on disk.

Module Contents

src.dackar.RCA.orchestrators.artifact_store.JsonDict

class src.dackar.RCA.orchestrators.artifact_store.NoOpSchemaValidator

Permissive SchemaValidator that only enforces JSON-object shape.

Implements all three validator styles but performs no schema checks — useful as a default when schema validation is not wired. Every artifact is reported as valid provided it is a JSON object.

validate(artifact_name, payload)

Legacy style: raise TypeError unless payload is a JSON object.

Parameters:

• artifact_name (str)

• payload (JsonDict)

Return type:

None

validate_artifact(artifact_name, payload)

Per-artifact style: shape-check payload and return an ok report.

Parameters:

• artifact_name (str)

• payload (JsonDict)

Return type:

JsonDict

validate_run_bundle(**kwargs)

Bundle style: shape-check each non-None keyword artifact and return ok.

Parameters:

kwargs (Any)

Return type:

JsonDict

class src.dackar.RCA.orchestrators.artifact_store.FileArtifactStore(root_dir)

ArtifactStore that persists artifacts as JSON files on disk.

Each artifact is written atomically to <root_dir>/<run_id>/<name>.json so a reader never observes a partial write.

Parameters:

root_dir (str | pathlib.Path)

root_dir

save(run_id, artifact_name, payload)

Persist a single-object artifact and return its file path.

Parameters:

• run_id (str)

• artifact_name (str)

• payload (JsonDict)

Return type:

str

save_list(run_id, artifact_name, payload)

Persist a list-valued artifact and return its file path.

Parameters:

• run_id (str)

• artifact_name (str)

• payload (List[JsonDict])

Return type:

str

load(run_id, artifact_name)

Load and return a previously saved artifact, or None if absent.

Parameters:

• run_id (str)

• artifact_name (str)

Return type:

Any

is_run_complete(run_id)

Return True only if run_status.json exists and run_complete is True.

External callers (notebooks, replay scripts) should check this before loading or re-validating artifacts from a run directory. A run that crashed mid-pipeline will have run_complete=False (or no run_status.json at all), and its artifacts should not be treated as authoritative.

Parameters:

run_id (str)

Return type:

bool

_write_atomic(run_id, artifact_name, payload)

Write payload to <run_dir>/<artifact_name>.json via temp-file + rename.

The rename is atomic on POSIX and near-atomic on Windows (py3.3+). A reader can never observe a partial write.

Parameters:

• run_id (str)

• artifact_name (str)

• payload (Any)

Return type:

str