Escalation & Alert Routing for Excursion Events

from __future__ import annotations

import hashlib
import json
import logging
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Optional, Protocol

logger = logging.getLogger("excursion.escalation")


class Severity(str, Enum):
    INFO = "INFO"
    WARNING = "WARNING"
    CRITICAL = "CRITICAL"


class EscalationState(str, Enum):
    RAISED = "RAISED"
    NOTIFIED = "NOTIFIED"
    ACK = "ACK"
    ESCALATED = "ESCALATED"
    RESOLVED = "RESOLVED"


@dataclass(frozen=True)
class Tier:
    name: str
    channel: str          # abstract channel key resolved by the transport layer
    ack_window: timedelta  # bounded acknowledgement SLA for this rung


@dataclass(frozen=True)
class RoutingPolicy:
    # The escalation policy is a change-controlled artifact under ICH Q10; its
    # fingerprint is bound to every ledger entry so a reviewer can prove which
    # routing table produced a given escalation path (21 CFR Part 11 §11.10(k)).
    tiers: tuple[Tier, ...]
    dedup_window: timedelta = timedelta(minutes=10)
    # Score boundaries that map a disposition onto a response severity. These are
    # deliberately separate from the scoring engine's own state boundaries.
    warning_at: float = 26.0
    critical_at: float = 76.0

    def severity_for(self, score: float, critical_product: bool) -> Severity:
        # ICH Q9 proportionality: response tier scales with quantified risk, and a
        # narrow-therapeutic-index product is escalated one band harder.
        if score >= self.critical_at or (critical_product and score >= self.warning_at):
            return Severity.CRITICAL
        if score >= self.warning_at:
            return Severity.WARNING
        return Severity.INFO

    def fingerprint(self) -> str:
        payload = json.dumps(
            {
                "tiers": [(t.name, t.channel, t.ack_window.total_seconds()) for t in self.tiers],
                "dedup_window_s": self.dedup_window.total_seconds(),
                "warning_at": self.warning_at,
                "critical_at": self.critical_at,
            },
            sort_keys=True, separators=(",", ":"),
        )
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()


class NotificationChannel(Protocol):
    def send(self, channel: str, subject: str, body: str) -> str:
        """Deliver a notification and return a transport receipt id."""


def event_fingerprint(device_id: str, zone: str, rule_id: str, at: datetime,
                      bucket: timedelta) -> str:
    # A stable dedup key over the PHYSICAL + LOGICAL coordinates of the excursion.
    # Bucketing the timestamp collapses a burst from one fault into one key while
    # a different device, zone, or rule yields a distinct key — suppression must
    # never drop a genuinely separate quality event (ICH Q10 event handling).
    epoch = int(at.timestamp() // bucket.total_seconds())
    raw = f"{device_id}|{zone}|{rule_id}|{epoch}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


@dataclass
class Escalation:
    excursion_id: str
    fingerprint: str
    severity: Severity
    policy: RoutingPolicy
    escalation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    tier: int = 0
    state: EscalationState = EscalationState.RAISED
    ack_deadline_utc: Optional[datetime] = None
    acknowledged_by: Optional[str] = None
    _prev_hash: str = "0" * 64
    ledger: list[dict] = field(default_factory=list)

    def _append(self, transition: str, now: datetime) -> None:
        # §11.10(e): every state change is a secure, time-stamped entry that does
        # not obscure prior entries. Each record commits to its predecessor's
        # digest, so a retrospective edit breaks the chain (tamper evidence).
        entry = {
            "escalation_id": self.escalation_id,
            "excursion_id": self.excursion_id,
            "fingerprint": self.fingerprint,
            "transition": transition,
            "severity": self.severity.value,
            "tier": self.tier,
            "state": self.state.value,
            "ack_deadline_utc": self.ack_deadline_utc.isoformat() if self.ack_deadline_utc else None,
            "acknowledged_by": self.acknowledged_by,
            "config_version": self.policy.fingerprint(),
            "recorded_at": now.isoformat(),  # contemporaneous UTC stamp §11.10(e)
            "prev_hash": self._prev_hash,
        }
        canonical = json.dumps(entry, sort_keys=True, separators=(",", ":"))
        entry["entry_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        self._prev_hash = entry["entry_hash"]
        self.ledger.append(entry)

    def notify(self, channel: NotificationChannel, now: datetime) -> None:
        tier = self.policy.tiers[self.tier]
        self.ack_deadline_utc = now + tier.ack_window
        receipt = channel.send(
            tier.channel,
            subject=f"[{self.severity.value}] excursion {self.excursion_id}",
            body=f"Acknowledge by {self.ack_deadline_utc.isoformat()} (tier {tier.name}).",
        )
        self.state = EscalationState.NOTIFIED
        logger.info("notified %s tier=%s receipt=%s", self.escalation_id, tier.name, receipt)
        self._append("notified", now)

    def acknowledge(self, operator_id: str, now: datetime) -> None:
        # §11.10(g) authority check: acknowledgement must carry an authenticated
        # operator identity; an anonymous ack is not an accountable action.
        if not operator_id:
            raise ValueError("acknowledgement requires an authenticated operator id")
        if self.state == EscalationState.RESOLVED:
            raise ValueError("cannot acknowledge a resolved escalation")
        self.acknowledged_by = operator_id
        self.state = EscalationState.ACK
        self._append("acknowledged", now)

    def escalate_if_overdue(self, channel: NotificationChannel, now: datetime) -> bool:
        # Time-boxed accountability: an unacknowledged deadline climbs one rung.
        if self.state in (EscalationState.ACK, EscalationState.RESOLVED):
            return False
        if self.ack_deadline_utc is None or now < self.ack_deadline_utc:
            return False
        if self.tier + 1 >= len(self.policy.tiers):
            # Top rung reached: hold at max tier, re-notify, do not silently drop.
            self.state = EscalationState.ESCALATED
            self._append("escalated_top", now)
            self.notify(channel, now)
            return True
        self.tier += 1
        self.state = EscalationState.ESCALATED
        self._append("escalated", now)
        self.notify(channel, now)
        return True

    def resolve(self, now: datetime) -> dict:
        # An event is only resolvable once owned — proves an action was recorded,
        # not merely that the system alerted (ICH Q10 / §11.10(e)).
        if self.acknowledged_by is None:
            raise ValueError("cannot resolve an unacknowledged escalation")
        self.state = EscalationState.RESOLVED
        self._append("resolved", now)
        return self.ledger[-1]


class EscalationRouter:
    def __init__(self, policy: RoutingPolicy, channel: NotificationChannel):
        self.policy = policy
        self.channel = channel
        self._active: dict[str, Escalation] = {}  # fingerprint -> escalation

    def raise_event(self, excursion_id: str, device_id: str, zone: str, rule_id: str,
                    score: float, critical_product: bool, now: datetime) -> Optional[Escalation]:
        fp = event_fingerprint(device_id, zone, rule_id, now, self.policy.dedup_window)
        existing = self._active.get(fp)
        if existing is not None and existing.state != EscalationState.RESOLVED:
            # Duplicate within the live window: attach to the open event, do not
            # raise a second page (suppresses alert storms without losing signal).
            existing._append("deduplicated", now)
            logger.info("suppressed duplicate for fingerprint=%s", fp)
            return None
        severity = self.policy.severity_for(score, critical_product)
        esc = Escalation(excursion_id, fp, severity, self.policy)
        esc._append("raised", now)
        esc.notify(self.channel, now)
        self._active[fp] = esc
        return esc

    def sweep(self, now: datetime) -> None:
        # Called on a fixed cadence to enforce acknowledgement SLAs.
        for esc in list(self._active.values()):
            esc.escalate_if_overdue(self.channel, now)

from datetime import datetime, timedelta, timezone


class _RecordingChannel:
    def __init__(self):
        self.sent = []

    def send(self, channel, subject, body):
        self.sent.append((channel, subject))
        return f"receipt-{len(self.sent)}"


def _policy():
    return RoutingPolicy(tiers=(
        Tier("Technician", "chat", timedelta(minutes=15)),
        Tier("Supervisor", "sms", timedelta(minutes=10)),
        Tier("Quality", "phone", timedelta(minutes=5)),
    ))


def test_timeout_climbs_one_rung():
    ch = _RecordingChannel()
    router = EscalationRouter(_policy(), ch)
    t0 = datetime(2026, 7, 14, 2, 0, tzinfo=timezone.utc)
    esc = router.raise_event("EXC-9", "DEV-1", "ZONE-A", "R-HI", 82.0, True, t0)
    # Past the 15-minute tier-0 SLA the event must climb to tier 1 (§11.10(e)).
    esc.escalate_if_overdue(ch, t0 + timedelta(minutes=16))
    assert esc.tier == 1
    assert esc.severity is Severity.CRITICAL


def test_duplicate_is_suppressed_not_dropped():
    ch = _RecordingChannel()
    router = EscalationRouter(_policy(), ch)
    t0 = datetime(2026, 7, 14, 2, 0, tzinfo=timezone.utc)
    first = router.raise_event("EXC-1", "DEV-1", "ZONE-A", "R-HI", 40.0, False, t0)
    second = router.raise_event("EXC-2", "DEV-1", "ZONE-A", "R-HI", 40.0, False, t0)
    assert first is not None and second is None       # storm collapsed to one page
    assert first.ledger[-1]["transition"] == "deduplicated"