Security & Data Boundaries in Corporate Entity Compliance & Annual Filing Automation #

from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field, field_validator, ValidationError
from datetime import date
from enum import StrEnum

class ComplianceBoundaryError(Exception):
    """Raised when a payload violates a compliance boundary contract."""
    def __init__(self, code: str, detail: str, jurisdiction: str = "UNKNOWN"):
        self.code = code
        self.detail = detail
        self.jurisdiction = jurisdiction
        super().__init__(f"[{code}] {detail} (Jurisdiction: {jurisdiction})")

class Jurisdiction(StrEnum):
    DE = "DE"
    CA = "CA"
    NY = "NY"
    TX = "TX"

class EntityRecord(BaseModel, strict=True):
    entity_id: str = Field(pattern=r"^[A-Z]{2}-\d{8}$", description="State-prefixed statutory ID")
    jurisdiction: Jurisdiction
    entity_type: Literal["LLC", "CORP", "LP", "LLP"]
    formation_date: date
    registered_agent_id: str
    annual_report_due: date | None = None

    @field_validator("entity_id", "registered_agent_id")
    @classmethod
    def enforce_boundary_format(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Statutory identifier cannot be empty or whitespace")
        return v.strip()

def validate_ingestion_payload(raw: dict) -> EntityRecord:
    """Deterministic boundary validation. Rejects non-compliant payloads at the perimeter."""
    try:
        return EntityRecord.model_validate(raw)
    except ValidationError as exc:
        raise ComplianceBoundaryError(
            code="SCHEMA_VIOLATION",
            detail=str(exc),
            jurisdiction=raw.get("jurisdiction", "UNKNOWN")
        ) from exc

import asyncio
import time
from dataclasses import dataclass
from typing import Protocol
from enum import Enum
from .validation import EntityRecord  # from the validation block above

class TransportBoundaryError(Exception):
    def __init__(self, code: str, detail: str, fallback_triggered: bool = False):
        self.code = code
        self.detail = detail
        self.fallback_triggered = fallback_triggered
        super().__init__(f"[{code}] {detail}")

class CircuitState(Enum):
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    recovery_timeout: float = 30.0
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = time.monotonic()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

    @property
    def is_available(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            elapsed = time.monotonic() - self.last_failure_time
            if elapsed >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
        return False

class RoutingEndpoint(Protocol):
    async def submit(self, payload: EntityRecord) -> dict: ...

async def resilient_route(
    payload: EntityRecord,
    primary: RoutingEndpoint,
    fallback: RoutingEndpoint,
    breaker: CircuitBreaker
) -> dict:
    """Priority routing with circuit-breaker isolation and exponential backoff."""
    if not breaker.is_available:
        return await fallback.submit(payload)

    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = await primary.submit(payload)
            breaker.record_success()
            return response
        except (ConnectionError, TimeoutError, RuntimeError) as exc:
            breaker.record_failure()
            if attempt == max_retries - 1:
                raise TransportBoundaryError(
                    code="PRIMARY_ENDPOINT_FAILURE",
                    detail=str(exc),
                    fallback_triggered=True
                ) from exc
            await asyncio.sleep(2 ** attempt)  # Exponential backoff
    return await fallback.submit(payload)

import time
import logging
from typing import Optional

class ComplianceBoundaryError(Exception):
    """Base exception for all boundary-enforcement failures."""
    def __init__(self, code: str, detail: str, jurisdiction: str = "UNKNOWN", context: Optional[dict] = None):
        self.code = code
        self.detail = detail
        self.jurisdiction = jurisdiction
        self.context = context or {}
        super().__init__(f"[{code}] {detail} (Jurisdiction: {jurisdiction})")

class TransportBoundaryError(ComplianceBoundaryError):
    fallback_triggered: bool = False

class DeadlineBoundaryError(ComplianceBoundaryError):
    days_past_cutoff: int = 0

def log_boundary_event(error: ComplianceBoundaryError, logger: logging.Logger) -> None:
    """Structured audit logging for compliance boundary violations."""
    logger.error(
        "BOUNDARY_VIOLATION",
        extra={
            "error_code": error.code,
            "jurisdiction": error.jurisdiction,
            "compliance_impact": "HIGH" if "DEADLINE" in error.code else "MEDIUM",
            "audit_hash": hash(f"{error.code}:{error.detail}:{time.time()}")
        }
    )