Routing Excursion Alerts by Severity in Python

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from enum import Enum


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


@dataclass(frozen=True)
class SeverityPolicy:
    warning_at: float = 26.0
    critical_at: float = 76.0

    def classify(self, score: float, critical_product: bool) -> Severity:
        # ICH Q9 proportionality: response scales with quantified risk. A product
        # with a narrow therapeutic index is escalated one band harder so a
        # moderate score on a fragile biologic still rings the quality tier.
        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

p = SeverityPolicy()
assert p.classify(25.9, False) is Severity.INFO
assert p.classify(26.0, False) is Severity.WARNING     # exact boundary
assert p.classify(40.0, True) is Severity.CRITICAL      # elevated by product flag

@dataclass(frozen=True)
class Route:
    recipient: str        # on-call role, resolved to a person by the rota
    channel: str          # abstract channel key: "chat", "sms", "voice"
    ack_window: timedelta  # §11.10(e) time-boxed accountability


SEVERITY_ROUTES: dict[Severity, Route] = {
    Severity.INFO: Route("shift_technician", "chat", timedelta(minutes=30)),
    Severity.WARNING: Route("shift_supervisor", "sms", timedelta(minutes=10)),
    Severity.CRITICAL: Route("quality_on_call", "voice", timedelta(minutes=5)),
}

from typing import Protocol


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


class SeverityRouter:
    def __init__(self, policy: SeverityPolicy, channel: NotificationChannel,
                 routes: dict[Severity, Route]):
        self.policy = policy
        self.channel = channel
        self.routes = routes

    def dispatch(self, excursion_id: str, score: float, critical_product: bool,
                 now: datetime) -> dict:
        severity = self.policy.classify(score, critical_product)
        route = self.routes[severity]
        deadline = now + route.ack_window
        receipt = self.channel.send(
            route.channel, route.recipient,
            subject=f"[{severity.value}] excursion {excursion_id}",
            body=f"Acknowledge by {deadline.isoformat()}.",
        )
        # §11.10(e): the dispatch is a time-stamped record of a system action.
        return {
            "excursion_id": excursion_id,
            "severity": severity.value,
            "recipient": route.recipient,
            "channel": route.channel,
            "ack_deadline_utc": deadline.isoformat(),
            "dispatched_at": now.isoformat(),
            "transport_receipt": receipt,
            "acknowledged_by": None,
        }

import hashlib
import hmac


class AckTracker:
    def __init__(self, signing_key: bytes):
        if len(signing_key) < 32:
            raise ValueError("signing_key must be at least 32 bytes")
        self.signing_key = signing_key

    def issue_token(self, excursion_id: str) -> str:
        # Bind the ack token to the specific event so it cannot be replayed
        # against a different excursion (§11.10(g) authority / non-repudiation).
        return hmac.new(self.signing_key, excursion_id.encode("utf-8"),
                        hashlib.sha256).hexdigest()

    def acknowledge(self, record: dict, operator_id: str, token: str,
                    now: datetime) -> dict:
        if not operator_id:
            # §11.10(g): an acknowledgement without a verified identity is not an
            # accountable action and must be rejected.
            raise ValueError("acknowledgement requires an authenticated operator id")
        expected = self.issue_token(record["excursion_id"])
        if not hmac.compare_digest(expected, token):  # constant-time compare
            raise ValueError("acknowledgement token does not match this excursion")
        record["acknowledged_by"] = operator_id
        record["acknowledged_at"] = now.isoformat()   # contemporaneous §11.10(e)
        record["overdue"] = now.isoformat() > record["ack_deadline_utc"]
        return record

tracker = AckTracker(b"k" * 32)
now = datetime(2026, 7, 14, 2, 0, tzinfo=timezone.utc)
rec = {"excursion_id": "EXC-7", "ack_deadline_utc": "2026-07-14T02:05:00+00:00",
       "acknowledged_by": None}
good = tracker.issue_token("EXC-7")
acked = tracker.acknowledge(rec, "qa.jdoe", good, now)
assert acked["acknowledged_by"] == "qa.jdoe"
# A token minted for a different event must be rejected.
try:
    tracker.acknowledge(rec, "qa.jdoe", tracker.issue_token("EXC-OTHER"), now)
    assert False, "expected rejection of a mismatched token"
except ValueError:
    pass

def route_and_open(router: SeverityRouter, tracker: AckTracker,
                   excursion_id: str, score: float, critical_product: bool,
                   now: datetime) -> tuple[dict, str]:
    record = router.dispatch(excursion_id, score, critical_product, now)
    token = tracker.issue_token(excursion_id)  # handed to the recipient's ack link
    return record, token

class _Recorder:
    def __init__(self): self.sent = []
    def send(self, channel, recipient, subject, body):
        self.sent.append((channel, recipient)); return "rcpt-1"

router = SeverityRouter(SeverityPolicy(), _Recorder(), SEVERITY_ROUTES)
now = datetime(2026, 7, 14, 2, 0, tzinfo=timezone.utc)
rec, tok = route_and_open(router, AckTracker(b"k" * 32), "EXC-3", 88.0, True, now)
assert rec["severity"] == "CRITICAL" and rec["channel"] == "voice"