Tracking Multi-Party Submittal Approval Chains

from __future__ import annotations

import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Literal, Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
logger = logging.getLogger("submittal.approval_chain")

ApproverRole = Literal["general_contractor", "architect", "engineer"]
# Canonical routing order: GC coordinates, architect confirms intent, engineer signs off.
ROUTING_ORDER: tuple[ApproverRole, ...] = ("general_contractor", "architect", "engineer")

StepStatus = Literal["pending", "approved", "approved_as_noted", "rejected", "revise_and_resubmit"]
Decision = Literal["approved", "approved_as_noted", "rejected", "revise_and_resubmit"]
ADVANCING = ("approved", "approved_as_noted")
HALTING = ("rejected", "revise_and_resubmit")


class ApprovalStep(BaseModel):
    """One party's slot in the approval chain — immutable once decided."""

    model_config = ConfigDict(frozen=True)

    role: ApproverRole
    party_name: str = Field(min_length=2, max_length=120)
    status: StepStatus = "pending"
    decided_at: Optional[datetime] = None
    comment: Optional[str] = Field(default=None, max_length=400)

    @field_validator("decided_at")
    @classmethod
    def _require_tz(cls, v: Optional[datetime]) -> Optional[datetime]:
        if v is not None and v.tzinfo is None:
            raise ValueError("decided_at must be timezone-aware (ISO 8601)")
        return v.astimezone(timezone.utc) if v is not None else None

    @model_validator(mode="after")
    def _decided_has_timestamp(self) -> "ApprovalStep":
        decided = self.status != "pending"
        if decided and self.decided_at is None:
            raise ValueError(f"{self.role} is {self.status} but has no decided_at")
        if not decided and self.decided_at is not None:
            raise ValueError(f"{self.role} is pending but carries a decided_at")
        return self

WorkflowState = Literal["in_review", "approved", "returned"]


class ApprovalWorkflow(BaseModel):
    """An ordered, immutable multi-party approval chain for one submittal revision."""

    model_config = ConfigDict(frozen=True)

    submittal_number: str = Field(pattern=r"^\d{2} \d{2} \d{2}-\d{3}$")
    revision_id: str = Field(pattern=r"^R\d{2}$")
    wbs_node: str = Field(pattern=r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$")
    steps: tuple[ApprovalStep, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def _ordered_and_sequential(self) -> "ApprovalWorkflow":
        roles = tuple(s.role for s in self.steps)
        if roles != tuple(r for r in ROUTING_ORDER if r in roles):
            raise ValueError(f"steps must follow routing order {ROUTING_ORDER}, got {roles}")
        if len(set(roles)) != len(roles):
            raise ValueError("each party may appear at most once in the chain")
        require_pending = False
        for s in self.steps:
            if require_pending and s.status != "pending":
                raise ValueError(f"{s.role} decided before an earlier step finished")
            if s.status in HALTING or s.status == "pending":
                require_pending = True  # nothing after a halt or a pending step may be decided
        return self

    @property
    def state(self) -> WorkflowState:
        if any(s.status in HALTING for s in self.steps):
            return "returned"
        if all(s.status in ADVANCING for s in self.steps):
            return "approved"
        return "in_review"

    @property
    def next_approver(self) -> Optional[ApprovalStep]:
        if self.state != "in_review":
            return None
        return next((s for s in self.steps if s.status == "pending"), None)

    def apply_decision(
        self, role: ApproverRole, decision: Decision, decided_at: datetime,
        comment: Optional[str] = None,
    ) -> "ApprovalWorkflow":
        target = next((s for s in self.steps if s.role == role), None)
        if target is None:
            raise ValueError(f"{role} is not part of this chain")
        if target.status == decision:
            logger.info("idempotent replay: %s already %s", role, decision)
            return self  # retry-safe no-op
        current = self.next_approver
        if current is None or current.role != role:
            raise ValueError(f"out-of-turn decision: {role} is not the current approver")
        updated = tuple(
            s.model_copy(update={"status": decision, "decided_at": decided_at, "comment": comment})
            if s.role == role else s
            for s in self.steps
        )
        return ApprovalWorkflow(
            submittal_number=self.submittal_number, revision_id=self.revision_id,
            wbs_node=self.wbs_node, steps=updated,
        )

AUTO_ROUTE = Decimal("0.92")     # site-canonical: >= 0.92 applies automatically
HUMAN_REVIEW = Decimal("0.75")   # 0.75-0.92 waits for a coordinator; < 0.75 quarantines

RouteState = Literal["auto_route", "human_review", "quarantine"]


def route_inbound_approval(
    workflow: ApprovalWorkflow, inbound_number: str, match_confidence: Decimal
) -> RouteState:
    """Decide whether an inbound approval message binds to this workflow."""
    confidence = Decimal("0.99") if inbound_number == workflow.submittal_number else match_confidence
    if confidence >= AUTO_ROUTE:
        return "auto_route"
    if confidence >= HUMAN_REVIEW:
        return "human_review"
    return "quarantine"


if __name__ == "__main__":
    wf = ApprovalWorkflow(
        submittal_number="05 12 00-007", revision_id="R01", wbs_node="PROJ-014-STR-02",
        steps=(
            ApprovalStep(role="general_contractor", party_name="Turner Construction"),
            ApprovalStep(role="architect", party_name="HOK"),
            ApprovalStep(role="engineer", party_name="Thornton Tomasetti"),
        ),
    )
    now = datetime(2026, 6, 1, 10, 0, tzinfo=timezone.utc)
    print("next:", wf.next_approver.role)
    wf = wf.apply_decision("general_contractor", "approved", now)
    replay = wf.apply_decision("general_contractor", "approved", now)  # retry
    print("replay is a no-op:", replay is wf, "| next:", wf.next_approver.role)
    wf = wf.apply_decision("architect", "approved_as_noted", now)
    wf = wf.apply_decision("engineer", "approved", now)
    print("final state:", wf.state)