"""Onboarding state-machine service. The cockpit-aggregator entry point (``OnboardingCockpitService.build_queue``) lives in ``cockpit_service.py`` to keep this file focused on the state machine. T+0 advance + T+7 verify live in ``advance_service.py`` and ``t7_verify_service.py`` respectively. Public entry points ------------------- * ``start_process(person_id, hire_date)`` — idempotent. Called by the HR sync coordinator on every new-hire confirmation and on every HRIS sync thereafter (so a hire_date correction lands cleanly). * ``rescan(process_id)`` — admin manual refresh; same scanner as the start path. * ``list_processes(filters)`` — admin queue. * ``get_process(process_id)`` — admin detail (process + blockers). * ``set_hardware_allocation`` — manager-set per-slot device source map (admin endpoint in PR-1; manager endpoint in PR-3). * ``cancel_process(process_id, reason)`` — admin rollback (HRIS rescinded the offer). * ``advance_now(process_id)`` — admin manual hire_date arrival; flips ready → live ahead of the cron tick (rare; for emergency Day-1 push). * ``reconcile_blockers(process_id)`` — refresh blocker state from the linked provisioning_tasks (the 5-min cron drives this for every active process). * ``skip_blocker / mark_blocker_resolved`` — admin manual operations on individual blockers. """ from __future__ import annotations import json import logging import re from datetime import UTC, datetime, timedelta from datetime import date as date_type from typing import TYPE_CHECKING, Any from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import and_, delete, func, or_, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.models import AuditLog from app.audit.service import AuditService from app.core.coordinators.onboarding_hardware_coordinator import ( autoclaim_reserved_for_hire, resolve_hire_hardware_state, ) from app.core.exceptions import ( AppException, format_exception_for_log, is_transaction_aborting_db_error, ) from app.core.provision_match import policy_match_conditions from app.core.working_hours import ( hire_date_is_stale_backfill, resolve_timezone, ) from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.domains.identity_sync.models import ( IdentitySyncState, IdentitySyncTarget, ) from app.domains.onboarding.models import ( PROC_CANCELLED, PROC_LIVE, PROC_OPEN_STATUSES, PROC_PREPARING, PROC_READY, PROC_SCANNING, PROC_TERMINAL_STATUSES, RES_AI_TOOL_DEFAULT, RES_EMAIL_GROUP, RES_FEISHU_CONTACT, RES_HARDWARE_ALLOCATION, RES_SOFTWARE_SEAT, RES_SSO_PROVISION, RES_TRAINING_ACK, RES_VPN, RES_WORK_EMAIL, RESO_MANUAL_PROVISIONED, RESO_PENDING, RESO_PROVISIONED, RESO_SKIPPED, SEVERITY_ADVISORY, SEVERITY_BLOCKER, SLA_BREACH, SLA_NONE, SLA_URGENT, SLA_WARN, OnboardingAckItem, OnboardingAcknowledgement, OnboardingBlocker, OnboardingProcess, ) from app.domains.provisioning.models import ( TARGET_SEAT, ProvisionPolicy, SoftwareConnector, ) from app.domains.vpn_application.models import ( VPN_APP_STATUS_APPROVED, VpnApplication, ) from app.identity.person_models import ( IdentityExternalIdentity, IdentityOrgMembership, IdentityPerson, ) from app.integrations.models import Integration if TYPE_CHECKING: from app.approval.feishu_client import FeishuApprovalClient logger = logging.getLogger(__name__) _APP_TZ = ZoneInfo("Asia/Shanghai") # Asset categories that count as the Day-1 hardware allocation. _COMPUTER_CATEGORIES = ("laptop", "desktop", "computer") _FEISHU_TARGET_KIND = "feishu_contact" # Per-person identity_sync_state.last_status values that mean "this hire # is actually in Feishu contacts": ``success`` (just pushed) or # ``skipped`` (payload unchanged from a prior successful push). Excludes # failed / deactivated / excluded — those are not a live contact. _FEISHU_SYNCED_STATUSES = ("success", "skipped") def _now_utc() -> datetime: return datetime.now(UTC) def _today_app_tz() -> date_type: return datetime.now(_APP_TZ).date() def _local_today_for_person(person: IdentityPerson | None) -> date_type: """Today's date in ``person``'s local TZ. Per-employee state-transition / escalation guards must answer "has this hire's hire_date arrived in THEIR wall-clock?" — a single Asia/Shanghai "today" misjudges US/EU hires near midnight by up to a full calendar day (advanced to live / escalated / scored a day early or late). Mirrors offboarding's ``_local_today_for_person``. Falls back to UTC date when person / TZ resolution fails — the same conservative shape as the offboarding twin. """ now_utc = datetime.now(UTC) if person is None: return now_utc.date() try: tz = resolve_timezone(person) except Exception: # noqa: BLE001 — TZ resolution must never break onboarding return now_utc.date() return now_utc.astimezone(tz).date() # A real HRIS employee number is an optional ``P`` prefix + digits, e.g. # ``P01572`` or the legacy all-numeric ``100113`` / ``80100057`` (a real # person on a different SF instance). SF ALSO seeds non-employee service / # workflow-admin accounts whose ``hris_person_id`` carries LETTERS — # ``LifecyleAdmin`` / ``TimeAdmin`` / ``ECadmin1`` / ``COMPadmin`` — which # are robots, not hires. Matching "looks like an employee number" (rather # than the stricter ``^P\d+$`` identity-sync push gate) is deliberate: the # strict form would wrongly flag the real all-numeric people. _EMPLOYEE_NO_RE = re.compile(r"^P?\d+$") def _is_hris_service_account(person: IdentityPerson | None) -> bool: """True iff this SF-sourced row is a non-employee service / admin account rather than a real hire — its ``hris_person_id`` contains letters instead of being a numeric employee number (2026-07-08 IT feedback: 'LifecyleAdmin' / 'TimeAdmin' kept paging IT「已入职 N 天仍未 配齐」). Only judges ``successfactors`` rows; a non-SF source is never classified here. Kept narrow (letters ⇒ robot) so real employees — ``P#####`` AND legacy all-numeric ids like 100113 — are never caught.""" if person is None: return False if (getattr(person, "source_system", None) or "") != "successfactors": return False hid = (getattr(person, "hris_person_id", None) or "").strip() return bool(hid) and _EMPLOYEE_NO_RE.match(hid) is None def _person_onboarding_active(person: IdentityPerson | None) -> bool: """True iff ``person`` is a live hire onboarding should still act on. A rescinded offer / early departure lands the HRIS row as ``employment_status == 'left'`` (or the row is soft-deleted). The SLA clock, IT escalation dispatchers, and the T+7 verify must NOT treat such a person as an in-flight hire — otherwise IT is paged「已入职 N 天仍未配齐」 for someone who will never arrive, and T+7 opens a compliance incident for a departed person (#1746). Reads only the person's OWN identity columns (no cross-domain import — onboarding must not import offboarding/hr internals). ``hris_record_state == 'purged'`` (2026-07-12): the absent-person reconciliation tombstones a hire whose HRIS record vanished (HR deleting a botched entry) but deliberately leaves ``employment_status`` ``active``. The account-minting gates already skip purged rows; the state machine must too, or it keeps running SLA/escalation/T+7 for a record HRIS no longer carries — the roger.zhou02 shape surviving on the cockpit/compliance side (a false ``onboarding_day1_gap`` incident for a deleted hire). Un-purge on re-ingest flips this back to present. """ if person is None: return False if person.deleted_at is not None: return False if person.hris_record_state == "purged": return False return person.employment_status != "left" # Per-resource-type severity. Used by the scanner to pick the default # severity when creating a blocker. Anything not listed here is treated # as ``SEVERITY_BLOCKER`` (fail-closed) per the Round-14 convention from # offboarding. _SEVERITY_BY_RESOURCE: dict[str, str] = { RES_WORK_EMAIL: SEVERITY_BLOCKER, RES_SSO_PROVISION: SEVERITY_BLOCKER, RES_FEISHU_CONTACT: SEVERITY_ADVISORY, RES_HARDWARE_ALLOCATION: SEVERITY_BLOCKER, RES_SOFTWARE_SEAT: SEVERITY_ADVISORY, RES_VPN: SEVERITY_ADVISORY, RES_EMAIL_GROUP: SEVERITY_ADVISORY, RES_AI_TOOL_DEFAULT: SEVERITY_ADVISORY, RES_TRAINING_ACK: SEVERITY_ADVISORY, } # Human-readable label seed for each resource type. The scanner may # override this with a more specific label (e.g. naming the missing # software) — this is only the default. _DEFAULT_LABEL: dict[str, str] = { RES_WORK_EMAIL: "Work email", RES_SSO_PROVISION: "SSO provisioning", RES_FEISHU_CONTACT: "Feishu contact sync", RES_HARDWARE_ALLOCATION: "Hardware allocation", RES_SOFTWARE_SEAT: "Software seat", RES_VPN: "VPN access", RES_EMAIL_GROUP: "Email group membership", RES_AI_TOOL_DEFAULT: "AI tool access", RES_TRAINING_ACK: "Training acknowledgement", } def _severity_for(resource_type: str) -> str: return _SEVERITY_BY_RESOURCE.get(resource_type, SEVERITY_BLOCKER) def _label_for(resource_type: str, suffix: str | None = None) -> str: base = _DEFAULT_LABEL.get(resource_type, resource_type) label = f"{base}: {suffix}" if suffix else base # OnboardingBlocker.label is String(255) — truncate defensively so a # long missing-item list can't fail the INSERT. return label[:255] # Plain-language "what's missing" phrase per resource type, in the # recipient's language (zh — IT escalation cards are IT-internal). Used # by the IT escalation card so a reader sees "电脑设备未发放" instead of # the internal "[blocker] Hardware allocation" code+English label. Every # onboarding push must answer "what should the reader DO?" — these # phrases name the gap; the card's single CTA names the fix. #: resource_type → (zh, en, ja) gap phrase for the IT escalation card. #: Three languages because the card's recipients are the IT global_admins, and #: not all of them read Chinese — a 2026-08-17 render of the production outbox #: found this whole escalation family shipping Chinese-only. _IT_GAP_LABELS: dict[str, tuple[str, str, str]] = { RES_WORK_EMAIL: ( "企业邮箱未开通", "Work email not provisioned", "会社メール未開通", ), RES_SSO_PROVISION: ( "SSO 账号未创建", "SSO account not created", "SSO アカウント未作成", ), RES_FEISHU_CONTACT: ( "飞书通讯录未同步", "Feishu directory not synced", "Feishu 連絡先未同期", ), RES_SOFTWARE_SEAT: ( "软件席位未分配", "Software seat not assigned", "ソフトウェア席未割当", ), RES_HARDWARE_ALLOCATION: ( "电脑设备未发放", "Laptop not handed out", "PC 未配布", ), RES_VPN: ("VPN 未开通", "VPN not enabled", "VPN 未開通"), RES_EMAIL_GROUP: ( "邮件组未加入", "Mail groups not joined", "メーリングリスト未参加", ), RES_AI_TOOL_DEFAULT: ( "AI 工具未开通", "AI tools not enabled", "AI ツール未開通", ), RES_TRAINING_ACK: ( "入职须知未确认", "Onboarding notice not acknowledged", "入社案内が未確認", ), } def _it_gap_label(resource_type: str, fallback: str) -> tuple[str, str, str]: """Plain-language gap phrase for the IT card, in all three languages. Falls back to the raw blocker label (single-language, whatever the scanner wrote) only for an unmapped — i.e. future — resource type. """ return _IT_GAP_LABELS.get(resource_type, (fallback, fallback, fallback)) # The work_email + sso_provision dimensions both gate on the SAME Google # login identity (see _scan_dimensions — both map to ``has_login_identity``), # so an IT card that lists 「企业邮箱未开通」AND「SSO 账号未创建」as two # separate items is pure redundancy: SSO 账号就是 Google 邮箱账号 (2026-07-08 # IT feedback). Collapse the pair into ONE line. _LOGIN_IDENTITY_RES = (RES_WORK_EMAIL, RES_SSO_PROVISION) _LOGIN_IDENTITY_GAP = ( "企业邮箱 / SSO 账号未开通(同一 Google 账号)", "Work email / SSO account not provisioned (one Google account)", "会社メール / SSO アカウント未開通(同一 Google アカウント)", ) def _it_gap_lines( blockers: list[OnboardingBlocker], ) -> list[tuple[str, str, str]]: """Plain-language gap lines for the IT escalation card, one ``(zh, en, ja)`` triple per gap, with the work_email + sso_provision pair (same underlying Google account) collapsed into a single line. Order-preserving; the merged line takes the position of whichever of the two appears first. """ lines: list[tuple[str, str, str]] = [] login_gap_added = False for b in blockers: if b.resource_type in _LOGIN_IDENTITY_RES: if login_gap_added: continue login_gap_added = True lines.append(_LOGIN_IDENTITY_GAP) else: lines.append(_it_gap_label(b.resource_type, b.label)) return lines def _join_detail(missing: list[str], cap: int = 3) -> str | None: """Compact human detail for a blocker label: first N names + count.""" if not missing: return None head = ", ".join(missing[:cap]) extra = len(missing) - cap return f"{head} (+{extra})" if extra > 0 else head class OnboardingService: """State-machine driver for ``onboarding_processes``.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ───────────────────────── Process lifecycle ───────────────────────── async def start_process( self, person_id: UUID, hire_date: date_type, *, trace_id: str, actor_user_id: UUID | None = None, ) -> OnboardingProcess: """Create or reopen an onboarding process for a person. Idempotent re-entry semantics (mirrors offboarding): * existing open process with the same hire_date → return unchanged (HRIS re-syncs are no-ops). * existing open process with a different hire_date → update in place + audit + rescan (HRIS corrected a typed-wrong date). * existing terminal process (completed / cancelled) → reopen with scanning status + fresh scan (re-hired employee). The HR sync coordinator hook should call this for every confirmed hire event; ``hire_date`` may be in the past (HRIS-late detection) or the future (typical case). """ existing = ( await self.db.execute( select(OnboardingProcess) .where(OnboardingProcess.person_id == person_id) .with_for_update() ) ).scalar_one_or_none() if existing and existing.status in PROC_OPEN_STATUSES: if existing.hire_date != hire_date: prior_date = existing.hire_date existing.hire_date = hire_date # A corrected hire_date re-arms the date-anchored # single-shots: a typo'd past date may have burnt the # day0_it_gaps + SLA escalation slots already — without # this, the REAL arrival passes in silence (review-r3). # The t7_verify_failed marker is deliberately kept (it # dedups compliance incidents within the cycle, not a # date-anchored reminder). from app.domains.onboarding.models import ( REMINDER_DAY0_IT_GAPS, REMINDER_SLA_BREACH, REMINDER_SLA_URGENT, OnboardingReminderLog, ) await self.db.execute( delete(OnboardingReminderLog).where( OnboardingReminderLog.process_id == existing.id, OnboardingReminderLog.reminder_kind.in_( [ REMINDER_DAY0_IT_GAPS, REMINDER_SLA_URGENT, REMINDER_SLA_BREACH, ] ), ) ) # #1772: a typo'd PAST hire_date makes the T+0 advance cron # flip the process to ``live`` (and stamp ``live_at``). # When HR then corrects the date to the real FUTURE date, # ``_recompute_status`` deliberately never demotes ``live`` # (that guard protects an admin's intentional early-seat via # mark_live_manual / advance_now) — so the process stayed # ``live`` with a stale ``live_at``, and ~7 days later # ``verify_all_due`` (``live_at <= now-7d``) either marked a # not-yet-arrived hire ``completed`` (real Day-1 untracked) # or opened an ``onboarding_day1_gap`` incident for someone # who hadn't started. The corrected HRIS date is the # authoritative signal the person has NOT arrived: demote # back out of ``live`` and clear ``live_at`` so the T+7 # window re-arms off the real arrival. ``_rescan`` below then # recomputes ready/preparing from the live blocker set. _person = await self.db.get(IdentityPerson, existing.person_id) if ( existing.status == PROC_LIVE and hire_date > _local_today_for_person(_person) ): existing.status = PROC_READY existing.live_at = None await self.db.flush() await self._rescan(existing) await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="onboarding.hire_date_updated", status="success", resource_type="onboarding_process", resource_id=str(existing.id), detail={ "prior_hire_date": prior_date.isoformat(), "new_hire_date": hire_date.isoformat(), }, ) return existing if existing and existing.status in PROC_TERMINAL_STATUSES: # Re-hire path. Reset the state machine, keep the row (audit # trail integrity — the prior cycle is observable via the # audit log + step_log). prior_status = existing.status existing.status = PROC_SCANNING existing.hire_date = hire_date existing.scanned_at = None existing.ready_at = None existing.hard_ready_at = None existing.live_at = None existing.completed_at = None existing.cancelled_at = None existing.first_ungated_at = None existing.sla_breach_level = SLA_NONE existing.backfilled = False # Re-hire is a brand-new employment cycle: drop the prior # cycle's blocker rows so ``_rescan`` re-evaluates every # dimension from scratch. Without this, a ``skipped`` (or any # terminal) blocker from the first stint would be silently # inherited — ``_rescan`` never reopens a non-PROVISIONED # blocker — suppressing a real Day-1 gap for the boomerang # hire (the audit trail of the prior cycle survives in the # audit log + the process row's history, not these rows). await self.db.execute( delete(OnboardingBlocker).where( OnboardingBlocker.process_id == existing.id ) ) # Same for the reminder log: every dispatcher is single-shot # per (process, kind), so inherited rows from the first # stint would silently mute ALL reminders + escalations for # the boomerang hire's second cycle (review-r1). from app.domains.onboarding.models import OnboardingReminderLog await self.db.execute( delete(OnboardingReminderLog).where( OnboardingReminderLog.process_id == existing.id ) ) await self.db.flush() await self._rescan(existing) await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="onboarding.process_reopened", status="success", resource_type="onboarding_process", resource_id=str(existing.id), detail={ "prior_status": prior_status, "hire_date": hire_date.isoformat(), }, ) return existing # Fresh process. Wrap the INSERT in a SAVEPOINT so a unique- # constraint race (a concurrent start_process for the same # person — the top-of-method ``with_for_update()`` locks nothing # when no row exists yet) rolls back ONLY this insert. A bare # ``self.db.rollback()`` here would nuke the entire outer # transaction — and ``start_process`` is called by # ``hr_sync_coordinator.provision_for_new_hire`` BEFORE the HRIS # batch's per-event commit, so the caller's session still holds # the person upsert + ``hr_lifecycle_events`` row + sibling # coordinator writes. Nuking those silently drops the new hire # from the pipeline. # A hire_date far in the past means this is an existing employee # first synced into Helix long after their start (HRIS-late # detection / bulk backfill), NOT a live Day-1 onboarding. Mark it # ``backfilled`` at birth so the escalation dispatchers, T+7 # compliance verify, and analytics all treat it as a 补录 row — # the same lenient handling migration 229 gave the one-time # historical seed — instead of paging IT「已入职 N 天仍未配齐」about # someone who has worked here for years. (The symmetric # provisioning gate lives in hr_sync_coordinator.) proc = OnboardingProcess( person_id=person_id, hire_date=hire_date, status=PROC_SCANNING, backfilled=hire_date_is_stale_backfill(hire_date), ) try: async with self.db.begin_nested(): self.db.add(proc) await self.db.flush() except IntegrityError: # Race lost: the concurrent caller created an equivalent # fresh scanning process for this same hire. The savepoint # rolled back our orphan insert; reload and return the # winner's row. Use ``scalar_one_or_none`` (not # ``scalar_one``) so a NON-race IntegrityError — e.g. a bad # FK to identity_persons — surfaces as the real error instead # of crashing the HR-sync coordinator with a confusing # NoResultFound. existing = ( await self.db.execute( select(OnboardingProcess).where( OnboardingProcess.person_id == person_id ) ) ).scalar_one_or_none() if existing is None: raise return existing await self._rescan(proc) await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="onboarding.process_started", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "hire_date": hire_date.isoformat(), }, ) return proc async def cancel_process( self, process_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, reason: str | None = None, ) -> OnboardingProcess: """Admin abort — typically because HRIS rescinded the offer.""" proc = await self._get_or_raise(process_id, for_update=True) if proc.status in PROC_TERMINAL_STATUSES: return proc proc.status = PROC_CANCELLED proc.cancelled_at = _now_utc() await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.process_cancelled", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={"reason": (reason or "").strip()[:512] or None}, ) return proc async def advance_now( self, process_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, ) -> OnboardingProcess: """Force ``ready`` → ``live`` ahead of the daily T+0 advance cron. Used for emergency Day-1 push (e.g. hire_date corrected to today an hour before the cron runs). Refuses if the process is not in ``ready`` (the scanner-driven gate is the only correct entrance to ``live``). """ proc = await self._get_or_raise(process_id, for_update=True) if proc.status != PROC_READY: raise AppException( code="ONBOARDING_NOT_READY", message=f"process status must be 'ready' to advance, got '{proc.status}'", status_code=409, ) # Same active-person gate the T+0 advance cron enforces # (advance_service, #1782): a hire whose offer was rescinded # (employment_status='left') / whose HRIS row was purged / # soft-deleted must NOT be flipped ``live`` — that would count a # never-arriving person as in-seat on the cockpit + drive SLA / # Day-1 analytics until the next reconcile self-heals it. person = await self.db.get(IdentityPerson, proc.person_id) if not _person_onboarding_active(person): raise AppException( code="ONBOARDING_PERSON_INACTIVE", message=( "cannot advance onboarding: the hire is departed / " "purged / soft-deleted" ), status_code=409, ) # Same Day-1 ledger conversion the T+0 cron performs: an # emergency manual push must not leave the hire's earmarked # device stuck in ``reserved`` while they are marked in-seat. claimed_asset_id = await autoclaim_reserved_for_hire( self.db, proc.person_id, trace_id=trace_id, process_id=proc.id, actor_user_id=actor_user_id, ) proc.status = PROC_LIVE proc.live_at = _now_utc() await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.advanced_now", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "autoclaimed_asset_id": ( str(claimed_asset_id) if claimed_asset_id else None ), }, ) return proc async def mark_live_manual( self, process_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, ) -> OnboardingProcess: """Manual admin override: force an open process straight to ``live``. Distinct from :meth:`advance_now`, which only promotes a scanner-gated ``ready`` process. The cockpit's 「直接就位」 action lets an admin who has confirmed a hire is set up out-of-band mark them in-seat regardless of remaining dimension gaps. Refuses only on already-``live`` or terminal (``completed`` / ``cancelled``) processes — both are no-ops that should surface as a clear 409 rather than silently re-stamping ``live_at``. """ proc = await self._get_or_raise(process_id, for_update=True) if proc.status == PROC_LIVE: raise AppException( code="ONBOARDING_ALREADY_LIVE", message="process is already live", status_code=409, ) if proc.status in PROC_TERMINAL_STATUSES: raise AppException( code="ONBOARDING_TERMINAL", message=f"cannot mark a {proc.status} process live", status_code=409, ) # Never mark a departed / purged / soft-deleted hire live (same gate # as advance_now / the T+0 cron). An admin's out-of-band 「直接就位」 # must not resurrect a rescinded hire into the in-seat population. person = await self.db.get(IdentityPerson, proc.person_id) if not _person_onboarding_active(person): raise AppException( code="ONBOARDING_PERSON_INACTIVE", message=( "cannot mark live: the hire is departed / purged / " "soft-deleted" ), status_code=409, ) prior_status = proc.status # 「直接就位」 also converts a reservation into a claim: the admin # is asserting the hire is set up, and an earmarked machine is # part of that setup. claimed_asset_id = await autoclaim_reserved_for_hire( self.db, proc.person_id, trace_id=trace_id, process_id=proc.id, actor_user_id=actor_user_id, ) proc.status = PROC_LIVE proc.live_at = _now_utc() await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.marked_live_manual", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "prior_status": prior_status, "autoclaimed_asset_id": ( str(claimed_asset_id) if claimed_asset_id else None ), }, ) return proc async def set_hardware_allocation( self, process_id: UUID, allocation: dict[str, Any] | None, *, trace_id: str, actor_user_id: UUID | None = None, ) -> OnboardingProcess: proc = await self._get_or_raise(process_id) if proc.status in PROC_TERMINAL_STATUSES: raise AppException( code="ONBOARDING_TERMINAL", message=f"cannot edit a {proc.status} process", status_code=409, ) # Size guard (review-r3): the body is free-form manager-written # JSON; cap it so a script can't bloat the row (20 slots / 8KB # serialized — far above any real declaration). if allocation is not None: if len(allocation) > 20: raise AppException( code="HARDWARE_ALLOCATION_TOO_LARGE", message="at most 20 hardware slots", status_code=400, ) import json as _json if len(_json.dumps(allocation, ensure_ascii=False)) > 8192: raise AppException( code="HARDWARE_ALLOCATION_TOO_LARGE", message="allocation payload exceeds 8KB", status_code=400, ) proc.hardware_allocation = allocation await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.hardware_allocation_set", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "slot_count": len(allocation) if allocation else 0, }, ) return proc # ───────────────────────── Blocker mutations ───────────────────────── async def skip_blocker( self, blocker_id: UUID, *, reason: str, trace_id: str, actor_user_id: UUID | None = None, ) -> OnboardingBlocker: clean_reason = (reason or "").strip() if not clean_reason: raise AppException( code="REASON_REQUIRED", message="skip reason is required", status_code=400, ) blocker = await self._get_blocker_or_raise(blocker_id) if blocker.resolution != RESO_PENDING: return blocker blocker.resolution = RESO_SKIPPED blocker.resolution_ref = clean_reason[:255] blocker.notes = ( (blocker.notes or "") + f"\n[skipped] {clean_reason}" ).strip() await self.db.flush() # Reload process for status recompute. proc = await self.db.get(OnboardingProcess, blocker.process_id) if proc is not None: await self._recompute_status(proc) await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.blocker_skipped", status="success", resource_type="onboarding_blocker", resource_id=str(blocker.id), detail={ "resource_type": blocker.resource_type, "reason": clean_reason[:512], }, ) return blocker async def mark_blocker_resolved( self, blocker_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, note: str | None = None, ) -> OnboardingBlocker: blocker = await self._get_blocker_or_raise(blocker_id) if blocker.resolution != RESO_PENDING: return blocker # STICKY manual confirmation — NOT the scanner's ``provisioned``. # The 5-min rescan re-pends only rows IT itself auto-wrote as # ``provisioned`` (resource read absent again); a value distinct # from that keeps this admin decision from being silently reverted # and re-escalated every tick (#1742). blocker.resolution = RESO_MANUAL_PROVISIONED if note: blocker.resolution_ref = note.strip()[:255] await self.db.flush() proc = await self.db.get(OnboardingProcess, blocker.process_id) if proc is not None: await self._recompute_status(proc) await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="onboarding.blocker_resolved", status="success", resource_type="onboarding_blocker", resource_id=str(blocker.id), detail={"resource_type": blocker.resource_type}, ) return blocker # ───────────────────────── Query helpers ───────────────────────── async def get_process( self, process_id: UUID, ) -> tuple[OnboardingProcess, list[OnboardingBlocker]]: proc = await self._get_or_raise(process_id) blockers = ( ( await self.db.execute( select(OnboardingBlocker) .where(OnboardingBlocker.process_id == proc.id) .order_by(OnboardingBlocker.resource_type) ) ) .scalars() .all() ) return proc, list(blockers) async def get_account_action_timeline( self, person_id: UUID, *, process_id: UUID | None = None, limit: int = 200, ) -> list[dict[str, Any]]: """Chronological account-action trail for one hire. Reads the append-only ``audit_logs`` — the single source of truth for "what Helix actually DID to this person's account and whether it succeeded" — instead of a second event store (same rationale as :mod:`cmdb.timeline_service`). Admins previously saw only the cockpit's green/red lights; when a hire had no account (P10002) there was no way to see *why* — the Google error text sits in ``audit_logs.detail.reason`` of a ``google_workspace. account_create_failed`` row that nothing surfaced. Two audit scopes are unioned: * ``resource_type='identity_person'`` keyed on ``person_id`` — google_workspace.account_created / account_linked / account_create_failed / reconcile_lookup_failed (welcome-email outcome rides in the create row's detail), plus per-person provisioning rows (onboarding_tasks_created, role_change). * ``resource_type='onboarding_process'`` keyed on ``process_id`` (when known) — process_started / advanced_now / cancelled / hardware_allocation_set. Both predicates hit the composite ``(resource_type, resource_id, created_at)`` index (migration 180), so this is an index range scan, not a table scan. Newest-first, capped at ``limit``. """ scope = or_( and_( AuditLog.resource_type == "identity_person", AuditLog.resource_id == str(person_id), ), *( [ and_( AuditLog.resource_type == "onboarding_process", AuditLog.resource_id == str(process_id), ) ] if process_id is not None else [] ), ) rows = ( ( await self.db.execute( select(AuditLog) .where(scope) .order_by(AuditLog.created_at.desc()) .limit(limit) ) ) .scalars() .all() ) audit_events = [ { "id": str(r.id), "action": r.action, "status": r.status, "actor_type": r.actor_type, "resource_type": r.resource_type, # detail is writer-curated + small for these actions (no # provisioning-task vendor blobs, which key on a different # resource_type and aren't in scope) — pass it through so # the UI can render the Google error / primary_email / # welcome-email outcome verbatim. "detail": r.detail or {}, "created_at": ( r.created_at.isoformat() if r.created_at else None ), } for r in rows ] person = await self.db.get(IdentityPerson, person_id) sf_events: list[dict[str, Any]] = [] if person is not None and person.hris_person_id: instance_clause = "" params: dict[str, Any] = { "user_id": person.hris_person_id, "sf_limit": max(limit * 5, 200), } if person.hris_instance_id is not None: instance_clause = ( "AND (b.hris_instance_id = :instance_id " "OR b.hris_instance_id IS NULL)" ) params["instance_id"] = str(person.hris_instance_id) staged_rows = ( await self.db.execute( text( "SELECT se.id, se.event_type, se.action, " "se.person_data, se.created_at, " "b.trigger_source, b.sync_mode " "FROM hr_sync_staged_events se " "JOIN hr_sync_batches b ON b.id = se.batch_id " "WHERE se.user_id = :user_id " f"{instance_clause} " "ORDER BY se.created_at ASC " "LIMIT :sf_limit" ), params, ) ).mappings().all() seen_snapshots: set[tuple[Any, ...]] = set() placeholder_tokens = {"", "unknown", "null", "n/a", "na", "-"} for staged in staged_rows: pdata = staged["person_data"] or {} if isinstance(pdata, str): pdata = json.loads(pdata) detail = { "trigger_source": staged["trigger_source"], "sync_mode": staged["sync_mode"], "event_type": staged["event_type"], "staged_action": staged["action"], "display_name": pdata.get("display_name"), "first_name": pdata.get("first_name"), "last_name": pdata.get("last_name"), "email_name": pdata.get("email_name"), "work_email": pdata.get("work_email"), "employee_no": pdata.get("employee_no"), "hire_date": pdata.get("hire_date"), "source_updated_at": pdata.get("source_updated_at"), } fingerprint = tuple(detail.get(key) for key in ( "event_type", "staged_action", "display_name", "first_name", "last_name", "email_name", "work_email", "employee_no", "hire_date", )) if fingerprint in seen_snapshots: continue seen_snapshots.add(fingerprint) risk_flags: list[str] = [] display = str(detail.get("display_name") or "").strip().casefold() email_name = str(detail.get("email_name") or "").strip().casefold() if display in placeholder_tokens: risk_flags.append("placeholder_display_name") if email_name and email_name.split("@", 1)[0] in placeholder_tokens: risk_flags.append("placeholder_email_name") detail["risk_flags"] = risk_flags created_at = staged["created_at"] sf_events.append({ "id": f"hris-stage:{staged['id']}", "action": "hris.sync_snapshot_received", "status": "warning" if risk_flags else "success", "actor_type": "system", "resource_type": "hr_sync_staged_event", "detail": detail, "created_at": created_at.isoformat() if created_at else None, }) return sorted( [*audit_events, *sf_events], key=lambda event: event.get("created_at") or "", reverse=True, )[:limit] async def list_processes( self, *, status_in: list[str] | None = None, hire_date_from: date_type | None = None, hire_date_to: date_type | None = None, limit: int = 100, offset: int = 0, ) -> tuple[list[OnboardingProcess], int]: clauses = [] if status_in: clauses.append(OnboardingProcess.status.in_(status_in)) if hire_date_from: clauses.append(OnboardingProcess.hire_date >= hire_date_from) if hire_date_to: clauses.append(OnboardingProcess.hire_date <= hire_date_to) total = int( ( await self.db.execute( select(func.count()) .select_from(OnboardingProcess) .where(and_(*clauses)) if clauses else select(func.count()).select_from(OnboardingProcess) ) ).scalar_one() ) query = ( select(OnboardingProcess) .order_by(OnboardingProcess.hire_date.desc()) .limit(limit) .offset(offset) ) if clauses: query = query.where(and_(*clauses)) rows = (await self.db.execute(query)).scalars().all() return list(rows), total # ───────────────────────── Reconcile ───────────────────────── async def reconcile_blockers( self, process_id: UUID, *, feishu_client: FeishuApprovalClient | None = None, ) -> int: """Refresh blocker state by rescanning the resource layer. Cheap-ish — runs all per-dimension queries once. Returns the number of blocker rows whose resolution flipped this call. ``feishu_client`` (cron path) enables the live 飞书 directory check. """ # #1781: lock the row for the duration of the reconcile. The scan # window includes a live 飞书 HTTP call (``resolve_user_id_by_email``), # and an admin ``cancel_process`` (which also locks the row) can # commit while we're mid-scan. Without a locking read this method # read a pre-cancel snapshot (REPEATABLE READ pins the txn snapshot # at the first read), passed the terminal short-circuit, then # ``_recompute_status`` flushed ``status='ready'/'live'`` off the # stale in-memory row — resurrecting the just-cancelled process. A # FOR UPDATE read sees the winner's commit: if cancel landed first # we observe the terminal status and return; otherwise cancel waits # until this tick releases and then cancels the reconciled row. proc = await self._get_or_raise(process_id, for_update=True) if proc.status in PROC_TERMINAL_STATUSES: return 0 before_pending = await self._count_pending_blockers(proc.id) await self._rescan(proc, feishu_client=feishu_client) await self._recompute_status(proc) after_pending = await self._count_pending_blockers(proc.id) return max(0, before_pending - after_pending) async def _build_feishu_client_or_none( self, ) -> FeishuApprovalClient | None: """Best-effort Feishu client for the live 飞书 directory check. Reuses the shared chat_approval credentials (same app the approval / notification layer uses) via ``get_feishu_config`` — the exact pattern every other domain follows. Lazy-imported so the module graph and the HTTP scan paths stay free of the approval client. Returns ``None`` (never raises) if credentials are missing so a connector hiccup can't abort the reconcile tick. """ try: from app.approval.feishu_client import FeishuApprovalClient from app.integrations.compat import get_feishu_config cfg = await get_feishu_config(session=self.db) app_id = str(cfg.get("app_id") or "") app_secret = str(cfg.get("app_secret") or "") if not app_id or not app_secret: return None return FeishuApprovalClient(app_id=app_id, app_secret=app_secret) except Exception as exc: # noqa: BLE001 — best-effort logger.warning( "onboarding.reconcile: feishu client unavailable, 飞书 " "dimension falls back to DB signal: %s", format_exception_for_log(exc), ) return None async def reconcile_all_active_processes(self) -> dict[str, Any]: """5-min cron entry point. Returns ``{scanned_count, blockers_flipped, failed_count, errors}``. Per-process failures are collected into ``errors`` (not propagated) so one bad process can't abort the tick; the handler re-raises on a non-zero ``failed_count`` so the run records 'failed' and ops is paged (handler-must-raise contract — mirrors offboarding's reconcile). """ ids = ( ( await self.db.execute( select(OnboardingProcess.id).where( OnboardingProcess.status.in_(PROC_OPEN_STATUSES) ) ) ) .scalars() .all() ) # Build the Feishu directory client ONCE per tick (its in-process # token / email→id cache is then reused across every process). The # live 飞书 lookup runs only on this cron path; the HTTP scan paths # pass no client. Build is best-effort — if the chat_approval # integration is absent / misconfigured, the 飞书 dimension simply # falls back to its DB signal instead of aborting the whole tick. feishu_client = await self._build_feishu_client_or_none() total = 0 errors: list[dict[str, str]] = [] for pid in ids: # Per-process SAVEPOINT so one process whose rescan raises # (transient DB error, deadlock, an integrity edge) doesn't # poison the AsyncSession for the rest of the batch. Without # it, the first failure leaves the session in a failed- # transaction state and every subsequent ``reconcile_blockers`` # — plus the handler's final ``session.commit()`` — raises # PendingRollbackError, discarding the whole 5-min tick. # Mirrors ProcessAdvanceService.run_tick's per-person isolation. try: async with self.db.begin_nested(): total += await self.reconcile_blockers( pid, feishu_client=feishu_client ) except Exception as exc: # A transaction-aborting error (deadlock/lock-wait/orphaned # savepoint) kills the whole txn on TiDB; swallowing it would # cascade onto every later process and make the handler's # final commit raise PendingRollbackError, masking the real # cause. Re-raise so the handler's lock-retry wrapper retries # the whole tick in a fresh session (reconcile is idempotent). if is_transaction_aborting_db_error(exc): raise logger.exception( "onboarding.reconcile_blockers failed for %s", pid ) errors.append({ "process_id": str(pid), "error": format_exception_for_log(exc, max_len=500), }) return { "scanned_count": len(ids), "blockers_flipped": total, "failed_count": len(errors), "errors": errors, } # ───────────────────────── Scanner ───────────────────────── async def _rescan( self, proc: OnboardingProcess, *, feishu_client: FeishuApprovalClient | None = None, ) -> None: """Re-evaluate every dimension and reconcile blocker rows. For each resource_type, look at the live resource layer: * if the dimension is satisfied: mark the blocker (if it exists) ``provisioned``; otherwise no-op. * if the dimension is NOT satisfied: create a pending blocker if none exists; otherwise leave the pending one in place. This is the single canonical place that decides "what's a Day-1 blocker" — both ``start_process`` and the 5-min reconcile cron call it. The scanner is read-only on the resource layer; it never creates provisioning tasks (that's PR-3's bundle-apply + cockpit inline-action surface). """ person = await self.db.get(IdentityPerson, proc.person_id) if person is None: # The person row went away mid-flight. Cancel the process so # the cron stops looping over it. Log it — this is a terminal # transition on a rare path (person hard-deleted during # onboarding) and otherwise leaves no trace for operators. logger.warning( "onboarding._rescan: person %s gone mid-flight; " "cancelling process %s", proc.person_id, proc.id, ) proc.status = PROC_CANCELLED proc.cancelled_at = _now_utc() return if not _person_onboarding_active(person): # Rescinded offer / early departure: HRIS marked the hire # ``left`` (or the row was soft-deleted) while onboarding was # still open. Nothing downstream will ever provision this # person, so auto-cancel to stop the 5-min reconcile loop AND # every escalation surface (SLA / day0 / urgent / breach cards, # T+7 incident) — all of which select open/live processes. # This is the clean in-domain seam requested by #1746: the # reconcile scanner already owns "cancel when the person is no # longer a valid hire" (see the person-gone branch above), and # it reads only the person's own identity columns. Audited so # the terminal transition is traceable (unlike the rare # person-gone path, ``left`` is a routine HRIS outcome). logger.info( "onboarding._rescan: person %s is left/soft-deleted; " "auto-cancelling open process %s", proc.person_id, proc.id, ) prior_status = proc.status proc.status = PROC_CANCELLED proc.cancelled_at = _now_utc() await self.audit.log( trace_id=f"system:onboarding:auto_cancel:{proc.id}", actor_type="system", action="onboarding.process_cancelled", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "reason": "person_left_or_soft_deleted", "via": "rescan_auto_cancel", "prior_status": prior_status, "employment_status": person.employment_status, "soft_deleted": person.deleted_at is not None, }, ) return # Pull existing blockers indexed by resource_type so we can # diff. There's at most one open row per resource_type per # process (the scanner doesn't dup-create). existing_blockers = { b.resource_type: b for b in ( await self.db.execute( select(OnboardingBlocker).where( OnboardingBlocker.process_id == proc.id ) ) ).scalars().all() } # Resolve every dimension's current state. Carry whether the feishu # dimension was already directory-confirmed so the scanner keeps it # green (sticky) without re-hitting Feishu on this pass. feishu_bl = existing_blockers.get(RES_FEISHU_CONTACT) feishu_provisioned = ( feishu_bl is not None and feishu_bl.resolution in (RESO_PROVISIONED, RESO_MANUAL_PROVISIONED) ) dim_state = await self._scan_dimensions( proc, person, feishu_client=feishu_client, feishu_blocker_provisioned=feishu_provisioned, ) # Diff: ensure each dimension has the right blocker row. for resource_type, (satisfied, detail) in dim_state.items(): blocker = existing_blockers.get(resource_type) if satisfied: # Mark provisioned if pending. Don't reopen a SKIPPED # blocker — that's an explicit admin choice. if blocker and blocker.resolution == RESO_PENDING: blocker.resolution = RESO_PROVISIONED continue label = _label_for(resource_type, detail) # Not satisfied. Create blocker if missing. SAVEPOINT-wrap # the INSERT: a concurrent rescan (ack-triggered reconcile # in the HTTP session vs the 5-min worker cron) can race # this create — the UNIQUE (process_id, resource_type) # constraint (migration 281) makes the loser's insert fail; # the savepoint confines the rollback to this row and the # winner's row stands (review-r3). if blocker is None: try: async with self.db.begin_nested(): self.db.add( OnboardingBlocker( process_id=proc.id, resource_type=resource_type, resource_id=None, label=label, severity=_severity_for(resource_type), resolution=RESO_PENDING, ) ) await self.db.flush() except IntegrityError: logger.info( "onboarding._rescan: concurrent blocker create " "for %s/%s — keeping the winner's row", proc.id, resource_type, ) elif blocker.resolution == RESO_PROVISIONED: # The resource was previously provisioned but the scan # now shows it's gone (e.g. seat was revoked, asset # unassigned). Re-pend so the cockpit surfaces the gap. # NOTE: keyed on the scanner-written ``provisioned`` ONLY. # A ``manual_provisioned`` row (admin 「标记已发放」) is # deliberately NOT re-pended here — the resource may live # off-books (device handed out but AssetItem not entered), # so reverting the admin's confirmation every 5-min tick # would re-escalate the same gap forever (#1742). blocker.resolution = RESO_PENDING blocker.label = label blocker.last_error_message = None blocker.last_error_at = None blocker.error_count = 0 elif blocker.resolution == RESO_PENDING and blocker.label != label: # Keep the pending blocker's detail fresh — e.g. one of # two missing email groups landed; the label should now # name only the remaining one. blocker.label = label proc.scanned_at = _now_utc() await self.db.flush() await self._recompute_status(proc) async def _scan_dimensions( self, proc: OnboardingProcess, person: IdentityPerson, *, feishu_client: FeishuApprovalClient | None = None, feishu_blocker_provisioned: bool = False, ) -> dict[str, tuple[bool, str | None]]: """Return ``{resource_type: (is_satisfied, detail)}`` for every dim. ``detail`` (only meaningful when unsatisfied) names the concrete missing resources — e.g. the email groups a policy promises but the hire isn't seated in — so the blocker label tells the operator exactly what to fix instead of a generic dimension name. ``feishu_client`` (cron path only) enables a LIVE Feishu directory check for the 飞书 dimension; ``feishu_blocker_provisioned`` carries whether the feishu blocker was already directory-confirmed so we can keep it green without re-hitting Feishu. Both default off on the HTTP paths (start_process / manual rescan), which fall back to the DB-only signal. """ person_id = person.id person_id_str = str(person_id) # SSO / mailbox: only a real LOGIN identity counts. Every person # synced from HRIS carries a ``provider='successfactors'`` external # identity — but that row is a directory-source record # (``is_login_identity=False``), NOT a login account. Counting it # greened SSO + 账号 + 邮箱 for hires who had no Google account at # all (the P10002 false-green bug). The only rows that mint a real # mailbox / SSO login are ``is_login_identity=True`` (google / # feishu — see google_workspace.account_provisioner and # auth.service). So the SSO dimension — and, below, the work_email # "live mailbox" dimension — gate on that flag, not mere presence. login_identity_count = int( ( await self.db.execute( select(func.count()).select_from(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == person_id, IdentityExternalIdentity.is_login_identity.is_(True), IdentityExternalIdentity.identity_status == "active", IdentityExternalIdentity.deleted_at.is_(None), ) ) ).scalar_one() ) has_login_identity = login_identity_count > 0 # Software seat (any active). seat_count = int( ( await self.db.execute( select(func.count()).select_from(SoftwareSeatAllocation).where( SoftwareSeatAllocation.user_id == person_id, SoftwareSeatAllocation.status == "active", ) ) ).scalar_one() ) # Hardware: a computer-category asset either already handed over # to the hire (``assigned``) or earmarked for them by hardware # prep (``reserved``). Both satisfy the dimension — see # ``onboarding_hardware_coordinator`` for why the earmark counts # (2026-09-05: 29 of 31 Day-1 misses in production were a device # that existed but whose ledger claim happened after the hire # date). The T+0 advance converts the earmark into a real claim, # and ``_recompute_status`` un-banks the credit if the earmark is # released before the hire lands, so this cannot be gamed. hw_state = await resolve_hire_hardware_state(self.db, person_id) # VPN: any approved application. Note: VpnApplication.person_id # is String — bind str(uuid). vpn_count = int( ( await self.db.execute( select(func.count()).select_from(VpnApplication).where( VpnApplication.person_id == person_id_str, VpnApplication.status == VPN_APP_STATUS_APPROVED, ) ) ).scalar_one() ) # Feishu: did THIS hire actually land in a feishu_contact target? # Read the per-person projection (identity_sync_state) rather than # channel-level health — channel health is the wrong proxy: it # stays red while a healthy channel simply hasn't run yet, and # green for hires the sync never covered. A 'success' (pushed) or # 'skipped' (payload unchanged from a prior success) row means the # person is in Feishu, so the advisory auto-resolves on the next # scan instead of sitting pending forever. feishu_synced_count = int( ( await self.db.execute( select(func.count()) .select_from(IdentitySyncState) .join( IdentitySyncTarget, IdentitySyncTarget.id == IdentitySyncState.target_id, ) .where( IdentitySyncTarget.kind == _FEISHU_TARGET_KIND, IdentitySyncTarget.deleted_at.is_(None), IdentitySyncState.person_id == person_id, IdentitySyncState.last_status.in_( _FEISHU_SYNCED_STATUSES ), ) ) ).scalar_one() ) feishu_ready = feishu_synced_count > 0 # The outbound feishu_contact sync is the wrong proxy for "is this # person in Feishu": it can be disabled / never-run (the switch is # held OFF until Authing-parity cutover) while the hire is very # much in the Feishu 通讯录 already. So when the sync signal is # absent, fall back to a LIVE directory lookup by work_email — the # same batch_get_id call the approval / notification layer uses. # * already directory-confirmed (blocker provisioned) → stay # green without re-hitting Feishu. Membership doesn't get # revoked mid-onboarding, and a transient resolve blip must # not re-pend a confirmed hire. # * FEISHU_USER_NOT_FOUND → definitive: leave pending. # * auth / rate-limit / transport error → indeterminate: leave # the DB signal's verdict (never flap a hire red on a blip). if not feishu_ready and feishu_blocker_provisioned: feishu_ready = True if not feishu_ready and feishu_client is not None and person.work_email: try: await feishu_client.resolve_user_id_by_email(person.work_email) feishu_ready = True except AppException as exc: if exc.code != "FEISHU_USER_NOT_FOUND": logger.warning( "onboarding: feishu directory resolve indeterminate " "for %s (%s): %s", person_id, person.work_email, format_exception_for_log(exc), ) except Exception as exc: # noqa: BLE001 — network/transport logger.warning( "onboarding: feishu directory resolve error for %s " "(%s): %s", person_id, person.work_email, format_exception_for_log(exc), ) # Policy-promised plan subsets: which seat plans do active # ProvisionPolicy rows promise this person, and of those, which # are (a) email-group-backed (Google Workspace group_member # binding) and (b) AI tools (software category='ai'). The # generic RES_SOFTWARE_SEAT dim stays "any active seat"; these # two carve out the specific policy promises so the cockpit can # say "the dev-all@ group / the Claude seat hasn't landed yet". eg_missing, ai_missing = await self._scan_policy_plan_subsets(person) # Training acknowledgements: every active+required 入职须知 item # must be confirmed by the hire. No items configured → satisfied. ack_missing = await self._scan_missing_acks(person_id) return { # work_email here is the LIVE-mailbox gate for the state # machine: a hire isn't Day-1 ready on a merely-reserved SF # address — the mailbox has to actually exist, which is exactly # what a live login identity proves. (The cockpit renders a # softer "已预留" middle state for the reserved-but-not-live case; # the state machine only cares live-vs-not.) RES_WORK_EMAIL: (has_login_identity, None), RES_SSO_PROVISION: (has_login_identity, None), RES_FEISHU_CONTACT: (feishu_ready, None), RES_SOFTWARE_SEAT: (seat_count > 0, None), # No detail: ``_rescan`` only reads it on the UNsatisfied # branch, so an "已预留 · " string here would be dead # code. The reserved-vs-assigned distinction reaches the # cockpit through its own ``hardware_state`` field # (``cockpit_service``), which is where the UI needs it. RES_HARDWARE_ALLOCATION: (hw_state.satisfied, None), RES_VPN: (vpn_count > 0, None), RES_EMAIL_GROUP: ( not eg_missing, _join_detail(eg_missing), ), RES_AI_TOOL_DEFAULT: ( not ai_missing, _join_detail(ai_missing), ), RES_TRAINING_ACK: ( not ack_missing, _join_detail(ack_missing), ), } async def _scan_policy_plan_subsets( self, person: IdentityPerson, ) -> tuple[list[str], list[str]]: """Return ``(missing_email_group_labels, missing_ai_labels)``. A "promise" is an active ProvisionPolicy seat target matching the person's (department × job_title × location). Of the promised plans: * email-group subset — the plan's software has an active connector binding with ``operation='group_member'`` on a ``google_workspace`` integration (the platform's canonical "membership in a Google group as an entitlement" shape, same as google.seat_sync Phase B); * AI subset — the plan's software has ``category='ai'``. A promised plan with no active seat allocation for the person is "missing". Plans satisfy via the normal provisioning queue — this scanner only reads; it never creates tasks. """ # ``person.primary_department`` walks the lazy org_memberships # relationship (MissingGreenlet in async context unless eager- # loaded) — query the primary membership explicitly instead. department = ( await self.db.execute( select(IdentityOrgMembership.org_node_name) .where( IdentityOrgMembership.person_id == person.id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) # Deterministic pick when a data-quality edge leaves >1 # is_primary row — without ORDER BY the engine could # return a different membership per call and the # "promised plans" set would flap (review-r1). .order_by(IdentityOrgMembership.created_at) .limit(1) ) ).scalar_one_or_none() conditions = policy_match_conditions( ProvisionPolicy, department=department, job_title=person.job_title, location=person.work_location_code, ) conditions.append(ProvisionPolicy.target_type == TARGET_SEAT) plan_ids = { row for row in ( await self.db.execute( select(ProvisionPolicy.target_id).where(and_(*conditions)) ) ).scalars() } if not plan_ids: return [], [] plans = ( ( await self.db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.id.in_(plan_ids) ) ) ) .scalars() .all() ) if not plans: return [], [] software_ids = {p.software_id for p in plans} softwares = { s.id: s for s in ( await self.db.execute( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.id.in_(software_ids) ) ) ).scalars() } connectors = ( ( await self.db.execute( select(SoftwareConnector).where( SoftwareConnector.software_id.in_(software_ids), SoftwareConnector.is_active.is_(True), ) ) ) .scalars() .all() ) integration_ids = {c.integration_id for c in connectors} integration_kinds: dict[UUID, str] = {} if integration_ids: rows = ( await self.db.execute( select(Integration.id, Integration.kind).where( Integration.id.in_(integration_ids) ) ) ).all() integration_kinds = {r.id: r.kind for r in rows} # binding_config is JSON — classify in Python rather than with a # JSON SQL predicate (MySQL vs TiDB JSON-boolean semantics # diverge; Python-side is also trivially cheap at this volume). eg_software_ids = { c.software_id for c in connectors if ( integration_kinds.get(c.integration_id) == "google_workspace" and str((c.binding_config or {}).get("operation") or "") == "group_member" ) } active_plan_ids = { row for row in ( await self.db.execute( select(SoftwareSeatAllocation.plan_id).where( SoftwareSeatAllocation.user_id == person.id, SoftwareSeatAllocation.plan_id.in_( [p.id for p in plans] ), SoftwareSeatAllocation.status == "active", ) ) ).scalars() } eg_missing: list[str] = [] ai_missing: list[str] = [] for plan in plans: if plan.id in active_plan_ids: continue software = softwares.get(plan.software_id) label = software.name if software else plan.name if plan.software_id in eg_software_ids: eg_missing.append(label) if software is not None and (software.category or "") == "ai": ai_missing.append(label) return sorted(eg_missing), sorted(ai_missing) async def _scan_missing_acks(self, person_id: UUID) -> list[str]: """Titles of active+required ack items the person hasn't confirmed.""" required_items = ( ( await self.db.execute( select(OnboardingAckItem) .where( OnboardingAckItem.is_active.is_(True), OnboardingAckItem.required.is_(True), ) .order_by( OnboardingAckItem.sort_order, OnboardingAckItem.created_at, ) ) ) .scalars() .all() ) if not required_items: return [] acked_ids = { row for row in ( await self.db.execute( select(OnboardingAcknowledgement.item_id).where( OnboardingAcknowledgement.person_id == person_id, OnboardingAcknowledgement.item_id.in_( [i.id for i in required_items] ), ) ) ).scalars() } return [i.title for i in required_items if i.id not in acked_ids] # ───────────────────────── State machine ───────────────────────── async def _recompute_status(self, proc: OnboardingProcess) -> None: """Drive the state machine off the current blocker set. Allowed transitions: scanning → preparing (≥1 pending blocker, hire_date future) scanning → ready (no pending blockers, hire_date future) scanning → live (no pending blockers, hire_date passed) preparing → ready (last pending blocker resolved) preparing → live (hire_date passed; the T+0 cron does it too, but this scanner usually gets there first — see the promote branch, which therefore carries the same T+0 asset auto-claim) ready → live (hire_date passed; same as above) live → completed (T+7 verify-driven, not here) Terminal statuses (``completed`` / ``cancelled``) short-circuit. """ if proc.status in PROC_TERMINAL_STATUSES: return pending_blockers = await self._count_pending_blockers(proc.id) # Per-employee local "today": a US/EU hire near midnight must not # be scored hire_passed a calendar day early/late off a single # Asia/Shanghai reference. One extra db.get per recompute is # acceptable (called from start/rescan/reconcile/advance paths). person = await self.db.get(IdentityPerson, proc.person_id) today = _local_today_for_person(person) hire_passed = proc.hire_date <= today # Stamp ``hard_ready_at`` (once) when the BLOCKER-severity # pending count first hits zero — the Day-1 scorecard's honest # timestamp (migration 280). Advisory dims (须知确认 etc.) must # not gate it; they gate ``ready_at`` / the ``ready`` status. hard_pending = await self._count_pending_blockers( proc.id, severity=SEVERITY_BLOCKER, ) if proc.hard_ready_at is None: if hard_pending == 0: proc.hard_ready_at = _now_utc() elif hard_pending > 0 and not hire_passed: # Un-bank pre-arrival readiness (2026-09-05). Since the # hardware dimension accepts a mere reservation, a device # earmarked at T-5 and released at T-2 would otherwise leave # ``hard_ready_at`` stamped and score the hire Day-1 ready # with no machine anywhere. Clearing it while the hire has # NOT yet arrived keeps the timestamp honest and lets the # next satisfied scan re-stamp it. # # Deliberately scoped to ``not hire_passed``: once the hire # date has passed, ``hard_ready_at`` is a HISTORICAL # attainment record. Later churn — the person swaps laptops, # returns a device, an account is deactivated months on — # re-pends a blocker but must not retroactively rewrite what # was true on day 1, which is exactly what the Day-1 and SLA # metrics read. logger.info( "onboarding.hard_ready_reset: process %s lost a hard " "dimension before hire_date %s; clearing hard_ready_at", proc.id, proc.hire_date, ) proc.hard_ready_at = None # Capture transitions explicitly so we set the right timestamp. prior = proc.status if pending_blockers > 0: # Re-pend drives only the PRE-live states (scanning/ready) # back to ``preparing``. A ``live`` process — the person is # already in seat, T+0 advance fired — must NOT be demoted: # there is no documented ``live → preparing`` transition, and # demoting it made a seated-but-still-blocked hire ping-pong # ``live`` ↔ ``preparing`` every 5-min reconcile tick AND # starved the T+7 verify sweep (which selects # ``status == live``), so the process could never complete. # Pending blockers on a live process are an SLA/escalation # concern, tracked via ``first_ungated_at`` + the 30-min # escalation tick (which already selects [preparing, live]). if proc.status in (PROC_SCANNING, PROC_READY): proc.status = PROC_PREPARING if ( proc.status in (PROC_PREPARING, PROC_LIVE) and proc.first_ungated_at is None ): proc.first_ungated_at = _now_utc() else: # All clear. Promote based on date. if hire_passed: if proc.status != PROC_LIVE: # T+0 ledger truth, on THIS path too. The T+0 # advance cron performs the same conversion, but it # skips a process that is ALREADY ``live`` — and # this reconcile branch is what usually makes it # live first: a properly prepared hire has zero # pending blockers, so whichever 5-minute reconcile # tick first sees ``hire_passed`` promotes them # right here. Without this call the earmark→claim # conversion never ran for exactly the hires IT had # prepared correctly, and the nightly # stale-reservation sweep then released the earmark # once T+7 completed the process — putting a laptop # that is physically in the hire's hands back into # stock, and losing the only ledger link between # the two. await autoclaim_reserved_for_hire( self.db, proc.person_id, trace_id=( f"system:onboarding:recompute_live:{proc.id}" ), process_id=proc.id, ) proc.status = PROC_LIVE if proc.live_at is None: proc.live_at = _now_utc() # Stamp ready_at even when the process was ALREADY live # (the canonical path: hire arrives with the 须知 dim # pending, acks on Day 1, all-clear lands post-live). # Without this the timestamp stayed NULL forever on a # completed, fully-green process (review-r3). if proc.ready_at is None: proc.ready_at = _now_utc() else: # All clear but hire_date hasn't arrived yet. Promote the # PRE-live states to ``ready`` — but do NOT demote a # ``live`` process back to ``ready`` (mirror the pending>0 # branch above, which already protects live). An admin can # seat a hire early via mark_live_manual / advance_now # (#1695) before hire_date; without this guard the next # 5-min reconcile silently rolled a fully-green live process # back to ready (no audit, just logger.info), erasing the # admin action and pre-arming the T+7 window off a stale # live_at (#1741). Only the cleanest all-green case hit # this — a live process WITH pending blockers was already # safe. ``live`` still exits only via the T+7 verify sweep. if proc.status in (PROC_SCANNING, PROC_PREPARING): proc.status = PROC_READY if proc.ready_at is None: proc.ready_at = _now_utc() # Clear the SLA gate-timer since blockers are cleared, and # RE-ARM the single-shot escalation kinds: a process that # fully recovered and later regresses (seat revoked, asset # pulled) is a NEW gap that must page IT again — the # reminder-log rows would otherwise mute the whole chain # for the rest of the cycle (review-r3). if proc.sla_breach_level != SLA_NONE or proc.first_ungated_at: from app.domains.onboarding.models import ( REMINDER_DAY0_IT_GAPS, REMINDER_SLA_BREACH, REMINDER_SLA_URGENT, OnboardingReminderLog, ) await self.db.execute( delete(OnboardingReminderLog).where( OnboardingReminderLog.process_id == proc.id, OnboardingReminderLog.reminder_kind.in_( [ REMINDER_DAY0_IT_GAPS, REMINDER_SLA_URGENT, REMINDER_SLA_BREACH, ] ), ) ) proc.first_ungated_at = None proc.sla_breach_level = SLA_NONE if prior != proc.status: logger.info( "onboarding.process %s: %s → %s (pending=%d, hire_passed=%s)", proc.id, prior, proc.status, pending_blockers, hire_passed, ) async def _count_pending_blockers( self, process_id: UUID, *, severity: str | None = None, ) -> int: conditions = [ OnboardingBlocker.process_id == process_id, OnboardingBlocker.resolution == RESO_PENDING, ] if severity is not None: conditions.append(OnboardingBlocker.severity == severity) return int( ( await self.db.execute( select(func.count()) .select_from(OnboardingBlocker) .where(*conditions) ) ).scalar_one() ) # ───────────────────────── SLA evaluation ───────────────────────── async def evaluate_sla(self, proc: OnboardingProcess) -> str: """Compute and persist sla_breach_level for a single process. Called by the 30-min escalation tick. Levels (measured from how long blockers have been outstanding past hire_date): none — hire_date hasn't passed OR no pending blockers warn — 0–24h past hire_date with pending blockers urgent — 24–72h past hire_date breach — > 72h past hire_date """ if proc.status in PROC_TERMINAL_STATUSES: new_level = SLA_NONE else: # Only BLOCKER-severity gaps drive the SLA clock (2026-06-12). # Advisory dims (feishu sync health / VPN / policy promises) # are visibility items — letting them push a hire # to urgent/breach paged IT about non-actionable noise and, # worse, swept the whole historical in-flight backlog into # escalation cards on deploy day. pending = await self._count_pending_blockers( proc.id, severity=SEVERITY_BLOCKER, ) # Per-employee local "today" — the SLA clock starts when # hire_date arrives in the hire's OWN wall-clock, not HQ's. person = await self.db.get(IdentityPerson, proc.person_id) today = _local_today_for_person(person) if not _person_onboarding_active(person): # Rescinded offer / departed hire — never escalate. The # reconcile scanner auto-cancels these (#1746); this guards # the window between a leaver sync and the next reconcile # tick so the escalation dispatchers (which select on # sla_breach_level) never page IT for someone who left. new_level = SLA_NONE elif pending == 0 or proc.hire_date > today: new_level = SLA_NONE else: hours_past = max( 0, int( ( datetime.combine(today, datetime.min.time()) - datetime.combine( proc.hire_date, datetime.min.time() ) ).total_seconds() / 3600 ), ) if hours_past >= 72: new_level = SLA_BREACH elif hours_past >= 24: new_level = SLA_URGENT else: new_level = SLA_WARN if proc.sla_breach_level != new_level: proc.sla_breach_level = new_level await self.db.flush() return new_level # ───────────────────────── Internal helpers ───────────────────────── async def _get_or_raise( self, process_id: UUID, *, for_update: bool = False, ) -> OnboardingProcess: # ``for_update`` takes a row lock so a status-transition method # (advance_now / mark_live_manual / cancel / skip_blocker / # mark_blocker_resolved) re-reads and flips ``status`` atomically — # without it, a cron ``advance_now`` overlapping an admin # ``mark_live_manual`` double-click can both pass the status guard # and each re-stamp ``live_at`` + write a duplicate audit row. if for_update: # ``populate_existing`` is load-bearing when the caller (or a # cron batch) already has the row in the identity map: a bare # ``with_for_update()`` takes the DB lock but hands back the # STALE cached instance (expire_on_commit=False), so the # status guards below could read an out-of-date status. Mirror # the provisioning ``execute_task`` hardening. proc = await self.db.scalar( select(OnboardingProcess) .where(OnboardingProcess.id == process_id) .with_for_update() .execution_options(populate_existing=True) ) else: proc = await self.db.get(OnboardingProcess, process_id) if proc is None: raise AppException( code="ONBOARDING_NOT_FOUND", message=f"onboarding process {process_id} not found", status_code=404, ) return proc async def _get_blocker_or_raise(self, blocker_id: UUID) -> OnboardingBlocker: blocker = await self.db.get(OnboardingBlocker, blocker_id) if blocker is None: raise AppException( code="ONBOARDING_BLOCKER_NOT_FOUND", message=f"onboarding blocker {blocker_id} not found", status_code=404, ) return blocker # ───────────────────────── Manager view (PR-3) ───────────────────────── async def list_my_team_joiners( self, viewer_id: UUID, *, page: int = 1, page_size: int = 50, ) -> tuple[list[dict[str, Any]], int]: """Incoming hires the viewer manages (direct or skip-level). Mirrors offboarding's ``list_my_team_leavers`` for symmetry: a manager sees rows where the hire's ``manager_person_id == viewer_id`` (direct report) or the hire's manager's manager is the viewer (skip-level). HRBP fan-out is not included by default — managers, not HRBPs, are the onboarding stakeholder (device-inheritance preference + readiness visibility). Completed / cancelled processes are filtered out. """ direct_reports_subquery = ( select(IdentityPerson.id) .where(IdentityPerson.manager_person_id == viewer_id) .where(IdentityPerson.deleted_at.is_(None)) ) eligible_person_ids = ( select(IdentityPerson.id) .where( or_( IdentityPerson.manager_person_id == viewer_id, # Onboarding-cockpit skip-level SQL subquery (direct + # one level), not a transitive manager-tree BFS; hires # are filtered by process status below, not employment. # person-scope-exempt: skip-level onboarding subquery IdentityPerson.manager_person_id.in_(direct_reports_subquery), ) ) .where(IdentityPerson.id != viewer_id) .where(IdentityPerson.deleted_at.is_(None)) # A hire marked 'left' (rescinded/early departure) with a still-open # process must not surface as an incoming joiner on the manager view. .where(IdentityPerson.employment_status != "left") ) from app.domains.onboarding.models import ( PROC_CANCELLED as _PC, ) from app.domains.onboarding.models import ( PROC_COMPLETED as _PCMP, ) base_filters = [ OnboardingProcess.person_id.in_(eligible_person_ids), OnboardingProcess.status.notin_([_PCMP, _PC]), ] base = select(OnboardingProcess).where(*base_filters) count_q = ( select(func.count()) .select_from(OnboardingProcess) .where(*base_filters) ) total = (await self.db.execute(count_q)).scalar() or 0 q = ( base.order_by( # earliest hire_date first; null hire_date sinks to bottom OnboardingProcess.hire_date.asc(), ) .offset((page - 1) * page_size) .limit(page_size) ) rows = list((await self.db.execute(q)).scalars().all()) # Load identity rows once for display. target_person_ids: set[UUID] = {r.person_id for r in rows} people: dict[UUID, IdentityPerson] = {} if target_person_ids: person_rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(target_person_ids) ) ) ).scalars().all() people = {p.id: p for p in person_rows} # Pending blocker counts per process (for the at-a-glance badge). pending_counts: dict[UUID, int] = {} if rows: count_rows = ( await self.db.execute( select( OnboardingBlocker.process_id, func.count().label("n"), ) .where( OnboardingBlocker.process_id.in_([r.id for r in rows]), OnboardingBlocker.resolution == RESO_PENDING, ) .group_by(OnboardingBlocker.process_id) ) ).all() pending_counts = {r[0]: int(r[1]) for r in count_rows} responses: list[dict[str, Any]] = [] for r in rows: hire = people.get(r.person_id) responses.append( { "id": str(r.id), "person_id": str(r.person_id), "hire_date": r.hire_date.isoformat(), "status": r.status, "sla_breach_level": r.sla_breach_level, "hardware_allocation": r.hardware_allocation, "hire_display_name": ( (hire.display_name or hire.work_email) if hire else None ), "hire_work_email": hire.work_email if hire else None, "hire_job_title": hire.job_title if hire else None, "hire_department": hire.company_name if hire else None, "pending_blocker_count": pending_counts.get(r.id, 0), } ) return responses, total async def _viewer_can_manage_process( self, viewer_id: UUID, process: OnboardingProcess, ) -> bool: """True iff the viewer is the joiner's direct or skip-level manager. Used by the manager-gated mutation endpoint (set_hardware_allocation). Admin-gated paths bypass this check. """ hire = await self.db.get(IdentityPerson, process.person_id) if hire is None or hire.manager_person_id is None: return False if hire.manager_person_id == viewer_id: return True manager = await self.db.get(IdentityPerson, hire.manager_person_id) if manager is None: return False return manager.manager_person_id == viewer_id # ───────────────── Escalation chain (onboarding-parity PR2) ───────────────── # 打扰预算红线 (2026-06-12): every dispatcher below is IT-ONLY. # No manager, no skip-level, no chain walk-up — new notification # surfaces never page regular employees. async def _list_it_admins(self) -> list[IdentityPerson]: """Active identity persons holding global_admin (or legacy it_admin). ``management`` is deliberately excluded — leadership reads the platform but is out of every operational admin-broadcast fan-out. """ 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( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", AuthRole.name.in_(("global_admin", "it_admin")), ) .distinct() ) ).scalars().all() return list(rows) async def _batch_load_persons( self, person_ids: set[UUID], ) -> dict[UUID, IdentityPerson]: """Load identity rows for a set of person ids in ONE query. Used by the escalation scan to compute per-employee local-today without an N+1 ``db.get`` per process. """ if not person_ids: return {} rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(person_ids) ) ) ).scalars().all() return {p.id: p for p in rows} async def _pending_blocker_rows( self, process_id: UUID, ) -> list[OnboardingBlocker]: return list( ( await self.db.execute( select(OnboardingBlocker).where( OnboardingBlocker.process_id == process_id, OnboardingBlocker.resolution == RESO_PENDING, ) ) ) .scalars() .all() ) async def _reminder_already_logged( self, process_id: UUID, kind: str, ) -> bool: from app.domains.onboarding.models import OnboardingReminderLog row = ( await self.db.execute( select(OnboardingReminderLog).where( OnboardingReminderLog.process_id == process_id, OnboardingReminderLog.reminder_kind == kind, ) ) ).scalar_one_or_none() return row is not None async def dispatch_day0_it_gaps(self, *, trace_id: str) -> int: """IT worklist card when a hire's Day-1 arrives with hard gaps. For every open process whose ``hire_date`` is **today in the hire's own wall-clock** and that still has pending BLOCKER- severity dimensions, push one card (single-shot per process, ``day0_it_gaps`` reminder log) to every IT global_admin listing the concrete hard gaps + a deep link to the process detail. The inbound mirror of offboarding's T+0 auto-opened transfer requests: arrival is the moment readiness debt converts from "upcoming" to "the person is sitting there without access". Strictly an *arrival-day* signal (2026-06-24): a process that only becomes eligible after Day-1 (opened late / blockers surfaced late) is NOT given a day0 card — it is owned by the SLA urgent (24–72h) / breach (>72h) ladder instead. Firing day0 late used to emit a "今天入职" title on a body that read "已入职 N 天" (self-contradictory) AND duplicated the same-day SLA-urgent escalation for the identical gap. """ from app.domains.onboarding.models import ( REMINDER_DAY0_IT_GAPS, OnboardingReminderLog, ) # Widen the candidate window by +1 day so hires in TZs EAST of # Shanghai (whose local date is already ahead of HQ's) are not # missed; the per-proc local-today skip below then drops anyone # whose hire_date hasn't actually arrived in their own wall-clock. today = _today_app_tz() procs = ( ( await self.db.execute( select(OnboardingProcess).where( OnboardingProcess.status.in_(PROC_OPEN_STATUSES), OnboardingProcess.hire_date <= today + timedelta(days=1), # Backfilled processes are historical catch-up # rows (t7_verify already auto-completes them); # escalating them would fire a one-time IT # worklist burst for every past-hire backlog row # on the first tick after deploy. OnboardingProcess.backfilled.is_(False), ) ) ) .scalars() .all() ) if not procs: return 0 # Batch-load candidate persons once (avoid N+1) for the # per-employee local-today gate. people = await self._batch_load_persons( {p.person_id for p in procs} ) sent = 0 for proc in procs: # day0 is an ARRIVAL-DAY signal: fire ONLY when hire_date is # today in the hire's own wall-clock. The +1-day query # superset (TZ-east safety) is trimmed to exactly the arrival # day here. Two cases are dropped: # • future arrival (hire_date > local today) — not arrived. # • past arrival (hire_date < local today) — process # opened / blockers surfaced after Day-1; owned by the # SLA urgent/breach ladder, NOT a "今天入职" card whose # body would read "已入职 N 天" and which would duplicate # the same-day SLA-urgent escalation for the same gap. hire_person = people.get(proc.person_id) if not _person_onboarding_active(hire_person): # Rescinded offer / departed hire — reconcile auto-cancels # these (#1746); skip here to guard the leaver-sync→reconcile # window so IT is never paged for someone who left. continue if _is_hris_service_account(hire_person): # SF workflow-admin / vendor robot (LifecyleAdmin, TimeAdmin, # ECadmin…) — carries a hire_date but is not a real hire; never # page IT to "onboard" it (2026-07-08 IT feedback). continue if proc.hire_date != _local_today_for_person(hire_person): continue if await self._reminder_already_logged( proc.id, REMINDER_DAY0_IT_GAPS, ): continue hard_pending = [ b for b in await self._pending_blocker_rows(proc.id) if b.severity == SEVERITY_BLOCKER ] if not hard_pending: continue dispatched = await self._emit_escalation_card( proc, kind=REMINDER_DAY0_IT_GAPS, blockers=hard_pending, trace_id=trace_id, ) if dispatched: try: async with self.db.begin_nested(): self.db.add( OnboardingReminderLog( process_id=proc.id, reminder_kind=REMINDER_DAY0_IT_GAPS, channels=["feishu", "email"], detail={ "hard_pending": sorted( {b.resource_type for b in hard_pending} ), }, ) ) except IntegrityError: # Concurrent tick already claimed this slot — skip in # isolation (per-process SAVEPOINT, see _rescan). continue sent += 1 return sent async def dispatch_sla_escalations(self, *, trace_id: str) -> dict[str, int]: """Escalate urgent/breach SLA tiers — to IT admins ONLY. * ``urgent`` (24–72h past hire_date, hard dims pending) → IT. * ``breach`` (>72h) → IT. 打扰预算红线 (2026-06-12): no skip-level manager, no employee- facing recipients — escalation is an IT ops signal. Only BLOCKER-severity gaps are listed; advisory dims never page. Single-shot per (process, tier) via the reminder log — SLA can only move up while blockers persist, and once cleared ``_recompute_status`` resets the level so a NEW gap after a clean recovery is a different process state anyway. """ from app.domains.onboarding.models import ( REMINDER_SLA_BREACH, REMINDER_SLA_URGENT, OnboardingReminderLog, ) procs = ( ( await self.db.execute( select(OnboardingProcess).where( OnboardingProcess.status.in_(PROC_OPEN_STATUSES), OnboardingProcess.sla_breach_level.in_( [SLA_URGENT, SLA_BREACH] ), # See dispatch_day0_it_gaps — exclude historical # backfill rows from the escalation burst. OnboardingProcess.backfilled.is_(False), ) ) ) .scalars() .all() ) counts = {"sla_urgent": 0, "sla_breach": 0} # Batch-load candidate persons once (avoid N+1) for the # left/soft-deleted skip-guard. people = await self._batch_load_persons({p.person_id for p in procs}) for proc in procs: proc_person = people.get(proc.person_id) if not _person_onboarding_active(proc_person): # Rescinded offer / departed hire — reconcile auto-cancels # these (#1746); skip here to guard the leaver-sync→reconcile # window so IT is never paged「已入职 N 天仍未配齐」for a # person who will never arrive. continue if _is_hris_service_account(proc_person): # SF workflow-admin / vendor robot (LifecyleAdmin, TimeAdmin, # ECadmin…) — not a real hire; never page IT to onboard it. continue kind = ( REMINDER_SLA_BREACH if proc.sla_breach_level == SLA_BREACH else REMINDER_SLA_URGENT ) if await self._reminder_already_logged(proc.id, kind): continue hard_pending = [ b for b in await self._pending_blocker_rows(proc.id) if b.severity == SEVERITY_BLOCKER ] if not hard_pending: continue dispatched = await self._emit_escalation_card( proc, kind=kind, blockers=hard_pending, trace_id=trace_id, ) if dispatched: try: async with self.db.begin_nested(): self.db.add( OnboardingReminderLog( process_id=proc.id, reminder_kind=kind, channels=["feishu", "email"], detail={ "pending": sorted( {b.resource_type for b in hard_pending} ), "sla_breach_level": proc.sla_breach_level, }, ) ) except IntegrityError: # Concurrent tick already claimed this slot — skip in # isolation (per-process SAVEPOINT, see _rescan). continue counts[kind] += 1 return counts async def _emit_escalation_card( self, proc: OnboardingProcess, *, kind: str, blockers: list[OnboardingBlocker], trace_id: str, ) -> bool: """Fan an escalation card out to IT admins ONLY (打扰预算红线 — never employees / managers). True if at least one recipient was dispatched to.""" from app.notifications.service import ( LocalizedText, NotificationEnvelope, NotificationService, ) hire = await self.db.get(IdentityPerson, proc.person_id) hire_label = ( (hire.display_name or hire.work_email) if hire else str(proc.person_id) ) recipients: dict[UUID, IdentityPerson] = { p.id: p for p in await self._list_it_admins() } if not recipients: logger.warning( "onboarding.escalation: no recipients for %s/%s", proc.id, kind, ) return False # Every IT card must answer "what should I, the reader, DO?": # title — who + how urgent, in plain words (no internal codes) # body — the concrete gaps in plain language (not "[blocker] # Hardware allocation"), then ONE explicit next action # naming the buttons on the linked page. # Per-employee local "today" — count days-past in the hire's own # wall-clock so a US/EU card doesn't read "1 天" early/late. days_past = max(0, (_local_today_for_person(hire) - proc.hire_date).days) # Collapse the work_email + sso_provision pair into one gap line # (same Google account) BEFORE counting, so the「有 N 项」headline # matches the de-duplicated body. gap_lines = _it_gap_lines(blockers) gap_count = len(gap_lines) if kind == "day0_it_gaps": title = f"🚨 {hire_label} 今天入职,有 {gap_count} 项 IT 工作没完成" elif kind == "sla_breach": title = ( f"🚨 {hire_label} 已入职 {days_past} 天仍未配齐,已超时 —— 请立即处理" ) else: # sla_urgent title = f"⚠️ {hire_label} 已入职 {days_past} 天,还有 {gap_count} 项 IT 工作没完成" when = "今天入职" if days_past == 0 else f"已入职 {days_past} 天" body_lines = [ f"{hire_label}（{when}，入职日 {proc.hire_date.isoformat()}）", "", "以下工作需要 IT 完成,该员工才能正常开工：", ] for line in gap_lines[:8]: body_lines.append(f" • {line[0]}") if gap_count > 8: body_lines.append(f" • …等共 {gap_count} 项") body_lines.append("") body_lines.append( "👉 请点「去 Helix 查看」打开该员工的入职详情，逐项分配 / 标记已发放；" "确实不需要的可「跳过」并填写原因。" ) # The registered三语 template interpolates ``{gap_lines}`` — one # rendered bullet block per language, resolved per recipient by # ``_localized_context``. The zh variant is byte-identical to the # ``body_lines`` block above, which stays as the envelope fallback. def _gap_block(idx: int, overflow: str) -> str: block = [f" • {line[idx]}" for line in gap_lines[:8]] if gap_count > 8: block.append(f" • {overflow}") return "\n".join(block) gap_lines_text = LocalizedText( zh=_gap_block(0, f"…等共 {gap_count} 项"), en=_gap_block(1, f"… {gap_count} items in total"), ja=_gap_block(2, f"…全 {gap_count} 件"), ) notif = NotificationService(self.db) any_sent = False for recipient in recipients.values(): env = NotificationEnvelope( recipient=recipient, title=title, body_text="\n".join(body_lines), channels=["feishu", "email"], template_key=f"onboarding.{kind}", url=f"/admin/onboarding/process/{proc.id}", related_resource_type="onboarding_process", related_resource_id=proc.id, context={ "hire_display_name": hire_label, "hire_date": proc.hire_date.isoformat(), "days_past": str(days_past), "pending_count": str(gap_count), "sla_breach_level": proc.sla_breach_level, "gap_lines": gap_lines_text, }, ) try: await notif.dispatch(env, trace_id=trace_id) any_sent = True except Exception: logger.exception( "onboarding.escalation dispatch failed for %s/%s → %s", proc.id, kind, recipient.id, ) return any_sent