Source code for src.dackar.RCA.cross_pattern.models

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, List, Optional


@dataclass

[docs]
class HistoricalDocExtraction:
    """Canonical cross-pattern representation of a single doc extraction record.

    Produced by converting a SemanticMatch (from DocExtractionStore.query()) into
    a richer object that carries temporal and FM resolution metadata needed for
    the linkage rules in rules.py.

    Instances are treated as immutable inputs by CrossPatternLinker.run(): the
    linker never writes back onto them.  Episode↔doc linkage is reported in the
    run() result (per-candidate ``linked_episode_ids`` / ``linked_doc_ids`` and
    the flat ``all_links``), not by mutating these objects.
    """

[docs]
    doc_id: str


[docs]
    doc_type: str


[docs]
    asset_id: Optional[str]


[docs]
    event_time_start: Optional[datetime]


[docs]
    event_time_end: Optional[datetime]


[docs]
    event_time_confidence: str           # "explicit" | "inferred" | "absent"


[docs]
    identified_effect: Optional[str]


[docs]
    assessed_cause: Optional[str]


[docs]
    inferred_fm_label: Optional[str]


[docs]
    fm_id_candidate: Optional[str]


[docs]
    fm_id_candidate_alt: Optional[str]


[docs]
    fm_resolution_status: str            # "auto_resolved" | "ambiguous" | "unresolved"


[docs]
    fm_resolution_score: Optional[float]


[docs]
    confidence: str                      # "high" | "medium" | "low"


[docs]
    cause_is_symptom: bool

    # Epistemic annotation fields — carried from SemanticMatch (Phase A)

[docs]
    epistemic_class: Optional[str] = None


[docs]
    classification_resolution_level: Optional[str] = None


[docs]
    degraded_classification: bool = False




@dataclass

[docs]
class CrossPatternLink:
    """A single link between one HistoricalSignalEpisode and one HistoricalDocExtraction.

    Carries full provenance so analysts can trace exactly which checks passed
    (direct reference, temporal, or semantic/FM; asset compatibility is scored
    separately as asset_match).
    """

[docs]
    link_id: str


[docs]
    episode_id: str


[docs]
    doc_id: str


[docs]
    asset_match: bool


[docs]
    time_overlap_hours: Optional[float]


[docs]
    temporal_link_skipped: bool


[docs]
    linkage_precedence_level: int        # 1=direct, 2=temporal, 3=semantic/FM


[docs]
    component_overlap: List[str]


[docs]
    fm_alignment_score: Optional[float]


[docs]
    signal_similarity_score: float


[docs]
    document_similarity_score: Optional[float]


[docs]
    link_confidence: float


[docs]
    provenance: Dict[str, Any]




@dataclass

[docs]
class CandidateCrossPatternEvidence:
    """Aggregated cross-pattern evidence for a single causality candidate.

    Produced by CrossPatternLinker.run() — one instance per candidate in
    causality_candidates["candidates"].
    """

[docs]
    candidate_id: str


[docs]
    component_id: str


[docs]
    fm_id: str


[docs]
    linked_episode_ids: List[str]


[docs]
    linked_doc_ids: List[str]


[docs]
    best_link_score: float


[docs]
    support_posture: str                # "reinforcing" | "conflicting" | "weakly_supporting" | "unresolved"


[docs]
    reinforcement_strength: Optional[str]   # "single" | "multiple_consistent" | "mixed" | None


[docs]
    linkage_outcome: str                # "linked" | "no_data" | "no_match" | "below_threshold"


[docs]
    evidence_paths: List[CrossPatternLink]