models.py #1

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