"""SLA escalation for leaver-blocked processes. Triggered by the ``leaver.escalation_tick`` scheduled job (every 30 min). Walks every ``OffboardingProcess`` whose ``leaver_date`` has passed AND whose status is still ``blocked``, and advances the SLA level: T+0 (just blocked) → first_blocked_at = NOW, sla_breach_level = 'none' +24h after first_blocked_at → sla = 'warn', manager-chain DM +48h after first_blocked_at → sla = 'urgent', HRBP + it_admin DM +72h after first_blocked_at → sla = 'breach', critical incident + HRBP + it_admin DM The HRBP is resolved live from Feishu's per-user ``custom_attrs`` (see ``app.integrations.feishu_hrbp``) and cached on ``identity_persons.hrbp_person_id`` for subsequent ticks. When HRBP resolution fails (no Feishu account, no HRBP attr value, HRBP email doesn't match a Helix person), an audit row is written and the SLA escalation falls back to ``it_admin``-only — never silently dropped. The leaver themselves is always excluded from every dispatch path. Notifications are sent at most once per (process, level) — the escalation idempotency table is ``OffboardingReminderLog`` records with ``reminder_kind = 'sla_warn' / 'sla_urgent' / 'sla_breach'``. The pending-secondary dual-sign overrides are also expired here (``leaver_dual_sign_overrides.expires_at`` < NOW) so abandoned override requests don't sit forever in the dashboard. """ from __future__ import annotations import logging from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.compliance.models import ( SEVERITY_CRITICAL, TYPE_LEAVER_ORPHAN, ) from app.compliance.service import ComplianceIncidentService from app.core.exceptions import is_transaction_aborting_db_error from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, HR_ROLE, MANAGEMENT_ROLE, ) from app.core.txn import best_effort from app.core.working_hours import resolve_timezone from app.domains.leaver_blocked.models import ( DUAL_SIGN_APPROVED, DUAL_SIGN_EXPIRED, DUAL_SIGN_PENDING_SECONDARY, LeaverDualSignOverride, ) from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_CANCELLED, PROC_COMPLETED, RES_HARDWARE, RESO_PENDING, SLA_BREACH, SLA_NONE, SLA_URGENT, SLA_WARN, OffboardingBlocker, OffboardingProcess, OffboardingReminderLog, ) from app.identity.person_models import IdentityPerson from app.integrations.feishu_hrbp import resolve_hrbp_for_person from app.notifications.service import NotificationEnvelope, NotificationService logger = logging.getLogger(__name__) SLA_REMINDER_KIND_WARN = "sla_warn" SLA_REMINDER_KIND_URGENT = "sla_urgent" SLA_REMINDER_KIND_BREACH = "sla_breach" # #1788: the BREACH DM fan-out gets its OWN idempotency kind, distinct from # the ``sla_breach`` slot that guards the (non-idempotent, fire-once) # critical-incident raise. That lets the DM roll back + retry on an # all-channels-failed delivery WITHOUT re-firing the incident (which would # mint a duplicate ticket). Mirrors the WARN/URGENT empty-delivered rollback. SLA_REMINDER_KIND_BREACH_DM = "sla_breach_dm" SLA_WARN_DELTA = timedelta(hours=24) SLA_URGENT_DELTA = timedelta(hours=48) SLA_BREACH_DELTA = timedelta(hours=72) # Roles that can READ the per-process offboarding detail page # (``/admin/offboarding/process/:processId``, gated by # ``require_lifecycle_read`` on the backend since 2026-07-08). SLA card # recipients holding any of these land on that detail page — where the # blocker list is fully visible — instead of the counts-only # ``/my/team-leavers`` lookout. This is what lets the leaver's HRBP (who # holds ``hr`` but NOT ``global_admin`` — e.g. Robert Wood's HRBP Zedan # Liu) actually SEE which resource is outstanding when they click the # card's action button, closing the "点进去看不了详情" dead-end. # Legacy aliases (``it_admin`` / ``security_auditor``) are included so a # tenant with un-canonicalised role rows still routes correctly. _LIFECYCLE_DETAIL_READ_ROLE_NAMES: tuple[str, ...] = ( GLOBAL_ADMIN_ROLE, "it_admin", HR_ROLE, AUDITOR_ROLE, "security_auditor", MANAGEMENT_ROLE, ) class LeaverBlockedSlaService: """Post-T+0 SLA escalation. Run from the ``leaver.escalation_tick`` scheduled job.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.notify = NotificationService(db) self.incidents = ComplianceIncidentService(db) self.audit = AuditService(db) async def run_tick(self, *, now: datetime | None = None) -> dict[str, Any]: ts_now = now or datetime.now(UTC) # Live processes only procs = list((await self.db.execute( select(OffboardingProcess) .where(OffboardingProcess.status.notin_( [PROC_COMPLETED, PROC_CANCELLED] )) )).scalars().all()) warn_count = 0 urgent_count = 0 breach_count = 0 first_blocked_set = 0 expired_overrides = 0 for proc in procs: # Past leaver_date and still blocked → SLA clock is on if proc.status != PROC_BLOCKED: continue # Pre-launch P1 fix (Round 45, CR-406): use the leaver's # local date, not UTC. PR #443 fixed this in process_advance # + LeaverEscalationService but the SLA cron was missed. # UTC-based gate fired LA manager-chain alerts ~7h before # the employee finished their last working day in PDT. # Round-27 pre-launch fix (2026-05-25): when Person is missing # (hard-deleted — see ``reference_prod_person_cleanup``) or # ``resolve_timezone`` raises, the previous code fell back to # UTC, which gated the SLA tier ~7h early for PDT employees. # Default to Asia/Shanghai (the platform's office timezone) # rather than UTC. ``resolve_timezone`` already returns # Asia/Shanghai when the Person has no explicit tz, so this # only affects the "Person missing" branch. from zoneinfo import ZoneInfo # noqa: PLC0415 tz = ZoneInfo("Asia/Shanghai") person = await self.db.get(IdentityPerson, proc.person_id) if person is not None: try: tz = resolve_timezone(person) or tz except Exception: # noqa: BLE001 pass today = ts_now.astimezone(tz).date() if proc.leaver_date and proc.leaver_date > today: continue # Capture the id up front: ``_evaluate_one_process`` dirties + # flushes ``proc`` inside the SAVEPOINT, so after a rollback the # ORM instance is expired and reading ``proc.id`` in the except # handler would emit a sync lazy-load → MissingGreenlet in the # async context. Mirrors ProcessAdvanceService's guard. proc_id = str(proc.id) # Per-process SAVEPOINT isolation (mirrors the onboarding # escalation tick + ``reconcile_all_active_processes``): one # process's flush failure — most concretely a 1062 on the # ``offboarding_reminder_log`` unique key under a lock-TTL-expiry # race — must not poison the shared session and discard the whole # tick's commit, which would also throw away the SLA level-ups # already applied to earlier processes and the override-expiry # sweep below. The leaver SLA tick was the asymmetric outlier # (its onboarding twin already had this isolation). try: async with self.db.begin_nested(): deltas = await self._evaluate_one_process(proc, ts_now) except Exception as exc: # noqa: BLE001 if is_transaction_aborting_db_error(exc): # deadlock 1213 / lock-wait 1205 / lost-connection kills # the txn server-side on TiDB; swallowing it would # cascade onto every later begin_nested and make the # trailing commit raise PendingRollbackError. Re-raise so # the lock-retry wrapper restarts the whole tick on a # fresh session (idempotent: level-up-only + reminder-log # dedup). raise logger.exception( "leaver SLA tick: per-process step failed for %s — " "isolated by SAVEPOINT, continuing", proc_id, ) continue first_blocked_set += deltas["first_blocked_set"] warn_count += deltas["warn"] urgent_count += deltas["urgent"] breach_count += deltas["breach"] # Expire stale dual-sign overrides expired_overrides = await self._expire_stale_overrides(ts_now) return { "processes_scanned": len(procs), "first_blocked_set": first_blocked_set, "warn_sent": warn_count, "urgent_sent": urgent_count, "breach_sent": breach_count, "expired_overrides": expired_overrides, } async def _evaluate_one_process( self, proc: OffboardingProcess, ts_now: datetime, ) -> dict[str, int]: """Evaluate + escalate the SLA for a single blocked process. Extracted from ``run_tick`` so the caller can wrap it in a per-process SAVEPOINT (see the isolation comment there). Dirties and flushes ``proc`` (level-up) and the reminder-log idempotency rows; returns the per-tier dispatch deltas the loop accumulates. """ deltas = {"first_blocked_set": 0, "warn": 0, "urgent": 0, "breach": 0} # Initialise first_blocked_at on the first tick after T+0 if proc.first_blocked_at is None: proc.first_blocked_at = ts_now proc.sla_breach_level = SLA_NONE deltas["first_blocked_set"] += 1 # MySQL/TiDB DATETIME(6) round-trips as a naive datetime # even though we wrote a UTC-aware value. Normalise on # read so subtracting from ``ts_now`` (always aware) does # not raise ``can't subtract offset-naive and offset-aware``. # Same fix pattern as PR #138 (close-month) and PR #139 # (Feishu MFA datetime); see DECISIONS.md "TZ-naive # DATETIME(6) round-trip". first_blocked = proc.first_blocked_at if first_blocked.tzinfo is None: first_blocked = first_blocked.replace(tzinfo=UTC) elapsed = ts_now - first_blocked target_level = SLA_NONE if elapsed >= SLA_BREACH_DELTA: target_level = SLA_BREACH elif elapsed >= SLA_URGENT_DELTA: target_level = SLA_URGENT elif elapsed >= SLA_WARN_DELTA: target_level = SLA_WARN # Only level-up; never level-down (a level once raised # stays raised even if the SLA delta calc would somehow # regress — defensive against clock skew). if _level_rank(target_level) > _level_rank(proc.sla_breach_level): proc.sla_breach_level = target_level # Trigger the corresponding notification at most once. # Pre-launch P2 fix (Round 1, CR-005): on dispatch failure # roll back the idempotency log row so the next tick # retries. Without this, a transient Feishu/email outage # at the moment the SLA crosses a level meant the # notification was permanently lost (log row blocked the # retry, dispatch had already failed). Per-recipient # try/except inside ``_dm_*`` already prevents single-bad- # recipient blowing the whole fan-out — this catches the # rarer "entire dispatch raised" case. # 2026-05-30 — User feedback: don't nudge the manager at T+24h if the # only thing left is hardware (laptop return). Hardware return is on # its own physical-pickup track owned by IT + Office Admin via the # hardware_disposition console, and managers can't accelerate it. # # 2026-07-08 fix: the ``_has_non_hardware_pending`` gate MUST be # evaluated BEFORE ``_send_once`` (which stamps+flushes the WARN # idempotency slot). The old order stamped the slot on a hardware-only # tick, so if a NON-hardware blocker appeared LATER within the WARN # window (idle-reclaim / HRIS resync / manual add) ``_send_once`` then # returned False and the manager was NEVER warned about the now- # actionable blocker. Short-circuit ``and`` stamps the slot only when # we actually have something to DM — mirroring escalation.py. Hardware- # only still never DMs (the check is False), it just no longer poisons # a future legitimate WARN. # CR-005 + issue #1751: only RETAIN the idempotency slot when the # fan-out actually delivered / enqueued at least one channel. The # per-recipient ``_dispatch_to`` swallows exceptions (including the # DB error ``NotificationService.dispatch`` re-raises after its # outbox-INSERT SAVEPOINT rolls back) and reports delivered # channels, so the outer ``except`` alone can't detect a total # dispatch failure — we must also roll back when the aggregate # delivered set is empty. Otherwise the stamped slot blocks every # future retry AND there is no outbox row for the resend worker to # dead-letter → the 24h manager DM is silently, permanently lost. # Mirrors ``escalation.py``'s CR-314 ``if not delivered:`` guard. if ( target_level == SLA_WARN and await self._has_non_hardware_pending(proc) and await self._send_once(proc, SLA_REMINDER_KIND_WARN) ): delivered: list[str] = [] try: delivered = await self._dm_manager_chain(proc, level=SLA_WARN) except Exception: # noqa: BLE001 logger.exception( "SLA WARN dispatch raised for process %s — " "rolling back idempotency log for retry", proc.id, ) if delivered: deltas["warn"] += 1 else: logger.warning( "SLA WARN for process %s delivered to 0 channels — " "rolling back idempotency log so the next tick retries", proc.id, ) await self._rollback_send_log( proc.id, SLA_REMINDER_KIND_WARN, ) elif target_level == SLA_URGENT and await self._send_once( proc, SLA_REMINDER_KIND_URGENT, ): delivered = [] try: delivered = await self._dm_hrbp_and_it_admins( proc, level=SLA_URGENT, ) except Exception: # noqa: BLE001 logger.exception( "SLA URGENT dispatch raised for process %s — " "rolling back idempotency log for retry", proc.id, ) if delivered: deltas["urgent"] += 1 else: logger.warning( "SLA URGENT for process %s delivered to 0 channels — " "rolling back idempotency log so the next tick retries", proc.id, ) await self._rollback_send_log( proc.id, SLA_REMINDER_KIND_URGENT, ) elif target_level == SLA_BREACH: # Round 2 P1 fix (LG2-003) + #1788: the BREACH branch has two # INDEPENDENT idempotency guards. Step 1 (incident) is # "fire-and-forget once" — critical-incident creation is NOT # vendor-side idempotent, so it is never rolled back. Step 2 # (DM fan-out) has its OWN ``sla_breach_dm`` slot so it can roll # back + retry on failure WITHOUT re-firing the incident. if await self._send_once(proc, SLA_REMINDER_KIND_BREACH): try: await self._raise_critical_incident(proc) except Exception: # noqa: BLE001 logger.exception( "SLA BREACH critical-incident raise failed " "for process %s — NOT rolling back the log: " "incident creation is NOT idempotent vendor-" "side. Operator should manually open the " "incident if missing.", proc.id, ) # #1788: the DM fan-out is guarded separately AND checks the # delivered-channel count. Previously the BREACH DM lived under # the same ``sla_breach`` slot and never checked ``delivered``, # so when BOTH feishu + email failed silently (``_dispatch_to`` # swallows and returns []), the slot stayed stamped and the # HRBP/IT breach DM was lost forever (no outbox row to # dead-letter). WARN/URGENT already had this rollback; BREACH # was the gap (#1751 only fixed WARN/URGENT). if await self._send_once(proc, SLA_REMINDER_KIND_BREACH_DM): delivered: list = [] try: delivered = await self._dm_hrbp_and_it_admins( proc, level=SLA_BREACH, ) except Exception: # noqa: BLE001 logger.exception( "SLA BREACH DM dispatch raised for process %s — " "rolling back the DM idempotency log for retry " "(incident, if raised, is untouched).", proc.id, ) if delivered: deltas["breach"] += 1 else: logger.warning( "SLA BREACH DM for process %s delivered to 0 " "channels — rolling back the DM idempotency log so " "the next tick retries.", proc.id, ) await self._rollback_send_log( proc.id, SLA_REMINDER_KIND_BREACH_DM, ) return deltas # ── Manager-DM gate ────────────────────────────────────────── async def _has_non_hardware_pending( self, proc: OffboardingProcess, ) -> bool: """Does this process have any pending blocker that is NOT ``RES_HARDWARE``? 2026-05-30 — Used to gate the T+24h WARN manager DM. Hardware return is on its own physical-pickup track (IT + Office Admin via ``hardware_disposition`` console); the direct manager has no operational lever there. If hardware is the only thing left when the SLA WARN tier crosses, skip the manager DM entirely. Pending = ``resolution == RESO_PENDING``. """ row = await self.db.scalar( select(OffboardingBlocker.id) .where( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, OffboardingBlocker.resource_type != RES_HARDWARE, ) .limit(1) ) return row is not None # ── Idempotency on reminder dispatch ───────────────────────── async def _send_once( self, proc: OffboardingProcess, reminder_kind: str, ) -> bool: """Returns True if the reminder hasn't been sent before (in which case the caller should send it now). Writes the log row in either case so a parallel tick won't double-send.""" with self.db.no_autoflush: existing = await self.db.scalar( select(OffboardingReminderLog).where( OffboardingReminderLog.process_id == proc.id, OffboardingReminderLog.reminder_kind == reminder_kind, ) ) if existing is not None: return False log = OffboardingReminderLog( process_id=proc.id, reminder_kind=reminder_kind, channels=["feishu", "email"], detail={ "leaver_date": proc.leaver_date.isoformat() if proc.leaver_date else None, "first_blocked_at": proc.first_blocked_at.isoformat() if proc.first_blocked_at else None, "sla_level": proc.sla_breach_level, }, ) self.db.add(log) await self.db.flush() return True async def _rollback_send_log( self, process_id: UUID, reminder_kind: str, ) -> None: """Delete the idempotency row written by ``_send_once`` so a failed dispatch can be retried on the next tick. Called only when the dispatch helper raised — per-recipient try/except inside ``_dm_*`` handles the more common partial- success case. A double-send on tick crash between rollback and dispatch start is bounded by per-channel idempotency (Feishu open_id / email subject hash). """ from sqlalchemy import delete as _sa_delete # noqa: PLC0415 # SAVEPOINT-isolate (2026-08-06): the comment below promises this # "must not abort the caller's SLA tick loop". Without the savepoint # it did exactly that — the failed delete deactivated the # transaction and every remaining process in the tick blew up. async with best_effort( self.db, label=f"_rollback_send_log {process_id}/{reminder_kind}", log=logger, ) as rollback_step: await self.db.execute( _sa_delete(OffboardingReminderLog).where( OffboardingReminderLog.process_id == process_id, OffboardingReminderLog.reminder_kind == reminder_kind, ) ) await self.db.flush() if not rollback_step.ok: logger.error( "_rollback_send_log: failed to delete reminder log " "for process %s kind %s (%s) — next tick may not retry", process_id, reminder_kind, rollback_step.reason(), ) # Do NOT re-raise: a DB error here must not abort the caller's # SLA tick loop (which runs all processes in one shot). The # consequence is acceptable: the log row persists, so the next # tick won't retry the dispatch for this process/kind. # Surfaced as logger.exception so on-call sees it in logs. # ── Recipients ─────────────────────────────────────────────── async def _dm_manager_chain( self, proc: OffboardingProcess, *, level: str, ) -> list[str]: """Notify the leaver's DIRECT manager only (not skip-level). 2026-05-30 — Skip-level (manager's manager) is intentionally out of the recipient set per user feedback on the lifecycle visualization: "经理链上级 / level+2 不需要知道这个事情". The skip-level has no operational lever over their indirect report's offboarding blockers, and the alert reads as 越界 to them. Only the direct manager — who actually knows the leaver, owns the work-transition, and can pick a successor — gets the WARN DM. URGENT / BREACH tiers escalate to HRBP + IT Admin instead (see ``_dm_hrbp_and_it_admins``). """ person = await self.db.get(IdentityPerson, proc.person_id) if person is None: return [] recipients: list[IdentityPerson] = [] if person.manager_person_id: mgr = await self.db.get(IdentityPerson, person.manager_person_id) if ( mgr is not None and mgr.deleted_at is None and mgr.employment_status == "active" and mgr.id != person.id # never DM the leaver about their own SLA ): recipients.append(mgr) # Pick the card CTA per recipient by lifecycle-read access: a # manager who ALSO holds a lifecycle-read role (global_admin / # hr / auditor / management) lands on the per-process detail # page where the blocker list — and, for global_admin, the # override / advance controls — are visible. Pure managers (no # such role) route to ``/my/team-leavers`` — the admin detail # route would 403 for them otherwise. Mirrors the URGENT-tier # ``_dm_hrbp_and_it_admins`` per-recipient switch. reader_ids = await self._lifecycle_detail_readers( [m.id for m in recipients] ) pending_resources = await self._pending_resource_lines(proc) delivered: list[str] = [] for recipient in recipients: delivered.extend(await self._dispatch_to( person, proc, recipient, level=level, can_read_detail=recipient.id in reader_ids, pending_resources=pending_resources, )) return delivered async def _dm_hrbp_and_it_admins( self, proc: OffboardingProcess, *, level: str, ) -> list[str]: """Notify the leaver's HRBP plus the IT admin pool. At T+48h (URGENT) / T+72h (BREACH) we used to spray the entire ``hr`` role pool — typically 25+ HR admins enterprise-wide, the leaver included if they happen to hold the ``hr`` role. That produced flood-level noise on every blocked leaver and forced Olivia He to be notified about her own offboarding (same email had ``hr`` role membership, see PR digest 2026-05-02). New rule: deliver to **exactly two cohorts**: 1. The leaver's own HRBP, resolved via Feishu's per-user ``custom_attrs`` (see :mod:`app.integrations.feishu_hrbp`). This is the single named person responsible for shepherding the offboarding through. 2. Every active ``it_admin`` (global_admin) — the small pool with the platform-level authority to unblock the process. Typically 2–4 people, not 25. The leaver themselves is **always** filtered out of the recipient set, even if they hold ``it_admin`` somehow, because notifying someone about their own SLA breach is absurd and was an active source of confusion. When HRBP resolution fails (no Feishu account for the leaver, no HRBP configured in Feishu, HRBP email doesn't match a Helix person, etc.), an audit row tagged ``leaver_sla.hrbp_resolve_failed`` records the reason. The dispatch still proceeds to the IT admin pool so SLA escalation isn't silently dropped — escalation has compliance value even without HRBP routing. """ from app.auth.models import AuthRole, AuthUserRole person = await self.db.get(IdentityPerson, proc.person_id) if person is None: return [] recipients: list[IdentityPerson] = [] seen_ids: set = set() # 1. HRBP — resolve from Feishu (or cached column). resolution = await resolve_hrbp_for_person(self.db, person) if resolution.person_id is not None and resolution.person_id != person.id: hrbp = await self.db.get(IdentityPerson, resolution.person_id) if ( hrbp is not None and hrbp.deleted_at is None and hrbp.employment_status == "active" ): recipients.append(hrbp) seen_ids.add(hrbp.id) else: # Audit when HRBP routing falls through to admin-only — # ops can spot tenants where the Feishu HRBP attr is empty # or stale before it becomes a compliance blind spot. await self.audit.log( trace_id=str(proc.id), actor_type="system", actor_user_id=None, action="leaver_sla.hrbp_resolve_failed", status="failure", resource_type="offboarding_process", resource_id=str(proc.id), detail={ "leaver_person_id": str(person.id), "leaver_email": person.work_email, "level": level, "reason": resolution.reason, "hrbp_email_seen": resolution.hrbp_email, }, ) # 2. IT admin pool — global_admin only. ``hr`` role is no # longer a recipient; HR routing is HRBP-targeted above. admin_ids = list((await self.db.execute( select(AuthUserRole.user_id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthRole.name == GLOBAL_ADMIN_ROLE) )).scalars().all()) for admin_id in admin_ids: if admin_id in seen_ids: continue if admin_id == person.id: # Self-exclusion: the leaver may also hold global_admin. continue admin = await self.db.get(IdentityPerson, admin_id) if ( admin is None or admin.deleted_at is not None or admin.employment_status != "active" ): continue recipients.append(admin) seen_ids.add(admin.id) # Route each recipient's card CTA by whether they can open the # per-process detail page. The leaver's HRBP typically holds # ``hr`` (not ``global_admin``) — before 2026-07-15 they were # dead-ended on the counts-only /my/team-leavers even though # ``hr`` can read the detail page. Now they land on the detail # page and see the blocker list. (``admin_ids`` above is still # the global_admin recipient pool; this reader set is the # broader lifecycle-read cohort used only for CTA routing.) reader_ids = await self._lifecycle_detail_readers( [r.id for r in recipients] ) pending_resources = await self._pending_resource_lines(proc) delivered: list[str] = [] for recipient in recipients: delivered.extend(await self._dispatch_to( person, proc, recipient, level=level, can_read_detail=recipient.id in reader_ids, pending_resources=pending_resources, )) return delivered async def _dispatch_to( self, person: IdentityPerson, proc: OffboardingProcess, recipient: IdentityPerson, *, level: str, can_read_detail: bool = True, pending_resources: str, ) -> list[str]: """Dispatch one SLA reminder and return the channels that actually delivered / were enqueued to the outbox. ``can_read_detail`` — the recipient holds a role that can open the per-process offboarding detail page (``require_lifecycle_read``: global_admin / hr / auditor / management). Such recipients land directly on ``/admin/offboarding/process/{id}`` where the blocker list is visible; everyone else routes to the counts-only ``/my/team-leavers`` lookout (the admin detail route 403s for them). Before 2026-07-15 this switch keyed on ``global_admin`` alone, so the leaver's HRBP — who holds ``hr`` but not ``global_admin`` — was dead-ended on the counts-only page even though ``hr`` can now read the detail page (opened 2026-07-08). ``pending_resources`` — a pre-rendered bullet list of the outstanding blocker labels, injected into the card / email so a recipient understands *what* is blocking (e.g. the specific laptop) without having to click through. Computed once per process by the caller and shared across recipients. Returns ``[]`` when the dispatch raised (e.g. ``NotificationService.dispatch`` re-raising a 1205 lock-wait after its outbox-INSERT SAVEPOINT rolled back). The per-recipient try/except stays — one dirty recipient must not abandon the rest of the fan-out — but the caller now needs the delivered count so it only retains the ``_send_once`` idempotency row when at least one recipient truly received a channel. Mirrors ``escalation.py``'s CR-314 ``res.delivered_channels`` accounting (previously this swallowed everything and returned ``None``, so the WARN/URGENT ``except`` + ``_rollback_send_log`` compensation was unreachable and a total dispatch failure permanently lost the DM with no outbox dead-letter — issue #1751). """ try: label = ( "[警告]" if level == SLA_WARN else "[紧急]" if level == SLA_URGENT else "[严重]" ) # Lifecycle-read recipients (global_admin / hr / auditor / # management) land on the per-process detail page — it shows # the full blocker list AND, for global_admin, the override / # advance controls (which 403 cleanly for pure-read roles). # Recipients with no lifecycle-read role route to the # read-only /my/team-leavers lookout (the admin route 403s # for them — PR 2026-05-19 fix for the unusable "Handle in # Helix" button on the 24h SLA Feishu card). portal_url = ( f"/admin/offboarding/process/{proc.id}" if can_read_detail else "/my/team-leavers" ) leaver_name = person.display_name or person.work_email res = await self.notify.dispatch( NotificationEnvelope( recipient=recipient, title=( f"{label} {leaver_name} " f"离职流程 SLA {level} — 仍有未清资源" ), body_text=( f"员工 {leaver_name} 离职流程已" f"卡住超过 {_level_label(level)}。\n\n" f"仍需处理的资源:\n{pending_resources}\n\n" f"{{cta_line}}\n\n" f"未清资源会持续阻断离职完成,影响合规。" ), channels=["feishu", "email"], template_key=f"leaver_blocked_sla_{level}", # ``url`` drives the Feishu card's action button. # Without this set, the card body referenced the # path as plain text and recipients had to copy / # paste — found 2026-05-02 alongside the empty-card # template fallback bug. url=portal_url, context={ "leaver_name": leaver_name, "leaver_email": person.work_email, "level": level, # Locale fragment triple — each language's template # interpolates its own. Frozen into the outbox row so # the bell still renders correctly if the reader later # switches their language preference. "level_label_zh": _level_label(level), "level_label_en": _level_label_en(level), "level_label_ja": _level_label_ja(level), "pending_resources": pending_resources, "first_blocked_at": ( proc.first_blocked_at.isoformat() if proc.first_blocked_at else None ), "portal_url": portal_url, }, related_resource_type="offboarding_process", related_resource_id=proc.id, ), trace_id="leaver_blocked.sla", ) return list(res.delivered_channels) except Exception: # noqa: BLE001 logger.exception( "leaver_blocked SLA dispatch to %s for %s failed", recipient.work_email, person.work_email, ) return [] async def _lifecycle_detail_readers( self, person_ids: list[UUID], ) -> set[UUID]: """Subset of ``person_ids`` holding a role that can open the per-process offboarding detail page (``require_lifecycle_read``). Used to decide each SLA recipient's card CTA target: readers → ``/admin/offboarding/process/{id}`` (blocker list visible), others → the counts-only ``/my/team-leavers`` lookout. """ if not person_ids: return set() from app.auth.models import AuthRole, AuthUserRole rows = (await self.db.execute( select(AuthUserRole.user_id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthRole.name.in_(_LIFECYCLE_DETAIL_READ_ROLE_NAMES)) .where(AuthUserRole.user_id.in_(person_ids)) )).scalars().all() return set(rows) async def _pending_resource_lines( self, proc: OffboardingProcess, *, limit: int = 12, ) -> str: """Render the outstanding (``RESO_PENDING``) blocker labels of a process into a bullet list for the SLA card / email body. Mirrors ``escalation.py``'s ``leaver_reminder`` ``pending_blockers`` block so recipients see *which* resource is blocking (e.g. the specific laptop asset tag) directly in the notification, rather than the generic "仍有未清资源" that forced everyone to click through to find out. Capped at ``limit`` lines with a truncation tail so a pathological blocker count can't balloon the card. The ONE pre-rendered block is injected into BOTH the zh ``body_text`` and en ``body_text_en`` template bodies (the context dict is shared across recipients; templates pick the body per recipient ``preferred_language``). The blocker labels themselves are locale-neutral resource identifiers, so every fixed string this renderer adds (the empty-fallback line and the truncation tail) must be bilingual — a Chinese-only sentence would land verbatim in the English email body. Returns a safe fallback line when nothing is pending (a benign race — the last blocker resolved between the SLA decision and this render); the SLA body still ships. """ rows = (await self.db.execute( select(OffboardingBlocker.label) .where( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, ) .order_by(OffboardingBlocker.resource_type) )).scalars().all() labels = [str(r) for r in rows if r] if not labels: return " · (请进入 Helix 查看当前未清项 / see pending items in Helix)" shown = labels[:limit] lines = [f" · {label}" for label in shown] if len(labels) > limit: lines.append( f" · …等 {len(labels)} 项 / {len(labels)} items in total" ) return "\n".join(lines) async def _raise_critical_incident( self, proc: OffboardingProcess, ) -> None: person = await self.db.get(IdentityPerson, proc.person_id) if person is None: return pending_blockers = list((await self.db.execute( select(OffboardingBlocker).where( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, ) )).scalars().all()) # Round-58 logic scan F2 (2026-05-30): race where the SLA tick # decided to fire BREACH but a concurrent reconcile or task # callback resolved the last blocker between then and here. # Without this guard the empty list falls into the "vendor # stall" branch below and we raise a CRITICAL compliance # incident titled "软件 revoke pipeline 72h+ 未完成,0 项" — a # contradictory false alarm. ``_send_once`` already wrote the # idempotency log so we won't re-fire; just skip the incident. if not pending_blockers: logger.info( "SLA breach raise skipped for process %s: all pending " "blockers resolved between _send_once gate and " "_raise_critical_incident (race with reconcile or " "provisioning callback). No incident needed.", proc.id, ) return # Round-58 (2026-05-30) — split pending blockers into the two # ownership domains so the incident text doesn't conflate them: # # - ``human_pending`` (hardware / network owner / API key / # IAM / scope owner / email group / …) → operator-actionable. # The "X 个未清 blocker" headline counts THESE. # # - ``scheduled_pending`` (software_seat / software_permission / # software_account) → ProvisioningTask in-flight or retrying. # A leaver with only scheduled_pending at 72h means the # vendor pipeline is genuinely slow — still worth a # critical incident (vendor outage / quota exhaustion), but # the title needs to say "software revoke vendor stall" # not "未清 blocker", otherwise the on-call admin reads it # as "go transfer assets" and bounces back confused. from app.hr_lifecycle.advance_service import ( # noqa: PLC0415 SCAN_SCHEDULED_RESOURCE_TYPES, ) human_pending = [ b for b in pending_blockers if b.resource_type not in SCAN_SCHEDULED_RESOURCE_TYPES ] scheduled_pending = [ b for b in pending_blockers if b.resource_type in SCAN_SCHEDULED_RESOURCE_TYPES ] if human_pending: title = ( f"{person.display_name or person.work_email} 离职流程 SLA 突破 72h " f"({len(human_pending)} 个未清 blocker)" ) else: # All pending is vendor-side software revoke retrying past # 72h. Different operator action: investigate the # ProvisioningTask error, not chase asset handover. title = ( f"{person.display_name or person.work_email} 离职流程 SLA 突破 72h " f"(软件 revoke pipeline 72h+ 未完成,{len(scheduled_pending)} 项)" ) await self.incidents.create_or_upsert_incident( incident_type=TYPE_LEAVER_ORPHAN, severity=SEVERITY_CRITICAL, person_id=person.id, # Dedup per offboarding PROCESS — without resource_type/ # resource_id the upsert WHERE matches ANY open LEAVER_ORPHAN # incident regardless of person, so a second SLA-breached # leaver overwrites the first's title/detail (forensics loss). # ``str(proc.id)`` not the raw UUID: resource_id is CHAR(36) # and asyncmy binds a uuid.UUID as bytes → silent 0-row match. # Shares the per-process key with advance_service's force- # transfer-failure incident so one stuck leaver = one ticket. resource_type="offboarding_process", resource_id=str(proc.id), resource_label=person.display_name or person.work_email, title=title, detail={ "process_id": str(proc.id), "first_blocked_at": ( proc.first_blocked_at.isoformat() if proc.first_blocked_at else None ), "leaver_date": ( proc.leaver_date.isoformat() if proc.leaver_date else None ), "pending_blocker_ids": [str(b.id) for b in pending_blockers], "pending_blocker_types": [b.resource_type for b in pending_blockers], "human_pending_count": len(human_pending), "scheduled_pending_count": len(scheduled_pending), }, ) async def _expire_stale_overrides( self, ts_now: datetime, ) -> int: """Mark pending_secondary AND approved-but-unexecuted overrides whose ``expires_at`` has passed as ``expired``. Returns the count of rows transitioned. Pre-launch P1 fix: previously only ``pending_secondary`` was swept, so an approved override could sit for months waiting for execution and still flip a person to ``left`` long after the cited blockers / business context drifted. The execute path also enforces the TTL at execute-time as defense-in- depth (``override_service.execute``). """ rows = list((await self.db.execute( select(LeaverDualSignOverride).where( LeaverDualSignOverride.status.in_( [DUAL_SIGN_PENDING_SECONDARY, DUAL_SIGN_APPROVED] ), LeaverDualSignOverride.expires_at.isnot(None), LeaverDualSignOverride.expires_at <= ts_now, ) )).scalars().all()) for row in rows: row.status = DUAL_SIGN_EXPIRED if rows: await self.db.flush() return len(rows) # ── helpers ──────────────────────────────────────────────────────── _LEVEL_RANK = { SLA_NONE: 0, SLA_WARN: 1, SLA_URGENT: 2, SLA_BREACH: 3, } def _level_rank(level: str | None) -> int: return _LEVEL_RANK.get(level or SLA_NONE, 0) def _level_label(level: str) -> str: return { SLA_WARN: "24 小时", SLA_URGENT: "48 小时", SLA_BREACH: "72 小时", }.get(level, level) def _level_label_en(level: str) -> str: return { SLA_WARN: "24 hours", SLA_URGENT: "48 hours", SLA_BREACH: "72 hours", }.get(level, level) def _level_label_ja(level: str) -> str: return { SLA_WARN: "24 時間", SLA_URGENT: "48 時間", SLA_BREACH: "72 時間", }.get(level, level) __all__ = [ "LeaverBlockedSlaService", "SLA_REMINDER_KIND_WARN", "SLA_REMINDER_KIND_URGENT", "SLA_REMINDER_KIND_BREACH", "SLA_REMINDER_KIND_BREACH_DM", ]