Routing Compliance Tasks to Regional Legal-Ops Teams #

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import AsyncGenerator, Optional, Protocol

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("routing.regional_dispatch")


class RouteStatus(Enum):
    ROUTED = "routed"          # placed on a regional queue (primary or fallback)
    ESCALATED = "escalated"    # fallback chain exhausted — sent to human review


class QueueClient(Protocol):
    async def push(self, queue: str, task: "ComplianceTask") -> None: ...
    async def depth(self, queue: str) -> int: ...


@dataclass(frozen=True)
class RegionRoute:
    jurisdiction: str          # e.g. "US-DE"
    primary: str               # owning regional queue
    fallbacks: tuple[str, ...] # ordered chain tried when primary is unavailable
    depth_ceiling: int         # backpressure threshold for the primary queue


@dataclass
class ComplianceTask:
    entity_id: str
    jurisdiction: str
    statutory_deadline: float          # Unix timestamp
    routing_hash: str = ""
    assigned_queue: Optional[str] = None
    fallback_depth: int = 0

    def __post_init__(self) -> None:
        # Stable hash over identity + deadline: the same task always routes the
        # same way, so a retry is idempotent rather than a second enqueue.
        if not self.routing_hash:
            payload = f"{self.entity_id}:{self.jurisdiction}:{self.statutory_deadline}"
            self.routing_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]


@dataclass(frozen=True)
class RoutingRecord:
    routing_hash: str
    entity_id: str
    jurisdiction: str
    status: RouteStatus
    queue: str
    fallback_depth: int
    timestamp: float
    integrity_hash: str = field(default="")

    def sealed(self) -> "RoutingRecord":
        # Tamper-evident seal over everything but the seal field itself.
        body = {k: v for k, v in self.__dict__.items() if k != "integrity_hash"}
        body["status"] = self.status.value
        canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
        digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        return RoutingRecord(**{**body, "status": self.status, "integrity_hash": digest})


class RegionalRouter:
    ESCALATION_QUEUE = "global-compliance-escalation"

    def __init__(self, table: dict[str, RegionRoute], client: QueueClient) -> None:
        self.table = table
        self.client = client

    async def _try_queue(self, queue: str, ceiling: int, task: ComplianceTask) -> bool:
        # Probe backpressure before committing: a saturated queue would silently
        # delay a statutory deadline, so we treat it as unavailable and fall back.
        try:
            if await self.client.depth(queue) >= ceiling:
                return False
            await asyncio.wait_for(self.client.push(queue, task), timeout=5.0)
            return True
        except (asyncio.TimeoutError, ConnectionError) as exc:
            logger.warning(json.dumps({"event": "queue_unavailable", "queue": queue,
                                       "entity_id": task.entity_id, "error": str(exc)}))
            return False

    async def route(self, task: ComplianceTask) -> RoutingRecord:
        route = self.table.get(task.jurisdiction)
        if route is None:  # unknown jurisdiction is a data defect, not a routing target
            return self._seal(task, RouteStatus.ESCALATED, self.ESCALATION_QUEUE)

        # Primary first, then each fallback in declared order — deterministic chain.
        candidates = (route.primary, *route.fallbacks)
        for depth, queue in enumerate(candidates):
            ceiling = route.depth_ceiling if depth == 0 else route.depth_ceiling * 2
            if await self._try_queue(queue, ceiling, task):
                task.assigned_queue, task.fallback_depth = queue, depth
                return self._seal(task, RouteStatus.ROUTED, queue)

        # Chain exhausted: never drop a deadline — hand to supervised escalation.
        await self.client.push(self.ESCALATION_QUEUE, task)
        task.fallback_depth = len(candidates)
        return self._seal(task, RouteStatus.ESCALATED, self.ESCALATION_QUEUE)

    def _seal(self, task: ComplianceTask, status: RouteStatus, queue: str) -> RoutingRecord:
        record = RoutingRecord(task.routing_hash, task.entity_id, task.jurisdiction,
                               status, queue, task.fallback_depth, time.time()).sealed()
        logger.info(json.dumps({
            "level": "WARNING" if status is RouteStatus.ESCALATED else "INFO",
            "event": "task_routed", "routing_hash": record.routing_hash,
            "entity_id": record.entity_id, "jurisdiction": record.jurisdiction,
            "status": record.status.value, "queue": record.queue,
            "fallback_depth": record.fallback_depth, "seal": record.integrity_hash,
        }))
        return record

    async def run(self, source: AsyncGenerator[ComplianceTask, None]) -> list[RoutingRecord]:
        return [await self.route(task) async for task in source]