"""Data models shared across all p4rca layers. These are the contracts between layers โ€” changing them requires updating all consumers. Define clearly before implementing layer logic. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any, Optional class DetectorType(str, Enum): WEDGE = "wedge" SLOW_COMMAND = "slow_command" CONNECTION_SPIKE = "connection_spike" class LockType(str, Enum): READ = "read" WRITE = "write" class ActionTier(int, Enum): """Corrective action autonomy ladder (ยง6.1 of briefing).""" OBSERVE = 0 ALERT = 1 RECOMMEND = 2 ACT_WITH_TIMEOUT = 3 AUTONOMOUS = 4 @dataclass class LockEvent: """A single table lock record parsed from P4LOG.""" timestamp: datetime pid: int table: str lock_type: LockType duration_ms: float command: str user: str client: str = "" @dataclass class MonitorEntry: """One row from `p4 monitor show -ale`.""" pid: int status: str # I=idle, R=running, T=terminated, W=waiting user: str client: str command: str args: str = "" duration_s: float = 0.0 @dataclass class MonitorSnapshot: """Point-in-time snapshot from the collector.""" timestamp: datetime entries: list[MonitorEntry] = field(default_factory=list) lsof_lines: list[str] = field(default_factory=list) lslocks_lines: list[str] = field(default_factory=list) @dataclass class CandidateIncident: """Structured trigger event emitted by the anomaly detection layer.""" incident_id: str detected_at: datetime detector_type: DetectorType confidence: float # 0.0โ€“1.0 log_window_start: datetime log_window_end: datetime lock_events: list[LockEvent] = field(default_factory=list) monitor_snapshots: list[MonitorSnapshot] = field(default_factory=list) raw_signals: dict[str, Any] = field(default_factory=dict) description: str = "" @dataclass class RCAResult: """Structured output from the SLM reasoning layer.""" incident_id: str analyzed_at: datetime diagnosis: str confidence: float # 0.0โ€“1.0 recommended_action: str reasoning_summary: str action_tier: ActionTier raw_slm_response: str = "" action_taken: Optional[str] = None outcome: Optional[str] = None