"""IT worklist for hires whose Day-1 computer never reached the ledger. Why this exists (2026-09-05). Production's six-month Day-1 readiness window held 40 hires; 31 missed, and 29 of those missed on the hardware dimension alone. Eleven of them are still open with **no computer in Helix at all** — overseas self-purchase / direct-ship cases where the machine exists in the world but no one ever created the asset row. The cockpit shows them, but only as a red dimension: closing the gap meant leaving Helix (create asset → find the hire → claim → come back), which is exactly why they sat for up to 108 days. This service is the closing surface: * :meth:`OnboardingHardwareGapService.list_gaps` — every hire whose hire date has passed and whose hardware blocker is still pending, **including ``completed`` processes** (the cockpit's window is open processes only, so a completed-with-gap hire fell off every board). Each row carries the overdue clock, the person's location, the MDM fleet hint (a Jamf/Intune device matching their email is strong evidence of which machine they actually hold), and any earmark. * :meth:`register_device` — one call that creates the asset row, claims it to the hire, and resolves the blocker. The whole gap closes from one form. The skip path deliberately reuses ``OnboardingService.skip_blocker`` — a hire who genuinely needs no company computer is an existing, audited decision, not a new concept. No employee-facing notification is sent from here: the platform closes these gaps through IT, not by asking 11 people to self-report (explicit product decision, 2026-09-05). """ from __future__ import annotations import logging from datetime import date, datetime, timedelta from typing import Any from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.coordinators.onboarding_hardware_coordinator import ( resolve_hire_hardware_states, ) from app.core.exceptions import AppException, NotFoundError from app.domains.onboarding.models import ( PROC_CANCELLED, PROC_TERMINAL_STATUSES, RES_HARDWARE_ALLOCATION, RESO_PENDING, RESO_PROVISIONED, OnboardingBlocker, OnboardingProcess, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) _APP_TZ = ZoneInfo("Asia/Shanghai") # How far back the worklist looks. Long enough to cover the Day-1 # metric's own six-month scoring window (so anything the health score # can still blame IT for is actionable here), and bounded so the queue # doesn't accumulate ancient history nobody will ever reconcile. GAP_LOOKBACK_DAYS = 400 def _today() -> date: return datetime.now(_APP_TZ).date() class OnboardingHardwareGapService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def list_gaps( self, *, lookback_days: int = GAP_LOOKBACK_DAYS, ) -> dict[str, Any]: """Hires past their start date with no computer on the ledger. Returns ``{"rows": [...], "summary": {...}}``. Rows are ordered most-overdue first — the 108-day case must never sit below a two-day-old one. """ today = _today() window_start = today - timedelta(days=max(1, lookback_days)) # Pending hardware blockers joined to their process. ``pending`` # is the only unresolved value; ``skipped`` (IT decided no device # is needed) and every ``*provisioned`` value are closed states # and must not resurface here. rows = ( await self.db.execute( select(OnboardingBlocker, OnboardingProcess) .join( OnboardingProcess, OnboardingProcess.id == OnboardingBlocker.process_id, ) .where( OnboardingBlocker.resource_type == RES_HARDWARE_ALLOCATION, OnboardingBlocker.resolution == RESO_PENDING, OnboardingProcess.status != PROC_CANCELLED, OnboardingProcess.hire_date <= today, OnboardingProcess.hire_date >= window_start, OnboardingProcess.backfilled.is_(False), ) ) ).all() if not rows: return { "rows": [], "summary": { "total": 0, "no_device": 0, "reserved_only": 0, "overdue_30d_plus": 0, "already_holds_device": 0, }, } person_ids = [proc.person_id for _blocker, proc in rows] persons = { p.id: p for p in ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(person_ids), IdentityPerson.deleted_at.is_(None), ) ) ).scalars() } hardware = await resolve_hire_hardware_states(self.db, person_ids) mdm_hints = await self._mdm_hints(list(persons.values())) out: list[dict[str, Any]] = [] already_holds = 0 for blocker, proc in rows: person = persons.get(proc.person_id) # A departed or purged hire is offboarding's problem, not a # Day-1 gap: chasing a laptop for someone who already left # belongs on the leaver-recovery board, and listing them here # would make the queue un-closable. if person is None or person.employment_status == "left": continue hw = hardware.get(proc.person_id) # 🔴 The hire ALREADY holds a computer — this is a stale # blocker, not a gap. It happens because a ``completed`` # process's blocker rows are frozen: ``reconcile_blockers`` # short-circuits on terminal statuses and the 5-minute # reconcile cron only selects open processes, so a hire who # got their laptop through the ordinary claim flow after # their process closed keeps a pending hardware blocker # forever. Listing them here rendered a red 「无设备」 tag # against someone who demonstrably has a device — a false # statement on the one page whose entire job is naming who # has none — and every action on the row 409s # (``HIRE_ALREADY_HAS_DEVICE``). Counted rather than # silently dropped, because "how many of these are # bookkeeping rather than missing hardware" is exactly the # kind of number this series refuses to hide. # # ⚠️ This comment first claimed "nothing else reads these # rows". That was wrong: ``/me/welcome`` read them too and # told the employee 「IT 正在为你准备剩下的 N 项」 about work # nobody was doing (fixed alongside). Every OTHER reader is # bounded to ``PROC_OPEN_STATUSES`` — verified by # enumerating them — and the Day-1 metric scores # ``hard_ready_at``, not live blockers. if hw is not None and hw.state == "assigned": already_holds += 1 continue out.append( { "process_id": str(proc.id), "blocker_id": str(blocker.id), "person_id": str(proc.person_id), "display_name": person.display_name or person.work_email, "work_email": person.work_email, "employee_no": person.employee_no, "employment_status": person.employment_status, "job_title": person.job_title, "department": person.company_name, "work_location": person.work_location_name, "hire_date": proc.hire_date.isoformat(), "process_status": proc.status, "days_overdue": (today - proc.hire_date).days, # ``reserved`` here means a device is earmarked but # the T+0 conversion has not happened — the pending # blocker plus an earmark is a contradiction worth # showing, since it means the claim failed rather # than never being prepared. "hardware_state": hw.state if hw else "missing", "reserved_asset_tag": hw.asset_tag if hw else None, "mdm_hint": mdm_hints.get(proc.person_id), } ) out.sort(key=lambda r: (-r["days_overdue"], r["display_name"] or "")) return { "rows": out, "summary": { "total": len(out), "no_device": sum( 1 for r in out if r["hardware_state"] == "missing" ), "reserved_only": sum( 1 for r in out if r["hardware_state"] == "reserved" ), "overdue_30d_plus": sum( 1 for r in out if r["days_overdue"] >= 30 ), # Pending blockers whose hire demonstrably HAS a # computer — bookkeeping residue on closed processes, # not missing hardware. Not listed above; shown as a # number so it is visible rather than swallowed. "already_holds_device": already_holds, }, } async def _mdm_hints( self, persons: list[IdentityPerson], ) -> dict[UUID, dict[str, Any] | None]: """Fleet evidence of what machine each hire actually holds. A Jamf / Intune record matching the hire by person link or work email is the strongest available signal for the serial number an admin should type into the register form. Match on ``assigned_person_id`` OR ``mdm_assigned_email`` because the MDM link table is only partially resolved to Helix people. Returns ``None`` for hires with no fleet record — an explicit "no evidence", never a guess. """ from app.domains.asset.models import MdmManagedDevice if not persons: return {} emails = { p.work_email.lower(): p.id for p in persons if p.work_email } person_ids = [p.id for p in persons] conditions = [MdmManagedDevice.assigned_person_id.in_(person_ids)] if emails: conditions.append( MdmManagedDevice.mdm_assigned_email.in_(list(emails)) ) devices = ( await self.db.execute( select(MdmManagedDevice).where(or_(*conditions)) ) ).scalars().all() hints: dict[UUID, dict[str, Any] | None] = {} for dev in devices: key = dev.assigned_person_id if key is None and dev.mdm_assigned_email: key = emails.get(dev.mdm_assigned_email.lower()) if key is None or key in hints: continue hints[key] = { "serial_number": dev.serial_number, "model": dev.model, "device_name": dev.device_name, "provider": dev.provider, "already_linked": dev.asset_id is not None, } return hints async def register_device( self, process_id: UUID, *, payload: Any, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Create the missing asset row and hand it to the hire. One transaction, three effects: intake the device into the asset ledger, claim it to the hire, and let the onboarding scanner see a satisfied hardware dimension. Refuses when the hire already has a computer — that is a no-op the caller should see as a 409, not a duplicate ledger row. The asset write goes through ``AssetService`` (not raw ORM) so intake keeps every invariant the interactive path has: asset-tag generation, brand/model lookup upsert, legal-entity resolution and managed-location normalisation. """ from app.domains.asset.schemas import AssetClaimRequest, AssetIntakeRequest from app.domains.asset.service import AssetService from app.domains.onboarding.service import OnboardingService proc = ( await self.db.execute( select(OnboardingProcess).where( OnboardingProcess.id == process_id ) ) ).scalar_one_or_none() if proc is None: raise NotFoundError("Onboarding process", str(process_id)) if proc.status == PROC_CANCELLED: raise AppException( code="ONBOARDING_TERMINAL", message="onboarding process is cancelled", status_code=409, ) person = await self.db.get(IdentityPerson, proc.person_id) if person is None or person.deleted_at is not None: raise NotFoundError("Person", str(proc.person_id)) if person.employment_status == "left": raise AppException( code="PERSON_LEFT", message=( "the hire has already departed — recover the device " "through the leaver flow instead" ), status_code=409, ) existing = await resolve_hire_hardware_states( self.db, [proc.person_id], ) state = existing.get(proc.person_id) if state is not None and state.state == "assigned": raise AppException( code="HIRE_ALREADY_HAS_DEVICE", message=( f"{person.display_name or person.work_email} already " f"holds {state.asset_tag or 'a computer'}; nothing to " "register" ), status_code=409, ) asset_service = AssetService(self.db) # The hire's own work location seeds the asset location so a # back-registered device lands in the right region instead of # NULL. The intake service routes it through the managed location # resolver, which returns None for anything it can't canonicalise # — so a bad value can't corrupt the column. resolved_location = payload.location or person.work_location_name # ``model_validate`` on a dict projection, not keyword # construction: this is a MAPPING from one request model to # another, the dict makes exactly which fields cross the boundary # legible, and — because this repo runs mypy WITHOUT the pydantic # plugin — keyword construction of a model whose optional fields # use ``Field(None, …)`` raises a ``call-arg`` error per omitted # field (14 of them here). Validation still runs in full. intake = AssetIntakeRequest.model_validate( { "asset_tag": payload.asset_tag, "category": payload.category, "brand": payload.brand, "model": payload.model, "brand_id": payload.brand_id, "model_id": payload.model_id, "serial_number": payload.serial_number, "purchase_date": payload.purchase_date, "purchase_cost": payload.purchase_cost, "currency": payload.currency, "purchase_entity": payload.purchase_entity, "purchase_entity_id": payload.purchase_entity_id, "invoice_number": payload.invoice_number, "location": resolved_location, "notes": payload.notes, } ) asset, _op = await asset_service.intake_asset(intake, actor_user_id) # Claim in the same transaction: an intake row with no claim # would leave the device idle AND the blocker pending — the exact # half-finished state this surface exists to eliminate. await asset_service.claim_asset( asset.id, AssetClaimRequest.model_validate( { "to_user_id": proc.person_id, "claim_date": payload.claim_date, "location": resolved_location, "notes": ( "Back-registered from the Day-1 hardware worklist " f"(onboarding process {proc.id})" ), } ), actor_user_id, ) # Re-scan rather than hand-resolving the blocker row: the scanner # is the single writer of dimension state, and it now sees an # assigned computer. Hand-flipping the row would drift the two # views apart the moment the scan disagrees. # # 🔴 EXCEPT on a terminal process, where there is no scanner any # more. ``reconcile_blockers`` returns 0 immediately for # ``completed`` / ``cancelled`` and the 5-minute reconcile cron # only selects open processes, so a completed process's blocker # rows are frozen forever. This queue deliberately INCLUDES # completed processes — they are the hires who fell off every # other board, the reason it exists — so leaving it at the # re-scan meant the one-click close fixed the ledger and left # the row in the worklist permanently, 409ing on the next click. # With the scanner retired for this row, writing it here is not # a second writer; it is the only one left, and it writes the # value the scanner itself would have written. if proc.status in PROC_TERMINAL_STATUSES: blocker = ( await self.db.execute( select(OnboardingBlocker).where( OnboardingBlocker.process_id == proc.id, OnboardingBlocker.resource_type == RES_HARDWARE_ALLOCATION, OnboardingBlocker.resolution == RESO_PENDING, ) ) ).scalar_one_or_none() if blocker is not None: # Exactly what ``_rescan`` writes when a pending # dimension becomes satisfied — same value, same single # field, so the two paths cannot produce different rows. blocker.resolution = RESO_PROVISIONED await self.db.flush() else: await OnboardingService(self.db).reconcile_blockers(proc.id) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="onboarding.hardware_gap_device_registered", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "asset_id": str(asset.id), "asset_tag": asset.asset_tag, "serial_number": asset.serial_number, "person_id": str(proc.person_id), "days_overdue": (_today() - proc.hire_date).days, }, ) logger.info( "onboarding.hardware_gap: registered asset %s for person %s " "(process %s, %d days overdue)", asset.id, proc.person_id, proc.id, (_today() - proc.hire_date).days, ) return { "asset_id": str(asset.id), "asset_tag": asset.asset_tag, "person_id": str(proc.person_id), "process_id": str(proc.id), }