"""Successor resolution for TERMINAL leaver dispositions (2026-08-13). Why this module exists ---------------------- Helix cuts a leaver's vendor accounts in two stages: a day-of **suspend** (reversible) and, ``delete_lead_days`` later, a **hard DELETE**. The delete is where data dies — Google's Drive/Calendar content and Zoom's cloud recordings live under the account being removed. Until 2026-08-13 the terminal delete took *no successor and did no data transfer*. That was safe only under an assumption that turned out to be false: "the blocker cascade already transferred everything before the process completed". The cascade only fires for resources Helix still sees on the leaver at scan time, and the Google Workspace seat — the ONE blocker whose revoke performs the Drive/Calendar transfer — is routinely gone by then: * ``google.seat_sync`` auto-revokes the Helix seat the morning after the account stops being active in the tenant ("account no longer active in Google Workspace tenant"). A leaver whose account was suspended by hand before Helix ran loses the seat row, so the offboarding scan finds no GW seat and creates no blocker (prod: alison.u@pingcap.com, 2026-08-12 — deleted with zero data transfer). * A blocker that DID exist gets ``[stale_after_rescan]`` auto-skipped when the seat vanishes mid-flight (prod: hice@pingcap.com, 2026-08-02). Both leavers were hard-deleted with no handover. This resolver is half of the fix: the delete gates in ``OffboardingService`` now REFUSE to delete without a successor, and use the resolution below to perform the transfer themselves. The other half is the evidence check in those gates. Resolution order (first active, non-leaver candidate wins) ---------------------------------------------------------- 1. ``OffboardingProcess.default_handover_person_id`` — IT's explicit choice on the offboarding console, and the value prefilled from the leaver's HRIS manager at process start. 2. The leaver's direct manager (``manager_person_id``) — covers processes created before the manager link resolved, where the prefill ran dry. 3. The manager chain upward, bounded — a leaver whose own lead left in the same reorg still has a live owner above them. 4. ``offboarding.default_handover_fallback_person_id`` platform setting — the org-level backstop (typically the IT ops lead) so an orphaned leaver at the top of a tree still resolves. A candidate must be an ACTIVE, non-deleted person with a work_email, and must never be the leaver themselves. When nothing resolves, callers are expected to refuse the irreversible action and page IT — never to delete anyway. "Better a stuck account than silently destroyed data" is the whole point. """ from __future__ import annotations import logging from dataclasses import dataclass from typing import TYPE_CHECKING, Any from uuid import UUID from app.identity.person_models import IdentityPerson if TYPE_CHECKING: # pragma: no cover - typing only from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # Platform setting holding the org-wide fallback handover person (a person # UUID as a string). Unset (the default) simply drops tier 4 — the resolver # then returns "none" and the caller refuses the terminal action. SETTING_FALLBACK_HANDOVER_PERSON = ( "offboarding.default_handover_fallback_person_id" ) # Bound on the upward manager walk. Real org trees are shallow; the cap only # guards a pathological / cyclic ``manager_person_id`` chain (which prod HRIS # data HAS produced — a two-person mutual-manager loop in 2026-06). MANAGER_CHAIN_MAX_DEPTH = 5 # Resolution tiers, most-specific first. Surfaced in audit detail + the # ledger so an operator can tell "IT chose this person" from "we fell back # to the org default" without re-deriving it. TIER_DEFAULT_HANDOVER = "default_handover" TIER_MANAGER = "manager" TIER_MANAGER_CHAIN = "manager_chain" TIER_PLATFORM_FALLBACK = "platform_fallback" TIER_NONE = "none" # ``reason`` values when nothing resolves — these reach the IT notification # body, so they are diagnoses, not generic failures. REASON_NO_CANDIDATE = "no_active_candidate_in_chain" @dataclass(frozen=True, slots=True) class SuccessorResolution: """Outcome of a terminal-disposition successor lookup. ``person`` is ``None`` exactly when no tier produced an eligible candidate; ``tier`` is then :data:`TIER_NONE` and ``reason`` explains what was tried. ``email`` is a convenience mirror of ``person.work_email`` (vendor APIs address people by email). """ person: IdentityPerson | None tier: str reason: str | None = None @property def ok(self) -> bool: return self.person is not None @property def email(self) -> str | None: if self.person is None: return None return (self.person.work_email or "").strip() or None @property def person_id(self) -> UUID | None: return None if self.person is None else self.person.id def as_detail(self) -> dict[str, Any]: """Compact audit/ledger payload (safe to persist in JSON columns).""" return { "successor_tier": self.tier, "successor_email": self.email, "successor_person_id": ( str(self.person_id) if self.person_id else None ), "reason": self.reason, } def _eligible(candidate: IdentityPerson | None, *, leaver_id: UUID) -> bool: """A candidate may receive a leaver's data only if they are a live employee with a mailbox — and are not the leaver themselves. The self-check matters: a mis-linked HRIS record where ``manager_person_id`` points at the person's own row would otherwise "transfer" the data to the account we are about to delete, which reads as a successful handover in every log while destroying the content. """ if candidate is None: return False if candidate.id == leaver_id: return False if candidate.deleted_at is not None: return False if (candidate.employment_status or "").lower() != "active": return False return bool((candidate.work_email or "").strip()) async def resolve_terminal_successor( db: AsyncSession, *, leaver: IdentityPerson, default_handover_person_id: UUID | None, ) -> SuccessorResolution: """Resolve who inherits ``leaver``'s vendor-side data before a hard delete. See the module docstring for the tier order and rationale. Never raises: a lookup failure degrades to the next tier, and an exhausted chain returns ``tier=none`` so the caller can refuse the delete and page IT. Callers MUST treat a non-``ok`` resolution as "do not delete". """ # ── Tier 1: IT's explicit / prefilled choice on the process. ── if default_handover_person_id is not None: candidate = await db.get(IdentityPerson, default_handover_person_id) if _eligible(candidate, leaver_id=leaver.id): return SuccessorResolution(candidate, TIER_DEFAULT_HANDOVER) logger.info( "successor_resolver: default_handover %s not eligible for " "leaver %s — falling through to the manager chain", default_handover_person_id, leaver.work_email, ) # ── Tiers 2+3: the manager chain, bounded and cycle-guarded. ── seen: set[UUID] = {leaver.id} current_id = leaver.manager_person_id depth = 0 while current_id is not None and depth < MANAGER_CHAIN_MAX_DEPTH: if current_id in seen: # Cyclic manager data (HRIS has produced mutual-manager pairs); # stop rather than loop to the depth cap. logger.warning( "successor_resolver: manager cycle at %s while resolving " "successor for %s", current_id, leaver.work_email, ) break seen.add(current_id) candidate = await db.get(IdentityPerson, current_id) if candidate is None: break if _eligible(candidate, leaver_id=leaver.id): return SuccessorResolution( candidate, TIER_MANAGER if depth == 0 else TIER_MANAGER_CHAIN, ) # Manager themselves left / was deleted — keep walking upward. current_id = candidate.manager_person_id depth += 1 # ── Tier 4: org-wide platform fallback. ── from app.integrations.service import get_platform_setting # noqa: PLC0415 raw = await get_platform_setting( SETTING_FALLBACK_HANDOVER_PERSON, default=None, session=db, ) fallback_id = coerce_person_id(raw) if fallback_id is not None: candidate = await db.get(IdentityPerson, fallback_id) if _eligible(candidate, leaver_id=leaver.id): return SuccessorResolution(candidate, TIER_PLATFORM_FALLBACK) logger.warning( "successor_resolver: platform fallback handover person %s is " "not an eligible recipient (missing / inactive / deleted)", fallback_id, ) return SuccessorResolution(None, TIER_NONE, REASON_NO_CANDIDATE) def coerce_person_id(raw: Any) -> UUID | None: """Parse the platform setting into a UUID. The ``platform_settings.value`` column is JSON, so ops can land the value as a bare string or as ``{"person_id": "..."}`` depending on which admin surface wrote it. Both shapes are accepted; anything else is ignored with a warning rather than raising into a cron tick. """ if raw is None: return None value = raw if isinstance(raw, dict): value = raw.get("person_id") or raw.get("value") if isinstance(value, UUID): return value if not isinstance(value, str) or not value.strip(): return None try: return UUID(value.strip()) except ValueError: logger.warning( "successor_resolver: %s is not a valid person UUID: %r", SETTING_FALLBACK_HANDOVER_PERSON, value, ) return None __all__ = [ "MANAGER_CHAIN_MAX_DEPTH", "REASON_NO_CANDIDATE", "SETTING_FALLBACK_HANDOVER_PERSON", "TIER_DEFAULT_HANDOVER", "TIER_MANAGER", "TIER_MANAGER_CHAIN", "TIER_NONE", "TIER_PLATFORM_FALLBACK", "SuccessorResolution", "coerce_person_id", "resolve_terminal_successor", ]