src.dackar.RCA.validation.schema_validator

Attributes

JsonDict
Severity

Classes

ValidationIssue	A single validation finding against one artifact.
ValidationReport	Aggregate result of validating one artifact or a run bundle.
RCAArtifactValidator	Two-layer validator:

Module Contents

src.dackar.RCA.validation.schema_validator.JsonDict

src.dackar.RCA.validation.schema_validator.Severity

class src.dackar.RCA.validation.schema_validator.ValidationIssue

A single validation finding against one artifact.

artifact

The artifact type the issue was raised against (e.g. "event").

severity

"error" or "warning". In warn_only mode the validator emits every finding as "warning".

code

A stable machine-readable identifier for the issue (e.g. "schema_validation_error", "event_time_order_invalid").

message

Human-readable description of the problem.

path

JSON path (as string segments) to the offending field, or an empty list for artifact-level issues.

artifact: str

severity: Severity

code: str

message: str

path: List[str] = []

to_dict()

Return type:

Dict[str, Any]

class src.dackar.RCA.validation.schema_validator.ValidationReport

Aggregate result of validating one artifact or a run bundle.

ok

True when the report contains no "error"-severity issues. Callers should treat this as authoritative only after the report is fully populated; recompute_ok() refreshes it from issues.

issues

All findings collected during validation, both errors and warnings.

The :attr:`errors` and :attr:`warnings` properties partition ``issues`` by

severity. In ``warn_only`` mode every finding is a warning, so ``ok`` stays

``True`` regardless of the problems found.

ok: bool

issues: List[ValidationIssue] = []

property errors: List[ValidationIssue]

Return type:

List[ValidationIssue]

property warnings: List[ValidationIssue]

Return type:

List[ValidationIssue]

add(issue)

Parameters:

issue (ValidationIssue)

Return type:

None

extend(issues)

Parameters:

issues (Sequence[ValidationIssue])

Return type:

None

recompute_ok()

Return type:

None

to_dict()

Return type:

Dict[str, Any]

class src.dackar.RCA.validation.schema_validator.RCAArtifactValidator(schema_dir, *, mode='compat')

Two-layer validator:

1. per-artifact Draft7 JSON Schema validation

2. cross-artifact semantic consistency checks

Modes:

• strict: no legacy field aliases; schema mismatches are errors

• compat: normalize common legacy aliases before validation

• warn_only: same checks, but schema/semantic failures downgraded to warnings

Parameters:

• schema_dir (str | pathlib.Path)

• mode (Literal['strict', 'compat', 'warn_only'])

CORE_ARTIFACTS

schema_dir

mode = 'compat'

schemas: Dict[str, Dict[str, Any]]

validators: Dict[str, jsonschema.Draft7Validator]

validate_artifact(artifact_type, payload)

Validate a single artifact payload against its schema and semantics.

The payload is first normalized (a deep copy — the caller’s payload is never mutated) and validated against its Draft 7 schema. Semantic consistency checks run only when schema validation passes: those checks assume schema-valid object types, so running them on malformed input could raise instead of reporting an issue.

Parameters:

• artifact_type (str) – One of CORE_ARTIFACTS. Whitespace and case are normalized.

• payload (Dict[str, Any]) – The artifact object to validate.

Returns:

All findings for this artifact. An unknown artifact_type yields a single schema_missing issue. This method does not raise on invalid payloads — every problem is returned as a ValidationIssue.

Return type:

ValidationReport

validate_run_bundle(*, event=None, telemetry_summary=None, kg_context=None, signal_evidence=None, tskr_patterns=None, causality_candidates=None, evidence_bundle=None, ishikawa_matrix=None, barrier_analysis=None, rca_card=None, operational_context=None, pm_compliance=None, cmms_context=None)

Validate a run bundle: each member individually, then cross-artifact.

Every supplied member is validated on its own via validate_artifact(), then the schema-valid members are checked for cross-artifact consistency (shared event_id / asset_id, evidence and candidate cross-references, full-mode RCA-card requirements, etc.).

Parameters:

• event (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• telemetry_summary (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• kg_context (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• signal_evidence (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• tskr_patterns (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• causality_candidates (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• evidence_bundle (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• ishikawa_matrix (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• barrier_analysis (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• rca_card (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• operational_context (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• pm_compliance (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

• cmms_context (Optional[Dict[str, Any]]) – Keyword-only bundle members. Each is optional; a member left as None is skipped entirely. Caller payloads are never mutated.

Returns:

Aggregate of every member’s issues plus the cross-artifact findings. Members that fail their own schema validation are excluded from the cross-artifact checks so malformed input is reported, not raised on.

Return type:

ValidationReport

_load_schemas()

Return type:

None

static _load_json(path)

Parameters:

path (pathlib.Path)

Return type:

Dict[str, Any]

_normalize_payload(artifact_type, payload)

Parameters:

• artifact_type (str)

• payload (Dict[str, Any])

Return type:

Dict[str, Any]

_semantic_checks_single(artifact_type, payload)

Parameters:

• artifact_type (str)

• payload (Dict[str, Any])

Return type:

List[ValidationIssue]

_semantic_checks_bundle(bundle)

Parameters:

bundle (Dict[str, Optional[Dict[str, Any]]])

Return type:

List[ValidationIssue]

_norm_artifact_type(artifact_type)

Parameters:

artifact_type (str)

Return type:

str

static _parse_instant(value)

Parse an ISO-8601 date-time string into a timezone-aware datetime.

Returns None when the value is not a parseable string, so callers skip the comparison rather than raising. A naive datetime (no offset) is assumed to be UTC, so two timestamps are always compared as instants and offsets like +01:00 vs Z order chronologically, not lexicographically.

Parameters:

value (Any)

Return type:

Optional[datetime.datetime]

_sev(base)

Parameters:

base (Severity)

Return type:

Severity

static _issue(*, artifact, severity, code, message, path=None)

Parameters:

• artifact (str)

• severity (Severity)

• code (str)

• message (str)

• path (Optional[List[str]])

Return type:

ValidationIssue

_semantic_checks_rca_card(payload)

Parameters:

payload (JsonDict)

Return type:

List[ValidationIssue]

_semantic_checks_run_manifest(payload)

Parameters:

payload (JsonDict)

Return type:

List[ValidationIssue]

_semantic_checks_run_context(payload)

Parameters:

payload (JsonDict)

Return type:

List[ValidationIssue]

_semantic_checks_causality_candidates(payload)

Parameters:

payload (JsonDict)

Return type:

List[ValidationIssue]

_bundle_checks_rca_card_consistency(rca_card, evidence, candidates, kg_context=None)

Parameters:

• rca_card (JsonDict)

• evidence (Optional[JsonDict])

• candidates (Optional[JsonDict])

• kg_context (Optional[JsonDict])

Return type:

List[ValidationIssue]