Implementing Jurisdictional Fallback Rules for Compliance Data #

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import time
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional

from pydantic import BaseModel, Field

# Structured JSON logging so every fallback decision is machine-parseable for audit.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("compliance.fallback_engine")


class Jurisdiction(str, Enum):
    DE = "DE"
    CA = "CA"
    NY = "NY"
    TX = "TX"


class FallbackSource(str, Enum):
    PRIMARY_API = "primary_api"
    COMMERCIAL_REGISTRY = "commercial_registry"
    STATUTORY_DEFAULT = "statutory_default"
    MANUAL_OVERRIDE = "manual_override"


class CompliancePayload(BaseModel):
    entity_id: str
    jurisdiction: Jurisdiction
    filing_deadline: Optional[datetime] = None
    statutory_fee: Optional[float] = None
    registered_agent_required: Optional[bool] = None
    source: FallbackSource = FallbackSource.PRIMARY_API
    confidence_score: float = Field(ge=0.0, le=1.0, default=1.0)
    audit_hash: str = ""
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    def generate_audit_hash(self) -> str:
        # mode="json" serializes datetime -> ISO 8601 and Enum -> .value, so json.dumps
        # never raises on a raw datetime/Enum. This digest is the non-repudiation anchor.
        serializable = self.model_dump(mode="json", exclude={"audit_hash"})
        payload_bytes = json.dumps(serializable, sort_keys=True).encode()
        return hashlib.sha256(payload_bytes).hexdigest()


class CircuitBreaker:
    """Trips open after repeated primary-source failures to stop hammering a dead portal."""

    def __init__(self, failure_threshold: int = 3, cooldown_seconds: float = 60.0):
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.failures = 0
        self.last_failure_time = 0.0
        self.state = "closed"

    def record_success(self) -> None:
        self.failures = 0
        self.state = "closed"

    def record_failure(self) -> None:
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        # After cooldown, allow a single probe (half-open) before fully reopening.
        if self.state == "open" and (time.time() - self.last_failure_time) > self.cooldown_seconds:
            self.state = "half-open"
            return True
        return False


# Codified baseline per statute — the known-good value used when live sources fail.
STATUTORY_DEFAULTS: dict[Jurisdiction, dict[str, Any]] = {
    Jurisdiction.DE: {"deadline": "2026-03-01", "fee": 300.0, "ra_required": True},   # DGCL § 503
    Jurisdiction.CA: {"deadline": "2026-04-01", "fee": 25.0, "ra_required": False},   # Corp. Code § 1502
    Jurisdiction.NY: {"deadline": "2026-02-01", "fee": 9.0, "ra_required": True},     # BSC § 408 (biennial)
    Jurisdiction.TX: {"deadline": "2026-05-15", "fee": 0.0, "ra_required": True},     # Tax Code § 171 margin report
}

# Confidence degrades by tier; anything below HALT_THRESHOLD must not auto-file.
CONFIDENCE = {
    FallbackSource.PRIMARY_API: 1.0,
    FallbackSource.COMMERCIAL_REGISTRY: 0.85,
    FallbackSource.STATUTORY_DEFAULT: 0.60,
    FallbackSource.MANUAL_OVERRIDE: 0.40,
}
HALT_THRESHOLD = 0.50


class FallbackEngine:
    def __init__(self, cache_client: Optional[Any] = None):
        self.cache = cache_client
        self.breakers: dict[Jurisdiction, CircuitBreaker] = {j: CircuitBreaker() for j in Jurisdiction}

    async def _fetch_primary(self, entity_id: str, jurisdiction: Jurisdiction) -> Optional[dict[str, Any]]:
        try:
            # Hard timeout: a slow portal must not block the scheduler thread.
            return await asyncio.wait_for(self._call_portal(jurisdiction), timeout=3.0)
        except asyncio.TimeoutError:
            logger.warning(json.dumps({"event": "primary_timeout", "entity_id": entity_id,
                                       "jurisdiction": jurisdiction.value}))
            return None
        except Exception as exc:  # network/parse failure -> treat as a tier miss, not a crash
            logger.error(json.dumps({"event": "primary_failure", "entity_id": entity_id,
                                     "jurisdiction": jurisdiction.value, "error": str(exc)}))
            return None

    async def _fetch_secondary(self, entity_id: str, jurisdiction: Jurisdiction) -> Optional[dict[str, Any]]:
        # Licensed commercial-registry aggregator feed; slightly less authoritative than the state itself.
        return {"deadline": "2026-04-16", "fee": 310.0, "ra_required": True}

    async def resolve_compliance_data(self, entity_id: str, jurisdiction: Jurisdiction) -> CompliancePayload:
        breaker = self.breakers[jurisdiction]

        # Tier 1 — Primary state API, gated by the circuit breaker.
        if breaker.allow_request():
            primary = await self._fetch_primary(entity_id, jurisdiction)
            if primary:
                breaker.record_success()
                return self._finalize(entity_id, jurisdiction, primary, FallbackSource.PRIMARY_API)
            breaker.record_failure()

        # Tier 2 — Commercial registry.
        secondary = await self._fetch_secondary(entity_id, jurisdiction)
        if secondary:
            return self._finalize(entity_id, jurisdiction, secondary, FallbackSource.COMMERCIAL_REGISTRY)

        # Tier 3 — Statutory default baseline (always available).
        return self._finalize(entity_id, jurisdiction, STATUTORY_DEFAULTS[jurisdiction],
                              FallbackSource.STATUTORY_DEFAULT)

    def _finalize(self, entity_id: str, jurisdiction: Jurisdiction, raw: dict[str, Any],
                  source: FallbackSource) -> CompliancePayload:
        payload = CompliancePayload(
            entity_id=entity_id,
            jurisdiction=jurisdiction,
            filing_deadline=datetime.strptime(raw["deadline"], "%Y-%m-%d").replace(tzinfo=timezone.utc),
            statutory_fee=raw.get("fee"),
            registered_agent_required=raw.get("ra_required"),
            source=source,
            confidence_score=CONFIDENCE[source],
        )
        if source is not FallbackSource.PRIMARY_API:
            # A fallback fired: purge stale cache so downstream reads see the corrected value.
            self._invalidate_cache(entity_id, jurisdiction)
        payload.audit_hash = payload.generate_audit_hash()
        logger.info(json.dumps({
            "event": "fallback_resolved", "entity_id": entity_id, "jurisdiction": jurisdiction.value,
            "source": source.value, "confidence": payload.confidence_score, "audit_hash": payload.audit_hash,
            "halt_required": payload.confidence_score < HALT_THRESHOLD,
        }))
        return payload

    def _invalidate_cache(self, entity_id: str, jurisdiction: Jurisdiction) -> None:
        if not self.cache:
            return
        for key in (f"compliance:{entity_id}:{jurisdiction.value}:deadline",
                    f"compliance:{entity_id}:{jurisdiction.value}:fee"):
            self.cache.delete(key)

    async def _call_portal(self, jurisdiction: Jurisdiction) -> dict[str, Any]:
        await asyncio.sleep(0.1)  # stand-in for the real authenticated request
        return {"deadline": "2026-04-15", "fee": 300.0, "ra_required": True}