"""T+7 verify service — onboarding case-file health check. Daily cron entry: ``OnboardingT7VerifyService.verify_all_due()``. For every process in status ``live`` whose ``live_at`` is ≥ 7 days ago: 1. Rescan the resource layer (``OnboardingService._rescan``) so blocker state reflects current truth. 2. If all blockers are resolved/skipped → flip to ``completed``. 3. If at least one blocker remains pending AND the process is NOT ``backfilled``: a. Open a ``ComplianceIncident`` (type=``onboarding_day1_gap``, severity=``warning``) — one per process per cycle, dedup-aware via the reminder log. b. Write a ``t7_verify_failed`` reminder log row (idempotent marker so we don't re-incident every daily tick). 4. Backfilled rows: silently flip to ``completed`` regardless of pending blockers — the platform never observed their Day-1 readiness end-to-end so the scan result is meaningless. 5. Incident lifecycle upkeep (2026-08-18), both halves of which run off the same nightly tick: * **Re-arm** — a ``resolved`` incident whose process is still ``live`` with the same hard gap ``T7_REOPEN_GRACE_DAYS`` later means the close was premature. Drop the dedup marker and open a fresh incident so the governance queue stops saying "resolved" over a live gap. ``accepted`` incidents are the admin's explicit "known exception" and are never reopened. * **Auto-close** — an ``open`` incident whose process has since reached ``completed`` / ``cancelled`` can never be worked; the sweep resolves it as a system close. The T+7 grace window is long enough to absorb ordinary Day-1/2/3 IT catch-up; if a hire is still missing core access after a week, that's operationally interesting. """ from __future__ import annotations import logging from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.compliance.models import ( INCIDENT_ACCEPTED, INCIDENT_OPEN, INCIDENT_RESOLVED, SEVERITY_WARNING, TYPE_ONBOARDING_DAY1_GAP, ComplianceIncident, ) from app.core.exceptions import is_transaction_aborting_db_error from app.domains.onboarding.models import ( PROC_COMPLETED, PROC_LIVE, PROC_TERMINAL_STATUSES, REMINDER_T7_VERIFY_FAILED, RESO_PENDING, SEVERITY_BLOCKER, OnboardingBlocker, OnboardingProcess, OnboardingReminderLog, ) from app.domains.onboarding.service import OnboardingService logger = logging.getLogger(__name__) T7_GRACE_DAYS = 7 # How long a *resolved* day-1-gap incident is trusted before the nightly # sweep re-opens it over a still-pending hard gap. Long enough that an # admin who closes the ticket while the laptop is in transit isn't paged # again the next morning; short enough that a premature close can't hide # a live gap for months (prod: process 4c9b8daa was "resolved" # 2026-06-23 and still gapped 56 days later with no open incident). T7_REOPEN_GRACE_DAYS = 3 class OnboardingT7VerifyService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.svc = OnboardingService(db) async def verify_all_due(self, *, trace_id: str) -> dict[str, Any]: """Verify every due process. Returns counts. Returns ``{"completed": N, "incidents_opened": M, "incidents_reopened": R, "stale_incidents_closed": S, "still_live": K, "failed_count": F, "errors": [...]}`` (``incidents_reopened`` is the subset of ``incidents_opened`` raised again over a prematurely-closed incident). """ cutoff = datetime.now(UTC) - timedelta(days=T7_GRACE_DAYS) procs = ( ( await self.db.execute( select(OnboardingProcess).where( OnboardingProcess.status == PROC_LIVE, OnboardingProcess.live_at <= cutoff, ) ) ) .scalars() .all() ) completed = 0 incidents_opened = 0 incidents_reopened = 0 still_live = 0 failed_count = 0 errors: list[dict[str, str]] = [] for proc in procs: # Per-process SAVEPOINT isolation: ``_verify_one`` flushes # (``_rescan``) and inserts ComplianceIncident / # OnboardingReminderLog rows (the latter carries a UNIQUE on # ``(process_id, reminder_kind)``). Without a savepoint, the # first flush failure leaves the shared AsyncSession in a # failed-transaction state and every subsequent ``_rescan`` — # plus the trailing ``flush`` and the handler's # ``session.commit()`` — raises PendingRollbackError, # discarding the whole night's completions/incidents. Mirrors # ``OnboardingService.reconcile_all_active_processes``. try: async with self.db.begin_nested(): # Locking re-read before we act. The candidate SELECT # above is lock-free and the shared session is # expire_on_commit=False, so ``proc.status`` is a stale # cached ``live`` even if a concurrent admin # ``cancel_process`` (which takes this same FOR UPDATE # lock) committed ``cancelled`` in the meantime. Re-fetch # FOR UPDATE + ``populate_existing`` and re-assert # ``live`` under the lock, else ``_mark_completed`` would # overwrite the admin's ``cancelled`` with ``completed``. locked = ( await self.db.execute( select(OnboardingProcess) .where(OnboardingProcess.id == proc.id) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() if locked is None or locked.status != PROC_LIVE: # Concurrently cancelled/completed — skip entirely. continue outcome = await self._verify_one(locked, trace_id=trace_id) except Exception as exc: # Transaction-aborting errors (deadlock/lock-wait/orphaned # savepoint) kill the whole txn on TiDB; swallowing one would # poison the shared session and make every later _rescan plus # the handler commit raise PendingRollbackError, masking the # real cause and discarding the night's completions. Re-raise # so the handler's lock-retry wrapper reruns the tick in a # fresh session (verify is idempotent — UNIQUE reminder rows). if is_transaction_aborting_db_error(exc): raise logger.exception( "onboarding.t7_verify failed for process %s", proc.id ) failed_count += 1 if len(errors) < 50: errors.append({ "process_id": str(proc.id), "error": f"{type(exc).__name__}: {exc!s:.200}", }) continue if outcome == "completed": completed += 1 elif outcome in ("incident", "reopened"): incidents_opened += 1 still_live += 1 if outcome == "reopened": incidents_reopened += 1 else: still_live += 1 # Stale-incident sweep. Runs AFTER the per-process loop so a # process this very tick flipped to ``completed`` closes its own # open incident in the same run. Own savepoint for the same # reason the per-process work has one: a flush failure here must # not poison the session and discard the night's completions. stale_closed = 0 try: async with self.db.begin_nested(): stale_closed = await self._close_incidents_on_terminal_processes( trace_id=trace_id, ) except Exception as exc: if is_transaction_aborting_db_error(exc): raise logger.exception("onboarding.t7_verify stale-incident sweep failed") # The savepoint rolled the whole sweep back — don't report # closes that no longer exist. stale_closed = 0 failed_count += 1 if len(errors) < 50: errors.append({ "process_id": "-", "error": f"stale_incident_sweep: {type(exc).__name__}: {exc!s:.200}", }) if completed or incidents_opened or stale_closed: await self.db.flush() logger.info( "onboarding.t7_verify: completed=%d incidents=%d (reopened=%d) " "stale_closed=%d still_live=%d", completed, incidents_opened, incidents_reopened, stale_closed, still_live, ) return { "completed": completed, "incidents_opened": incidents_opened, "incidents_reopened": incidents_reopened, "stale_incidents_closed": stale_closed, "still_live": still_live, "failed_count": failed_count, "errors": errors, } async def _verify_one( self, proc: OnboardingProcess, *, trace_id: str, ) -> str: # Refresh blockers off the live resource layer first. await self.svc._rescan(proc) # ``_rescan`` cancels the process when the person row vanished # mid-flight (review-r2) OR when the hire is now left / soft-deleted # (rescinded offer / early departure — #1746). Either way don't # overwrite that terminal state with ``completed`` or open a # day-1-gap compliance incident for a person who left / no longer # exists. if proc.status != PROC_LIVE: return "skipped" pending_rows = ( ( await self.db.execute( select(OnboardingBlocker).where( OnboardingBlocker.process_id == proc.id, OnboardingBlocker.resolution == RESO_PENDING, ) ) ) .scalars() .all() ) # Only BLOCKER-severity gaps are incident-worthy (打扰预算红线, # 2026-06-12 — same口径 as SLA/escalation). Advisory leftovers # (须知未确认 / 策略承诺) close the case with the # leftover list recorded in the completion audit row instead of # opening a compliance incident for every slow ack. hard_pending = [ r for r in pending_rows if r.severity == SEVERITY_BLOCKER ] if not hard_pending or proc.backfilled: return await self._mark_completed( proc, trace_id=trace_id, advisory_leftovers=sorted( {r.resource_type for r in pending_rows} ) if pending_rows else None, ) return await self._open_incident_if_new( proc, pending_rows=list(hard_pending), trace_id=trace_id, ) async def _mark_completed( self, proc: OnboardingProcess, *, trace_id: str, advisory_leftovers: list[str] | None = None, ) -> str: proc.status = PROC_COMPLETED proc.completed_at = datetime.now(UTC) await self.audit.log( trace_id=trace_id, actor_type="system", action="onboarding.process_completed", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "via": "t7_verify", "backfilled": proc.backfilled, "advisory_leftovers": advisory_leftovers, }, ) return "completed" async def _open_incident_if_new( self, proc: OnboardingProcess, *, pending_rows: list[OnboardingBlocker], trace_id: str, ) -> str: # Dedup: at most ONE day-1-gap incident per process per onboarding # cycle. The ``t7_verify_failed`` reminder-log row is the permanent # marker (UNIQUE on (process_id, kind)); it is deliberately kept # across a re-hire's hire_date correction (see OnboardingService # .start_process) and is NOT reset when an admin closes the # incident — a still-pending hard gap keeps paging IT through the # SLA escalation chain (day0_it_gaps / sla_urgent / sla_breach), so # re-incidenting every nightly tick here would be redundant noise. existing_marker = ( await self.db.execute( select(OnboardingReminderLog).where( OnboardingReminderLog.process_id == proc.id, OnboardingReminderLog.reminder_kind == REMINDER_T7_VERIFY_FAILED, ) ) ).scalar_one_or_none() reopened = False if existing_marker is not None: if not await self._rearm_if_resolution_was_premature( proc, marker=existing_marker, trace_id=trace_id, ): return "skipped" reopened = True incident = ComplianceIncident( incident_type=TYPE_ONBOARDING_DAY1_GAP, severity=SEVERITY_WARNING, status=INCIDENT_OPEN, person_id=proc.person_id, resource_type="onboarding_process", resource_id=proc.id, resource_label=f"hire_date={proc.hire_date.isoformat()}", title=( f"Day-1 onboarding gap at T+7 " f"({len(pending_rows)} pending dimension" f"{'s' if len(pending_rows) != 1 else ''})" ), detail={ "process_id": str(proc.id), "hire_date": proc.hire_date.isoformat(), "pending_resource_types": sorted( {r.resource_type for r in pending_rows} ), }, detected_at=datetime.now(UTC), ) self.db.add(incident) self.db.add( OnboardingReminderLog( process_id=proc.id, reminder_kind=REMINDER_T7_VERIFY_FAILED, channels=["compliance_incident"], detail={ "pending_count": len(pending_rows), "pending_resource_types": sorted( {r.resource_type for r in pending_rows} ), }, ) ) await self.audit.log( trace_id=trace_id, actor_type="system", action="onboarding.t7_verify_failed", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "pending_count": len(pending_rows), "pending_resource_types": sorted( {r.resource_type for r in pending_rows} ), "reopened": reopened, }, ) return "reopened" if reopened else "incident" async def _rearm_if_resolution_was_premature( self, proc: OnboardingProcess, *, marker: OnboardingReminderLog, trace_id: str, ) -> bool: """Drop the dedup marker when a *closed* incident's gap is still open. The marker is a one-shot per onboarding cycle, so an incident an admin resolved while the hard gap was still pending silenced the governance queue permanently: prod process ``4c9b8daa`` was resolved 2026-06-23 and was still ``live`` with the same pending ``hardware_allocation`` blocker 56 days later — the only gapped hire with no open incident. Mirrors the re-arm precedent in ``OnboardingService._recompute_status``, which deletes the single-shot escalation markers on full recovery so a later regression can page again. Returns True when the caller should open a fresh incident. Never re-opens over: * an ``open`` incident — the queue already shows the gap; * an ``accepted`` incident — the admin's explicit "known exception" decision, which this sweep must not overrule. """ incidents = ( ( await self.db.execute( select(ComplianceIncident).where( ComplianceIncident.incident_type == TYPE_ONBOARDING_DAY1_GAP, ComplianceIncident.resource_type == "onboarding_process", ComplianceIncident.resource_id == proc.id, ) ) ) .scalars() .all() ) if any( i.status in (INCIDENT_OPEN, INCIDENT_ACCEPTED) for i in incidents ): return False # Grace anchor: the most recent close (or, if the incident rows # were purged, the marker itself). An admin who closes the ticket # because the laptop ships tomorrow gets the grace window before # the queue re-raises it, instead of being re-paged the next # morning. closes = [i.resolved_at or i.detected_at for i in incidents] anchor = max(closes) if closes else marker.sent_at if anchor is None: return False if anchor.tzinfo is None: # MySQL/TiDB hands back naive datetimes on some drivers; the # comparison below must not raise TypeError (see # ``reference_naive_aware_datetime_mix_in_composed_queries``). anchor = anchor.replace(tzinfo=UTC) if anchor > datetime.now(UTC) - timedelta(days=T7_REOPEN_GRACE_DAYS): return False # ORM delete + flush (not a bulk ``delete()``): the DELETE must # reach the DB before ``_open_incident_if_new`` INSERTs the fresh # marker, or the UNIQUE (process_id, reminder_kind) rejects it. await self.db.delete(marker) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="system", action="onboarding.t7_verify_rearmed", status="success", resource_type="onboarding_process", resource_id=str(proc.id), detail={ "prior_incident_ids": [str(i.id) for i in incidents], "prior_close_at": anchor.isoformat(), "reason": "gap_still_pending_after_resolution", }, ) return True async def _close_incidents_on_terminal_processes( self, *, trace_id: str ) -> int: """Resolve open day-1-gap incidents whose process is terminal. Nothing closed a ``onboarding_day1_gap`` incident when its process reached ``completed`` / ``cancelled``, so the governance queue kept two unworkable rows (prod: ``6153b7a4`` on a completed process, ``7a8691fc`` on a cancelled one). Sweeping here rather than hooking each transition covers every path that can end a process — the T+7 completion above, ``cancel_process``, the ``_rescan`` leaver/rescinded-offer cancel — plus rows that already went stale before this shipped. """ rows = ( await self.db.execute( select(ComplianceIncident, OnboardingProcess.status) .join( OnboardingProcess, OnboardingProcess.id == ComplianceIncident.resource_id, ) .where( ComplianceIncident.incident_type == TYPE_ONBOARDING_DAY1_GAP, ComplianceIncident.status == INCIDENT_OPEN, ComplianceIncident.resource_type == "onboarding_process", OnboardingProcess.status.in_(PROC_TERMINAL_STATUSES), ) ) ).all() now = datetime.now(UTC) for incident, proc_status in rows: incident.status = INCIDENT_RESOLVED incident.resolved_at = now # resolved_by_person_id stays NULL — a system close, so the # admin UI doesn't attribute it to a person who never acted. done = proc_status == PROC_COMPLETED incident.resolution_note = ( f"系统自动关闭:入职流程已{'完成' if done else '取消'}" f" (onboarding.t7_verify)" ) await self.audit.log( trace_id=trace_id, actor_type="system", action="compliance_incident.resolve", status="success", resource_type="compliance_incident", resource_id=str(incident.id), detail={ "incident_type": incident.incident_type, "severity": incident.severity, "accepted": False, "via": "onboarding.t7_verify", "process_status": proc_status, }, ) if rows: await self.db.flush() return len(rows)