Classifying Entities Into a Deterministic, Single-Intent Filing Taxonomy #

from __future__ import annotations
import re
import logging
from enum import Enum
from typing import Optional
from pydantic import BaseModel, field_validator

logger = logging.getLogger("compliance.taxonomy")

class JurisdictionCode(str, Enum):
    DE = "US-DE"
    CA = "US-CA"
    NY = "US-NY"
    TX = "US-TX"
    # Extend with the full ISO 3166-2 registry in production

class EntityType(str, Enum):
    DOMESTIC_CORP = "domestic_c_corp"
    FOREIGN_QUALIFIED = "foreign_qualified"
    LLC = "limited_liability_company"
    PARTNERSHIP = "partnership"

_JURISDICTION_MAP = {
    "DE": JurisdictionCode.DE, "DELAWARE": JurisdictionCode.DE,
    "CA": JurisdictionCode.CA, "CALIFORNIA": JurisdictionCode.CA,
    "NY": JurisdictionCode.NY, "NEW YORK": JurisdictionCode.NY,
    "TX": JurisdictionCode.TX, "TEXAS": JurisdictionCode.TX,
}

_ENTITY_ALIAS_MAP = {
    "c-corp": EntityType.DOMESTIC_CORP,
    "llc": EntityType.LLC,
    "foreign": EntityType.FOREIGN_QUALIFIED,
    "lp": EntityType.PARTNERSHIP,
}

class RawEntityPayload(BaseModel):
    entity_name: str
    formation_state: str
    entity_type_raw: str
    ein_prefix: Optional[str] = None
    fiscal_year_end_month: Optional[int] = None

class NormalizedEntity(BaseModel):
    entity_id: str
    canonical_name: str
    jurisdiction_iso: JurisdictionCode
    entity_type: EntityType
    ein_prefix: Optional[str] = None
    fiscal_year_end_month: Optional[int] = None

    @field_validator("canonical_name")
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return re.sub(r"\s+", " ", v.strip().upper())

    @classmethod
    def from_raw(cls, entity_id: str, raw: RawEntityPayload) -> "NormalizedEntity":
        """Normalize a RawEntityPayload into a typed NormalizedEntity."""
        jkey = re.sub(r"\s+", " ", raw.formation_state.strip().upper())
        jurisdiction = _JURISDICTION_MAP.get(jkey)
        if jurisdiction is None:
            raise ValueError(f"Unsupported jurisdiction code: {jkey!r}")

        etype_key = raw.entity_type_raw.strip().lower()
        entity_type = _ENTITY_ALIAS_MAP.get(etype_key)
        if entity_type is None:
            raise ValueError(f"Unrecognized entity type alias: {etype_key!r}")

        return cls(
            entity_id=entity_id,
            canonical_name=raw.entity_name,
            jurisdiction_iso=jurisdiction,
            entity_type=entity_type,
            ein_prefix=raw.ein_prefix,
            fiscal_year_end_month=raw.fiscal_year_end_month,
        )

from dataclasses import dataclass

class ClassificationIntent(str, Enum):
    ANNUAL_REPORT = "annual_report"
    FRANCHISE_TAX = "franchise_tax"
    STATEMENT_OF_INFO = "statement_of_info"
    FOREIGN_QUALIFICATION = "foreign_qualification"

@dataclass(frozen=True)
class ClassificationResult:
    intent: ClassificationIntent
    confidence: float
    rule_applied: str
    metadata: dict

class SingleIntentClassifier:
    def __init__(self, conflict_threshold: float = 0.65) -> None:
        self.conflict_threshold = conflict_threshold

    def evaluate(self, entity: NormalizedEntity) -> ClassificationResult:
        if entity.entity_type == EntityType.DOMESTIC_CORP:
            return ClassificationResult(
                intent=ClassificationIntent.ANNUAL_REPORT,
                confidence=1.0,
                rule_applied="DOMESTIC_CORP_DEFAULT",
                metadata={"filing_template": "corp_annual_report_v3"},
            )
        if entity.entity_type == EntityType.LLC:
            return ClassificationResult(
                intent=ClassificationIntent.STATEMENT_OF_INFO,
                confidence=0.95,
                rule_applied="LLC_SOS_DEFAULT",
                metadata={"filing_template": "llc_soi_v2"},
            )
        if entity.entity_type == EntityType.FOREIGN_QUALIFIED:
            return ClassificationResult(
                intent=ClassificationIntent.FOREIGN_QUALIFICATION,
                confidence=0.90,
                rule_applied="FOREIGN_QUAL_RULE",
                metadata={"requires_registered_agent": True},
            )
        return self._probabilistic_fallback(entity)

    def _probabilistic_fallback(self, entity: NormalizedEntity) -> ClassificationResult:
        # Lightweight heuristic scoring for hybrid/edge entities.
        # Confidence is held below 0.70 to force Tier 2/3 routing.
        return ClassificationResult(
            intent=ClassificationIntent.ANNUAL_REPORT,
            confidence=0.55,
            rule_applied="HEURISTIC_FALLBACK",
            metadata={"requires_manual_review": True},
        )

class RoutingErrorType(str, Enum):
    CONFLICTING_JURISDICTION = "CONFLICTING_JURISDICTION"
    MISSING_FISCAL_DECLARATION = "MISSING_FISCAL_DECLARATION"
    LOW_CONFIDENCE_THRESHOLD = "LOW_CONFIDENCE_THRESHOLD"
    SCHEMA_MUTATION = "SCHEMA_MUTATION"

class ClassificationRouter:
    def __init__(self, classifier: SingleIntentClassifier) -> None:
        self.classifier = classifier

    def route(self, entity: NormalizedEntity) -> ClassificationResult | None:
        result = self.classifier.evaluate(entity)

        if result.confidence >= 0.95:
            logger.info(
                "tier_1_match",
                extra={"entity_id": entity.entity_id, "intent": result.intent.value},
            )
            return result

        if 0.70 <= result.confidence < 0.95:
            logger.warning(
                "tier_2_heuristic",
                extra={"entity_id": entity.entity_id, "confidence": result.confidence},
            )
            self._enqueue_async_validation(entity, result)
            return result

        self._raise_compliance_alert(entity, RoutingErrorType.LOW_CONFIDENCE_THRESHOLD)
        return None

    def _enqueue_async_validation(
        self, entity: NormalizedEntity, result: ClassificationResult
    ) -> None:
        # Production: publish to SQS/Kafka with a deterministic idempotency key.
        logger.info("async_validation_queued", extra={"entity_id": entity.entity_id})

    def _raise_compliance_alert(
        self, entity: NormalizedEntity, error_type: RoutingErrorType
    ) -> None:
        logger.critical(
            "tier_3_blocked",
            extra={
                "entity_id": entity.entity_id,
                "error_type": error_type.value,
                "requires_legal_review": True,
            },
        )

import pytest

def make_entity(jurisdiction: JurisdictionCode, etype: EntityType) -> NormalizedEntity:
    return NormalizedEntity(
        entity_id="ent_test",
        canonical_name="ACME HOLDINGS",
        jurisdiction_iso=jurisdiction,
        entity_type=etype,
    )

def test_domestic_corp_is_tier_1_annual_report() -> None:
    clf = SingleIntentClassifier()
    result = clf.evaluate(make_entity(JurisdictionCode.DE, EntityType.DOMESTIC_CORP))
    assert result.intent is ClassificationIntent.ANNUAL_REPORT
    assert result.confidence >= 0.95  # Tier 1: direct execution

def test_partnership_falls_back_below_threshold() -> None:
    clf = SingleIntentClassifier()
    result = clf.evaluate(make_entity(JurisdictionCode.CA, EntityType.PARTNERSHIP))
    assert result.confidence < 0.70  # forces Tier 3 review
    assert result.metadata["requires_manual_review"] is True

def test_unknown_jurisdiction_raises() -> None:
    with pytest.raises(ValueError, match="Unsupported jurisdiction"):
        NormalizedEntity.from_raw(
            "ent_x",
            RawEntityPayload(
                entity_name="X", formation_state="Atlantis", entity_type_raw="llc"
            ),
        )

def test_low_confidence_returns_none_from_router() -> None:
    router = ClassificationRouter(SingleIntentClassifier())
    assert router.route(make_entity(JurisdictionCode.TX, EntityType.PARTNERSHIP)) is None