"""ORM models for the Leaver Blocked hard-gate. Two tables: * ``leaver_block_audit`` — every rejected attempt to flip a person to ``employment_status='left'``. Append-only; rows accumulate forever (small volume, high audit value). Powers the auditor's "who tried to bypass leaver-block" report and the daily digest's "rejected status flips" line. * ``leaver_dual_sign_overrides`` — escape hatch. An IT admin initiates an override naming the specific blocker IDs they want waived; a second signer (department director, typically) reviews and either approves or rejects. On approval the override is "executed" and the ProcessAdvanceService is invoked under the override flag, which causes ``LeaverBlockGuard`` to permit the flip-to-left for exactly the listed blockers. """ from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import ( JSON, DateTime, ForeignKey, Index, Integer, String, Text, ) from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid # ── leaver_block_audit denied_reason values ───────────────────────── DENIED_PENDING_BLOCKERS = "pending_blockers" DENIED_NO_FALLBACK_OWNER = "no_fallback_owner" DENIED_PROCESS_NOT_FOUND = "process_not_found" DENIED_PROCESS_CANCELLED = "process_cancelled" # ── attempted_via values ──────────────────────────────────────────── LEAVER_VIA_ADMIN_API = "admin_api" LEAVER_VIA_HRIS_SYNC = "hris_sync" LEAVER_VIA_PROCESS_ADVANCE = "process_advance" LEAVER_VIA_DUAL_SIGN_OVERRIDE = "dual_sign_override" LEAVER_VIA_CLI = "cli" LEAVER_VIA_LISTENER_TRAP = "listener_trap" # ── Dual-sign override statuses ───────────────────────────────────── DUAL_SIGN_PENDING_SECONDARY = "pending_secondary" DUAL_SIGN_APPROVED = "approved" DUAL_SIGN_REJECTED = "rejected" DUAL_SIGN_EXECUTED = "executed" DUAL_SIGN_CANCELLED = "cancelled" DUAL_SIGN_EXPIRED = "expired" ALL_DUAL_SIGN_STATUSES = frozenset({ DUAL_SIGN_PENDING_SECONDARY, DUAL_SIGN_APPROVED, DUAL_SIGN_REJECTED, DUAL_SIGN_EXECUTED, DUAL_SIGN_CANCELLED, DUAL_SIGN_EXPIRED, }) # Statuses that block the creation of a NEW override on the same # process. Any process in this state already has work in flight. BLOCKING_DUAL_SIGN_STATUSES = frozenset({ DUAL_SIGN_PENDING_SECONDARY, DUAL_SIGN_APPROVED, # approved but not yet executed }) # Override TTL — how long a pending_secondary override stays valid # before auto-expiring. A director who hasn't responded in 7 days # the IT admin must re-initiate. OVERRIDE_TTL_DAYS = 7 # Minimum length on free-text reason fields. Prevents drive-by # overrides with one-character justifications. MIN_REASON_CHARS = 30 class LeaverBlockAudit(Base): """Every rejected flip-to-left attempt is recorded here.""" __tablename__ = "leaver_block_audit" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False, ) attempted_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) attempted_by_user_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), ) attempted_via: Mapped[str] = mapped_column(String(64), nullable=False) target_status: Mapped[str] = mapped_column(String(32), nullable=False) denied_reason: Mapped[str] = mapped_column(String(64), nullable=False) pending_blocker_count: Mapped[int] = mapped_column( Integer, nullable=False, default=0, ) pending_blockers: Mapped[list[dict[str, Any]]] = mapped_column( JSON, nullable=False, ) detail: Mapped[dict[str, Any] | None] = mapped_column(JSON) trace_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False) client_ip: Mapped[str | None] = mapped_column(String(64)) user_agent: Mapped[str | None] = mapped_column(String(512)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) __table_args__ = ( Index("idx_leaver_block_audit_person_at", "person_id", "attempted_at"), Index("idx_leaver_block_audit_reason", "denied_reason"), Index("idx_leaver_block_audit_via", "attempted_via"), ) def __repr__(self) -> str: return ( f"" ) class LeaverDualSignOverride(Base): """Two-person co-signed override of the leaver-block invariant. Workflow: 1. IT admin initiates: writes a row with status='pending_secondary', names the specific blocker IDs to waive, and writes a >=30-char reason. This SLA-clocks immediately for the secondary signer. 2. Secondary signer (typically the leaver's department director) opens the row, reviews the waived blockers, and either approves (status -> 'approved') with a >=30-char reason of their own, or rejects (status -> 'rejected') with a reason. 3. On approval, the IT admin returns and clicks "execute" — the row's status flips to 'executed', and the ProcessAdvanceService is invoked with this override's id in scope, which causes LeaverBlockGuard to permit the flip-to- left specifically for the named blockers. 4. On TTL expiry without secondary action, status -> 'expired' (idempotent, run by leaver.escalation_tick). Every transition writes an audit_logs row and an event in the associated WorkflowRun's step log. Override rows are immutable once executed/rejected/expired. """ __tablename__ = "leaver_dual_sign_overrides" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) process_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("offboarding_processes.id", ondelete="CASCADE"), nullable=False, ) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False, ) status: Mapped[str] = mapped_column( String(32), nullable=False, default=DUAL_SIGN_PENDING_SECONDARY, ) primary_signer_user_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False, ) primary_signed_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) primary_reason: Mapped[str] = mapped_column(Text, nullable=False) secondary_signer_user_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), ) secondary_signed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) secondary_reason: Mapped[str | None] = mapped_column(Text) overridden_blocker_ids: Mapped[list[str]] = mapped_column( JSON, nullable=False, ) overridden_blockers_snapshot: Mapped[list[dict[str, Any]]] = mapped_column( JSON, nullable=False, ) executed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) rejected_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) rejected_reason: Mapped[str | None] = mapped_column(Text) expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) trace_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) __table_args__ = ( Index("idx_dual_sign_process_status", "process_id", "status"), Index("idx_dual_sign_status", "status"), ) def is_active(self) -> bool: """True while the override could still affect the leaver-block gate — pending_secondary or approved-not-yet-executed.""" return self.status in BLOCKING_DUAL_SIGN_STATUSES def __repr__(self) -> str: return ( f"" )