"""Leaver notice-period predicate — the single source of truth for "is this ``leaver_pending`` person still working?". HRIS-confirm flips a future-dated leaver to ``employment_status= 'leaver_pending'`` the moment IT confirms the termination — often weeks before the last working day (LWD). That status drops the person out of every ``active``/``on_leave`` filter, so any sweep that treats "not active" as "gone" acts weeks early: * identity_sync's out-of-scope sweep would deactivate the vendor account (the 2026-05-28 Thalie Zhuang incident, #1735). * seat-sync reconcilers would drift-revoke every ``software_seat_ allocations`` row for a still-working employee, and on the next tick that re-matches them, re-import the seat as a BRAND-NEW row — which in turn produced duplicate offboarding cancel events (2026-08-11, P0779 / P01361 / Angelyn cohort). The invariant, stated once: ``leaver_pending`` means "scheduled to leave", NOT "gone". Only ``left`` — or ``leaver_pending`` whose LWD has fully ended in the employee's own wall-clock — is gone. """ from __future__ import annotations from datetime import UTC, datetime, timedelta, tzinfo from app.core.working_hours import resolve_timezone from app.identity.person_models import IdentityPerson #: Statuses every seat/account reconciler treats as "no longer holds #: company resources". ``leaver_pending`` is in the list, but callers #: MUST pass it through :func:`leaver_last_working_day_passed` rather #: than excluding it outright — see module docstring. LEAVER_STATUSES: tuple[str, ...] = ("left", "leaver_pending") STATUS_LEAVER_PENDING = "leaver_pending" def leaver_last_working_day_passed(person: IdentityPerson) -> bool: """True iff the person's last working day has fully ended in their LOCAL timezone — i.e. we are at/after the first moment of ``termination_date + 1`` local. Mirrors offboarding's ``_leaver_revoke_not_before`` semantics (revoke fires AFTER the last working day ends in the employee's wall-clock), so no caller acts on a leaver before their notice period closes. A missing ``termination_date`` is treated as NOT passed (conservative: never act early on a ``leaver_pending`` row whose last day HR has not populated) — the status flip to ``left`` clears the guard. """ td = person.termination_date if td is None: return False next_day = td + timedelta(days=1) tz: tzinfo try: tz = resolve_timezone(person) except Exception: # noqa: BLE001 — TZ resolution must never break a sweep tz = UTC not_before = datetime.combine( next_day, datetime.min.time(), tzinfo=tz, ).astimezone(UTC) return datetime.now(UTC) >= not_before def is_within_leaver_notice_period(person: IdentityPerson) -> bool: """True iff ``person`` is a confirmed future leaver who is STILL working — ``leaver_pending`` with a last working day that has not yet ended locally. Callers that exclude leaver statuses from a reconcile should keep these people IN scope: they legitimately still hold their seats, accounts and access until the LWD passes. """ return ( person.employment_status == STATUS_LEAVER_PENDING and not leaver_last_working_day_passed(person) )