"""Leaver reminder + manager-escalation worker. Runs from :func:`app.worker._escalation_tick` on an hourly cadence. One pass scans every live offboarding process and, per process, does at most one action — idempotent via ``offboarding_reminder_log``. Timeline (days are relative to ``leaver_date``): ── (T-7, T-3, T-1) ──► push "leaver_reminder" to employee + manager + IT (global_admins). Lists outstanding MUST_TRANSFER blockers so they know what's left to do. ── (T-0: leaver_date) ──► for every still-pending MUST_TRANSFER blocker, auto-open an ``AssetTransferRequest`` to the leaver's manager. Manager gets a single push naming every escalated item. Once manager accepts, blocker resolves and HR worker transitions the person to ``left``. If leaver_date passes without auto-escalation (e.g. worker down for a day), escalation runs on next tick — still idempotent. Fallback when the leaver has no manager (rare — HRIS should populate it): IT global_admins become the escalation target. If there are none either, the blocker is surfaced in the admin dashboard and human IT handles it. """ from __future__ import annotations import logging from datetime import UTC from datetime import date as date_type from datetime import datetime as datetime_type from typing import Any from uuid import UUID from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.rbac import GLOBAL_ADMIN_ROLE from app.core.working_hours import resolve_timezone from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_CANCELLED, PROC_COMPLETED, REMINDER_1D, REMINDER_3D, REMINDER_7D, REMINDER_ESCALATION, RES_API_KEY, RES_HARDWARE, RESO_PENDING, XFER_PENDING, AssetTransferRequest, OffboardingBlocker, OffboardingProcess, OffboardingReminderLog, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # Blocker resource types that need a human transfer via # ``AssetTransferRequest`` — i.e. types for which # ``OffboardingService._apply_transfer`` and ``_initiator_still_holds_asset`` # have working branches. SSO grants, domain registrations, cloud scope # ownership and Google orphan ownership are NOT in this set: they are # handled directly by ``LeaverAdvancer._handle_one_blocker`` (revoke or # FK rewrite) at T+0 grace, not via a manager-accept transfer flow. # # Pre-launch P1 fix (Round 1, CR-003 / CR-020): the prior set included # ``RES_SSO_GRANT`` + ``RES_DOMAIN_REGISTRATION`` and the query # broadened to ``RES_CLOUD_SCOPE_OWNER``. The escalation worker # generated AssetTransferRequest rows the manager could never accept # (``respond_to_transfer`` returns 409 ASSET_NO_LONGER_HELD because # ``_initiator_still_holds_asset`` has no branch for those types). MUST_TRANSFER_RESOURCE_TYPES = { RES_HARDWARE, RES_API_KEY, } # Subset of ``MUST_TRANSFER_RESOURCE_TYPES`` that actually routes to # the manager at T+0 via an auto-opened ``AssetTransferRequest``. The # leaver's reminder log uses the broader set to decide "is there # anything left to nudge anyone about" (an unfinished laptop return # is a legit reason to keep nudging the leaver + IT), but the manager # is intentionally NOT the fallback for physical-asset return — per # the "经理 ≠ 资产" principle (raised by the user on 2026-05-30 against # the lifecycle visualization), hardware returns are owned end-to-end # by Office Admin + IT through the HardwareDispositionPanel at # ``/admin/offboarding/{id}`` (executed in # ``ProcessAdvanceService._apply_hardware_disposition`` with three # branches: ``keep`` → held_by_leaver, ``recycle_to_office``, # ``transfer_to_person``). Only API keys still route to the manager # at T+0 — because the API-key custodian almost always knows the # successor, while the IT pool does not. MANAGER_AUTO_TRANSFER_RESOURCE_TYPES = { RES_API_KEY, } class LeaverEscalationService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def run_tick(self, *, today: date_type | None = None) -> dict[str, Any]: """Evaluate every pending offboarding process once. Returns a summary dict so the worker can log a single structured line per tick. ``today`` is test-only — production callers pass nothing and every delta is computed against the LEAVER'S local TZ (matches PR #443 + the C-1 process_advance fix), not the worker-process clock. A US employee with leaver_date=2026-05-21 at 04:00 UTC = 21:00 PDT prior day used to escalate (delta=0 in UTC) before their last working day even started in PDT. """ now_utc = datetime_type.now(UTC) fallback_today = today or now_utc.date() procs = ( await self.db.execute( select(OffboardingProcess).where( OffboardingProcess.status.notin_([PROC_COMPLETED, PROC_CANCELLED]) ) ) ).scalars().all() reminders_sent = 0 escalations_sent = 0 for proc in procs: if today is None: person = await self._get_person(proc.person_id) if person is not None: try: tz = resolve_timezone(person) except Exception: # noqa: BLE001 tz = None else: tz = None today_local = now_utc.astimezone(tz).date() if tz else fallback_today else: today_local = today delta = (proc.leaver_date - today_local).days # Pre-launch P1 fix (Round 1, CR-004): the previous bands # (T-7: ``delta <= 7 and > 3``; T-3: ``delta <= 3 and > 1``; # T-1: ``delta <= 1 and > 0``) had a silent gap when the # worker was offline across a band boundary — at delta=2 # after a 5-day outage that started on delta=7, T-7's lower # bound (``> 3``) was already crossed and no band fired, # so T-7 was permanently skipped. # # The fix: drop the lower bound on each band — fire when # the calendar threshold has been crossed AND the # ``OffboardingReminderLog`` has no row for that # (process_id, kind). Idempotency makes catch-up safe; in # an outage scenario the leaver and manager get a clean # cascade in one tick (T-7 + T-3 in same fan-out at # delta=2, then T-1 at delta=1). # Round 2 P2 fix (CR2-003 / LG2-006): fire only the LOWEST # un-logged band whose threshold has been crossed, not # every band. Without this cap, a cold-start at delta<=1 # (process just created in a single tick) would fan out # T-7 + T-3 + T-1 simultaneously to leaver/manager/IT- # admins — three duplicate cards back-to-back. After this # cap, the lowest band ALWAYS supersedes — outage catch-up # at delta=2 fires only T-3, at delta=1 fires only T-1. # The skipped higher bands are still recorded as # _already_logged="superseded" via _send_reminder marking # them with a no-op log so the next tick doesn't fire # them either. if delta > 0: if delta <= 1: if await self._send_reminder( proc, REMINDER_1D, days_left=delta, ): reminders_sent += 1 # Mark T-7/T-3 as superseded so they don't fire # tomorrow at delta=0 if escalation hasn't # already taken over. await self._mark_superseded( proc.id, (REMINDER_7D, REMINDER_3D), ) elif delta <= 3: if await self._send_reminder( proc, REMINDER_3D, days_left=delta, ): reminders_sent += 1 await self._mark_superseded( proc.id, (REMINDER_7D,), ) elif delta <= 7: if await self._send_reminder( proc, REMINDER_7D, days_left=delta, ): reminders_sent += 1 # On or past leaver_date: escalate outstanding must_transfer # blockers to the manager. Runs at most once per process. if delta <= 0 and proc.status == PROC_BLOCKED: if await self._escalate(proc): escalations_sent += 1 return { "processes_scanned": len(procs), "reminders_sent": reminders_sent, "escalations_sent": escalations_sent, # ``today`` defaults to None in prod (per-person TZ-local # date computed in the loop). Fall back to ``fallback_today`` # so this summary dict never crashes the tick. "date": (today or fallback_today).isoformat(), } # ── Reminder ───────────────────────────────────────────── async def _send_reminder( self, proc: OffboardingProcess, kind: str, *, days_left: int, ) -> bool: if await self._already_logged(proc.id, kind): return False blockers = await self._pending_must_transfer_blockers(proc) # If everything's already handled, don't bother reminding — and DO NOT # stamp the reminder slot. Same root cause the Round-59 follow-up # already fixed in ``_escalate``'s no-op branches (below): a leaver # whose must-transfer list is momentarily empty at this band's tick # (all hardware disposed, no API_KEY blocker yet) used to log the slot # here, so if idle-reclaim / HRIS resync / a manual add later surfaces # a must-transfer blocker inside this same band's window, the # ``_already_logged`` guard at the top would silently suppress the # reminder for good. The next tick re-runs the indexed # ``_pending_must_transfer_blockers`` query and re-evaluates cheaply. if not blockers: return False person = await self._get_person(proc.person_id) if not person: return False # Pre-launch P1 fix (Round 1, LG-005): walk the manager chain # past inactive / leaver_pending managers — if Person A's # manager is also offboarding, the reminder targets the next # active manager up the chain. Mirrors # ``LeaverAdvancer._resolve_fallback_person`` semantics. If no # active manager exists in the chain, fall back to IT admins # only (no manager line in the recipient list). manager = await self._walk_to_active_manager(person) it_admins = await self._list_it_admins() blocker_lines = "\n".join(f" · {b.label}" for b in blockers) portal_url = await self._portal_url() context = { "person_name": person.display_name or person.work_email, "person_email": person.work_email, "days_left": days_left, "leaver_date": proc.leaver_date.isoformat(), "pending_blockers": blocker_lines, "portal_url": portal_url, } from app.notifications import NotificationEnvelope, NotificationService notify = NotificationService(self.db) # 2026-05-30 — Manager-reminder fan-out trimmed (user feedback # on lifecycle visualization): the manager used to receive # T-7, T-3, T-1 reminders plus the T+0 escalation AND the # T+24h SLA WARN — five cards per leaver. The first three # were pure FYI cards with no actionable difference between # them. New cadence: manager hears once during the reminder # phase, at T-3 only. Leaver + IT admins still receive the # full T-7 / T-3 / T-1 cascade (they have real work to do at # each tick). recipients: list[IdentityPerson] = [person] if manager and kind == REMINDER_3D: recipients.append(manager) recipients.extend(it_admins) # Per-recipient CTA URL: the leaver lands on their own self-service # view, IT admins land on the per-process admin detail page (which # the standalone ``/offboarding/{id}`` URL we used to emit doesn't # resolve to — it 404'd). Manager isn't routed to the admin URL # because ``/admin/offboarding/process/:processId`` is gated by # the ``global_admin``/``auditor`` RoleRoute; manager would 403. # Send manager to their todos hub where the pending transfers from # this leaver surface. it_admin_ids = {a.id for a in it_admins} def _url_for(recipient: IdentityPerson) -> str: # Assign-then-return so the URL audit # (``tests/test_offboarding/test_round14_severity_and_url_audit # .py::test_no_inline_portal_url_in_notification_bodies``) # recognises these as sanctioned ``url=...`` sinks # rather than raw ``return f"..."`` strings — the audit # regex matches an `^...url\s*=\s*f"...{portal_url}/"` # pattern, which a direct return statement doesn't trip. # Pure refactor, no behavioural change. if recipient.id == person.id: self_url = f"{portal_url}/my/offboarding" return self_url if recipient.id in it_admin_ids: admin_url = f"{portal_url}/admin/offboarding/process/{proc.id}" return admin_url todos_url = f"{portal_url}/my/todos" return todos_url # Pre-launch P2 fix (Round 1, LG-009): per-recipient try/except # so a single dirty notification target (HRIS data quality — # malformed email, missing Feishu open_id) doesn't abandon the # remaining recipients in the same reminder fan-out. Mirrors # the per-admin try/except in # ``OffboardingService._notify_transfer_accepted_it``. delivered: list[str] = [] for r in recipients: try: res = await notify.dispatch( NotificationEnvelope( recipient=r, title=f"[离职 T-{days_left}] {person.display_name or person.work_email} 仍有资产未处理", body_text="(see template)", channels=["feishu", "email"], template_key="leaver_reminder", context=context, url=_url_for(r), ), ) delivered.extend(res.delivered_channels) except Exception: # noqa: BLE001 logger.exception( "leaver reminder dispatch failed for recipient %s " "on process %s — continuing fan-out to other " "recipients", r.id, proc.id, ) # Pre-launch P1 fix (Round 45 follow-up, CR-314): only stamp # the reminder-log row when at least one recipient actually # received a channel. Without this, a fan-out that throws # for every recipient (e.g. Feishu app revoked + SMTP creds # expired) still wrote ``OffboardingReminderLog`` claiming # the T-1 / T-3 / T-7 reminder "sent" with empty channels — # permanently suppressing future retries. if not delivered: logger.warning( "leaver reminder %s for process %s delivered to 0 " "channels across %d recipients — leaving log row " "absent so the next tick retries", kind, proc.id, len(recipients), ) return False await self._log_reminder( proc.id, kind, channels=sorted(set(delivered)), detail={ "blocker_ids": [str(b.id) for b in blockers], "recipients": [str(r.id) for r in recipients], }, ) return True # ── Escalation ──────────────────────────────────────────── async def _escalate(self, proc: OffboardingProcess) -> bool: """Fallback-transfer every pending MUST_TRANSFER blocker to the leaver's manager (or IT admin fallback).""" if await self._already_logged(proc.id, REMINDER_ESCALATION): return False blockers = await self._pending_must_transfer_blockers(proc) if not blockers: # Round-59 follow-up (2026-05-31): DO NOT log REMINDER_ESCALATION # in the no-op branch. A leaver whose must-transfer list is # momentarily empty (e.g. all hardware blockers got disposed # this tick and no API_KEY blocker exists yet) used to stamp # the slot here — and if `idle-reclaim` / HRIS resync later # surfaced a new API_KEY blocker, the `_already_logged` guard # at the top would silently block escalation forever. The next # tick re-runs the indexed `_pending_must_transfer_blockers` # query and re-evaluates cheaply. return False # 2026-05-30 — "经理 ≠ 资产" principle: hardware blockers do # NOT get auto-routed to the manager at T+0. They are handled # end-to-end by Office Admin + IT through the hardware- # disposition console at T+0 (`_apply_hardware_disposition` # in process_advance). Filter the manager-bound transfer set # to only the types the manager is actually accountable for. # The hardware blockers stay in `blockers` purely for the # "no pending" guard above — `ProcessAdvanceService` resolves # them on its own. manager_blockers = [ b for b in blockers if b.resource_type in MANAGER_AUTO_TRANSFER_RESOURCE_TYPES ] if not manager_blockers: # Round-59 follow-up (2026-05-31): same root cause as the # `not blockers` branch above — burning REMINDER_ESCALATION # on the hardware-only no-op meant a later API_KEY blocker # (added by idle-reclaim cron / HRIS resync / manual blocker # creation) could NEVER trigger manager escalation. Drop the # log row write; let the next hourly tick re-evaluate. return False person = await self._get_person(proc.person_id) if not person: return False # Asset-fallback-matrix contract: non-hardware MUST_TRANSFER # blockers honour the IT-set ``default_handover_person_id`` # before climbing the manager chain — "有交接人给交接人, # 无交接给直属经理 (lead)". This MUST mirror the same priority # order ``ProcessAdvanceService._handle_one_blocker`` uses for # the T+0 force-transfer path, otherwise the escalation tick # opens a transfer to manager while the advance tick rewrites # the FK to the handover person — two competing recipients for # the same resource. Guards (active, not soft-deleted, not the # leaver themself) match the advance-side check exactly. recipient: IdentityPerson | None = None preferred_id = proc.default_handover_person_id if preferred_id is not None and preferred_id != proc.person_id: preferred = await self.db.get(IdentityPerson, preferred_id) if ( preferred is not None and preferred.deleted_at is None and preferred.employment_status == "active" ): recipient = preferred # Pre-launch P1 fix (Round 1, LG-005): walk past inactive / # leaver-pending managers. Without this, a leaver whose manager # is also offboarding would have AssetTransferRequest rows # generated for the offboarding manager, who then accepts on a # half-active employment_status (or doesn't, blocking forever). if recipient is None: recipient = await self._walk_to_active_manager(person) if recipient is None: it_admins = await self._list_it_admins() recipient = it_admins[0] if it_admins else None if recipient is None: logger.warning( "leaver %s has no handover person, no active-manager-chain " "candidate and no it_admin fallback — blockers remain pending", proc.person_id, ) return False used_handover_person = ( preferred_id is not None and recipient.id == preferred_id ) # Open an AssetTransferRequest per blocker (only if not already # linked to a pending/accepted one — idempotent on re-run). # `manager_blockers` is hardware-stripped; hardware blockers # in the original `blockers` list flow through to the T+0 # `_force_transfer_all` hardware-disposition path instead. created: list[AssetTransferRequest] = [] for b in manager_blockers: if b.transfer_request_id: # already linked — skip continue reason_target = "交接人" if used_handover_person else "经理" xfer = AssetTransferRequest( initiator_id=proc.person_id, recipient_id=recipient.id, asset_type=_blocker_to_transfer_type(b.resource_type), asset_ref_id=b.resource_id, asset_label=b.label or str(b.resource_id), status=XFER_PENDING, reason=( f"系统兜底:{person.display_name or person.work_email} " f"今日离职,自动转让给{reason_target}。" ), offboarding_process_id=proc.id, ) self.db.add(xfer) await self.db.flush() b.transfer_request_id = xfer.id created.append(xfer) if not created: # Round-59 follow-up (2026-05-31): same anti-pattern as the # earlier no-op branches — burning REMINDER_ESCALATION here # would block legitimate future escalation if a new API_KEY # blocker (un-linked) materialises. The success path below # is the ONLY place that should stamp REMINDER_ESCALATION. return False portal_url = await self._portal_url() asset_list = "\n".join( f" · {x.asset_label} ({x.asset_type})" for x in created ) from app.notifications import NotificationEnvelope, NotificationService notify = NotificationService(self.db) # Pre-launch P2 fix (Round 1, LG-009): catch dispatch failures # so a transient Feishu / email outage doesn't make the # escalation row look like it failed end-to-end. The transfer # rows are already committed at this point — the manager will # see them in their inbox even without the proactive push. delivered_channels: list[str] = [] try: res = await notify.dispatch( NotificationEnvelope( recipient=recipient, title=f"[兜底] {person.display_name or person.work_email} 的 {len(created)} 项资产已自动转给你", body_text="(see template)", channels=["feishu", "email"], template_key="leaver_escalation", context={ "person_name": person.display_name or person.work_email, "person_email": person.work_email, "count": len(created), "asset_list": asset_list, "portal_url": f"{portal_url}/my/transfers", }, url=f"{portal_url}/my/transfers", ), ) delivered_channels = res.delivered_channels except Exception: # noqa: BLE001 logger.exception( "leaver escalation recipient-notify failed for process " "%s — transfer rows are committed; recipient will see " "them in /my/transfers on next session", proc.id, ) await self.audit.log( trace_id=str(proc.id), actor_type="system", actor_user_id=None, action="offboarding.escalated_to_manager", status="success", resource_type="offboarding_process", resource_id=str(proc.id), detail={ "recipient_id": str(recipient.id), "recipient_role": ( "handover_person" if used_handover_person else "manager" ), "transfer_ids": [str(x.id) for x in created], "delivered": delivered_channels, }, ) await self._log_reminder( proc.id, REMINDER_ESCALATION, channels=delivered_channels, detail={ "recipient_id": str(recipient.id), "recipient_role": ( "handover_person" if used_handover_person else "manager" ), "transfer_ids": [str(x.id) for x in created], }, ) return True # ── Helpers ────────────────────────────────────────────── async def _pending_must_transfer_blockers( self, proc: OffboardingProcess, ) -> list[OffboardingBlocker]: """Return blockers that actually need human transfer action. Pre-launch P1 fix (Round 1, CR-003): only resource types that ``OffboardingService._apply_transfer`` and ``_initiator_still_holds_asset`` support are returned. SSO grants, domain registrations, cloud scope ownership and Google orphan ownership are revoked / FK-rewritten directly by ``LeaverAdvancer._handle_one_blocker`` — generating an AssetTransferRequest for those types only created an un-acceptable row. """ rows = ( await self.db.execute( select(OffboardingBlocker).where( and_( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, OffboardingBlocker.resource_type.in_( list(MUST_TRANSFER_RESOURCE_TYPES) ), ) ) ) ).scalars().all() return list(rows) async def _mark_superseded( self, process_id: UUID, kinds: tuple[str, ...], ) -> None: """Insert no-op idempotency rows so a higher band never fires after a lower band already did. Used by the lowest-band cap (Round 2 CR2-003 fix). Without this, the leaver could receive T-1 today and then T-3 / T-7 catch-ups tomorrow when the same outage scenario re-evaluates — same spam class as the original CR2-003 finding. Idempotent: skips kinds that already have a row. """ for kind in kinds: if await self._already_logged(process_id, kind): continue self.db.add(OffboardingReminderLog( process_id=process_id, reminder_kind=kind, channels=None, detail={"superseded_by_lower_band": True}, )) await self.db.flush() async def _walk_to_active_manager( self, person: IdentityPerson, ) -> IdentityPerson | None: """Walk up ``manager_person_id`` chain, skipping any manager whose ``employment_status != 'active'`` or who is soft-deleted. Returns the first active candidate, or None if the chain has no active member. Cycle-protected via ``visited``. Mirrors ``LeaverAdvancer._resolve_fallback_person``'s chain walk so the two cron paths (advance + escalation) agree on recipient validation. """ visited: set[UUID] = {person.id} current = person while current.manager_person_id and current.manager_person_id not in visited: visited.add(current.manager_person_id) mgr = await self.db.get(IdentityPerson, current.manager_person_id) if mgr is None: return None if ( mgr.employment_status == "active" and mgr.deleted_at is None ): return mgr current = mgr return None async def _already_logged(self, process_id: UUID, kind: str) -> bool: row = ( await self.db.execute( select(OffboardingReminderLog).where( and_( OffboardingReminderLog.process_id == process_id, OffboardingReminderLog.reminder_kind == kind, ) ) ) ).scalar_one_or_none() return row is not None async def _log_reminder( self, process_id: UUID, kind: str, *, channels: list[str] | None, detail: dict[str, Any] | None, ) -> None: self.db.add(OffboardingReminderLog( process_id=process_id, reminder_kind=kind, channels=list(channels) if channels else None, detail=detail, )) await self.db.flush() async def _get_person(self, person_id: UUID | None) -> IdentityPerson | None: if not person_id: return None return ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id == person_id) ) ).scalar_one_or_none() async def _list_it_admins(self) -> list[IdentityPerson]: """Identity persons with role=global_admin / it_admin, enabled.""" from app.auth.models import AuthRole, AuthUserRole rows = ( await self.db.execute( select(IdentityPerson) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( and_( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", or_( AuthRole.name == "it_admin", AuthRole.name == GLOBAL_ADMIN_ROLE, ), ) ) .distinct() ) ).scalars().all() return list(rows) async def _portal_url(self) -> str: # Single source of truth for the public frontend origin is # ``app.frontend_base_url`` (the same key the centralized # ``NotificationService._resolve_absolute_url`` reads). The # earlier parallel ``portal.base_url`` key drifted out of sync # on prod (set to ``http://localhost:3000`` long after the real # ``app.frontend_base_url`` was pointed at the production # origin), so every escalation card / email body leaked # ``localhost`` while the action button rendered correctly — # the user-visible bug that triggered this consolidation. from app.integrations.service import get_platform_setting v = await get_platform_setting( "app.frontend_base_url", default="", session=self.db, ) return str(v or "").rstrip("/") def _blocker_to_transfer_type(resource_type: str) -> str: """Map blocker resource_type → AssetTransferRequest.asset_type. Mirrors the inverse of ``OffboardingService._transfer_type_to_blocker_type`` but lives here so the escalation module doesn't import the fat service. Pre-launch P1 fix (Round 1, CR-020): the prior ``.get(rt, rt)`` fallback silently returned the raw blocker resource_type for unmapped kinds. Combined with ``MUST_TRANSFER_RESOURCE_TYPES`` growing without an entry here, that produced AssetTransferRequest rows the manager could never accept (the accept path's ``_initiator_still_holds_asset`` had no matching branch). Raise on unmapped types so a future contributor catches the gap at runtime instead of in production. """ mapping = { RES_HARDWARE: "hardware", RES_API_KEY: "api_key", } if resource_type not in mapping: raise KeyError( f"_blocker_to_transfer_type: no AssetTransferRequest " f"asset_type mapping for blocker resource_type " f"{resource_type!r}. Either extend this mapping AND add an " f"_apply_transfer / _initiator_still_holds_asset branch in " f"OffboardingService, or remove the type from " f"MUST_TRANSFER_RESOURCE_TYPES." ) return mapping[resource_type]