from __future__ import annotations import logging import math import secrets from collections.abc import Awaitable, Callable, Iterable from dataclasses import asdict, dataclass from datetime import UTC, date, datetime, timedelta from decimal import Decimal, InvalidOperation from typing import Any, cast from uuid import UUID from zoneinfo import ZoneInfo from redis.exceptions import RedisError from sqlalchemy import and_, case, func, or_, select from sqlalchemy import inspect as sa_inspect from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.auth.models import AuthRole, AuthUserRole from app.core.config import settings from app.core.corporate_domains import ( email_is_corporate, get_corporate_email_domains, ) from app.core.distributed_lock import LockBusy, distributed_lock from app.core.exceptions import AppException, ConflictError, NotFoundError from app.core.geo import normalize_country_code from app.core.name_sanitize import strip_sentinel_name_tokens from app.core.person_filters import ( is_hris_non_employee_record, is_test_employee_no, ) from app.core.redis_client import get_redis from app.core.safe_logging import get_safe_logger from app.core.working_hours import resolve_timezone from app.domains.asset.models import AssetItem, AssetSoftwareCatalog from app.domains.hr.models import HrLifecycleEvent, HrSyncBatch, HrSyncStagedEvent from app.domains.hr.schemas import SuccessFactorsWebhookEvent from app.identity.person_models import ( IdentityExternalIdentity, IdentityOrgMembership, IdentityOrgNode, IdentityPerson, ) from app.identity.successfactors_auth import build_auth_from_config from app.identity.successfactors_client import SuccessFactorsClient from app.identity.successfactors_sync import ( _build_org_path, _department_name, _is_china_employee, _parse_org_level, ) from app.identity.successfactors_sync import ( _parse_date as _sf_parse_date, ) logger = get_safe_logger(__name__) RESOLVED_OFFBOARDING_STATUSES = {"returned", "reclaimed", "transferred", "disposed", "completed"} # --- SF status vocabulary (single source of truth) ------------------------- # SF ``User.status`` values that mean "currently employed / on their way in". # ``active_external_suite`` is SF Onboarding's pre-day-1 status for a REAL # upcoming hire (future-dated EmpJob + EmpEmployment exist) — never a leaver. # Both sync paths (full roster + incremental) MUST agree on this set; the full # path historically only special-cased bare ``t`` while incremental knew about # ``active_external_suite`` — exactly the kind of drift that produced the # leaver/future-hire misclassification bug class. SF_ACTIVE_STATUSES = frozenset({"t", "active_external_suite"}) def sf_status_is_active(status: str | None) -> bool: return str(status or "").strip().lower() in SF_ACTIVE_STATUSES # SF placeholder tokens that stand in for "no human identity here". SF # Onboarding emits a stripped ``User`` (displayName = a placeholder, no # firstName/lastName) for a requisition that has been cancelled / # withdrawn while its future-dated EmpEmployment/EmpJob rows still linger # in the OData feed. Ingesting one creates a nameless ghost person # (P10019, 2026-07-17: displayName "Unknown", hire_date 2026-07-26) that # the absent-person check's future-hire exemption can then never clean up. _SF_PLACEHOLDER_NAME_TOKENS = {"", "unknown", "null", "n/a", "na", "-"} def is_shell_sf_record(user: dict | None) -> bool: """True when an SF ``User`` payload carries NO usable human identity. A record is a shell when its ``displayName`` is empty or a known placeholder token AND both ``firstName`` and ``lastName`` are empty. The three-way AND keeps the false-positive rate at zero for real hires: a CJK-only employee whose SF config leaves firstName/lastName blank still carries a real composed ``displayName`` (e.g. "张三"), so ``placeholder`` is False and the record is NOT treated as a shell. Only a record with no name of any kind trips this gate. """ u = user or {} display = (strip_sentinel_name_tokens(u.get("displayName")) or "").strip() first = (u.get("firstName") or "").strip() last = (u.get("lastName") or "").strip() is_placeholder = display.casefold() in _SF_PLACEHOLDER_NAME_TOKENS return is_placeholder and not first and not last # Lifecycle intents returned by ``classify_inactive_sf_user``. LIFECYCLE_OFFBOARD = "offboard" LIFECYCLE_FUTURE_HIRE = "future_hire" LIFECYCLE_NO_TERMINATION = "no_termination" # Substantiated termination whose reason code is on the configured # "do NOT close accounts" list (人员分类转换 / 调往中国·德国数据中心 / # 兼岗结束 — the person keeps working; only the SF employment slice ends). LIFECYCLE_REASON_EXCLUDED = "reason_excluded" # Ledger event_type for an offboard that Helix received and DELIBERATELY # did not act on (excluded termination reason). A first-class row — not a # log line — so both sides can audit "we saw it, we classified it as a # transfer, we kept the account open" months later. EVENT_TYPE_OFFBOARD_SKIPPED = "offboard_skipped" @dataclass(frozen=True, slots=True) class InactiveUserDecision: """Result of classifying an SF user that is NOT active (``status='f'`` / forfeited / onboarding-inactive).""" intent: str # one of LIFECYCLE_* termination_date: date | None reason: str @property def is_leaver(self) -> bool: return self.intent == LIFECYCLE_OFFBOARD def normalize_reason_codes(raw: object) -> frozenset[str]: """Parse an exclusion-code config value into a comparable set. Accepts a JSON list or a comma/space-separated string (the admin form stores the latter). Tokens are upper-cased so ``d03`` matches ``D03``; raw picklist option ids (``2566``) pass through untouched. """ if isinstance(raw, str): raw = raw.replace(",", " ").split() if not isinstance(raw, (list, tuple, set, frozenset)): return frozenset() return frozenset( token for item in raw if (token := str(item).strip().upper()) ) def classify_inactive_sf_user( *, hire_date: date | None, as_of: date, termination_end_date: date | None, termination_reason_code: str | None = None, excluded_reason_codes: frozenset[str] = frozenset(), ) -> InactiveUserDecision: """SINGLE SOURCE OF TRUTH for "is this inactive SF user a leaver?". SF ``status='f'`` conflates two populations — genuine terminations AND future/day-one hires that simply haven't started yet — and every past incident in this domain (Camila P10009 add-order race, Robert Wood P10013 Shanghai-tz boundary, the fabricated-``lastModifiedDateTime`` offboard) came from the full-roster and incremental paths each re-deciding this with subtly different logic. Both paths now call THIS function so they can never drift again; each path still decides what to DO with a non-leaver (full sync skips it, incremental onboards it), but the leaver-or-not verdict is centralized here. Rules, in order: 1. **Not yet started** — ``hire_date >= as_of``: a future / day-one hire, never a leaver. ``as_of`` is the HQ Asia/Shanghai date, our easternmost tz, whose calendar date rolls over FIRST; Shanghai only reaches ``hire_date + 1`` once the westmost point on Earth is already on ``hire_date``, so ``>=`` protects the hire through their entire pre-start window in EVERY timezone (no per-person tz needed). 2. **Excluded termination reason** — a substantiated termination whose ``termination_reason_code`` (SF ``EmpEmploymentTermination. customString85`` → EcEvenReason2 externalCode, e.g. ``D07``) is on the integration's configured exclusion list: NOT a leaver. These are internal transitions (人员分类转换 / 跨数据中心调动 / 兼岗结束) where the human keeps working — only the SF employment slice ends — so closing accounts would cut off an active employee. Matching is case-insensitive; the config may also pin raw option ids. 3. **Substantiated termination** — a real ``termination_end_date`` that is ``>= hire_date``: OFFBOARD on that date. 4. **Everything else** — no real termination endDate, OR an endDate that precedes ``hire_date`` (a logically impossible leaver_date-before-hire, which is exactly the fabricated shape that made P10013 a leaver): NOT a leaver. We NEVER invent a termination date; the ambiguity surfaces via the daily ``hris_status_divergence`` data-quality check for human review instead of a silent, destructive auto-offboard. """ if hire_date is not None and hire_date >= as_of: return InactiveUserDecision( LIFECYCLE_FUTURE_HIRE, None, f"hire_date {hire_date} >= as_of {as_of}: not yet started, not a leaver", ) if termination_end_date is not None and ( hire_date is None or termination_end_date >= hire_date ): reason_token = (termination_reason_code or "").strip().upper() if reason_token and reason_token in excluded_reason_codes: return InactiveUserDecision( LIFECYCLE_REASON_EXCLUDED, termination_end_date, f"termination reason {reason_token} is on the configured " "no-account-closure list (internal transfer / " "reclassification): not a leaver", ) return InactiveUserDecision( LIFECYCLE_OFFBOARD, termination_end_date, "substantiated termination endDate", ) if termination_end_date is not None: return InactiveUserDecision( LIFECYCLE_NO_TERMINATION, None, f"termination endDate {termination_end_date} precedes hire_date " f"{hire_date}: impossible, refusing offboard", ) return InactiveUserDecision( LIFECYCLE_NO_TERMINATION, None, "SF status inactive but no substantiated termination endDate; " "not fabricating a leaver_date", ) @dataclass(slots=True) class SuccessFactorsWebhookSummary: events_received: int = 0 events_created: int = 0 people_upserted: int = 0 hr_employees_upserted: int = 0 # Per-event apply failures isolated by the per-event try/except in # ``process_webhook_events``. Reported back to the SF caller so the # iflow can surface non-fatal partial outcomes (HTTP 200 with # ``errors > 0`` rather than retrying the entire batch and risking # the mass-leaver brake on the resend). errors: int = 0 # Offboard events acknowledged but deliberately NOT acted on because # their termination reason is on the configured no-account-closure # list (人员分类转换 / 跨DC调动 / 兼岗结束). Each writes an # ``offboard_skipped`` ledger row. offboards_skipped_reason_excluded: int = 0 def to_dict(self) -> dict: return asdict(self) # ── HRIS small-delta auto-confirm settings (platform_settings keys) ── # When a preview produces a small, low-risk delta the admin would # rubber-stamp anyway, skip the manual confirm click. Master switch is # OFF by default — opt-in per deployment. SETTING_AUTO_CONFIRM_ENABLED = "hris.auto_confirm_small_delta" SETTING_AUTO_CONFIRM_THRESHOLD = "hris.auto_confirm_threshold_count" SETTING_AUTO_CONFIRM_BLOCK_TERMINATION = "hris.auto_confirm_block_on_termination" SETTING_AUTO_CONFIRM_BLOCK_MANAGER = "hris.auto_confirm_block_on_manager_change" AUTO_CONFIRM_DEFAULT_THRESHOLD = 3 # action used on the audit row written when a batch auto-confirms. AUDIT_ACTION_AUTO_CONFIRMED = "hris.batch_auto_confirmed" class SuccessFactorsLifecycleService: def __init__( self, db: AsyncSession, sf_client: SuccessFactorsClient | None = None, *, today: date | None = None, hris_instance_id: UUID | None = None, ): self.db = db # Lazy: credentials live in platform_configs, so the first # SF call resolves the client via from_config. Tests can still # inject a pre-built ``sf_client`` to bypass the DB lookup. self._sf_client: SuccessFactorsClient | None = sf_client # "Today" is the company business day (Asia/Shanghai), NOT server # UTC: the daily leave-sync cron fires 06:30 CST = 22:30 UTC the # PRIOR calendar day, so date.today() would label the just-opened # current-FY TimeAccount as not-current on its first business day. # Matches leave_insights_service, which anchors the same way. self.today = today or datetime.now(ZoneInfo("Asia/Shanghai")).date() # An explicitly injected ``today`` is a test/caller anchor and must # win over any per-person timezone refinement (see # ``_local_today_for``); ``None`` means "resolve from the clock". self._today_explicit = today # Master-data caches — loaded once per batch via # ``_ensure_master_data_loaded`` so that confirm can turn the # codes stored on EmpJob (company=DP00012, position=PO00001, # location=USA4491, department=1809807) into the human-readable # names the UI renders. Without these the employee list shows # raw codes and memberships fall back to ``Unassigned``. # ``userId`` → SF ``personIdExternal``, memoized per service so the # concurrent-employment guard's re-resolution (see # ``_resolve_person_id_external``) costs at most ONE EmpEmployment # call per userId per batch. ``None`` values are cached too — a # userId SF has no employment row for must not be re-asked once # per staged event. self._person_id_external_cache: dict[str, str | None] = {} # userIds whose HRBP / HR Ops relationship pull has already been made # in this batch (see ``_resolve_hire_hr_relationships``). One onboard # event reaches that helper twice on the webhook path; the memo keeps # it at one SF round-trip without turning "column is set" into a # reason to skip a refresh. self._hr_rel_resolved: set[str] = set() self._companies: dict[str, dict] | None = None self._locations: dict[str, dict] | None = None self._departments: dict[str, dict] | None = None self._positions: dict[str, dict] | None = None # Per-preview org placement snapshot (migration 565). ``None`` # = not preloaded; the drift helpers fall back to a per-person # query so a directly-constructed service (tests, webhook path) # keeps working. Seeded by ``_refresh_org_snapshot``. self._hris_id_by_person: dict[UUID, str | None] | None = None self._primary_dept_by_person: dict[UUID, str | None] | None = None # SF-verbatim label caches for Google Employee_Details (loaded in # ``_ensure_master_data_loaded``). office_code → EmailGroup label, # JF code → FOJobFunction name, JFM code → cust_JobFamily name. self._email_group_labels: dict[str, str] | None = None self._job_functions: dict[str, str] | None = None self._job_families: dict[str, str] | None = None # Personal contact caches (migration 048): PerPhone / PerEmail # keyed by personIdExternal. Mobile = first matching PerPhone # (phoneType='1043'), personal email = first matching PerEmail # (emailType='1047', prefer isPrimary). Loaded once per batch # alongside the other master data so confirm can flatten the # values onto identity_persons without per-row SF calls. self._personal_mobiles: dict[str, str] | None = None # personIdExternal → ISO-3166 alpha-2 of the mobile's OWN country # (SF PerPhone.countryCode picklist option → externalCode). Lets the # phone formatter use the phone's dialing code (host country for an # international assignee) instead of the person's home country. self._personal_mobile_countries: dict[str, str] | None = None self._personal_emails: dict[str, str] | None = None # Cross-instance personal-phone supplement (2026-08-03). A person # employment-OWNED by instance A (``hris_instance_id``) can have # their PerPhone maintained ONLY in instance B — PingCAP employs # many foreign nationals via the CN tenant while their personal # contact (mobile) lives in the EU tenant. The owning-instance # PerPhone pull then returns nothing and ``mobile_phone`` stays # NULL (root-caused 2026-08-03: 115 active people affected). This # map holds ``personIdExternal → {number, cc, email_lp}`` gathered # from every OTHER enabled instance; it is applied at write time # ONLY when the source instance's ``User.email`` local-part matches # the person's ``work_email`` (SF ``userId`` is not unique across # tenants — the guard prevents binding a colliding different # person's number). self._crossinstance_phones: dict[str, dict[str, str]] | None = None # personIdExternal → raw SF date string (e.g. ``/Date(...)/``). # Sourced from PerPersonal.customDate1 — PingCAP's tenant # stores DoB there, not on User.dateOfBirth. Caller passes # through ``_parse_date``. self._personal_birthdays: dict[str, str] | None = None # personIdExternal → SF PerPersonal.gender (``M`` / ``F``). # Same master-data pull as birthdays (migration 344). Projected # to Feishu's gender enum by the feishu_contact target. self._personal_genders: dict[str, str] | None = None # personIdExternal → HR-assigned corporate 邮箱名 (email local-part, # e.g. "san.zhang"). Authoritatively stored on SF # ``PerPersonal.customString4`` (matches SF's outbound # ``emailName <<-- customString4`` mapping). Loaded per-batch by the # pull-sync flows and read in ``_stage_event`` so the staged # ``email_name`` is the same value the webhook path receives as # ``emailName``. Empty dict = not yet loaded / nothing found. self._email_names_by_pid: dict[str, str] = {} # employee_no (externalCode) → SF cust_Other.cust_id = GitHub # handle (migration 369). Sparse (onboarding 「其他声明」form); # junk values filtered at load. → Feishu ``GitHub ID`` custom attr. self._github_ids: dict[str, str] | None = None # When running against a specific HRIS integration (multi-region # deployment with CN + EU SF side by side), this FK is stamped # onto every IdentityPerson / HrSyncBatch the service creates so # the scheduler can route the next tick to the right instance # and the admin UI can filter history by region. self.hris_instance_id = hris_instance_id # company_code → owning hris integration id, built lazily from # each enabled ``hris`` integration's # ``config_public["owned_company_codes"]`` (see # ``_company_owner_for``). Drives the onboard-poach guard: a # mirrored hire must be CREATED by the instance that canonically # owns their legal entity, no matter which tenant's sync sees # them first. self._company_owner_map: dict[str, UUID] | None = None # Upper-cased termination-reason exclusion set, loaded lazily from # the integration's ``config_public`` (see # ``_excluded_termination_reason_codes``). self._excluded_reason_codes: frozenset[str] | None = None # #2220 — see ``_consume_pending_dispatch``. self._pending_dispatch_event_id: UUID | None = None # #2215 — set when the stale-endDate gate refused the termination; # consumed by ``_dispatch_is_suppressed``. self._offboard_refused_stale: bool = False self._dispatch_suppressed: bool = False def _dispatch_is_suppressed(self) -> bool: """True when the last apply deliberately did NOT apply a termination (#2215) — the coordinators must not run for it. Cleared on read, so a monkeypatched ``_apply_event`` reads False. """ suppressed, self._dispatch_suppressed = self._dispatch_suppressed, False return suppressed def _consume_pending_dispatch(self) -> UUID | None: """Ledger row id whose coordinators have never run, or ``None``. Set by ``_apply_event`` from ``hr_lifecycle_events.dispatched_at`` and cleared on read, so each apply's answer is used exactly once (and a monkeypatched ``_apply_event`` reads as "nothing pending"). """ pending, self._pending_dispatch_event_id = ( self._pending_dispatch_event_id, None, ) return pending async def _mark_lifecycle_dispatched(self, event_id: UUID | None) -> None: """Stamp ``dispatched_at`` once the coordinators have run (#2220). Not stamped when dispatch raised — that is exactly the state the column exists to remember, so the next confirm/webhook re-arms. """ if event_id is None: return row = await self.db.get(HrLifecycleEvent, event_id) if row is not None and row.dispatched_at is None: row.dispatched_at = datetime.now(UTC) def _local_today_for(self, person: IdentityPerson | None) -> date: """Today's date in the PERSON's local timezone. ``self.today`` is anchored to Asia/Shanghai (the company business day). That anchor is wrong for date comparisons about a specific employee: for an EU-tenant person whose termination is entered on their last working day, Shanghai has already rolled over to tomorrow for ~6 hours every evening (CEST 18:00 = CST 00:00), so ``endDate < self.today`` reads TRUE for a termination that is in fact today's — the "is this endDate stale?" gate then misreads a live termination as a rehire artifact (#2215). Project rule: date logic about a person uses that person's local timezone (``resolve_timezone``), same as ``list_overdue_leavers``. An explicitly injected ``today`` (tests, backfills) still wins so callers keep a deterministic clock. """ if self._today_explicit is not None: return self._today_explicit if person is None: return self.today return datetime.now(UTC).astimezone(resolve_timezone(person)).date() def _is_inflight_scheduled_leaver( self, person: IdentityPerson | None, ) -> bool: """True iff ``person`` is a confirmed leaver whose last working day is still ahead of them (in THEIR timezone). SF keeps ``User.status='t'`` until the last working day passes, so every incremental tick AND the weekly full sync re-stage an ``onboard`` event for a person HR already confirmed as a future leaver. Treating that onboard as a rehire (``reactivate``) is wrong twice over: * ``_upsert_person``'s onboard branch reset ``employment_status='active'`` + ``termination_date=None``, erasing the confirmed termination. ``hr.process_advance`` can then never flip them to ``left`` (nothing to advance), they drop out of the offboarding cockpit / overdue board, and every ``employment_status`` gate counts them as fully employed. Downstream, the seat reconcilers flip-flopped their seat rows, which duplicated offboarding cancel events (2026-08-11, P0779 / P01361 / Angelyn cohort). * ``action='reactivate'`` fires the new-hire coordinator chain (email reserve + Google / Feishu account provisioning + welcome mail) at a departing employee. This is the ``onboard``-branch twin of the #2215 guard on the offboard branch — that one refuses to REVERT a leaver on a stale endDate; this one refuses to revert them on a re-pushed onboard. A genuine rehire is not affected: rehires come with the person already ``left`` (LWD passed), and the "un-leave" path for a still-scheduled exit is HR cancelling the offboarding process, which clears both fields deliberately. """ if person is None: return False if person.employment_status != "leaver_pending": return False if person.termination_date is None: return False return person.termination_date >= self._local_today_for(person) async def _excluded_termination_reason_codes(self) -> frozenset[str]: """Termination-reason codes that must NOT close accounts. Sourced from the hris integration's ``config_public["excluded_termination_reason_codes"]`` (admin form field on the SuccessFactors driver): EcEvenReason2 externalCodes (``D03 D04 D07 D08`` per the 2026-07-30 HRIS alignment) and/or raw picklist option ids. Instance-scoped when this service is bound to one (``hris_instance_id``); the legacy unbound service unions every enabled hris integration's list (the two prod tenants share one picklist, so union == either). Empty set = no exclusion = pre-change behaviour. """ if self._excluded_reason_codes is not None: return self._excluded_reason_codes from app.integrations.models import Integration # noqa: PLC0415 stmt = select(Integration.config_public).where( Integration.kind == "hris", Integration.enabled.is_(True), ) if self.hris_instance_id is not None: stmt = stmt.where(Integration.id == self.hris_instance_id) codes: set[str] = set() for (cfg,) in (await self.db.execute(stmt)).all(): codes |= normalize_reason_codes( (cfg or {}).get("excluded_termination_reason_codes") ) self._excluded_reason_codes = frozenset(codes) return self._excluded_reason_codes @staticmethod def person_id_external_is_usable( person_id_external: str | None, user_id: str | None, ) -> bool: """True when ``person_id_external`` can act as SF's *person* key. SF identifies a **person** by ``personIdExternal`` and a single **段雇佣** by ``userId``. A 兼岗 / concurrent employment is a second ``userId`` under the SAME ``personIdExternal`` — that inequality is the ONLY reliable signal Helix has for "this is not a new hire", and it is what the concurrent-employment guard in ``_upsert_person`` keys on. Two shapes make the carried value unusable AS A PERSON KEY: * **empty** — the historical case the webhook resolver already handled; * **self-referential** (``person_id_external == user_id``) — which is NOT a person key at all, it's the employment key echoed back. Two producers emit exactly this: (a) the SF outbound iflow, whose Integration Center mapping fills ``person_id_external`` from the assignment's own ``userId`` (prod 2026-08-25: the 兼岗 push for Hiroki Shinagawa carried ``person_id_external: "80100063"`` while SF's own ``EmpEmployment`` row said ``P01008``); and (b) Helix's own full-sync synthetic employment dicts, which stuff ``userId`` into the field as a placeholder for the master-data caches. Treating a self-referential value as authoritative silently DEFEATS the guard — ``str(pid) != user_id`` can never be true — so the 兼岗 was minted as a brand-new person, given ``hiroki.shinagawa01@pingcap.com`` and a real Google mailbox. Callers must re-resolve from ``EmpEmployment`` in both shapes. NOTE the deliberate asymmetry: for a normal single-employment person ``personIdExternal == userId`` legitimately (P01008 → P01008), so "unusable" here does not mean "wrong" — it means "not proof of anything", and the SF round-trip that follows returns the very same value for those rows. """ from app.identity.sf_assignment import ( person_id_external_is_usable as _usable, ) # Delegates so the ingest guard and the account minter # (``duplicate_employment_for``) cannot drift apart — two copies # of a predicate this load-bearing is a net liability. return _usable(person_id_external, user_id) async def _resolve_person_id_external( self, user_id: str, carried: str | None, *, force: bool = False, ) -> str | None: """``userId`` → SF ``personIdExternal``, preferring a usable carry. Returns ``carried`` untouched when it already passes :meth:`person_id_external_is_usable`, or when it equals ``user_id`` and ``force`` is False. ``force=True`` means "the value I hold is a stand-in, not something SF told me" — the webhook path (whose producer demonstrably echoes ``userId``) and Helix's own synthetic employment dicts (flagged ``_person_id_external_is_user_id``) pass it. Everything else carries a value that came straight off an ``EmpEmployment`` row, where ``pid == userId`` is the *correct* answer for a single-employment person and re-asking SF would just buy the same string back — 500+ sequential round-trips on a bootstrap full sync for zero information. Otherwise asks SF's ``EmpEmployment`` (the entity that owns the person↔employment mapping and has it populated from the moment the employment is created — verified in prod: ``get_person_id_external('80100063')`` → ``'P01008'``) and returns that. **Fails safe**: on a lookup miss or error the original ``carried`` value is returned, so the worst case is exactly today's behaviour rather than losing the personal-data enrichment key. Cached per service instance — ``_upsert_person`` runs once per staged event in a confirm batch and must not re-ask SF for the same userId. """ uid = (user_id or "").strip() if not uid: return carried if self.person_id_external_is_usable(carried, uid): return carried if (carried or "").strip() and not force: # Self-referential, but it came off a real EmpEmployment row — # that IS this person's key. See the ``force`` contract above. return carried if uid in self._person_id_external_cache: return self._person_id_external_cache[uid] or carried resolved: str | None = None try: client = await self._get_client() resolved = await client.get_person_id_external(uid) except Exception as exc: # noqa: BLE001 logger.warning( "hr.sync: personIdExternal lookup failed for user_id=%s " "(%s) — keeping carried value %r; the concurrent-employment " "guard stays blind for this event", uid, exc, carried, ) self._person_id_external_cache[uid] = resolved if resolved and resolved.strip() != carried: logger.info( "hr.sync: resolved personIdExternal for user_id=%s from SF " "EmpEmployment (carried=%r → %r)", uid, carried, resolved, ) return resolved or carried async def _tombstone_belongs_to_live_person( self, user_id: str, employment: dict | None, *, matched: IdentityPerson | None, ) -> IdentityPerson | None: """The live person SF says owns ``user_id``, when ``matched`` is a grave. The concurrent-employment guard in :meth:`_upsert_person` lives inside its ``person is None`` branch — i.e. it only asks "whose employment is this?" when the userId is one Helix has never seen. That gate is structurally blind to the case where the userId HAS been seen: as a **different, long-dead human's** primary key. Prod 2026-09-04 (P01412 战捷): HR added one 兼岗 for him, which SF mirrors into both tenants with a per-tenant userId — CN minted ``80100064`` (unseen → aliased correctly, exactly as #3228 intended) and EU minted ``100075``. But ``100075`` was already the ``hris_person_id`` of a leaver purged on 2026-04-30 whose row still carried ``hris_instance_id IS NULL``, so the legacy-adoption arm of the userId lookup matched that grave for the EU sync. With ``person`` non-None the guard never ran, the onboard branch cleared ``deleted_at``, and the resurrected row — still holding the dead colleague's ``work_email`` — was provisioned as a Day-1 new hire: a real Google mailbox at the dead man's address, its password mailed to 战捷's personal inbox, and that address written back to SF as **P01412's** primary business email. SF userIds are NOT unique across the CN and EU tenants, so "this userId once belonged to someone" is never evidence about who it belongs to now. ``personIdExternal`` is. It is SF's *person* key, it rides along on the ``EmpEmployment`` row the event already carries, and here it said ``P01412`` — a live, active person who is NOT the grave we matched. Whenever those two disagree the employment is that live person's, and the grave must stay a grave. Returns that canonical person, or ``None`` when the question does not apply. Deliberately narrow so no working path changes: * only when ``matched`` is soft-deleted or not ``active`` — a healthy active row matched by its own userId is the normal update path and must not pay an SF round-trip; * only when ``personIdExternal`` is usable as a person key (see :meth:`person_id_external_is_usable`) — a self-referential value proves nothing, and re-asking SF for one is paid for only when the match is an actual tombstone (see the ``force`` comment); * only when it resolves to a DIFFERENT live active person. A genuine rehire resolves to the same (non-active) row and so returns ``None`` — rehire handling is untouched. """ if matched is None: return None if matched.deleted_at is None and matched.employment_status == "active": return None # Re-resolving a stand-in value costs one sequential SF round-trip # per userId per batch, and EVERY leaver row a full sync refreshes # lands in this method (they are all non-``active``) — paying it # for all of them would put ~130 round-trips on the nightly sync # for a shape that has never once collided. Spend it only on the # shape that actually did: a TOMBSTONE. A live-but-departed row # is judged on the free evidence the event already carries, so it # is still caught whenever SF handed us the real key. pid_ext = await self._resolve_person_id_external( user_id, (employment or {}).get("personIdExternal"), force=( bool((employment or {}).get("_person_id_external_is_user_id")) and matched.deleted_at is not None ), ) if not self.person_id_external_is_usable(pid_ext, user_id): return None canonical = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == str(pid_ext), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) ) if canonical is None or canonical.id == matched.id: return None return canonical async def _claim_tombstoned_user_id( self, user_id: str, employment: dict | None, *, matched: IdentityPerson | None, event_type: str, ) -> bool: """Alias a grave's userId onto its real owner. True ⇒ skip the event. Companion to :meth:`_tombstone_belongs_to_live_person`; shared by the preview (``_stage_event``) and the apply (``_upsert_person``) so a manual confirm cannot walk past a verdict the preview reached. Registering the alias is what makes the fix *stick*: the alias short-circuit at the top of both methods precedes the userId lookup, so every later tick resolves this userId to the canonical person instead of re-litigating the grave. The skip itself is unconditional: SF has told us this employment belongs to somebody else, so there is nothing legitimate to apply to the row we matched, and resurrecting it is the exact harm. What IS conditional is the alias — ``record_hris_user_alias`` refuses a userId that a LIVE person holds as their own primary key (aliasing it would stop syncing that human entirely), which is the case for a terminated-but-not-tombstoned row. That refusal only costs us the memo: the verdict is re-derived on the next tick and comes out the same. The two outcomes get two event codes rather than one boolean field, because every field value reaching the log sink is hashed — "did the alias land?" has to be readable off the code. """ canonical = await self._tombstone_belongs_to_live_person( user_id, employment, matched=matched, ) if canonical is None: return False from app.identity.identity_reconcile import ( hris_user_alias_owner, record_hris_user_alias, ) await record_hris_user_alias( self.db, person=canonical, alias_user_id=user_id, reason="concurrent_employment", person_id_external=canonical.hris_person_id, hris_instance_id=self.hris_instance_id, note=( "EmpEmployment.personIdExternal belongs to an active " "person; this userId was a retired row's primary key" ), ) aliased = await hris_user_alias_owner(self.db, user_id) is not None logger.event( "HRIS_TOMBSTONE_USERID_CLAIMED" if aliased else "HRIS_TOMBSTONE_USERID_UNALIASABLE", { "user_id": user_id, "event_type": event_type, "canonical_hris_person_id": canonical.hris_person_id, "tombstone_person_id": str(matched.id) if matched else None, }, level=logging.WARNING, ) return True async def _termination_reasons_for( self, user_ids: Iterable[str], ) -> dict[str, str]: """Best-effort ``userId → termination-reason code`` for classifying. Cost-gated three ways so the pre-config / steady-state overhead is zero-to-tiny: * empty exclusion config → return ``{}`` without any SF call (byte-identical pre-change behaviour); * userIds whose Helix person is already ``left`` are dropped — their verdict can't change anymore (the nightly full sync would otherwise re-fetch every historical leaver's reason forever); * failures log + return ``{}`` — the classifier then falls back to the pre-change "substantiated endDate = leaver" verdict. Fail-open-to-offboard is deliberate: a broken reason lookup must never leave a REAL leaver's access alive; a wrongly offboarded transferee is recoverable and loudly visible. """ uids = sorted({u for u in user_ids if u}) if not uids: return {} if not await self._excluded_termination_reason_codes(): return {} rows = ( await self.db.execute( select( IdentityPerson.hris_person_id, IdentityPerson.employment_status, ).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id.in_(uids), IdentityPerson.deleted_at.is_(None), ) ) ).all() already_left = {uid for uid, status in rows if status == "left"} candidates = [u for u in uids if u not in already_left] if not candidates: return {} try: client = await self._get_client() return await client.get_termination_reasons_for_users(candidates) except Exception: # noqa: BLE001 — best-effort by design (see docstring) logger.exception( "hr.sync: termination-reason lookup failed for %d user(s) — " "classifying WITHOUT reason exclusion this pass (an " "excluded-reason transfer may be offboarded; a real leaver " "is never missed)", len(candidates), ) return {} async def _company_owner_for(self, company_code: str | None) -> UUID | None: """Resolve which HRIS instance canonically OWNS an SF company code. CN (api30) + EU (api55) SF tenants MIRROR one org: ops enters an overseas hire in BOTH, so both tenants' syncs see the person and the first one to run would create the ``identity_persons`` row — with ITS ``hris_instance_id``, permanently poaching the person from the true owner (the Camila P10009 race, onboard flavour). The tie-breaker is the person's legal entity: each enabled ``hris`` integration may declare the FOCompany codes it owns via ``config_public["owned_company_codes"]`` (a JSON list, or a comma/space-separated string from the admin form). Returns the owning integration id, or ``None`` when the code is unmapped, claimed by 2+ instances (ambiguous — logged and dropped), or no instance declares ownership at all. ``None`` means "no verdict": callers must fall back to today's behaviour, never treat it as "not owned by me". """ if not company_code or self.hris_instance_id is None: return None if self._company_owner_map is None: from app.integrations.models import Integration rows = ( await self.db.execute( select(Integration.id, Integration.config_public).where( Integration.kind == "hris", Integration.enabled.is_(True), ) ) ).all() mapping: dict[str, UUID] = {} ambiguous: set[str] = set() for integration_id, cfg in rows: raw = (cfg or {}).get("owned_company_codes") or [] if isinstance(raw, str): raw = raw.replace(",", " ").split() for code in raw: code = str(code).strip() if not code: continue if code in mapping and mapping[code] != integration_id: ambiguous.add(code) mapping[code] = integration_id for code in ambiguous: mapping.pop(code, None) logger.warning( "hr.sync: company code %s is claimed by MULTIPLE hris " "integrations' owned_company_codes — ignoring the " "mapping for this code (fix the integration configs)", code, ) self._company_owner_map = mapping return self._company_owner_map.get(company_code) async def _assert_batch_volume_is_sane(self, batch: HrSyncBatch) -> None: """Refuse to auto-apply a batch whose volume looks like a bug. Two symmetric brakes, both scoped to the batch's own HRIS instance so a CN batch is never measured against EU's headcount (Round-20 audit A3F2): * **Mass-leaver** — ``offboard/deactivate`` rows. One tick that says "every active person is now terminated" must not auto-apply. Pre-launch hardening 2026-05-06 lifted this out of ``run_scheduled_incremental`` into the shared confirm path so an admin clicking auto-confirm on a corrupted preview cannot drive through it either. * **Mass-onboard** — ``onboard/create`` rows. New in migration 565 and NOT a theoretical addition: until now the only thing standing between a corrupted full-roster preview and 500 real Google accounts was the ``sync_mode='full'`` coordinator suppression this migration removes. Removing a safety net without replacing it is how a bug fix becomes an incident. Both are exempt on a SEED batch, whose defining shape is exactly "hundreds of creates and every historical leaver at once", and whose coordinators are suppressed anyway. Production distribution at the time of writing (prod TiDB, per-batch maxima): routine scheduled full ≤ 9 creates / ≤ 16 deactivates; incremental ≤ 13 creates; genuine April-2026 seed batches 137–336 creates. The shared ``max(25, 5%)`` floor separates them cleanly with no near-miss on either side. """ counts = ( await self.db.execute( select( func.sum( case( ( and_( HrSyncStagedEvent.event_type == "offboard", HrSyncStagedEvent.action == "deactivate", ), 1, ), else_=0, ) ), func.sum( case( ( and_( HrSyncStagedEvent.event_type == "onboard", HrSyncStagedEvent.action == "create", ), 1, ), else_=0, ) ), ).where(HrSyncStagedEvent.batch_id == batch.id) ) ).one() offboard_count = int(counts[0] or 0) onboard_count = int(counts[1] or 0) active_total = await self._count_active_persons_for_instance() threshold = max(25, int(active_total * 0.05)) is_seed = bool(batch.is_seed) for label, count in ( ("offboard", offboard_count), ("onboard", onboard_count), ): if count <= threshold: continue if is_seed: # Log the exemption explicitly so an admin reviewing the # batch can verify the seed intent rather than inferring # it from the absence of a refusal. logger.info( "HRIS batch %s: %d %s events on a SEED batch " "(threshold %d) — brake exempted, this is the " "expected initial-import shape.", batch.id, count, label, threshold, ) continue logger.warning( "HRIS confirm refused: %d %s events vs %d active " "(threshold=%d). batch %s left in pending_review for " "explicit human review (sync_mode=%s, is_seed=False).", count, label, active_total, threshold, batch.id, batch.sync_mode, ) raise ConflictError( f"Mass-{label} brake: {count} {label} events exceeds " f"threshold {threshold} (active_total={active_total}). " f"Refusing to auto-confirm — review staged events " f"manually, or re-run as a seed import if this really is " f"a historical backfill." ) async def _count_active_persons_for_instance(self) -> int: """Count active ``IdentityPerson`` rows scoped to this service's HRIS instance (or globally if ``self.hris_instance_id`` is None). Used by every mass-leaver brake (confirm_batch / scheduled_incremental / webhook). Scoping the denominator prevents a CN-tenant batch's threshold from being inflated by EU's active population (and vice-versa) — Round-20 audit A3F2. Legacy single-instance rows (``hris_instance_id IS NULL``) are counted into every instance's scope so installs that haven't backfilled the column yet still get a reasonable threshold. """ q = ( select(func.count()) .select_from(IdentityPerson) .where(IdentityPerson.employment_status == "active") ) if self.hris_instance_id is not None: q = q.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) return await self.db.scalar(q) or 0 @classmethod async def fan_out( cls, db: AsyncSession, *, today: date | None = None, ) -> list[SuccessFactorsLifecycleService]: """Build one service per enabled HRIS integration. The multi-instance entry point: callers that need to run the same sync operation (full / window / incremental) across every configured SF instance (CN + EU + …) iterate the returned list instead of reimplementing the per-instance loop. Falls back to a legacy single-instance service (``hris_instance_id=None``, credentials from ``get_primary``) when no ``hris`` integration is registered yet — keeps tests and bare-metal installs working. """ from app.integrations.compat import list_hris_instances instances = await list_hris_instances(session=db) if not instances: # No rows — fall back to the legacy "primary" path. The # caller may get an AppException when it actually tries to # hit SF; that surfaces the config gap clearly. return [cls(db, today=today)] services: list[SuccessFactorsLifecycleService] = [] for row, config in instances: try: services.append( cls.for_instance( db, integration_id=row.id, config=config, today=today, ) ) except ValueError as exc: logger.warning( "skipping HRIS instance %s (%s): %s", row.slug, row.id, exc, ) return services @classmethod def for_instance( cls, db: AsyncSession, *, integration_id: UUID, config: dict, today: date | None = None, ) -> SuccessFactorsLifecycleService: """Build a service bound to one specific SF integration row. Used by the multi-instance scheduler: iterate over every enabled ``hris`` integration, build a service scoped to that instance, and run an incremental sync. Keeps concerns split — the caller already loaded the config (so we don't re-hit the DB) and knows the integration id (so we can stamp it on outputs). """ base_url = (config.get("base_url") or "").rstrip("/") if not base_url: raise ValueError("SuccessFactors integration is missing base_url") auth = build_auth_from_config(config) client = SuccessFactorsClient(base_url=base_url, auth=auth) return cls( db, sf_client=client, today=today, hris_instance_id=integration_id, ) async def _get_client(self) -> SuccessFactorsClient: if self._sf_client is None: self._sf_client = await SuccessFactorsClient.from_config(session=self.db) return self._sf_client @property def sf_client(self) -> SuccessFactorsClient: """Back-compat accessor for tests that assert against the client. Async callers inside this service should use ``_get_client()`` directly so credentials load on first use; this property only works after the lazy init has happened (or when a client was injected in ``__init__``). """ if self._sf_client is None: raise RuntimeError( "SuccessFactorsClient has not been initialized; " "await _get_client() first" ) return self._sf_client @staticmethod def _email_local_part(email: str | None) -> str: """Lowercased local-part of an email, or ``""`` when malformed. Used by the cross-instance phone same-person guard. A ``.com`` vs ``.com.cn`` domain difference between SF tenants must not defeat the match, so we compare local-parts only. """ e = (email or "").strip().lower() return e.split("@", 1)[0] if "@" in e else "" @staticmethod def _pick_primary_first( rows: list[dict], value_key: str, ) -> dict[str, str]: """personIdExternal → first non-empty ``value_key``, preferring ``isPrimary`` rows then any. Shared by the owning-instance pull and the cross-instance supplement so both resolve a person's multiple PerPhone rows identically. """ by_pid: dict[str, str] = {} for r in rows: pid = (r or {}).get("personIdExternal") val = (r or {}).get(value_key) if pid and val and r.get("isPrimary") and pid not in by_pid: by_pid[pid] = val for r in rows: pid = (r or {}).get("personIdExternal") val = (r or {}).get(value_key) if pid and val and pid not in by_pid: by_pid[pid] = val return by_pid @classmethod def _crossinstance_mobile_for( cls, work_email: str | None, person_id_external: str | None, crossinstance: dict[str, dict[str, str]] | None, ) -> dict[str, str] | None: """Return ``{"number", "cc"}`` from another instance's PerPhone for this person, or ``None``. Guards SAME-PERSON identity by requiring the source instance's ``User.email`` local-part (stored as ``email_lp``) to equal the person's ``work_email`` local-part. SF ``userId`` collides across the CN/EU tenants for genuinely different people, so an unguarded cross-instance lookup could bind the wrong person's number. """ if not crossinstance or not person_id_external: return None rec = crossinstance.get(person_id_external) if not rec or not rec.get("number"): return None wl = cls._email_local_part(work_email) if not wl or wl != rec.get("email_lp"): return None return {"number": rec["number"], "cc": rec.get("cc") or ""} async def _load_crossinstance_phones(self) -> dict[str, dict[str, str]]: """Best-effort: gather PerPhone (mobile, phoneType 1043) from every ENABLED HRIS instance OTHER than this sync's own, keyed by ``personIdExternal`` and tagged with the source ``User.email`` local-part for the same-person guard applied at write time. Any instance that fails to pull is logged and skipped — personal- contact enrichment must NEVER break the core roster sync. Returns ``{}`` when there is only one instance or all others fail. """ from app.identity.successfactors_client import ( # noqa: PLC0415 build_auth_from_config, ) from app.integrations.compat import list_hris_instances # noqa: PLC0415 out: dict[str, dict[str, str]] = {} try: instances = await list_hris_instances(session=self.db) except Exception as exc: # noqa: BLE001 logger.warning("hr.sync: list_hris_instances failed: %s", exc) return out for row, cfg in instances: # Skip our own instance — the owning-instance PerPhone is # already merged into ``_personal_mobiles``. if ( self.hris_instance_id is not None and getattr(row, "id", None) == self.hris_instance_id ): continue base = (cfg.get("base_url") or "").strip() if not base or not getattr(row, "enabled", False): continue slug = getattr(row, "slug", getattr(row, "id", "?")) try: client = SuccessFactorsClient( base_url=base, auth=build_auth_from_config(cfg), ) phones = await client.list_personal_phones() numbers = self._pick_primary_first(phones, "phoneNumber") if not numbers: continue cc_opts = self._pick_primary_first(phones, "countryCode") try: opt_to_iso = await client.list_country_code_picklist() except Exception: # noqa: BLE001 opt_to_iso = {} emails = await client.list_user_emails(list(numbers.keys())) for pid, num in numbers.items(): # First enabled other-instance with a matchable email # wins; a later instance never overwrites. if pid in out: continue lp = self._email_local_part(emails.get(pid)) if not lp: continue opt = cc_opts.get(pid) out[pid] = { "number": num, "cc": opt_to_iso.get(str(opt), "") if opt is not None else "", "email_lp": lp, } except Exception as exc: # noqa: BLE001 logger.warning( "hr.sync: cross-instance phone pull for %s failed: %s", slug, exc, ) continue return out # ---- Staged sync (preview → confirm) ---- async def _load_email_names_by_pid( self, employments: Iterable[dict], ) -> None: """Populate ``self._email_names_by_pid`` from SF PerPersonal.customString4. Called once per pull-sync batch (window / full / incremental) BEFORE the ``_stage_event`` loop so each staged onboard carries the HR-assigned 邮箱名 as ``email_name`` — the same value SF pushes to the webhook path as ``emailName``. Keyed by ``personIdExternal`` gathered from the EmpEmployment rows already in hand; best-effort — a fetch failure leaves the cache as-is and ``_stage_event`` falls back to the legacy ``User.custom07`` then the given.family derivation. """ pids = sorted( { pid for emp in employments if isinstance(pid := (emp or {}).get("personIdExternal"), str) and pid.strip() } ) if not pids: return try: client = await self._get_client() self._email_names_by_pid = await client.get_personal_email_names(pids) except Exception: # noqa: BLE001 — best-effort enrichment logger.exception( "hr.sync: failed to load PerPersonal.customString4 email " "names; falling back to User.custom07 for %d person(s)", len(pids), ) async def preview_import( self, *, start_date: date, end_date: date, triggered_by: str, ) -> HrSyncBatch: """Fetch data from SF, compute diff, stage events — do NOT write to identity_persons.""" await self._refresh_org_snapshot() client = await self._get_client() hires = await client.list_hire_employments(start_date, end_date) terminations = await client.list_termination_employments(start_date, end_date) user_ids = sorted( {row["userId"] for row in [*hires, *terminations] if row.get("userId")} ) users: dict[str, dict] = {} jobs_by_user: dict[str, list[dict]] = {} if user_ids: users = { user["userId"]: user for user in await client.get_users(user_ids) if user.get("userId") } jobs_by_user = await self._jobs_by_user_id(user_ids) # HR-assigned 邮箱名 (PerPersonal.customString4), keyed by # personIdExternal off the hire/termination employment rows — # staged into each event so the pull path uses the same email # local-part the webhook path receives as ``emailName``. await self._load_email_names_by_pid([*hires, *terminations]) batch = HrSyncBatch( triggered_by=triggered_by, trigger_source="manual", sync_mode="window", hris_instance_id=self.hris_instance_id, start_date=start_date, end_date=end_date, status="pending_review", ) self.db.add(batch) await self.db.flush() # get batch.id staged: list[HrSyncStagedEvent] = [] for employment in hires: event = await self._stage_event( batch_id=batch.id, event_type="onboard", employment=employment, user=users.get(employment.get("userId", "")), jobs=jobs_by_user.get(employment.get("userId", ""), []), ) if event is not None: staged.append(event) # Termination-reason exclusion (2026-07-30): the manual window # import bypasses ``classify_inactive_sf_user`` (it trusts the # date-window termination pull), so it needs its own gate — a # 人员分类转换 / 跨DC调动 / 兼岗结束 row in the imported window must # not stage a deactivate. Same config + fetch as the other paths. _import_reasons = await self._termination_reasons_for( {uid for e in terminations if (uid := e.get("userId"))} ) _import_excluded = await self._excluded_termination_reason_codes() for employment in terminations: _reason = ( _import_reasons.get(employment.get("userId") or "") or "" ).strip().upper() if _reason and _reason in _import_excluded: logger.info( "hr.sync: window import NOT staging offboard for " "user_id=%s — termination reason %s is on the " "no-account-closure exclusion list", employment.get("userId"), _reason, ) continue event = await self._stage_event( batch_id=batch.id, event_type="offboard", employment=employment, user=users.get(employment.get("userId", "")), jobs=jobs_by_user.get(employment.get("userId", ""), []), ) if event is not None: staged.append(event) for evt in staged: self.db.add(evt) batch.events_staged = len(staged) await self.db.flush() return batch async def evaluate_auto_confirm(self, batch) -> dict: """Decide whether ``batch`` qualifies for small-delta auto-confirm. Returns a verdict dict ``{eligible, reason, settings, signals}``. Does NOT mutate — the caller runs ``confirm_batch`` when eligible. Gate (all must hold): * master switch ``hris.auto_confirm_small_delta`` is on * new-hire count ≤ ``hris.auto_confirm_threshold_count`` * if ``block_on_termination``: zero terminations in the batch * if ``block_on_manager_change``: zero org-line moves (manager OR department change — both are org moves needing human review; the single setting governs both) ``batch.staged_events`` must be loaded by the caller. """ from app.integrations.service import get_platform_setting def _as_bool(v, default): if isinstance(v, bool): return v if isinstance(v, str): return v.strip().lower() in ("1", "true", "yes", "on") return default def _as_int(v, default): try: return int(v) except (TypeError, ValueError): return default enabled = _as_bool( await get_platform_setting( SETTING_AUTO_CONFIRM_ENABLED, default=False, session=self.db, ), False, ) threshold = _as_int( await get_platform_setting( SETTING_AUTO_CONFIRM_THRESHOLD, default=AUTO_CONFIRM_DEFAULT_THRESHOLD, session=self.db, ), AUTO_CONFIRM_DEFAULT_THRESHOLD, ) block_term = _as_bool( await get_platform_setting( SETTING_AUTO_CONFIRM_BLOCK_TERMINATION, default=True, session=self.db, ), True, ) block_mgr = _as_bool( await get_platform_setting( SETTING_AUTO_CONFIRM_BLOCK_MANAGER, default=True, session=self.db, ), True, ) events = list(batch.staged_events or []) # Count reactivate (rehire) alongside create: the onboarding chain # (Google account create + welcome email + identity-sync push) fires # for BOTH actions (event_type=="onboard" and action in # {"create","reactivate"}). Counting only "create" let a pure-rehire # batch score new_count=0, pass the threshold, and auto-confirm — # silently re-opening accounts / re-sending welcome mail / re-pushing # downstream for every returning employee with no human review. new_count = sum(1 for e in events if e.action in ("create", "reactivate")) termination_count = sum( 1 for e in events if e.event_type == "offboard" and e.action == "deactivate" ) manager_changed = sum( 1 for e in events if (e.person_data or {}).get("manager_changed") is True ) dept_changed = sum( 1 for e in events if (e.person_data or {}).get("dept_changed") is True ) signals = { "new_count": new_count, "termination_count": termination_count, "manager_changed_count": manager_changed, "dept_changed_count": dept_changed, "events_staged": len(events), } settings = { "enabled": enabled, "threshold": threshold, "block_on_termination": block_term, "block_on_manager_change": block_mgr, } if not enabled: return {"eligible": False, "reason": "disabled", "settings": settings, "signals": signals} if new_count > threshold: return {"eligible": False, "reason": "new_count_over_threshold", "settings": settings, "signals": signals} if block_term and termination_count > 0: return {"eligible": False, "reason": "termination_present", "settings": settings, "signals": signals} if block_mgr and (manager_changed > 0 or dept_changed > 0): return {"eligible": False, "reason": "org_change_present", "settings": settings, "signals": signals} return {"eligible": True, "reason": "small_delta", "settings": settings, "signals": signals} async def auto_confirm_if_eligible( self, batch, *, actor_id: str, trace_id: str = "", ) -> dict: """Evaluate ``batch`` and, if it qualifies, confirm it now + write an ``hris.batch_auto_confirmed`` audit row + push a Feishu summary card to the triggering admin (best-effort). Returns the verdict dict with an added ``confirmed`` bool (and ``confirm_result`` when confirmed). Never raises on the notification leg — a failed card must not roll back the confirm. """ verdict = await self.evaluate_auto_confirm(batch) verdict["confirmed"] = False if not verdict["eligible"]: return verdict result = await self.confirm_batch( batch.id, confirmed_by=f"auto:{actor_id}", ) verdict["confirmed"] = True verdict["confirm_result"] = result # Audit (own short txn — confirm_batch already committed). from app.audit.service import AuditService await AuditService(self.db).log( trace_id=trace_id, actor_type="system", action=AUDIT_ACTION_AUTO_CONFIRMED, status="success", resource_type="hr_sync_batch", resource_id=str(batch.id), detail={ "triggered_by": actor_id, "reason": verdict["reason"], "settings": verdict["settings"], "signals": verdict["signals"], }, ) await self.db.commit() # Feishu summary card to the triggering admin (best-effort). try: actor_uuid = UUID(str(actor_id)) except (ValueError, TypeError): actor_uuid = None if actor_uuid is not None: try: from app.notifications import ( NotificationEnvelope, NotificationService, ) actor = await self.db.get(IdentityPerson, actor_uuid) if actor is not None: sig = verdict["signals"] await NotificationService(self.db).dispatch( NotificationEnvelope( recipient=actor, title="HRIS 小增量已自动确认", body_text=( f"同步批次 {batch.id} 已自动确认:新增 " f"{sig['new_count']} 人,0 离职 / 0 转岗。" ), channels=["feishu"], template_key="hris.batch_auto_confirmed", context={ "batch_id": str(batch.id), "new_count": sig["new_count"], "events_staged": sig["events_staged"], }, url="/admin/hr/sync", related_resource_type="hr_sync_batch", related_resource_id=batch.id, ), trace_id=trace_id or AUDIT_ACTION_AUTO_CONFIRMED, actor_user_id=None, ) except Exception as exc: # noqa: BLE001 — best-effort card logger.warning( "hris auto-confirm summary card failed for batch %s: %s", batch.id, exc, ) return verdict async def confirm_batch( self, batch_id: UUID, confirmed_by: str, ) -> dict: """Apply all pending staged events, mark batch confirmed. **Per-event commit (2026-04-22)**: previously the entire batch ran in one transaction holding row-locks on every touched ``identity_persons`` row until the last event finished. On sf-cn-prod (~500+ employees) this held locks for minutes, colliding with the FastAPI backend's readers of the same rows and triggering ``Lock wait timeout exceeded`` in prod. Now: each staged event is processed inside its own short transaction — ``_apply_event`` + coordinator hooks + staged row status update, then ``commit()``. Locks release within ~100ms per event. **Idempotency**: re-entering ``confirm_batch`` after a partial run is safe — events already marked ``confirmed`` / ``skipped`` are fast-skipped, and ``_apply_event`` / coordinators are all idempotent (upserts by email, ``_upsert_lifecycle_event`` dedups by natural key, coordinator hooks check pre-conditions). **Concurrent callers**: in practice only the scheduler invokes this path (see ``scheduled_jobs/handlers.py:run_hris_incremental_sync``) and it's already serialized to one in-flight run at a time. If a racing caller did appear, per-event ``staged.status != 'pending'`` guards deduplicate work — both callers would process each event at most once. Resuming a partially-confirmed batch (status still ``pending_review``, some staged rows already ``confirmed``) is naturally supported: the loop fast-skips finished rows. """ # Phase A — gate check + short SELECT FOR UPDATE so a second # caller arriving at the same instant sees pending_review and # both advance (worst case both loop; staged.status dedups). result = await self.db.execute( select(HrSyncBatch).where(HrSyncBatch.id == batch_id).with_for_update() ) batch = result.scalar_one_or_none() if batch is None: raise NotFoundError("HrSyncBatch", str(batch_id)) if batch.status != "pending_review": raise ConflictError(f"Batch {batch_id} is already {batch.status}") # Pre-launch hardening 2026-05-06: mass-leaver brake — applied # on the manual confirm path too. The brake previously fired # only inside ``run_scheduled_incremental`` (cron), so an admin # manually clicking "auto_confirm=True" on a corrupted preview # could wipe through the brake at 03:00 UTC during a deploy. # Owner's exact fear scenario; lift the gate into the shared # confirm_batch helper so it triggers regardless of trigger # path. await self._assert_batch_volume_is_sane(batch) # Release the SELECT FOR UPDATE lock before the (potentially # minutes-long) apply, then run the heavy per-event loop. This # ``confirm_batch`` entry point is the SYNCHRONOUS path used by # the cron / auto-confirm caller (``run_scheduled_incremental``), # which already runs on the durable worker process. The # interactive admin-confirm HTTP path instead calls # ``start_async_confirm`` (gate + brake + flip to ``applying`` + # return ``202``) and runs ``_apply_staged_events`` on a # background task, so the browser never blocks on the full apply # (which timed out the frontend on the full-bootstrap batch — # 2026-06-16 prod incident). await self.db.commit() return await self._apply_staged_events(batch_id, confirmed_by) async def _apply_staged_events( self, batch_id: UUID, confirmed_by: str, ) -> dict: """Apply every pending staged event + run the org-wide tail passes + finalize the batch status. Extracted from ``confirm_batch`` (2026-06-16) so the exact same apply semantics are driven from three callers: * ``confirm_batch`` — synchronous cron / auto-confirm path. * ``start_async_confirm`` + ``background.spawn`` — interactive admin-confirm HTTP path (batch already flipped to ``applying``). * ``resume_stale_applying_batches`` — restart-safety reconcile cron that resumes an ``applying`` batch whose background task died mid-run. Caller contract: Phase A (status gate + mass-leaver brake) has already run and committed. This method is idempotent / resumable — already-processed staged rows fast-skip and ``_apply_event`` + coordinators dedupe, so re-entry after a partial run is safe. """ # Read-only snapshot we'll need throughout the loop. Re-loaded in # a fresh read because Phase A committed (expiring its handle) and # the resume-cron path enters here without a prior in-session row. batch = await self.db.get(HrSyncBatch, batch_id) if batch is None: raise NotFoundError("HrSyncBatch", str(batch_id)) # Migration 565. This used to read ``sync_mode == 'full'``, which # made the WEEKLY reconciliation sync behave like a one-time # historical import: every org move, hire and termination it # recovered was written to the person row and then had its # coordinators suppressed. ``is_seed`` is now decided once, at # batch creation, and persisted — see ``_full_sync_is_seed``. is_initial_seed = bool(batch.is_seed) # Heartbeat cadence — bump ``apply_heartbeat_at`` every N events # so the restart-safety reconcile cron can tell a healthy in- # flight apply from a dead one (see ``resume_stale_applying_batches``). # Stamp it once up front so a freshly-launched apply is never # immediately mistaken for stale. heartbeat_every = 25 batch.apply_heartbeat_at = datetime.now(UTC) await self.db.commit() processed_since_beat = 0 # Phase B — fetch staged events in a fresh read, then loop with # per-event commits. Each iteration is its own transaction. staged_result = await self.db.execute( select(HrSyncStagedEvent) .where(HrSyncStagedEvent.batch_id == batch_id) ) staged_events = list(staged_result.scalars().all()) # Apply CANONICAL employments before CONCURRENT ones (2026-08-25). # The query above has no ORDER BY, so the apply order was whatever # the storage engine felt like — and for a person whose main # position AND 兼岗 are both new to Helix in the same batch, that # order decides which one becomes the human: # * 兼岗 first → it is created (no canonical row exists yet to # alias against), and the REAL employment that follows is then # refused as the "duplicate" that collides on work_email — the # person ends up keyed by the 兼岗's userId, which is exactly # the identity we never want to mint; # * canonical first → the 兼岗 finds a live ACTIVE owner and is # aliased, which is the whole point of the guard. # A stable sort on "is this a second employment?" makes the outcome # order-INDEPENDENT without changing anything else about the loop # (Python's sort is stable, so same-class rows keep their prior # relative order). Bites on bootstrap and on any person Helix has # not ingested yet — e.g. the German-DC 兼岗 landing before the CN # roster has been pulled. staged_events.sort( key=lambda s: 1 if self.person_id_external_is_usable( (s.employment_data or {}).get("personIdExternal"), s.user_id, ) else 0 ) events_confirmed = 0 events_skipped = 0 people_upserted = 0 lifecycle_events_created = 0 new_count = 0 updated_count = 0 reactivated_count = 0 deactivated_count = 0 error_count = 0 last_event_error: str | None = None # #1785: accumulate the confirmed user_ids as PLAIN VALUES here, # while the instance is live. The post-commit annual-leave refresh # must NOT re-read ``staged.user_id`` off the ORM instances: a # per-event failure earlier in the loop calls ``db.rollback()`` # which expires every sibling (expire_on_commit=False → no auto- # refresh), so a post-commit ``staged.user_id`` read triggers a # sync lazy-load → MissingGreenlet → the whole batch's leave # refresh is silently swallowed as "leave-data sync failed". confirmed_user_ids: list[str] = [] for staged in staged_events: # #1740: a PRIOR iteration's rollback expires EVERY instance in # this session's identity map — the per-event failure handler # below calls ``db.rollback()``, and so does the offboarding # post-commit dispatch's inner except. ``expire_on_commit`` is # False so nothing lazily refreshes these siblings. Reading # ``staged.status`` on an expired sibling would then emit a SYNC # attribute reload on the AsyncSession → ``MissingGreenlet``, which # escapes this loop and aborts the whole apply (batch stranded in # 'applying' until the resume cron, or in 'pending_review' forever # on the scheduled path). Refresh a row that has been expired so # the read below is always against live state. Cheap: only the # rows following a genuine rollback are expired. if sa_inspect(staged).expired: await self.db.refresh(staged) # Already-processed rows short-circuit with no DB work (keeps # resume cheap when a prior run got partway through). if staged.status != "pending": events_skipped += 1 continue # Capture the PK up front: the per-event except handler rolls the # session back (expiring ``staged``), so reading ``staged.id`` # there would trigger a lazy reload on an expired instance # (MissingGreenlet under async). The failure marker re-fetches by # this cached id instead. staged_id = staged.id # #1740: cache the scalar fields read AFTER a mid-iteration # rollback too. The offboarding post-commit dispatch (below) # rolls back on failure, expiring ``staged``; the counting block # further down then reads ``staged.action`` — an expired sync # load. Snapshot it here while the instance is live. staged_action = staged.action # Pre-launch P2 fix (cancel-mid-loop race): re-fetch the # batch status before each event so a concurrent # ``cancel_batch`` call that landed between Phase A commit # and now is honored. The previous implementation kept # processing the loop using the in-memory ``batch`` snapshot # and silently confirmed events for an already-cancelled # batch. ``cancel_batch`` itself only flips the column # when status == 'pending_review' — so this re-fetch is # the symmetric guard on the apply side. await self.db.refresh(batch, attribute_names=["status"]) if batch.status == "cancelled": logger.info( "hr.confirm_batch: detected cancellation mid-loop " "for batch %s — stopping; processed events keep " "their committed state, remaining stay 'pending'.", batch.id, ) break # Liveness heartbeat — refreshed every ``heartbeat_every`` # events so the restart-safety reconcile cron treats this run # as alive. Flushed with the per-event commit below (no extra # round-trip). ``refresh`` above only reloads ``status``, so # the pending heartbeat write survives until that commit. processed_since_beat += 1 if processed_since_beat >= heartbeat_every: batch.apply_heartbeat_at = datetime.now(UTC) processed_since_beat = 0 try: # Onboard-skip means the lifecycle event was already # recorded — but scalar fields (job_title, manager, # dept, company_code, …) may still have drifted since # we stopped selecting them on the SF side. Let these # flow through ``_apply_event`` so the latest snapshot # is re-applied; ``_upsert_lifecycle_event`` is # idempotent so no duplicate event rows get created, # and the post-loop provisioning/offboarding hooks # keep their explicit action allow-lists so skip still # doesn't fire welcome emails or offboarding scans. # Offboard-skip keeps its original meaning (person # already fully ``left``) — genuinely nothing to do. if staged.action == "skip": if staged.event_type == "offboard": staged.status = "skipped" events_skipped += 1 await self.db.commit() continue # fall through as a refresh-only pass employment = dict(staged.employment_data or {}) person_data = dict(staged.person_data or {}) # Reconstruct user dict from staged person_data for _apply_event user = { "userId": staged.user_id, "email": person_data.get("work_email"), "displayName": person_data.get("display_name"), # Keep SF's authoritative structure across preview→confirm. # Dropping it allowed displayName=Unknown to become unknown@. "firstName": person_data.get("first_name"), "lastName": person_data.get("last_name"), "empId": person_data.get("employee_no"), "lastModifiedDateTime": person_data.get("source_updated_at"), # Preference staged at preview time; mobile + personal # email are looked up by personIdExternal in # ``_upsert_person`` from the master-data cache # (migration 048). "defaultLocale": person_data.get("default_locale"), } jobs = employment.get("_jobs_snapshot", []) # A start-date correction is an ``update`` event, not a new # hire. Preserve the pre-apply value so we can synchronize # the existing onboarding process without widening the # new-hire provisioning / welcome-email trigger below. prior_hire_date = None if ( staged.event_type == "onboard" and staged.action == "update" and staged.existing_person_id is not None ): prior_hire_date = await self.db.scalar( select(IdentityPerson.hire_date).where( IdentityPerson.id == staged.existing_person_id, ) ) event_created, person_upserted, _, prior_trigger_source = await self._apply_event( event_type=staged.event_type, employment=employment, user=user, jobs=jobs, trigger_source=( "initial_seed" if is_initial_seed else "sync_confirm" ), # Persist HR's 邮箱名 onto the golden record on EVERY apply, # not only when the coordinator gate below admits the # action. An onboard ``skip`` reaches here as a # refresh-only pass — which is precisely the tick that # carried P10026's late-filled ``nick.muniz`` for the # nine days before his hire date, while the create-only # reserve gate ignored every one of them. email_name=person_data.get("email_name"), ) # NOTE (#2220): ``staged.status`` is flipped to # ``confirmed`` only AFTER the coordinators have run — see # below. The pre-dispatch commit used to carry # ``confirmed`` with it, so a dispatch failure produced a # row that read "failed, retry next sync" while the batch # summary counted it as confirmed. if staged.user_id: # #1785: capture the value now (instance is live); the # post-commit leave refresh reads this list, not the # possibly-expired ORM instances. confirmed_user_ids.append(staged.user_id) # Issue #623: detect role change (dept / manager / title) # AFTER _apply_event so the "after" snapshot reflects the # just-applied state. We only fire for update/reactivate # events on existing persons (new hires can't be role- # changing; offboards are handled by their own coordinator). # The coordinator itself swallows exceptions so this won't # abort the batch. if ( not is_initial_seed and staged.event_type == "onboard" and staged.action in {"update", "reactivate"} and staged.existing_person_id is not None ): prior = person_data.get("_prior_role") or {} new_dept = (jobs[0] if jobs else {}).get("department") new_mgr_hris = (jobs[0] if jobs else {}).get("managerId") # Resolve the new manager's IdentityPerson.id so the # before/after blob compares apples to apples # (prior is stored as Helix person UUID, not HRIS id). new_mgr_person_id: str | None = None if new_mgr_hris: # Must resolve the SAME way ``_assign_manager`` # does (alias-aware), or a manager referenced by a # secondary userId reads as "manager removed" here # while the person row got it right — a phantom # role-change fired on every sync. from app.identity.identity_reconcile import ( resolve_person_by_hris_user_id, ) _new_mgr = await resolve_person_by_hris_user_id( self.db, str(new_mgr_hris), hris_instance_id=self.hris_instance_id, ) new_mgr_person_id = str(_new_mgr.id) if _new_mgr else None # Title: read the flag STAGED alongside manager and # dept rather than re-deriving it here. The old # expression compared ``person_data["job_title"]`` # against the prior title, but ``_stage_event`` never # wrote that key — so it was a constant ``False`` and # every before/after blob recorded ``job_title: null`` # on both sides. The single role_change row that # reached production is that bug's fingerprint. # # ``.get`` with a legacy fallback: staged rows created # before migration 565 carry neither key, and a # resumed batch must not crash on them. new_title = person_data.get("job_title") title_changed = bool(person_data.get("title_changed")) role_changed = ( person_data.get("manager_changed") or person_data.get("dept_changed") or title_changed ) if role_changed: from app.core.coordinators import role_change_coordinator await role_change_coordinator.handle_role_change( self.db, staged.existing_person_id, before=prior, after={ "dept_code": new_dept, "job_title": new_title, "manager_person_id": new_mgr_person_id, }, actor_user_id=None, ) # Onboarding hooks (provision, Google account, welcome # email, identity-sync push) and offboarding start are # delegated to the shared ``_dispatch_lifecycle_coordinators`` # helper so the webhook path and this sync confirm path # exercise identical downstream logic. # # Reconciliation gate — fires coordinators when ANY of: # # 1. ``event_created=True`` — brand-new lifecycle row; # no prior path has handled this yet. # 2. Prior row was ``initial_seed`` AND this isn't — # bootstrap full-sync intentionally suppresses # dispatch (line 581 below); a real webhook / # confirm arriving later MUST upgrade it, otherwise # a future hire bootstrapped today never gets their # welcome email when SF fires the real Employee # Hire event on hire_date. # # Without (2), ``event_created=False`` would silently # skip dispatch and the seeded employee lives forever # without provisioning. ``_upsert_lifecycle_event`` # upgrades the row's ``trigger_source`` in-place so # the second arrival sees the post-upgrade state. # # Without (1), every follow-up incremental sync that # walks the same ``lastModifiedDateTime`` window as a # prior webhook would re-touch every coordinator hook # (most are idempotent but each call still writes an # audit row + start_process touch). upgraded_from_seed = ( prior_trigger_source == "initial_seed" and not is_initial_seed ) # 3. The ledger row exists but its coordinators have never # run (#2220): dispatch is committed-then-fired, so a # raise (transient 1205, blocker-scan AppException) or a # process kill between the two leaves a durable row that # ``event_created`` can only ever report as False. Without # this term the staged row's "failed — retry next sync" # marker is a lie: the retry re-applies scalar fields and # dispatches nothing, so the leaver keeps # ``leaver_pending`` with no OffboardingProcess (no # T-7/T-3/T-1, no Leaver-Blocked, no force-logout) and a # new hire never gets the SF e-mail write-back / welcome # mail. Migration 439 backfilled history, so this only # ever fires for genuinely un-dispatched rows. pending_dispatch_event_id = self._consume_pending_dispatch() dispatch_suppressed = self._dispatch_is_suppressed() offboarding_dispatch_target: tuple[UUID, UUID] | None = None if not dispatch_suppressed and ( event_created or upgraded_from_seed or pending_dispatch_event_id is not None ): # Commit the person upsert BEFORE the coordinators run # (#2130). They make external calls inline — the SF # e-mail write-back, the Google Admin account # creation, the Day-1 Feishu + e-mail notification — # and ``identity_persons`` is the FK convergence # point of the schema, so holding its row lock # across three vendor round-trips is the #1837 # shape: any concurrent login's # ``UPDATE identity_persons`` queues behind it. # The webhook path has always been this way — its # comment reads "The person upsert is already # committed (above)" — and confirm_batch was the # copy that never caught up. Committing first also # makes a vendor failure survivable: the person # exists, and provisioning retries on its own. await self.db.commit() offboarding_dispatch_target = ( await self._dispatch_lifecycle_coordinators( event_type=staged.event_type, action=staged.action, hris_user_id=staged.user_id, employment=employment, is_initial_seed=is_initial_seed, # HR-assigned corporate 邮箱名 (SF # PerPersonal.customString4, custom07 fallback), # staged into person_data by ``_stage_event`` — # reserve uses it as the top-priority local part # (same value the webhook path receives as # ``emailName``). NULL ⇒ given.family fallback. email_name=person_data.get("email_name"), ) ) # Coordinators ran — record it on the ledger row so a # later apply of the same event does NOT re-dispatch # (#2220). Skipped for the bootstrap seed pass, which # deliberately suppresses dispatch inside # ``_dispatch_lifecycle_coordinators``: leaving the row # un-stamped is what lets the first real webhook / # confirm still fire it. if not is_initial_seed: await self._mark_lifecycle_dispatched( pending_dispatch_event_id, ) # An onboard ``update`` normally has no lifecycle dispatch # (correctly: it is not a second new hire). But when HRIS # changed the hire date, that would otherwise leave the # existing onboarding row on the old date — and a previously # due row may already have created accounts. Refresh only # the state machine; no welcome message or provisioning is # re-fired here. if ( not is_initial_seed and staged.event_type == "onboard" and staged.action == "update" and prior_hire_date is not None ): current_hire_date = await self.db.scalar( select(IdentityPerson.hire_date).where( IdentityPerson.id == staged.existing_person_id, ) ) if current_hire_date != prior_hire_date: from app.core.coordinators import hr_sync_coordinator await hr_sync_coordinator.refresh_onboarding_for_hire_date_change( self.db, staged.user_id, ) # Everything for this event succeeded, including dispatch — # only now is the staged row genuinely confirmed (#2220). staged.status = "confirmed" # Commit this event's work — releases identity_persons # and hr_lifecycle_events row locks immediately. await self.db.commit() # Pre-launch P0 fix (Round 1, LG-003): force-logout + # leaver/manager notifications fire AFTER the per-event # commit so the offboarding row is durable before any # external side effect runs. Failures here never raise — # they are logged and the next reconcile cron picks up # any partial state. Without this, every HRIS-driven # leaver received zero notifications and zero force- # logout — only the manual ``mark_person_leaver`` router # path was wired. Mirrors the manual path in # ``hr/router.py:1170``. if offboarding_dispatch_target is not None: _person_id, _proc_id = offboarding_dispatch_target try: from app.domains.offboarding.service import ( # noqa: PLC0415 OffboardingService, ) _off_svc = OffboardingService(self.db) await _off_svc.dispatch_post_commit_actions( _person_id, _proc_id, trace_id=str(_person_id), actor_user_id=None, ) await self.db.commit() except Exception: # noqa: BLE001 # Round 2 P2 fix (LG2-005): the original # comment claimed "Reconcile cron will retry" # but no such reconcile pass exists for # missing ``process_started_*`` reminder logs. # The escalation cron only handles T-7/T-3/ # T-1/REMINDER_ESCALATION; SLA service handles # warn/urgent/breach. Start-of-process # notifications have no automatic retry path. # Operator must use the offboarding admin # console's "Re-notify" action (HTTP route # ``POST /offboarding/processes/{id}/re-notify``) # which calls dispatch_post_commit_actions # directly. The process row itself remains # durable; only the start-of-cycle notifications # are at risk. logger.exception( "confirm_batch: HRIS offboarding post-commit " "dispatch failed for staged_event=%s person=%s " "process=%s — process is durable; force-logout " "/ notifications may be partial. NO automatic " "retry — operator must hit the offboarding " "console 'Re-notify' button.", staged.id, _person_id, _proc_id, ) try: await self.db.rollback() except Exception: # noqa: BLE001 pass events_confirmed += 1 people_upserted += int(person_upserted) lifecycle_events_created += int(event_created) # ``staged_action`` (cached above) instead of ``staged.action`` # — the offboarding post-commit dispatch may have rolled back # and expired ``staged`` by this point (#1740). if staged_action == "create": new_count += 1 elif staged_action == "reactivate": reactivated_count += 1 elif staged_action == "deactivate": deactivated_count += 1 elif staged_action == "update": updated_count += 1 except Exception as exc: # noqa: BLE001 # Per-event failure isolates to one employee; roll back # the current event's partial work, record the error on # the staged row, commit that marker, keep going. await self.db.rollback() logger.error( "confirm_batch event %s failed for batch %s: %s", staged_id, batch_id, exc, exc_info=True, ) error_count += 1 last_event_error = str(exc)[:1024] try: # Re-fetch the staged row in a fresh state so we can # mark it failed without the rollback-poisoned handle. # The per-event error is persisted on the staged row's # error_message column (migration 385) so admins can see # who/why failed without grepping the worker log. fresh = await self.db.get(HrSyncStagedEvent, staged_id) if fresh is not None: fresh.status = "failed" fresh.error_message = last_event_error # Advance the liveness heartbeat in this same committed # transaction. The in-memory bump at the top of the loop # is discarded by the rollback above, so without this a # run of consecutive failures (incl. right at a heartbeat # boundary) would let ``apply_heartbeat_at`` go stale # (>10 min) while the apply is genuinely alive — the # restart-safety cron would then start a redundant # concurrent resume. Stamping on every failed marker # closes the gap regardless of cadence. fresh_batch = await self.db.get(HrSyncBatch, batch_id) if fresh_batch is not None: fresh_batch.apply_heartbeat_at = datetime.now(UTC) processed_since_beat = 0 await self.db.commit() except Exception: # noqa: BLE001 await self.db.rollback() logger.warning( "confirm_batch could not mark staged %s as failed", staged_id, ) # Manager backfill — re-link any active person whose declared # ``manager_external_id`` finally resolves to a live Helix row # (e.g. their manager onboarded in this very batch, or in a # prior batch after this person's last event cleared the # pointer). Per-event ``_assign_manager`` clears on a miss; this # tail pass is what closes the gap instead of leaving it NULL # forever. Idempotent + best-effort: a failure here must not fail # an otherwise-confirmed batch. try: from app.identity.identity_reconcile import backfill_manager_links linked = await backfill_manager_links( self.db, hris_instance_id=self.hris_instance_id, ) # Commit the backfill's flushed links here (it only flushes, no # commit) so they're durable BEFORE the role-refresh tail pass # below — otherwise a role-refresh failure's ``rollback()`` would # also discard this pass's freshly-linked managers. await self.db.commit() if linked: logger.info( "hr.sync: manager-backfill linked %d late-arriving " "manager(s) after batch %s", linked, batch_id, ) except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "hr.sync: manager-backfill pass failed for batch %s — " "non-fatal; next sync retries", batch_id, ) # Phase B.2 — refresh HRBP / Budget-Owner role pointers from SF # ``EmpJobRelationships`` (工作关系) over EVERY active SF person, not # just this batch's diff. These columns feed the 汇联易 department # sync (per-BU member-shared HRBP / Budget-Owner). The relationship # data is org-wide and independent of which person changed in this # batch, so a per-event resolution would leave unchanged members' # columns NULL forever — hence the whole-roster tail pass here, # mirroring the manager-backfill above. One SF pull + one short # transaction. Best-effort: a pull/permission failure must not fail # an otherwise-confirmed batch (the next sync retries). try: from app.identity.successfactors_sync import ( apply_hris_role_relationships, ) client = await self._get_client() roster_result = await self.db.execute( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ) ) roster = list(roster_result.scalars().all()) refreshed = await apply_hris_role_relationships( self.db, client, roster, ) await self.db.commit() if refreshed: logger.info( "hr.sync: HRBP / Budget-Owner refresh updated %d " "person(s) after batch %s", refreshed, batch_id, ) except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "hr.sync: HRBP / Budget-Owner refresh failed for batch %s " "— non-fatal; next sync retries", batch_id, ) # Phase B.3 — derive department-based platform roles (Finance BU ⇒ # ``finance``, HR ⇒ ``hr``) over the whole roster. Like the HRBP # refresh above, derivation is an org-wide function of primary org # membership — a person's role can change because a *peer* moved or # because this batch onboarded/transferred them, so a per-event hook # would leave unchanged members stale. The standalone # ``SuccessFactorsSyncService._upsert_primary_membership`` carries a # per-person hook for the (not-yet-live) scheduled-sync path; the # live admin-confirm path runs the full reconciler here so derivation # is real-time instead of waiting up to a day for the # ``identity.department_role_reconcile`` backstop cron. Idempotent # (only touches ``grant_source='derived'`` rows, never manual grants) # + best-effort: a failure here must not fail an otherwise-confirmed # batch (the daily cron retries and pages on missing role rows). try: from app.identity.department_role_sync import ( reconcile_department_roles, ) dept_role_summary = await reconcile_department_roles(self.db) await self.db.commit() if dept_role_summary.get("role_missing"): logger.warning( "hr.sync: department-role derivation skipped after batch " "%s — missing role rows %s (daily cron will page)", batch_id, dept_role_summary["role_missing"], ) elif dept_role_summary.get("granted") or dept_role_summary.get( "revoked" ): logger.info( "hr.sync: department-role derivation granted %d / revoked " "%d after batch %s", dept_role_summary.get("granted", 0), dept_role_summary.get("revoked", 0), batch_id, ) except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "hr.sync: department-role derivation failed for batch %s — " "non-fatal; daily cron retries", batch_id, ) # Phase C — finalize batch status in a short transaction. batch = await self.db.get(HrSyncBatch, batch_id) if batch is None: # paranoia: batch was deleted mid-run raise NotFoundError("HrSyncBatch", str(batch_id)) if batch.status == "cancelled": # A concurrent cancel_batch landed mid-apply (the loop above broke # on detection). Don't let Phase C clobber the cancellation with # confirmed/failed — finalize as-is. Events committed before the # break keep their state; the remainder stay 'pending'. await self.db.commit() return { "batch_id": str(batch_id), "status": "cancelled", "events_confirmed": events_confirmed, "events_skipped": events_skipped, "people_upserted": people_upserted, "lifecycle_events_created": lifecycle_events_created, "new_count": new_count, "updated_count": updated_count, "reactivated_count": reactivated_count, "deactivated_count": deactivated_count, "error_count": error_count, } # Resume-safe terminal status: derive from the FINAL staged-event # states across ALL runs, not this run's counters. On a resumed # apply, events confirmed by a prior run are 'confirmed' in the DB # but fast-skip this run (counted in events_skipped, not # events_confirmed) — so a counter-only check would wrongly flip a # mostly-successful batch to 'failed' when only the leftover events # fail on resume. ``failed`` iff zero events ever confirmed AND at # least one ended failed; otherwise ``confirmed`` even with partial # errors (surfaced via error_count + per-row error_message). state_counts = dict((await self.db.execute( select(HrSyncStagedEvent.status, func.count()) .where(HrSyncStagedEvent.batch_id == batch_id) .group_by(HrSyncStagedEvent.status) )).all()) if state_counts.get("confirmed", 0) == 0 and state_counts.get( "failed", 0 ) > 0: batch.status = "failed" batch.error_message = (last_event_error or "all events failed")[:1024] else: batch.status = "confirmed" batch.reviewed_at = datetime.now(UTC) batch.reviewed_by = confirmed_by batch.new_count = new_count batch.updated_count = updated_count batch.reactivated_count = reactivated_count batch.deactivated_count = deactivated_count batch.skipped_count = events_skipped batch.error_count = error_count await self.db.commit() result: dict = { "batch_id": str(batch_id), # Reflect the real terminal status — a batch where every event # errored finalizes as ``failed`` (set just above), not # ``confirmed``. The old hard-coded "confirmed" lied on that path. "status": batch.status, "events_confirmed": events_confirmed, "events_skipped": events_skipped, "people_upserted": people_upserted, "lifecycle_events_created": lifecycle_events_created, "new_count": new_count, "updated_count": updated_count, "reactivated_count": reactivated_count, "deactivated_count": deactivated_count, "error_count": error_count, } # Post-commit: refresh annual-leave carryover for every userId # touched in this batch. Pulled here (not in ``_upsert_person``) # because SF's ``EmpTimeAccountBalance`` rejects ``or`` in # ``$filter`` (``COE_HRIS_QUERY_USING_OR_IN_FILTER``) — single- # user calls × 500 employees would dominate the sync window. # Instead we batch ``TimeAccount`` + ``TimeAccountDetail`` and # derive balance from detail aggregation. Defensive: on tenant # without EC Time-Off, helpers return empty and the four # ``identity_persons.annual_leave_*`` columns stay as-is (HR # manual input in the 年假计算器 page is the fallback). try: # #1785: use the values captured DURING the loop, never a # post-commit re-read of the ORM instances (which a mid-batch # rollback may have expired → MissingGreenlet). De-dup while # preserving order. touched_user_ids = list(dict.fromkeys(confirmed_user_ids)) if touched_user_ids: leave_summary = await self.sync_leave_data( user_ids=touched_user_ids, ) result["leave_synced"] = leave_summary # Back-compat key — derived inside sync_leave_data now. result["annual_leave_refreshed"] = leave_summary.get( "legacy_derived", 0, ) except Exception as exc: # noqa: BLE001 — never fail the batch on this logger.warning( "confirm_batch: post-commit leave-data sync failed for " "batch %s (%s). Persons keep their previous leave rows + " "annual_leave_* values; HR can still hand-edit via the " "calculator.", batch_id, exc, ) return result async def start_async_confirm( self, batch_id: UUID, confirmed_by: str, ) -> tuple[HrSyncBatch, bool]: """Gate-check + mass-leaver brake, then flip the batch to ``applying`` and return immediately — WITHOUT running the apply. Returns ``(batch, did_flip)`` where ``did_flip`` is ``True`` only when THIS call performed the ``pending_review → applying`` transition. The caller MUST spawn the background apply only when ``did_flip`` is ``True`` — an idempotent no-op return (batch was already ``applying``) means another runner already owns the apply, and spawning a second one races it on the same staged rows (#1743: double-runner lock contention mislabels events failed + zeroes the Phase C counts). This is the interactive admin-confirm HTTP entry point. The heavy per-event apply (``_apply_staged_events``) is dispatched onto a background task by the router so the browser gets a fast ``202`` instead of blocking for minutes on a full-bootstrap batch (the 2026-06-16 prod incident: the synchronous apply outran the axios timeout → nginx ``499`` → misleading error toast). The brake runs HERE, synchronously, so its refusal reaches the admin as an immediate ``409`` (mass-leaver protection must never be swallowed into a background task). Idempotent: a second click while the apply is in flight sees ``status='applying'`` and returns the batch as-is (the caller treats that as "already running", not an error). A single winner makes the ``pending_review → applying`` transition under ``SELECT FOR UPDATE``, guaranteeing exactly one background runner from this path. """ result = await self.db.execute( select(HrSyncBatch).where(HrSyncBatch.id == batch_id).with_for_update() ) batch = result.scalar_one_or_none() if batch is None: raise NotFoundError("HrSyncBatch", str(batch_id)) if batch.status == "applying": # Already in flight — idempotent no-op. Release the lock and # hand the current row back so the UI keeps polling. # ``did_flip=False``: another runner owns the apply, so the # caller must NOT spawn a second background task (#1743). await self.db.commit() return batch, False if batch.status != "pending_review": raise ConflictError(f"Batch {batch_id} is already {batch.status}") # Volume brakes — identical gate to ``confirm_batch`` so the # async path can't bypass it. await self._assert_batch_volume_is_sane(batch) sync_mode = (batch.sync_mode or "").lower() now = datetime.now(UTC) batch.status = "applying" batch.apply_started_at = now batch.apply_heartbeat_at = now # Stamp the human now so the resume-cron can preserve attribution # even if the background task dies before Phase C. batch.reviewed_by = confirmed_by await self.db.commit() await self.db.refresh(batch) logger.info( "HRIS batch %s flipped to 'applying' by %s (sync_mode=%s, " "is_seed=%s, events_staged=%d) — apply dispatched to " "background task.", batch.id, confirmed_by, sync_mode, bool(batch.is_seed), batch.events_staged, ) return batch, True async def resume_stale_applying_batches( self, *, stale_after: timedelta = timedelta(minutes=10), limit: int = 5, ) -> dict: """Find ``applying`` batches whose apply heartbeat went stale (the background task's backend process died mid-run) and resume them. A *healthy* in-flight apply refreshes ``apply_heartbeat_at`` every few events, so a heartbeat older than ``stale_after`` means the runner is gone. Resuming re-enters the idempotent ``_apply_staged_events`` — finished events fast-skip, the rest complete, and the batch finalizes. The original ``reviewed_by`` is preserved across the resume. Restart-safety backstop only — the happy path starts the apply immediately via ``background.spawn`` (see the router). Mirrors the inline-primary + cron-backstop shape of migration 239 / PR #279. """ cutoff = datetime.now(UTC) - stale_after rows = await self.db.execute( select( HrSyncBatch.id, HrSyncBatch.reviewed_by, HrSyncBatch.hris_instance_id, ) .where(HrSyncBatch.status == "applying") .where( or_( HrSyncBatch.apply_heartbeat_at < cutoff, and_( HrSyncBatch.apply_heartbeat_at.is_(None), or_( HrSyncBatch.apply_started_at < cutoff, HrSyncBatch.apply_started_at.is_(None), ), ), ) ) .order_by(HrSyncBatch.apply_started_at) .limit(limit) ) stale = rows.all() resumed = 0 failed = 0 for batch_id, reviewed_by, batch_instance_id in stale: confirmed_by = reviewed_by or "system:cron:resume_stuck_apply" logger.warning( "HRIS resume_stuck_apply: batch %s stale in 'applying' " "(heartbeat older than %s) — resuming idempotent apply " "as %s.", batch_id, stale_after, confirmed_by, ) # Isolate each batch: one failing resume must not block its # siblings this tick. The apply is idempotent, so a batch that # errors here (and stays 'applying' with a stale heartbeat) is # simply retried on the next 60s tick. try: # Resume on a service scoped to the batch's own HRIS # instance. The cron constructs this service with # ``hris_instance_id=None``, but ``_apply_staged_events``' # org-wide tail passes (manager-backfill, HRBP/Budget-Owner # refresh) filter by ``self.hris_instance_id`` — running them # unscoped would touch other instances' rows in a # multi-instance deployment. Per-event apply is batch-scoped # so it's unaffected either way. scoped = ( self if batch_instance_id == self.hris_instance_id else SuccessFactorsLifecycleService( self.db, hris_instance_id=batch_instance_id ) ) await scoped._apply_staged_events(batch_id, confirmed_by) resumed += 1 except Exception: # noqa: BLE001 failed += 1 logger.exception( "HRIS resume_stuck_apply: resume failed for batch %s — " "left 'applying' for the next tick to retry.", batch_id, ) return {"stale_found": len(stale), "resumed": resumed, "failed": failed} async def sync_annual_leave_carryover( self, *, user_ids: list[str] | None = None, ) -> int: """Refresh the four ``identity_persons.annual_leave_*`` columns from SF's EC Time-Off ledger. Carries forward the **prior fiscal year**'s residual balance (= carry_legal_days / carry_welfare_days) and the **current fiscal year**'s used-days total (= used_legal_days / used_welfare_days) for every userId in ``user_ids`` (defaults to every active person scoped to this service's HRIS instance when ``None``). Algorithm — single round-trip per ~16 users: 1. Batched ``TimeAccount`` fetch (50 users/call) → all leave accounts for {legal, welfare} ``accountType``. 2. Batched ``TimeAccountDetail`` fetch (16 uuids/call) → individual booking lines per account. 3. Per (user, accountType): pick PRIOR FY (endDate <= today) and CURRENT FY (startDate <= today < endDate) accounts. Sum ``bookingAmount`` for PRIOR FY → carry-forward balance (matches what ``EmpTimeAccountBalance`` would return). Sum |bookingAmount| of leave-taken booking types (``EMPLOYEE_TIME`` / ``DEDUCTION``) on CURRENT FY → used. 4. Write the four Decimal columns; commit once at the end. Returns the number of ``identity_persons`` rows whose annual-leave columns were touched. Zero is a valid outcome (tenant without EC Time-Off, empty user list, all NULL data). """ from decimal import Decimal as _Decimal legal_type = (settings.HR_TIMEOFF_LEGAL_ACCOUNT_TYPE or "").strip() welfare_type = (settings.HR_TIMEOFF_WELFARE_ACCOUNT_TYPE or "").strip() account_types = [t for t in (legal_type, welfare_type) if t] if not account_types: return 0 # Resolve which userIds to refresh + the Helix person rows # to write back to (scoped to this service's HRIS instance so # CN sync can't accidentally overwrite EU rows or vice versa). person_query = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.deleted_at.is_(None), ) if self.hris_instance_id is not None: person_query = person_query.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) if user_ids is not None: if not user_ids: return 0 person_query = person_query.where( IdentityPerson.hris_person_id.in_(user_ids) ) person_rows = (await self.db.execute(person_query)).scalars().all() person_by_user_id: dict[str, IdentityPerson] = { p.hris_person_id: p for p in person_rows if p.hris_person_id } if not person_by_user_id: return 0 client = await self._get_client() accounts = await client.list_time_accounts( user_ids=list(person_by_user_id.keys()), account_types=account_types, ) if not accounts: return 0 # Parse SF OData v2 epoch ``/Date(milliseconds)/`` literals # into ``date``. Returns None for the missing / un-parseable. def _parse_sf_date(value: object) -> date | None: if not isinstance(value, str): return None import re m = re.match(r"/Date\((-?\d+)(?:[+-]\d+)?\)/", value) if not m: return None return datetime.fromtimestamp(int(m.group(1)) / 1000, tz=UTC).date() # Index accounts: (user_id, account_type, role) → uuid where # role ∈ {"current", "prior"}. Skip ambiguous / overlapping rows. today_date = self.today # bookingType values that count as leave actually taken (used). # Both PingCAP CN tenant codes (``EMPLOYEE_TIME`` is auto-created # when an employee submits a time-off request; ``DEDUCTION`` is # the manual-deduction variant). ``ACCRUAL`` / ``MANUAL_ADJUSTMENT`` # are entitlement entries and do NOT count as used. used_booking_types = {"EMPLOYEE_TIME", "DEDUCTION"} # Group accounts per (user, accountType, role) — pick the one # whose range covers ``today`` as "current", the one with the # latest endDate strictly before today as "prior". Ignore any # other accounts (could be a third grace-period bucket on a # rolling tenant; not modelled here). grouped: dict[tuple[str, str], list[dict]] = {} for acct in accounts: user_id = acct.get("userId") atype = acct.get("accountType") if not user_id or not atype: continue grouped.setdefault((user_id, atype), []).append(acct) # account_uuid → (user_id, accountType, role) uuid_to_role: dict[str, tuple[str, str, str]] = {} for (user_id, atype), bucket in grouped.items(): current_acct: dict | None = None prior_acct: dict | None = None for acct in bucket: start = _parse_sf_date(acct.get("startDate")) end = _parse_sf_date(acct.get("endDate")) uuid = acct.get("externalCode") if not uuid or start is None or end is None: continue if start <= today_date <= end: # Take the most recently-starting current — usually # only one matches but defensive against overlap. if current_acct is None or start > _parse_sf_date( current_acct.get("startDate") ): # type: ignore[operator] current_acct = acct elif end < today_date: if prior_acct is None or end > _parse_sf_date( prior_acct.get("endDate") ): # type: ignore[operator] prior_acct = acct if current_acct is not None: uuid_to_role[current_acct["externalCode"]] = ( user_id, atype, "current", ) if prior_acct is not None: uuid_to_role[prior_acct["externalCode"]] = ( user_id, atype, "prior", ) if not uuid_to_role: return 0 # Fetch all relevant detail lines in one batched call. details = await client.list_time_account_details( time_account_uuids=list(uuid_to_role.keys()), ) # Aggregate per (user_id, accountType, role). balance_acc: dict[tuple[str, str, str], _Decimal] = {} used_acc: dict[tuple[str, str, str], _Decimal] = {} for d in details: uuid = d.get("TimeAccount_externalCode") if uuid not in uuid_to_role: continue role_key = uuid_to_role[uuid] try: amount = _Decimal(str(d.get("bookingAmount") or "0")) except Exception: # noqa: BLE001 — defensive on tenant drift continue balance_acc[role_key] = balance_acc.get(role_key, _Decimal(0)) + amount if d.get("bookingType") in used_booking_types: used_acc[role_key] = used_acc.get(role_key, _Decimal(0)) + abs(amount) # Write to identity_persons. ``carry_*`` = prior-FY balance, # ``used_*`` = current-FY used. Round to 1 decimal place to # match the Numeric(4,1) column precision. def _q(v: _Decimal) -> _Decimal: return v.quantize(_Decimal("0.1")) rows_touched = 0 for user_id, person in person_by_user_id.items(): carry_legal = balance_acc.get((user_id, legal_type, "prior")) carry_welfare = balance_acc.get((user_id, welfare_type, "prior")) used_legal = used_acc.get((user_id, legal_type, "current")) used_welfare = used_acc.get((user_id, welfare_type, "current")) changed = False if carry_legal is not None: new_val = _q(carry_legal) if person.annual_leave_carry_legal_days != new_val: person.annual_leave_carry_legal_days = new_val changed = True if carry_welfare is not None: new_val = _q(carry_welfare) if person.annual_leave_carry_welfare_days != new_val: person.annual_leave_carry_welfare_days = new_val changed = True if used_legal is not None: new_val = _q(used_legal) if person.annual_leave_used_legal_days != new_val: person.annual_leave_used_legal_days = new_val changed = True if used_welfare is not None: new_val = _q(used_welfare) if person.annual_leave_used_welfare_days != new_val: person.annual_leave_used_welfare_days = new_val changed = True if changed: rows_touched += 1 if rows_touched: await self.db.commit() return rows_touched async def sync_leave_data( self, *, user_ids: list[str] | None = None, commit: bool = True, ) -> dict: """Mirror SF EC Time-Off (all account types + every posting line) into ``hr_leave_accounts`` / ``hr_leave_postings`` and derive the four legacy ``annual_leave_*`` columns from the result. Supersedes :meth:`sync_annual_leave_carryover` for the post-sync refresh: that method only ever populated the four snapshot columns for {法定年假, 福利年假}; this one stores the full self-service- visible leave dataset and derives those same columns from it, so a single SF round-trip feeds both surfaces. Thin delegate to :func:`app.domains.hr.leave_service.sync_leave_data` (kept there so this large service file stays focused on the lifecycle saga). """ from app.domains.hr.leave_service import sync_leave_data return await sync_leave_data(self, user_ids=user_ids, commit=commit) # ---- Full-employee sync (seed all active users) ---- async def _full_sync_is_seed(self) -> bool: """Is a full sync of THIS HRIS instance an initial seed? Seed = Helix holds no people for the instance, so the roster arriving in one batch is history rather than news. **The two error directions are not symmetric, and the predicate is built around that.** Calling a real seed "routine" fires the coordinators over an imported roster — hundreds of real accounts, and an offboarding process per historical leaver. Calling a routine sync a "seed" is the bug migration 565 exists to fix: bad, but recoverable, and now also caught by the reconciler. So every ambiguous case resolves to *seed*. That is why the scope is a strict ``hris_instance_id`` match with no legacy-NULL fallback (unlike ``_count_active_persons_for_instance``, whose NULL-inclusive scope is right for a brake denominator): a newly-registered instance in an install that already holds unscoped legacy rows must still seed. Verified against production — both live instances carry the column (sf-cn-prod 505 alive, sf-eu-prod 37), so both correctly read as routine. Deleted people are excluded: an instance whose entire roster was removed and is being re-imported is, for coordinator purposes, a seed again. A ``hr_lifecycle_events`` clause was considered and dropped — events hang off persons by FK, so "no persons for this instance" already implies "no events for this instance". It would have been a second query that could never change the answer. """ return await self._count_persons_for_instance() == 0 async def _count_persons_for_instance(self) -> int: """Live people on this HRIS instance, whatever their employment status. Distinct from ``_count_active_persons_for_instance`` (the volume brakes' denominator) on two axes: leavers count here (a roster of departed people is still prior state), and the instance scope admits no legacy-NULL fallback — see ``_full_sync_is_seed``. """ stmt = ( select(func.count()) .select_from(IdentityPerson) .where(IdentityPerson.deleted_at.is_(None)) ) if self.hris_instance_id is not None: stmt = stmt.where( IdentityPerson.hris_instance_id == self.hris_instance_id, ) return int(await self.db.scalar(stmt) or 0) async def preview_full_employee_sync( self, *, triggered_by: str, is_seed: bool | None = None, ) -> HrSyncBatch: """Pull every SF user (active + forfeited) and stage one event per user. Unlike :meth:`preview_import` (which is bounded by a date window over hire / termination events), this path guarantees every SF-known employee is represented in ``identity_persons`` after the admin confirms the batch — the right tool for seeding a fresh production environment or reconciling drift. ``is_seed`` (migration 565) decides whether the batch behaves as a historical import (coordinators suppressed, mass-leaver brake exempt) or as routine reconciliation (coordinators fire exactly as they do for an incremental tick). ``None`` — the default — resolves it with :meth:`_full_sync_is_seed`. Pass it explicitly when a human is deliberately re-importing history; the word "full" must not make that call on their behalf, which is the bug this parameter exists to retire. Status routing: * ``user.status == 't'`` → stage an ``onboard`` event with ``endDate=None`` (the existing onboard handler creates the person with ``employment_status='active'``). * ``user.status == 'f'`` → stage an ``offboard`` event with ``endDate`` populated from ``EmpEmployment``. On a SEED batch the confirm loop fires no offboarding side effects (the ``not is_initial_seed`` gate); on a routine full sync it does, which is the point — a termination outside the incremental 30-day window reaches Helix through no other door. ``_upsert_person`` lands historical leavers (endDate ≥ 30 days past) at ``employment_status='left'`` directly via its INSERT-time cutoff — no separate sweep needed because the ``leaver_blocked.listener`` explicitly exempts ``session.new`` INSERTs (see ``leaver_blocked/listener.py:86-89``). Recent leavers (≤ 30 days) land at ``leaver_pending`` and the ``hr.process_advance`` cron handles their final flip with in-flight offboarding actions intact. Pre-launch P0 fix (2026-05-30 wangdi case): the previous version called ``list_active_users`` which silently dropped every SF leaver, leaving 48 PingCAP ex-employees as anchor-less owners on cmdb_servers / DNS / SaaS rows. The fix calls ``list_all_users`` and routes forfeited rows through the offboard branch so they land with ``employment_status='left'`` and historical anchors can re-link. """ await self._refresh_org_snapshot() client = await self._get_client() users = await client.list_all_users() user_ids = [u["userId"] for u in users if u.get("userId")] users_by_id = {u["userId"]: u for u in users if u.get("userId")} jobs_by_user = await self._jobs_by_user_id(user_ids) if user_ids else {} # Pull EmpEmployment for ALL users (not just forfeited). Two # distinct pieces of data come from this: # - termination_date (forfeited path): latest endDate-bearing row # - work_start_date / 参加工作日期 (every path): originalStartDate # Without the second use, the active-onboard synthesis below # falls back to ``user.hireDate`` for ``originalStartDate``, # silently equating career start with hire_date for every # active employee. Caught 2026-06-02 by the 年假计算器 PR # follow-up: prod TiDB showed 430/432 active rows with # ``work_start_date == hire_date`` because the seed full-sync # was the only path that ever wrote work_start_date and never # carried the real value. The cost is bounded (~chunks of 50 # per ``get_employments_for_users``, same primitive already # used for the forfeited subset). forfeited_user_ids = { u["userId"] for u in users if u.get("userId") and str(u.get("status") or "").lower() == "f" } terminations_by_user_id: dict[str, dict] = {} employments_by_user_id: dict[str, dict] = {} # CRITICAL fetch — forfeited subset only. This sources # ``termination_date`` for the offboard branch and MUST NOT be # made best-effort: a missing termination_date corrupts leaver # detection (the whole reason this sync exists — see the # wangdi-case note in the docstring). Failures here propagate. if forfeited_user_ids: for employment in await client.get_employments_for_users( sorted(forfeited_user_ids) ): uid = employment.get("userId") if not uid: continue # Terminations dict — forfeited users only, only rows # with a real endDate. Keep the LATEST endDate per # userId — SF can hold multiple EmpEmployment rows for # re-hires; the most recent termination is the # authoritative one for the current ``status='f'`` # flag. PR-audit fix (2026-07-16): select by max parsed # ``endDate``, NOT by ``lastModifiedDateTime`` — the # incremental path already picks max(endDate) and the two # paths feed the SAME ``classify_inactive_sf_user``; a # rehire whose OLD stint row was edited more recently # than the current stint's termination would otherwise # hand the classifier a STALE endDate here (wrong # NO_TERMINATION refusal when the stale endDate precedes # the current hireDate, or a years-old termination_date # when it doesn't). Parse via the shared helpers, never # compare raw ``/Date(ms)/`` strings: lexicographic # compare disagrees with chronological order across # digit-count boundaries. Tie-break on equal endDate by # latest lastModifiedDateTime so the winner is stable # regardless of SF row ordering (missing values sort # earliest so a real value always beats a None). if employment.get("endDate"): prior_term = terminations_by_user_id.get(uid) if prior_term is None: terminations_by_user_id[uid] = employment else: cur_key = ( _parse_date(employment.get("endDate")) or date.min, _parse_datetime( employment.get("lastModifiedDateTime") ) or datetime.min.replace(tzinfo=UTC), ) prior_key = ( _parse_date(prior_term.get("endDate")) or date.min, _parse_datetime( prior_term.get("lastModifiedDateTime") ) or datetime.min.replace(tzinfo=UTC), ) if cur_key > prior_key: terminations_by_user_id[uid] = employment # BEST-EFFORT enrichment — all users. This sources # ``originalStartDate`` (work_start_date) for the active-onboard # synthesis. Unlike the forfeited fetch above it is NOT critical: # the active path already falls back to ``user.hireDate`` when a # userId is absent from ``employments_by_user_id``. A raised # chunk here previously aborted the WHOLE sync — including leaver # detection — so we swallow failures, log, and fall back to an # empty dict (every active user then uses hire_date for # work_start_date, the pre-existing behaviour). if user_ids: try: all_employments = await client.get_employments_for_users(user_ids) except Exception: # noqa: BLE001 — enrichment is best-effort logger.exception( "full_sync: get_employments_for_users (work_start_date " "enrichment) failed; falling back to hire_date for all " "active users", ) all_employments = [] for employment in all_employments: uid = employment.get("userId") if not uid: continue # All-employments dict (any row) — used downstream to # read ``originalStartDate``. SF replicates the same # value across re-hire rows of one user, so the # latest-modified row is a deterministic canonical # winner. Compare as parsed datetimes (see the comment # block above for the lex-vs-chrono /Date(ms)/ trap). cur_lmd = _parse_datetime(employment.get("lastModifiedDateTime")) prior_emp = employments_by_user_id.get(uid) if prior_emp is None: employments_by_user_id[uid] = employment else: prior_lmd = _parse_datetime(prior_emp.get("lastModifiedDateTime")) if (cur_lmd or datetime.min.replace(tzinfo=UTC)) > ( prior_lmd or datetime.min.replace(tzinfo=UTC) ): employments_by_user_id[uid] = employment # Termination-reason exclusion inputs (2026-07-30): reasons for # every forfeited user with a real endDate (cost-gated — see # ``_termination_reasons_for``) + the configured exclusion set. # Feeds the classifier so 人员分类转换 / 跨DC调动 / 兼岗结束 are # never offboarded by this nightly backstop. termination_reasons = await self._termination_reasons_for( terminations_by_user_id ) excluded_reason_codes = await self._excluded_termination_reason_codes() # Synthesize employment records from the User objects themselves # so the existing _stage_event / _apply_event pipeline can process # them the same way as true hire / termination events. today_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") synthetic_events: list[tuple[str, dict]] = [] for user in users: user_id = user.get("userId") if not user_id: continue raw_status = str(user.get("status") or "").lower() termination = terminations_by_user_id.get(user_id) or {} if not sf_status_is_active(raw_status): # SF ``status='f'`` (inactive) conflates genuine terminations # with future / day-one hires that haven't started. The # leaver-or-not verdict is centralized in # ``classify_inactive_sf_user`` (see its docstring for the # Camila P10009 / Robert Wood P10013 history and the # Asia/Shanghai ``>=`` rationale) so this full-roster path and # the incremental path can never drift again. Full sync SKIPS # every non-leaver: future hires are created by the owning # instance's onboard event, and a forfeited-without-endDate SF # data gap is surfaced by the ``hris_status_divergence`` check # for review — we never fabricate a termination date. _hire_dt = _parse_date(user.get("hireDate")) decision = classify_inactive_sf_user( hire_date=_hire_dt, as_of=self.today, termination_end_date=_parse_date(termination.get("endDate")), termination_reason_code=termination_reasons.get(user_id), excluded_reason_codes=excluded_reason_codes, ) if not decision.is_leaver: logger.info( "full_sync: not offboarding user_id=%s — %s", user_id, decision.reason, ) continue # Substantiated leaver: use the REAL SF ``endDate`` (never a # lastModifiedDateTime fallback). ``endDate >= hireDate`` is # already guaranteed by the classifier. ``startDate`` / # ``originalStartDate`` are SF history only — we MUST NOT # synthesize ``today`` (``_upsert_person`` reads startDate into # ``person.hire_date``; today would corrupt a years-old leaver). synthetic_events.append(( "offboard", { "userId": user_id, # Real person key off the EmpEmployment row in hand; # ``user_id`` only as fallback. Same rationale as the # onboard branch below. "personIdExternal": ( termination.get("personIdExternal") or user_id ), "_person_id_external_is_user_id": not termination.get( "personIdExternal" ), "startDate": ( termination.get("startDate") or termination.get("originalStartDate") or user.get("hireDate") ), "endDate": termination.get("endDate"), "originalStartDate": ( termination.get("originalStartDate") or user.get("hireDate") ), # 参加工作日期 — preserve real SF firstDateWorked # if carried; the offboard path still seeds # 社会工龄 on the leaver row so retroactive # 年假 settlements have the correct N anchor. "firstDateWorked": termination.get("firstDateWorked"), # 司龄 S — SF's canonical seniority anchor. "seniorityDate": ( termination.get("seniorityDate") or termination.get("startDate") or user.get("hireDate") ), "lastModifiedDateTime": ( termination.get("lastModifiedDateTime") or user.get("lastModifiedDateTime") or today_iso ), "_full_sync": True, }, )) else: active_emp = employments_by_user_id.get(user_id) or {} synthetic_events.append(( "onboard", { "userId": user_id, # Prefer the REAL ``personIdExternal`` off the # EmpEmployment row already in hand — zero extra API # calls. ``user_id`` remains the fallback for rows # where no employment was fetched (see the # placeholder rationale below the ``_load_email_names_by_pid`` # call), but it must never be preferred: SF keys # PEOPLE by personIdExternal and EMPLOYMENTS by # userId, and echoing userId into this field is what # blinded the concurrent-employment guard on this # path (80100057 / 80100061 / 80100062 all had to be # aliased by a one-off script because # ``personIdExternal != userId`` could never be true # here). ``_upsert_person`` still re-resolves from SF # as a backstop when this lands self-referential. "personIdExternal": ( active_emp.get("personIdExternal") or user_id ), # Provenance, so ``_upsert_person`` knows whether a # ``pid == userId`` value is SF's answer or our # stand-in. Only the stand-in earns an EmpEmployment # round-trip on the create branch. "_person_id_external_is_user_id": not active_emp.get( "personIdExternal" ), # NEVER synthesize ``today`` here. ``_upsert_person`` # reads ``startDate`` straight into # ``person.hire_date``, so a fabricated "now" makes # any SF row with a null ``hireDate`` look like a # person STARTING TODAY — and the hire-day cron then # provisions them a real Google mailbox within the # hour. That is exactly how ``cbadmin@pingcap.com`` # was minted for the SF ``CBadmin`` admin account on # 2026-08-09 (the synthesized value even carried the # sync run's millisecond tail, ``/Date(…222897)/``, # where real SF dates are midnight-000). It also # re-fired on EVERY full sync: a fresh "today" # produced a new ``effective_date``, hence a brand-new # onboard event past the (person, type, date) dedup # key, so the person "onboarded" again and again and # their hire_date drifted forward each run. # # Leaving it None makes ``_effective_date`` return # None and ``_stage_event`` drop the event — the # correct outcome for a row SF gives no start date # for. The offboard branch above was already hardened # this way ("we MUST NOT synthesize ``today``"); this # is the onboard branch catching up. The incremental # path's synthetic fallback never had the bug — it # reads ``(user or {}).get("hireDate")`` bare. "startDate": user.get("hireDate"), "endDate": None, # ``originalStartDate`` kept for downstream code # paths that still consult it (rehire detection, # 司龄-for-rehire callers), but the canonical # 参加工作日期 source is now ``firstDateWorked`` # — see ``_upsert_person`` for the fallback chain. "originalStartDate": ( active_emp.get("originalStartDate") or user.get("hireDate") ), # 参加工作日期 (社会工龄 N 起点) — the field # PingCAP HR populates with the person's # first-ever career start. 328/500 (66%) # populated in prod survey 2026-06-02; the # remaining 172 land at NULL here and the # _upsert_person fallback chain catches them via # originalStartDate → user.hireDate. "firstDateWorked": active_emp.get("firstDateWorked"), # 司龄 S 起点 — SF's canonical seniority anchor. # 99.8% populated; diverges from startDate only # on rehire / employment-segment reset. "seniorityDate": ( active_emp.get("seniorityDate") or user.get("hireDate") ), "lastModifiedDateTime": user.get("lastModifiedDateTime") or today_iso, "_full_sync": True, }, )) # Migration 565 — resolve seed-vs-reconciliation ONCE, at # creation, and persist it. Deciding it at confirm time would # read a world the staged rows have already started changing. resolved_is_seed = ( await self._full_sync_is_seed() if is_seed is None else bool(is_seed) ) batch = HrSyncBatch( triggered_by=triggered_by, trigger_source="manual", sync_mode="full", is_seed=resolved_is_seed, hris_instance_id=self.hris_instance_id, start_date=date(1970, 1, 1), end_date=self.today, status="pending_review", ) self.db.add(batch) await self.db.flush() # HR-assigned 邮箱名 (PerPersonal.customString4). Key by the SAME # id ``_stage_event`` reads off these synthetic employments — # ``personIdExternal`` when the EmpEmployment row was fetched (the # normal case, and the key PerPersonal is ACTUALLY stored under), # falling back to ``userId`` only for rows without one. Both halves # read the same field off the same dict, so preview and this loader # can't drift; the fallback stays correct because ``userId == # personIdExternal`` on this tenant for every single-employment # person, which is the assumption the birthday/gender master-data # enrichment already relies on. await self._load_email_names_by_pid( [emp for _, emp in synthetic_events] ) staged: list[HrSyncStagedEvent] = [] for event_type, employment in synthetic_events: event = await self._stage_event( batch_id=batch.id, event_type=event_type, employment=employment, user=users_by_id.get(employment["userId"]), jobs=jobs_by_user.get(employment["userId"], []), ) if event is not None: staged.append(event) for evt in staged: self.db.add(evt) batch.events_staged = len(staged) await self.db.flush() return batch async def _stage_user_lifecycle_event( self, *, batch_id: UUID, user_id: str, user: dict | None, user_employments: list[dict], jobs: list[dict], ) -> HrSyncStagedEvent | None: """Decide + build the single staged event for one SF user. Extracted from the incremental preview loop (2026-07-22) so the incremental path and the single-person refresh path (:meth:`preview_single_person_sync`) share ONE staging decision and can never drift. Returns the staged event (NOT added to the session — the caller collects + adds) or ``None`` when nothing should be staged (unknown status, non-leaver inactive, or the event was filtered by ``_stage_event``). The employment-selection + active/inactive branching below is identical to the pre-extraction inline block; see the comments for the rehire / future-hire / NO_TERMINATION rationale. """ # Pick the employment record whose end date marks the # offboarding (if any); otherwise the employment that # represents the current hire. # # SF re-hires / termination-then-reactivate leave multiple # EmpEmployment rows for the same userId — one historical # terminated row with an endDate, and a fresh active row # with startDate but no endDate. The "any with endDate → # offboard" heuristic below mis-reads those stale rows as # a fresh termination and flips the person to ``left`` on # every tick. Guard against this by trusting SF's # ``User.status``: if SF says the user is currently ``'t'`` # (active), we do NOT stage an offboard no matter what # endDate shows on any EmpEmployment row — SF itself is # the source of truth for "is this person currently an # employee", and a past endDate is just history. _user_status = str((user or {}).get("status") or "").strip().lower() # ``active_external_suite`` (2026-07-12): SF Onboarding's # pre-day-1 status for future hires — a REAL upcoming employee # (future-dated EmpJob + EmpEmployment exist), never a leaver. # Route exactly like an active user; the future-hire gates # downstream defer any provisioning to the 24h lead window. user_active = sf_status_is_active(_user_status) # ``status='f'`` also covers future / day-one hires — SF Onboarding # marks pre-start hires inactive (P10013 Robert Wood 2026-07-15). # The leaver-or-not verdict for an inactive user is delegated to # ``classify_inactive_sf_user`` below so this incremental path and # the full-roster path share ONE definition and can't drift. _user_hire = _parse_date((user or {}).get("hireDate")) # A user can be handled with NO User record (``user is None``) on # the incremental path (employment-only change). In that case we # have no SF ``status`` signal — and we must NOT fall open to # "offboard": a real termination ALWAYS flips ``User.status`` t→f, # which modifies the User row, so it would be present with # ``status='f'``. ``user is None`` therefore means the person's # status did not change — inferring a termination from a stale # ``endDate`` (rehire history, a fixed-term contract's end, a # multi-row re-hire) would wrongly flip an active employee to # ``left`` and revoke their access. Treat unknown-status as # not-offboardable; the periodic full sync reconciles any real # drift. (The single-person refresh path always passes a real # User, so this only bites the incremental employment-only case.) # Review round-1 fix (2026-07-04): pick the LATEST endDate-bearing # employment, not the first. ``get_employments_for_users`` gives # no ordering guarantee; a termination-then-rehire-then-terminate # history returns multiple endDate rows, and "first" could stage # a years-old termination_date. Sort by parsed endDate (epoch-ms # ``/Date(...)/`` strings compare wrongly as text only when # lengths differ — parse properly via the shared helper). _end_bearing = [ e for e in user_employments if e.get("endDate") ] if not (user_active or user is None) else [] offboard_emp = ( max( _end_bearing, key=lambda e: _parse_date(e.get("endDate")) or date.min, ) if _end_bearing else None ) # #1784: pick the CURRENT stint, not ``user_employments[0]``. # ``get_employments_for_users`` gives no ordering guarantee, so # for a rehire (old terminated stint + new active stint) or an HR # edit of history, index 0 could be a years-old stint — staging # its startDate/seniorityDate would silently roll back the # employee's hire_date / 司龄 anchor and open an onboard event on # the wrong date. Mirror the offboard side (which already picks # the LATEST endDate): choose the employment with the latest # startDate (the current stint for a rehire). onboard_emp = ( max( user_employments, key=lambda e: _parse_date(e.get("startDate")) or date.min, ) if user_employments else { "userId": user_id, # No EmpEmployment row was returned, so this is a stand-in # for the master-data cache key, NOT SF's person key — # flagged so ``_upsert_person`` re-resolves it before the # concurrent-employment guard reads it. "personIdExternal": user_id, "_person_id_external_is_user_id": True, "startDate": (user or {}).get("hireDate"), "endDate": None, "lastModifiedDateTime": (user or {}).get( "lastModifiedDateTime" ), "_incremental_synthetic": True, } ) if user is None: # Employment-only change, no SF status signal → never # offboardable (a real termination always flips User.status # t→f, which would put the User row here). Stage nothing. return None if user_active: # Active user — stage an onboard (create / reactivate / # update) so attribute changes (e.g. displayName) flow. return await self._stage_event( batch_id=batch_id, event_type="onboard", employment=onboard_emp, user=user, jobs=jobs, ) # Inactive user present → the SHARED classifier decides # leaver-or-not (same function the full-roster path uses). # Termination-reason fetch is a single-user call, cost-gated to # zero when no exclusion codes are configured (and to nothing at # all unless this user actually carries an endDate). _stage_reasons: dict[str, str] = {} if offboard_emp is not None and offboard_emp.get("endDate"): _stage_reasons = await self._termination_reasons_for([user_id]) decision = classify_inactive_sf_user( hire_date=_user_hire, as_of=self.today, termination_end_date=( _parse_date(offboard_emp.get("endDate")) if offboard_emp else None ), termination_reason_code=_stage_reasons.get(user_id), excluded_reason_codes=( await self._excluded_termination_reason_codes() ), ) if decision.is_leaver: return await self._stage_event( batch_id=batch_id, event_type="offboard", employment=offboard_emp, user=user, jobs=jobs, ) if decision.intent == LIFECYCLE_FUTURE_HIRE: # Future / day-one hire — onboard so the person exists in # Helix pre-start; downstream 24h-lead gates defer any # provisioning. NEVER offboarded (even if SF carries a # nonsensical pre-start endDate — the classifier ignores it). return await self._stage_event( batch_id=batch_id, event_type="onboard", employment=onboard_emp, user=user, jobs=jobs, ) # LIFECYCLE_NO_TERMINATION: SF says forfeited but no # substantiated endDate (or an impossible one). Stage # NOTHING and never fabricate a termination_date; the daily # ``hris_status_divergence`` check surfaces this shape # (Helix active-ish + SF status='f') for review. # # LIFECYCLE_REASON_EXCLUDED lands here too: a substantiated # termination whose reason (人员分类转换 / 跨DC调动 / 兼岗结束) is # configured as "do not close accounts". Stage nothing — the # person keeps working; divergence surfaces the SF-side 'f' with # the reason attached for the operator. logger.warning( "hr.sync: user %s inactive in SF but %s — skipping staging; " "hris_status_divergence will surface this person", user_id, decision.reason, ) return None async def preview_single_person_sync( self, *, user_id: str, triggered_by: str, trigger_source: str = "manual", ) -> HrSyncBatch: """Pull ONE SF user by ``userId`` and stage a single event. Unlike :meth:`preview_incremental_sync` (bounded by a ``lastModifiedDateTime`` window), this force-fetches exactly one userId regardless of when it last changed — the HR "sync this person now" path. Fetches the same three entities the incremental path uses (User + EmpJob + EmpEmployment for the one id), runs the SHARED :meth:`_stage_user_lifecycle_event` decision, and leaves the batch at ``pending_review`` (``sync_mode='single_person'``) for the coordinator to gate + confirm. Raises ``NotFoundError`` when SF returns no record for the id (a purged / mistyped userId) so the caller surfaces a clean 404 instead of silently staging nothing. """ await self._refresh_org_snapshot() client = await self._get_client() users = await client.get_users([user_id]) user = next( (u for u in users if str(u.get("userId")) == str(user_id)), None ) if user is None: raise NotFoundError("SuccessFactors User", user_id) jobs_by_user = await self._jobs_by_user_id([user_id]) jobs = jobs_by_user.get(user_id, []) employments = await client.get_employments_for_users([user_id]) user_employments = [ e for e in employments if str(e.get("userId")) == str(user_id) ] # HR-assigned 邮箱名 (PerPersonal.customString4), keyed by # personIdExternal — same enrichment the incremental / full paths # load before staging so a re-derived work_email uses the HR value. await self._load_email_names_by_pid(user_employments) batch = HrSyncBatch( triggered_by=triggered_by, trigger_source=trigger_source, sync_mode="single_person", hris_instance_id=self.hris_instance_id, start_date=self.today, end_date=self.today, since_datetime=None, until_datetime=None, status="pending_review", ) self.db.add(batch) await self.db.flush() event = await self._stage_user_lifecycle_event( batch_id=batch.id, user_id=user_id, user=user, user_employments=user_employments, jobs=jobs, ) if event is not None: self.db.add(event) batch.events_staged = 1 if event is not None else 0 await self.db.flush() return batch # ---- Incremental sync (lastModifiedDateTime-based) ---- async def preview_incremental_sync( self, *, triggered_by: str, since: datetime, until: datetime | None = None, trigger_source: str = "scheduled", ) -> HrSyncBatch: """Stage every user touched in SF since ``since``. Full replacement is a one-time init operation; the daily / hourly tick runs this incremental path so data stays accurate without re-enumerating the whole roster. Strategy: 1. Fetch ``User`` rows with ``lastModifiedDateTime ≥ since`` (catches new hires, renames, status flips for leavers). 2. Fetch ``EmpEmployment`` rows in the same window so we know whose employment started or ended. 3. For each touched user, stage either an onboard event (if SF says they are currently active and we don't have them yet OR they are being reactivated) or an offboard event (if their employment end date is set). 4. Users where nothing meaningful changed get a ``skip`` action so the confirm step logs "N skipped" instead of silently dropping them. """ await self._refresh_org_snapshot() client = await self._get_client() users = await client.list_users_modified_since(since) employments = await client.list_employments_modified_since(since) users_by_id: dict[str, dict] = { u["userId"]: u for u in users if u.get("userId") } employments_by_user: dict[str, list[dict]] = {} for emp in employments: uid = emp.get("userId") if not uid: continue employments_by_user.setdefault(uid, []).append(emp) touched_user_ids = sorted(set(users_by_id) | set(employments_by_user)) # Leaver blind-spot fix (2026-07-03, P0051 韦万 post-mortem). # A user whose SF status just flipped ``t→f`` has their User row # re-modified (so — with the companion ``list_users_modified_since`` # fix — they now appear in ``users_by_id`` with ``status='f'``), # but the ``EmpEmployment`` row that carries ``endDate`` was often # last modified on an EARLIER tick (HR keys the termination days or # weeks ahead of the effective date; SF flips ``status`` only when # the effective date rolls over). When that ``endDate`` modification # predates this incremental window, the employment is absent from # ``employments_by_user`` and the offboard branch below has no # canonical ``termination_date`` to stage — the leaver would fall # through to an ``onboard`` and stay ``active``. Backfill the # employment for exactly those users (SF now says forfeited, no # endDate-bearing employment in-window) so the offboard stages with # the real termination_date. Keyed on ``status='f'`` so genuine # rehires (``status='t'`` with a stale terminated EmpEmployment) # are never pulled in — they must NOT offboard. terminated_needing_emp = [ uid for uid in touched_user_ids if str((users_by_id.get(uid) or {}).get("status") or "").strip().lower() == "f" and not any( e.get("endDate") for e in employments_by_user.get(uid, []) ) ] if terminated_needing_emp: for emp in await client.get_employments_for_users( terminated_needing_emp ): uid = emp.get("userId") if uid and emp.get("endDate"): employments_by_user.setdefault(uid, []).append(emp) jobs_by_user = ( await self._jobs_by_user_id(touched_user_ids) if touched_user_ids else {} ) until = until or datetime.now(UTC) batch = HrSyncBatch( triggered_by=triggered_by, trigger_source=trigger_source, sync_mode="incremental", hris_instance_id=self.hris_instance_id, start_date=since.date(), end_date=until.date(), since_datetime=since, until_datetime=until, status="pending_review", ) self.db.add(batch) await self.db.flush() # HR-assigned 邮箱名 (PerPersonal.customString4), keyed by # personIdExternal off every EmpEmployment row in hand (incl. the # terminated backfill) — staged into each event so the incremental # pull uses the same email local-part the webhook path receives. await self._load_email_names_by_pid( [e for rows in employments_by_user.values() for e in rows] ) staged: list[HrSyncStagedEvent] = [] for user_id in touched_user_ids: event = await self._stage_user_lifecycle_event( batch_id=batch.id, user_id=user_id, user=users_by_id.get(user_id), user_employments=employments_by_user.get(user_id, []), jobs=jobs_by_user.get(user_id, []), ) if event is not None: staged.append(event) for evt in staged: self.db.add(evt) batch.events_staged = len(staged) await self.db.flush() return batch async def run_scheduled_incremental( self, *, triggered_by: str = "system:scheduler", lookback: timedelta = timedelta(hours=25), auto_confirm: bool = True, ) -> dict: """End-to-end scheduled sync: preview → (optional) confirm. ``lookback`` is the "safety margin" beyond the last successful sync — for an hourly cadence 25 h is plenty of overlap so nothing is missed around a worker restart. If there is no previous successful batch for this instance, we fall back to ``today - lookback`` as the starting cursor. ``auto_confirm=True`` (default for scheduled runs) applies the staged events immediately; set False to require manual review. """ last_until = await self._last_successful_until() now = datetime.now(UTC) since = max( last_until - lookback if last_until else now - lookback, now - timedelta(days=30), # hard cap so a stale cursor can't ask SF for years of data ) batch = await self.preview_incremental_sync( triggered_by=triggered_by, since=since, until=now, trigger_source="scheduled", ) # Mass-leaver brake (2026-05-04): if SF returned an unusually large # number of offboard events for this incremental window — e.g. a # corrupted OData response, an upstream filter regression, or a # tenant-scope misconfiguration that suddenly excludes a whole BU — # halt the auto-confirm and leave the batch in pending_review for # an IT Admin to inspect. The threshold is intentionally generous # (5% of currently-active people OR 25 absolute leavers, whichever # is HIGHER) so a normal layoff round still flows through; anything # beyond that pattern is treated as suspicious and requires a human. offboard_count = await self.db.scalar( select(func.count()) .select_from(HrSyncStagedEvent) .where(HrSyncStagedEvent.batch_id == batch.id) .where(HrSyncStagedEvent.event_type == "offboard") .where(HrSyncStagedEvent.action == "deactivate") ) or 0 active_total = await self._count_active_persons_for_instance() threshold = max(25, int(active_total * 0.05)) if offboard_count > threshold: logger.warning( "scheduled HRIS sync staged %d offboard events vs %d active " "(threshold=%d) — halting auto-confirm; batch %s left in " "pending_review for manual IT Admin review", offboard_count, active_total, threshold, batch.id, ) return { "batch_id": str(batch.id), "status": batch.status, "events_staged": batch.events_staged, "since": since.isoformat(), "until": now.isoformat(), "auto_confirmed": False, "auto_confirm_blocked_reason": ( f"mass_leaver_brake: {offboard_count} offboard events " f"exceeds threshold {threshold} " f"(active_total={active_total})" ), } if not auto_confirm: return { "batch_id": str(batch.id), "status": batch.status, "events_staged": batch.events_staged, "since": since.isoformat(), "until": now.isoformat(), "auto_confirmed": False, } result = await self.confirm_batch(batch.id, confirmed_by=triggered_by) result["since"] = since.isoformat() result["until"] = now.isoformat() result["auto_confirmed"] = True return result async def run_scheduled_full( self, *, triggered_by: str = "system:scheduler", auto_confirm: bool = True, ) -> dict: """End-to-end scheduled FULL roster sync. Companion to :meth:`run_scheduled_incremental`. Incremental's ``lookback`` is hard-capped at 30 days so anyone whose ``lastModifiedDateTime`` is older than that lives in a permanent blind spot — historical SF leavers (``status='f'``) terminated more than 30 days ago can never be picked up by the hourly tick. Concrete prod fallout: wangdi@pingcap.com (P0175, terminated 2026-03-20) was reported missing from Helix on 2026-06-01 — 73 days after his SF termination. Even though PR #588 fixed the bootstrap path (``list_all_users`` now routes ``status='f'`` to offboard), the scheduled tick still only runs incremental, so the fix never executes automatically. This method is the scheduled companion that closes the gap permanently. Seed vs reconciliation (migration 565). The FIRST full sync of a fresh instance legitimately produces hundreds of creates and one offboard per historical leaver — the volume brakes' exact false-positive shape, and coordinators must stay silent because that roster is history, not news. Every LATER full sync is the opposite: whatever it recovers is precisely the news the 30-day incremental window could not see, and suppressing coordinators there throws the recovery away. ``preview_full_employee_sync`` resolves which one this is via ``_full_sync_is_seed`` and persists it on the batch; the cron never asserts either, so a newly-registered HRIS instance still seeds correctly on its first scheduled run. ``auto_confirm=True`` (default) applies the staged events immediately; set False to require manual review. """ now = datetime.now(UTC) batch = await self.preview_full_employee_sync(triggered_by=triggered_by) # Trigger_source on the row is hardcoded to 'manual' inside # preview_full_employee_sync. Flip it to 'scheduled' here so # the admin batch list distinguishes cron-driven runs from # manually-triggered ones. batch.trigger_source = "scheduled" await self.db.flush() if not auto_confirm: return { "batch_id": str(batch.id), "status": batch.status, "events_staged": batch.events_staged, "until": now.isoformat(), "auto_confirmed": False, } result = await self.confirm_batch(batch.id, confirmed_by=triggered_by) result["until"] = now.isoformat() result["auto_confirmed"] = True return result async def _last_successful_until(self) -> datetime | None: """Return ``until_datetime`` of the most recent confirmed batch for this instance (or any instance if not scoped). MySQL + asyncmy returns tz-naive datetimes even when the column is declared ``DateTime(timezone=True)`` — values are stored as UTC but tzinfo is stripped on read. Attach UTC back so callers can compare against ``datetime.now(UTC)`` without blowing up. """ stmt = select(HrSyncBatch).where( HrSyncBatch.status == "confirmed", HrSyncBatch.until_datetime.is_not(None), ) if self.hris_instance_id is not None: stmt = stmt.where(HrSyncBatch.hris_instance_id == self.hris_instance_id) stmt = stmt.order_by(HrSyncBatch.until_datetime.desc()).limit(1) batch = await self.db.scalar(stmt) if batch is None or batch.until_datetime is None: return None dt = batch.until_datetime return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt async def cancel_batch(self, batch_id: UUID, cancelled_by: str) -> None: batch = await self.db.scalar( select(HrSyncBatch).where(HrSyncBatch.id == batch_id).with_for_update() ) if batch is None: raise NotFoundError("HrSyncBatch", str(batch_id)) if batch.status != "pending_review": raise ConflictError(f"Batch {batch_id} is already {batch.status}") batch.status = "cancelled" batch.reviewed_at = datetime.now(UTC) batch.reviewed_by = cancelled_by await self.db.flush() async def list_batches( self, *, page: int = 1, page_size: int = 20, hris_instance_id: UUID | None = None, ) -> tuple[list[HrSyncBatch], int]: from sqlalchemy import func as sa_func # Pre-launch P2 fix: filter by hris_instance_id when caller # specifies it so multi-region admins can scope the listing # to a single tenant; secondary sort on id keeps pagination # deterministic when two batches share a created_at second. total_q = select(sa_func.count(HrSyncBatch.id)) list_q = select(HrSyncBatch) if hris_instance_id is not None: total_q = total_q.where(HrSyncBatch.hris_instance_id == hris_instance_id) list_q = list_q.where(HrSyncBatch.hris_instance_id == hris_instance_id) total = await self.db.scalar(total_q) or 0 result = await self.db.execute( list_q .order_by(HrSyncBatch.created_at.desc(), HrSyncBatch.id.desc()) .offset((page - 1) * page_size) .limit(page_size) ) return list(result.scalars().all()), total async def get_batch(self, batch_id: UUID) -> HrSyncBatch: from sqlalchemy.orm import selectinload batch = await self.db.scalar( select(HrSyncBatch) .options(selectinload(HrSyncBatch.staged_events)) .where(HrSyncBatch.id == batch_id) ) if batch is None: raise NotFoundError("HrSyncBatch", str(batch_id)) return batch async def _stage_event( self, *, batch_id: UUID, event_type: str, employment: dict, user: dict | None, jobs: list[dict], ) -> HrSyncStagedEvent | None: """Compute what would happen for a single event and return a staged record (not added to session).""" user_id = employment.get("userId") if not user_id: return None # HR test-record band (P99XX): HR reserves P9900-P9999 for their # own SF test records (confirmed 2026-07-13 — P9997/P9998/P9999 # were sitting `active` in prod). Drop them for EVERY event type # (onboard / update / offboard) so they never enter, update, or # churn identity_persons — the single ingest chokepoint that # starves every downstream path (accounts, outbound sync, hire- # day cron). Keyed on BOTH the SF userId (→ hris_person_id) and # empId (→ employee_no) since either can carry the test id. This # runs BEFORE the onboard-only `should_skip_user` gate below so # it also blocks a test record's stray offboard/update. if is_test_employee_no(user_id) or is_test_employee_no( (user or {}).get("empId") ): logger.info( "hr.sync: skipping %s event for user_id=%s — HR reserved " "P99XX test band (not a real employee)", event_type, user_id, ) return None effective_date = _effective_date(event_type, employment) if effective_date is None: return None # Config-gated preview skip — drop SF service accounts # (APIadmin/ONBadmin/…) and placeholder-email stubs so they # never land in identity_persons. Only applied to onboard # events; update/offboard paths still process existing rows # so terminations propagate correctly. if event_type == "onboard" and self.should_skip_user(user_id, user): logger.info( "hr.sync: skipping onboard for user_id=%s (email=%r) — " "matches HR_SKIP_USER_IDS or placeholder-email policy", user_id, (user or {}).get("email"), ) return None # Shape-based non-employee gate (2026-08-18). The two gates above # are ENUMERATIONS — a userId name list and a vendor-domain list — # and ``CBadmin`` defeated both on 2026-08-09: not in the name # list, and its SF email sits on the real corporate domain # (``cbadmin@pingcap.com``) so the domain gate had nothing to bite. # It was ingested as a hire and minted a real Google mailbox within # the hour. This gate reads the SHAPE instead: an SF ``User`` with # no EmpJob, no ``hireDate`` AND no employment ``startDate`` holds # no evidence of employment anywhere, so it is a tenant admin / # integration account, not a person. ALL THREE must be absent — any # one alone was falsified against the live roster (no-EmpJob alone # would have blocked real incoming hire P10025, # ``active_external_suite`` with its EmpJob not cut yet) — see the # prod-verified rationale in ``person_filters``. Onboard-only for # the same reason ``should_skip_user`` is: existing rows must keep # receiving update / offboard events. if event_type == "onboard" and is_hris_non_employee_record( hire_date=(user or {}).get("hireDate"), has_emp_job=bool(jobs), employment_start_date=employment.get("startDate"), ): logger.info( "hr.sync: skipping onboard for user_id=%s (email=%r) — SF " "carries no EmpJob and no hireDate for it, so it is an " "admin / integration account, not an employee", user_id, (user or {}).get("email"), ) return None # Duplicate-employment guard (2026-08-18). A userId registered in # ``identity_hris_user_aliases`` is a SECOND employment of a human # Helix already has a row for — SF keys employments by userId and # people by personIdExternal, and a concurrent assignment gets its # own userId. Staging anything for it produces a phantom new-hire # card on the onboarding page every single sync (prod: 80100061 / # 80100062 re-staged as ``create`` every 8h since 2026-06-26) and, # on an offboard, would retire a person whose other employment is # still open. Applies to every event type: the canonical userId is # the one that speaks for this human. from app.identity.identity_reconcile import hris_user_alias_owner alias_owner = await hris_user_alias_owner(self.db, user_id) if alias_owner is not None: logger.info( "hr.sync: skipping %s event for user_id=%s — registered as a " "duplicate employment of person %s (alias); the canonical " "userId drives this human's lifecycle", event_type, user_id, alias_owner.id, ) return None # Check if already exists in identity_persons — scope the lookup # to the owning instance in multi-region mode so the preview # doesn't conflate a CN user with an EU user of the same SF id. _stmt = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, ) if self.hris_instance_id is not None: _stmt = _stmt.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) existing_person = await self.db.scalar(_stmt) # Grave-collision guard (2026-09-04, P01412 战捷). The userId above # can match a row that belongs to a DIFFERENT, retired human — SF # userIds are not unique across the CN / EU tenants, and the # legacy-adoption arm of the lookup deliberately admits # ``hris_instance_id IS NULL`` rows. ``personIdExternal`` is the # authority on whose employment this is; when it names a live # active person other than the row we matched, alias the userId # onto them and stage nothing. See # ``_tombstone_belongs_to_live_person`` for the incident. if await self._claim_tombstoned_user_id( user_id, employment, matched=existing_person, event_type=event_type, ): return None # Shell-record guard (2026-07-17, P10019 post-mortem). SF # Onboarding emits a stripped ``User`` (placeholder displayName, # no first/last name) for a requisition that was cancelled / # withdrawn while its future-dated EmpEmployment rows still # linger in the incremental feed. Creating a person from it lands # a nameless ghost (``Unknown`` / ``unknown@pingcap.com``) that # the daily absent-person check can NEVER tombstone — its # future-hire exemption skips any hire whose start date is still # ahead, and a cancelled future hire is exactly that shape. Drop # the onboard for a genuinely-new person only: an UPDATE to an # existing row still flows so a transient SF blip can't block a # real employee's attribute sync (and won't overwrite their name # — ``_upsert_person`` only promotes a non-empty displayName). if ( event_type == "onboard" and existing_person is None and is_shell_sf_record(user) ): logger.warning( "hr.sync: skipping onboard for user_id=%s — SF returned a " "shell record (no displayName/first/last name); refusing to " "create a nameless person. A real hire re-stages once HR " "populates the name.", user_id, ) return None # Cross-instance ownership guard (added 2026-05-02). If the # instance-scoped lookup above returned ``None`` for an onboard # event, AND we're running a multi-instance sync, AND another # instance already owns this ``(source_system, hris_person_id)``, # we are NOT looking at a new hire — we're looking at the same # human that the other tenant already onboarded, surfaced in # this tenant's OData feed by a misconfigured filter or by # routine cross-tenant employee mobility. Without this guard, # the misclassification was ``create`` → welcome email + # provisioning re-fired every hourly tick (udi/kevin in prod # 2026-05-02). Skip the event entirely so the foreign-instance # sync remains the single owner. The staged_events table stays # clean of phantom ``create`` rows that confuse audit + admin # console diff. _reclaim_pending = False if ( event_type == "onboard" and existing_person is None and self.hris_instance_id is not None ): cross_instance_match = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, ) ) # Onboard-poach guard (2026-07-16, Camila P10009 follow-up). # The 2026-05-02 guard above only decides by "who created the # row first" — but for a MIRRORED pre-start hire neither # instance has a row yet, so whichever tenant's incremental # window surfaced them first would create the row with ITS # hris_instance_id and permanently own the person (the true # owner's sync then hits this very guard and skips forever). # Resolve the CANONICAL owner from the hire's legal entity # (EmpJob.company → ``owned_company_codes`` mapping) and: # * no row anywhere + another instance owns the code → # DEFER creation to that instance's own sync (it sees the # same mirrored hire in its window within the hour); # * row held by another instance but WE own the code → # let the event through — ``_upsert_person`` reclaims the # row onto this instance at apply time (heals races that # predate this fix or slipped past the defer). # Unmapped code / no config → ``None`` verdict → today's # behaviour, so single-instance installs and codes without # ownership declarations are untouched. _snap = self._select_job_snapshot(event_type, employment, jobs) or {} _owner_id = await self._company_owner_for(_snap.get("company")) if cross_instance_match is not None: if _owner_id == self.hris_instance_id: logger.warning( "hr.sync: onboard for user_id=%s — row currently " "held by hris_instance_id=%s but company code %s " "is canonically owned by THIS instance (%s); " "staging the event so apply reclaims the row " "(onboard-poach recovery)", user_id, cross_instance_match.hris_instance_id, _snap.get("company"), self.hris_instance_id, ) existing_person = cross_instance_match _reclaim_pending = True else: logger.info( "hr.sync: skipping onboard stage for user_id=%s — " "person already owned by hris_instance_id=%s " "(this sync runs on instance_id=%s); cross-tenant " "OData overlap, not a new hire", user_id, cross_instance_match.hris_instance_id, self.hris_instance_id, ) return None elif _owner_id is not None and _owner_id != self.hris_instance_id: logger.info( "hr.sync: deferring onboard CREATE for user_id=%s — " "company code %s is canonically owned by " "hris_instance_id=%s (this sync runs on %s); the " "owning instance's sync creates the person with the " "right ownership (mirrored cross-tenant hire)", user_id, _snap.get("company"), _owner_id, self.hris_instance_id, ) return None # Determine action # # ``org_drift`` is resolved BEFORE the action decision because it # is an INPUT to it (migration 565). Previously an already-active # employee whose onboard ledger row existed was unconditionally # staged ``skip``, and the org-change flags were only computed # for ``update``/``reactivate`` — so a manager or department move # on a normal employee (hire date unchanged, therefore ledger row # found) produced a ``skip`` row with no flags, and the confirm # loop's role-change gate, which admits only update/reactivate, # could never see it. The move was still APPLIED to the person # row by the skip fall-through, so Helix silently repointed the # manager and never reviewed the access that came with the old # one. That is why production holds 557 onboards, 133 offboards # and exactly one role_change. # Hoisted so the drift check and the staged payload below read the # SAME snapshot. Computing it twice from the same inputs happens to # agree today; a single binding makes disagreement impossible # rather than merely unlikely. onboard_job_snapshot = ( self._select_job_snapshot(event_type, employment, jobs) if event_type == "onboard" else None ) org_drift: tuple[bool, bool, bool] | None = None if ( event_type == "onboard" and existing_person is not None and existing_person.deleted_at is None ): org_drift = await self._compute_org_change_flags( existing_person, onboard_job_snapshot, ) if event_type == "onboard": if existing_person is None: action = "create" elif self._is_inflight_scheduled_leaver(existing_person): # Confirmed future leaver, LWD not reached — SF still # reports status='t', so this onboard is a routine # attribute refresh, NOT a rehire. Staging it as # ``reactivate`` fired the new-hire coordinator chain # (email reserve / Google + Feishu 建号 / welcome mail) # at a departing employee and counted them as a new # hire in the batch summary. See # ``_is_inflight_scheduled_leaver`` (2026-08-11). action = "update" elif existing_person.employment_status in ("left", "leaver_pending"): action = "reactivate" else: # Already active — check if lifecycle event already recorded existing_evt = await self.db.scalar( select(HrLifecycleEvent).where( HrLifecycleEvent.person_id == existing_person.id, HrLifecycleEvent.event_type == "onboard", HrLifecycleEvent.effective_date == effective_date, ) ) # An org move is news even when the onboard ledger row is # old news. ``update`` is the weakest action that reaches # the role-change gate: unlike ``create``/``reactivate`` # it fires no onboarding coordinator (the allow-list in # ``_dispatch_lifecycle_coordinators`` admits only those # two), so this widens detection without widening # provisioning. action = ( "skip" if existing_evt is not None and not any(org_drift or ()) else "update" ) if _reclaim_pending and action == "skip": # The poaching instance already recorded a lifecycle # onboard for this effective_date on the foreign-owned # row, which would derive "skip" — but a skipped event is # never applied, so the row would NEVER be reclaimed. # Force an update so apply runs and repoints ownership. action = "update" else: # offboard # Impossible-state guard (2026-07-15, Robert Wood P10013): a leaver # whose termination_date precedes their hire_date is logically # impossible — the signature of a fabricated / mis-derived offboard # (the old full-sync ``lastModifiedDateTime`` fallback produced # exactly this). The ``classify_inactive_sf_user`` classifier # already rejects it in the full + incremental paths; THIS is the # preview-time guard for the remaining STAGED path that skips the # classifier — the manual window import (``preview_import``). # (The webhook push path bypasses staging entirely — it is backed # by the twin apply-time guard in ``_upsert_person``.) Dropping the # event leaves the person untouched (safe); the daily # ``hris_status_divergence`` check surfaces the bad SF data. _off_end = _parse_date(employment.get("endDate")) _off_hire = _parse_date(employment.get("startDate")) or ( existing_person.hire_date if existing_person is not None else None ) if _off_end is not None and _off_hire is not None and _off_end < _off_hire: logger.error( "hr.sync: REFUSING impossible offboard for user_id=%s — " "termination_date %s precedes hire_date %s; dropping the " "staged event (person left untouched). Data-quality signal, " "not a real leaver.", user_id, _off_end, _off_hire, ) return None if existing_person is None: # Default: can't offboard someone not in the system. # Exception (2026-05-30 wangdi PR): when this staging # call is part of a bootstrap full sync, the person # genuinely doesn't exist in identity_persons yet — the # whole point of including ``status='f'`` users in # ``list_all_users()`` is to create them on the fly # with the offboard event as their FIRST sight. The # ``_full_sync`` marker on the synthesized employment # dict signals this case so ``_apply_event`` → # ``_upsert_person`` can create + offboard in one shot. # Action is ``deactivate`` (not ``skip``) so the confirm # loop actually invokes ``_upsert_person`` rather than # short-circuiting to a no-op. if not employment.get("_full_sync"): return None # Cross-instance ownership guard for the full-sync # offboard branch — added 2026-05-30 (review pass 1). # Multi-instance deployments (CN + EU SF) can both # return the SAME forfeited userId for a globally- # employed person; the instance-scoped lookup above # returned None because this instance has never seen # them, but the OTHER instance may already own the # row. Without this guard we would attempt an INSERT # in ``_upsert_person`` that crashes on # ``ConflictError`` via the empId-based cross-instance # dedup (service.py:2435-2458) — every cross-instance # forfeited user would error on bootstrap. Skip the # stage and let the foreign instance remain the # authoritative owner; the per-instance sync there # will set ``employment_status='left'`` for the same # person on its own pass. if self.hris_instance_id is not None: cross_instance_match = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, ) ) if cross_instance_match is not None: logger.info( "hr.sync: skipping full-sync offboard stage " "for user_id=%s — person already owned by " "hris_instance_id=%s (this sync runs on " "instance_id=%s); cross-tenant overlap, " "let the foreign-instance sync handle the " "leaver flip", user_id, cross_instance_match.hris_instance_id, self.hris_instance_id, ) return None # Build a synthesized event_data carrying ``deactivate`` # action and fall through to the standard offboard # build path below (which keys off ``action``). action = "deactivate" job_snapshot = self._select_job_snapshot( event_type, employment, jobs, ) _user_fs = user or {} person_data_fs: dict = { "user_id": user_id, "work_email": _user_fs.get("email") or f"{user_id}@successfactors.local", "display_name": strip_sentinel_name_tokens(_user_fs.get("displayName")) or user_id, "employee_no": _user_fs.get("empId") or user_id, "hire_date": employment.get("startDate"), "source_updated_at": _user_fs.get("lastModifiedDateTime"), "default_locale": _user_fs.get("defaultLocale"), "termination_date": employment.get("endDate"), # Always 'leaver_pending' at the staged level — the # 30-day historical cutoff that promotes to 'left' # lives in ``_upsert_person`` and only fires when # the row is brand-new (session.new). Putting 'left' # here would propagate it into a row that may # already exist by the time confirm runs (concurrent # incremental sync just created it) — the listener # would then block the UPDATE-to-'left'. "employment_status": "leaver_pending", } employment_data_fs = dict(employment) employment_data_fs["_jobs_snapshot"] = ( [job_snapshot] if job_snapshot else [] ) return HrSyncStagedEvent( batch_id=batch_id, event_type=event_type, user_id=user_id, person_data=person_data_fs, employment_data=employment_data_fs, action=action, existing_person_id=None, status="pending", ) # Defense in depth: if SF still reports the user as # currently active (``status='t'``), skip this offboard. # The ``endDate`` on the employment row is almost certainly # historical (a prior termination-then-rehire leaves an # orphaned EmpEmployment with endDate set). Without this # guard a stale endDate trips ``_upsert_person`` to flip # the person back to ``left`` every tick. Pairs with the # ``preview_incremental_sync`` guard above. user_status = str((user or {}).get("status") or "").strip().lower() if user_status == "t": logger.info( "hr.sync: skipping offboard stage for user_id=%s because " "SF User.status='t' (still active); endDate=%r is historical", user_id, employment.get("endDate"), ) action = "skip" elif existing_person.employment_status in ("left",): action = "skip" else: action = "deactivate" job_snapshot = ( onboard_job_snapshot if event_type == "onboard" else self._select_job_snapshot(event_type, employment, jobs) ) # Build person_data snapshot. ``defaultLocale`` rides on the # SF User entity; mobile_phone / personal_email live on the # PerPhone / PerEmail entities and are looked up in # ``_upsert_person`` (by ``personIdExternal``) from the # master-data caches loaded at confirm time. Staging them here # would be premature — ``_ensure_master_data_loaded`` hasn't # fired yet during preview. migration 048. _user = user or {} # HR-assigned corporate 邮箱名 (email local-part, e.g. "san.zhang"). # AUTHORITATIVE source is SF ``PerPersonal.customString4`` (matches # SF's own outbound ``emailName <<-- customString4`` Integration # Center mapping — the same value the webhook path receives as # ``emailName``), loaded per-batch into ``self._email_names_by_pid``. # ``User.custom07`` is a legacy secondary (a different entity+field # that only coincidentally held the value on PingCAP's tenant), kept # so a blank customString4 still resolves. NULL ⇒ reserve_work_email_for # falls back to the given.family pinyin derivation as before. _pid = employment.get("personIdExternal") email_name = ( (self._email_names_by_pid.get(_pid) if _pid else None) or _user.get("custom07") ) person_data: dict = { "user_id": user_id, "work_email": _user.get("email") or f"{user_id}@successfactors.local", "display_name": strip_sentinel_name_tokens(_user.get("displayName")) or user_id, "first_name": (_user.get("firstName") or "").strip() or None, "last_name": (_user.get("lastName") or "").strip() or None, "employee_no": _user.get("empId") or user_id, "hire_date": employment.get("startDate"), "source_updated_at": _user.get("lastModifiedDateTime"), # User-level preference — safe to stage (it's on User.$select). "default_locale": _user.get("defaultLocale"), "email_name": email_name, } if event_type == "offboard": person_data["termination_date"] = employment.get("endDate") # Store the actual target DB status: "left" if already departed (skip), # "leaver_pending" if we are about to deactivate them. person_data["employment_status"] = "left" if action == "skip" else "leaver_pending" else: person_data["employment_status"] = "active" # Org-change flags (HRIS small-delta auto-confirm gate, 2026-06-01). # For update/reactivate of an existing person, record whether this # event carries a manager-line, department or title change vs the # person's CURRENT state. These flags let ``evaluate_auto_confirm`` # block auto-confirm on org moves (which need human review) while # still auto-applying pure new-hire / no-op batches, and they are # what the confirm loop's role-change gate reads. Computed above # (``org_drift``) because the action decision depends on them; # create/offboard events have no prior placement to compare. if action in ("update", "reactivate") and existing_person is not None: manager_changed, dept_changed, title_changed = ( org_drift if org_drift is not None else await self._compute_org_change_flags( existing_person, job_snapshot, ) ) person_data["manager_changed"] = manager_changed person_data["dept_changed"] = dept_changed person_data["title_changed"] = title_changed # The incoming title, staged so the confirm-time before/after # blob has an "after" to record. Cache-free by construction — # see ``_direct_job_title``. person_data["job_title"] = self._direct_job_title(job_snapshot) # Issue #623: stash the PRIOR (dept_code, job_title, manager # person_id) so the role-change coordinator at confirm time # can write a before/after blob into the lifecycle event + # flag old-context cloud grants for review. We compute it # here because ``existing_person`` is in scope; by confirm # time the row will already carry the new values. from app.identity.person_models import IdentityOrgMembership prior_dept = await self.db.scalar( select(IdentityOrgMembership.org_node_source_id).where( IdentityOrgMembership.person_id == existing_person.id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ) person_data["_prior_role"] = { "dept_code": prior_dept, "job_title": existing_person.job_title, "manager_person_id": ( str(existing_person.manager_person_id) if existing_person.manager_person_id else None ), } # Build employment_data snapshot — include jobs for confirm phase employment_data = dict(employment) employment_data["_jobs_snapshot"] = [job_snapshot] if job_snapshot else [] return HrSyncStagedEvent( batch_id=batch_id, event_type=event_type, user_id=user_id, person_data=person_data, employment_data=employment_data, action=action, existing_person_id=existing_person.id if existing_person else None, status="pending", ) @staticmethod def _direct_job_title(job_snapshot: dict | None) -> str | None: """The incoming title, using only CACHE-FREE snapshot fields. ``_apply_job_fields`` resolves a title through five fallbacks, the last two of which need ``self._positions`` — a master-data cache that is loaded at CONFIRM time, not at staging. Staging must therefore compare like with like or not at all: resolving to ``position_code`` here and comparing it against a stored ``job_title`` that came from ``externalName_defaultValue`` would report a move on every sync for every employee. ``None`` means "this snapshot carries no directly comparable title", which the drift check treats as no change — the same absent-field rule ``_compute_org_change_flags`` applies to manager and department. """ snap = job_snapshot or {} title = ( snap.get("positionNameEN") or snap.get("positionName") or snap.get("jobTitle") ) title = (title or "").strip() return title or None async def _refresh_org_snapshot(self) -> None: """Preload the current org placement of every person on this instance, so per-employee drift checks cost no queries. Called at the head of each preview pass. A full roster preview asks "did this person move?" once per employee (~444 on prod); answering that with two scalar SELECTs each is ~900 round-trips for a question two queries can answer for everyone. Deliberately refreshed per preview rather than cached lazily once: a service instance is reused across preview → confirm, and confirm mutates exactly the columns this snapshot describes. A stale snapshot would compare the next preview against a world that already moved. """ from app.identity.person_models import IdentityOrgMembership hris_rows = await self.db.execute( select(IdentityPerson.id, IdentityPerson.hris_person_id) ) self._hris_id_by_person = {pid: hid for pid, hid in hris_rows} dept_rows = await self.db.execute( select( IdentityOrgMembership.person_id, IdentityOrgMembership.org_node_source_id, ).where( IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ) self._primary_dept_by_person = {pid: code for pid, code in dept_rows} async def _prior_manager_hris_id( self, existing_person: IdentityPerson, ) -> str | None: if existing_person.manager_person_id is None: return None if self._hris_id_by_person is not None: return self._hris_id_by_person.get(existing_person.manager_person_id) return cast( "str | None", await self.db.scalar( select(IdentityPerson.hris_person_id).where( IdentityPerson.id == existing_person.manager_person_id, ) ), ) async def _prior_primary_dept( self, existing_person: IdentityPerson, ) -> str | None: if self._primary_dept_by_person is not None: return self._primary_dept_by_person.get(existing_person.id) from app.identity.person_models import IdentityOrgMembership return cast( "str | None", await self.db.scalar( select(IdentityOrgMembership.org_node_source_id).where( IdentityOrgMembership.person_id == existing_person.id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ), ) async def _compute_org_change_flags( self, existing_person: IdentityPerson, job_snapshot: dict | None, ) -> tuple[bool, bool, bool]: """Return ``(manager_changed, dept_changed, title_changed)``. manager: incoming ``managerId`` (HRIS id) vs the HRIS id of the person's current ``manager_person_id``. dept: incoming ``department`` (org-node source code) vs the person's current primary ``org_node_source_id``. title: the cache-free incoming title vs the stored ``job_title``. Missing incoming values are treated as "no change" (a partial snapshot must not trigger a false org-move block); the confirm path is unaffected either way. ``title_changed`` joined this tuple with migration 565. It used to be derived at CONFIRM time from ``person_data["job_title"]`` — a key ``_stage_event`` never wrote, so the expression was a constant ``False`` and the before/after blob recorded ``job_title: null`` on both sides. The one role_change row that ever reached production shows exactly that. """ snap = job_snapshot or {} # ``is not None`` (not truthiness): a field PRESENT-but-empty in # the snapshot is a real removal (manager dropped / dept cleared) # — that IS an org move and must be detected. Only a field # ENTIRELY ABSENT (key missing → ``None``) is "not carried this # sync" → no change. Erring toward "changed" is the safe # direction for an auto-confirm gate: a false-positive just # routes the batch to manual review. manager_changed = False incoming_mgr = snap.get("managerId") if incoming_mgr is not None: prior_mgr_hris = await self._prior_manager_hris_id(existing_person) manager_changed = str(incoming_mgr) != str(prior_mgr_hris or "") dept_changed = False incoming_dept = snap.get("department") if incoming_dept is not None: prior_dept_code = await self._prior_primary_dept(existing_person) dept_changed = str(incoming_dept) != str(prior_dept_code or "") # Title is the ONE signal whose absent-vs-empty distinction we # cannot honour: the snapshot carries three alternative title # keys and a blank in all three is indistinguishable from a # partial payload. So an absent title is never a change — a # title cannot be "removed" in SF, only replaced. title_changed = False incoming_title = self._direct_job_title(snap) if incoming_title is not None: title_changed = incoming_title != (existing_person.job_title or None) return manager_changed, dept_changed, title_changed @staticmethod def _overlay_event_fields( event: SuccessFactorsWebhookEvent, base: dict | None, ) -> dict | None: """Merge the webhook event's direct-carry fields into the SF ``user`` dict so a hire can be created BEFORE the full SF data sync runs. The sync only fires on the hire date; the event arrives days earlier. The SF pull (``get_users`` / master data), when it has a value, WINS — the event only FILLS gaps (``setdefault``). ``firstName`` / ``lastName`` / ``displayName`` / ``empId`` / ``defaultLocale`` map to the keys ``_upsert_person`` already reads off ``user``; personal email / mobile / country normally come from PerEmail / PerPhone / company master data (absent pre-sync), so they're stashed under ``_event_*`` keys that ``_upsert_person`` reads as fallbacks. Returns ``None`` when there's nothing at all (preserves the prior ``users.get(...) -> None`` contract for the sync/enrichment path).""" u = dict(base or {}) # Fill-if-FALSY, not setdefault (2026-07-31, P01550 prod repro): # the SF ``get_users`` pull always returns every $select column, # so the key is PRESENT with ``None``/``""`` when the tenant has # no value — ``setdefault`` sees the key and silently discards # the event's carry field. Live consequence: the EU tenant's # User rows carry ``empId: null`` for mirrored people, the # event's ``empId: P01550`` never landed, and the offboard # adoption (which corroborates on empId) fell through to the # refuse-create guard — all five of HRIS's first real offboard # pushes were dropped with a clean summary. def _fill(key: str, value: str | None) -> None: if value and not u.get(key): u[key] = value _fill("firstName", event.first_name) _fill("lastName", event.last_name) if event.display_name: _fill("displayName", event.display_name) elif not u.get("displayName"): derived = " ".join( p for p in (event.first_name, event.last_name) if p ).strip() if derived: u["displayName"] = derived _fill("empId", event.employee_no) _fill("defaultLocale", event.default_locale) _fill("_event_personal_email", event.personal_email) _fill("_event_mobile_phone", event.mobile_phone) _fill("_event_country", event.country) return u or None async def enforce_mass_leaver_brake( self, events: list[SuccessFactorsWebhookEvent], ) -> None: """Raise ``ConflictError`` when a single webhook batch carries more apply-eligible offboard events than the mass-leaver threshold. Extracted so the fast-ACK HTTP handler can run it SYNCHRONOUSLY before spawning the background apply — a corrupted SF payload (or a leaked secret + forged body) that would mass-flip employees to ``leaver_pending`` in one POST must surface as an immediate 409 to SF, not vanish into a background task that only logs the refusal. Mirrors the batch-confirm path, which likewise keeps the brake in the synchronous request (``start_async_confirm``). Threshold matches the manual-confirm / scheduled-incremental paths: ``max(25, 5% of active)``. Cheap — a payload scan plus a single active-person COUNT, no SF round-trips. P1-A fix (2026-06-02 audit): count only events whose ``user_id`` would actually flow to ``_apply_event`` — exclude entries matching ``HR_SKIP_USER_IDS`` (which ``_upsert_person`` short-circuits before any state change). Otherwise 26 known vendor-account offboards + 4 real terminations reads as offboard_count=30 → brake fires → the 4 real leavers stay active past their last working day. The skip-email-domain check needs an SF round-trip and is intentionally NOT folded in here (still conservative) — it's deferred to ``_apply_event``. """ skip_user_ids = self._skip_user_id_set() applyable_offboards = [ e for e in events if e.event_type == "offboard" and (e.user_id or "").strip().lower() not in skip_user_ids ] offboard_count = len(applyable_offboards) if offboard_count == 0: return active_total = await self._count_active_persons_for_instance() threshold = max(25, int(active_total * 0.05)) if offboard_count > threshold: logger.warning( "HRIS webhook refused: %d apply-eligible offboard " "events vs %d active (threshold=%d). Forward to " "manual sync preview/confirm for human review.", offboard_count, active_total, threshold, ) raise ConflictError( f"Mass-leaver brake: webhook batch contains " f"{offboard_count} apply-eligible offboard events, " f"exceeds threshold {threshold} " f"(active_total={active_total}). Refusing to " f"auto-apply — investigate via the manual SF sync " f"preview." ) async def _record_offboard_skipped( self, *, event: SuccessFactorsWebhookEvent, reason_code: str, ) -> None: """Write the ``offboard_skipped`` ledger row for an excluded-reason offboard push. Idempotent via the shared ``_upsert_lifecycle_event`` natural key ``(person, 'offboard_skipped', effective_date)`` — an SF redelivery re-reads the existing row. A real offboard for the same person + date (HR correcting a mis-coded reason) still applies normally: the skip row lives under its OWN event_type, so it never collides with the ``offboard`` dedup key. When Helix has no person row for the userId there is nothing to anchor the ledger to — log only. """ # ``deleted_at IS NULL`` + a deterministic order (#2184). The # sibling helper added in the SAME PR (#2111), # ``_termination_reasons_for``, already filters tombstones; this # one did not, and used an unordered ``db.scalar``. With a # soft-deleted old row beside a live one for the same userId # (the soft-delete-then-reinsert shape is proven in the # external-identity domain, and identity_persons has no unique # constraint spanning soft deletes), the ledger row could anchor # to the tombstone — so a later audit asking "did we see and # skip this person?" finds nothing under the live person. stmt = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == event.user_id, IdentityPerson.deleted_at.is_(None), ).order_by(IdentityPerson.created_at.desc()) if self.hris_instance_id is not None: stmt = stmt.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) person = await self.db.scalar(stmt) if person is None and self.hris_instance_id is not None: # Cross-tenant fallback (2026-07-31): terminations are often # entered in the tenant that does NOT own the Helix row (the # P01550 cohort — see the offboard adoption branch in # ``_upsert_person``). The skip ledger row is pure audit — no # status change, no ownership change — so anchor it to the # person wherever they're pinned rather than dropping it. # #2233: mirror the primary query's tombstone filter and # ordering. Without them this fallback re-opens exactly the # failure #2184 closed — a soft-deleted/purged row (the # rogerzhou02 shape) wins the lookup and the ``offboard_skipped`` # ledger row anchors to a tombstone, invisible to every audit # query that filters live people. person = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == event.user_id, IdentityPerson.deleted_at.is_(None), ).order_by(IdentityPerson.created_at.desc()) ) if person is None: logger.info( "process_webhook_events: excluded-reason offboard for " "user_id=%s (reason %s) matches no Helix person — nothing " "to record", event.user_id, reason_code, ) return await self._upsert_lifecycle_event( person=person, event_type=EVENT_TYPE_OFFBOARD_SKIPPED, effective_date=event.effective_date, employment={ "userId": event.user_id, "personIdExternal": event.person_id_external, "endDate": event.effective_date.isoformat(), }, job=None, user=None, trigger_source="webhook", extra_detail={ "termination_reason": reason_code, "skip_reason": "excluded_termination_reason", }, ) async def process_webhook_events( self, events: list[SuccessFactorsWebhookEvent], ) -> SuccessFactorsWebhookSummary: """Apply a batch of SF-pushed lifecycle events end-to-end. For each onboard/offboard event: 1. ``_apply_event`` upserts person + lifecycle ledger row (tagged ``trigger_source='webhook'``). Idempotent on the natural key ``(person_id, event_type, effective_date)`` — a replay returns ``event_created=False`` and the rest of the per-event loop short-circuits coordinator dispatch. 2. Per-event commit so identity_persons row locks release within ~100ms and the ledger row is durable BEFORE any downstream side effect runs (matches the per-event commit pattern ``confirm_batch`` already uses for the same reason — see service.py:483 commentary). 3. ``_dispatch_lifecycle_coordinators`` fires provisioning / welcome / Google account / identity-sync hooks for onboard and ``start_offboarding_for`` for offboard. Skipped when ``event_created=False`` — the lifecycle has already been handled (replay or sync path got there first). 4. For offboard with a started process: per-event commit, then ``OffboardingService.dispatch_post_commit_actions`` for force-logout + leaver/manager notifications. Failures here log + roll back the dispatch but do not abort the batch — the offboarding row is durable, only start-of-cycle notifications may be partial. Operator uses the offboarding console's "Re-notify" action to retry. Mirrors ``confirm_batch``'s post-commit handling (service.py:818). """ summary = SuccessFactorsWebhookSummary(events_received=len(events)) user_ids = sorted({event.user_id for event in events if event.user_id}) if not user_ids: return summary # Belt-and-suspenders: the fast-ACK HTTP handler already runs the # brake synchronously (so a mass-leaver payload surfaces as an # immediate 409 to SF rather than vanishing into the background # task). Re-check here so every entrypoint — the manual/direct # service call and the background apply — is equally protected. # It's a single cheap COUNT; running it twice on the webhook path # is harmless. await self.enforce_mass_leaver_brake(events) client = await self._get_client() users = { user["userId"]: user for user in await client.get_users(user_ids) if user.get("userId") } jobs_by_user = await self._jobs_by_user_id(user_ids) # Termination-reason exclusion inputs (2026-07-30). The push # payload MAY carry ``termination_reason`` (trusted directly); # for offboards without it, resolve the reason from SF's # EmpEmploymentTermination — HRIS filters excluded reasons on # their side, but their filter is an implicit rule we can't # audit, so Helix re-derives the verdict itself (defence in # depth; a mis-mapped iflow must not close a transferee's # accounts). Cost-gated to zero when nothing is configured. excluded_reason_codes = await self._excluded_termination_reason_codes() webhook_reason_by_uid: dict[str, str] = {} if excluded_reason_codes: _needs_lookup = { e.user_id for e in events if e.event_type == "offboard" and e.user_id and not e.termination_reason } webhook_reason_by_uid = await self._termination_reasons_for( _needs_lookup ) for event in events: if event.event_type not in {"onboard", "offboard"}: continue # HR test-record band (P99XX) — same hard gate ``_stage_event`` # applies on the sync paths (#1917 全链路). Without it an SF # outbound push for a reserved P9900-9999 test record would # sail straight through ``_apply_event`` → ``_upsert_person`` # and create a REAL identity_persons row (webhook was the one # ingest chokepoint the band check missed). Keyed on the SF # userId, the event's own employee_no carry-field, AND the # pulled User's empId — any of the three can hold the test id. if ( is_test_employee_no(event.user_id) or is_test_employee_no(event.employee_no) or is_test_employee_no( (users.get(event.user_id) or {}).get("empId") ) ): logger.info( "process_webhook_events: skipping %s event for " "user_id=%s — HR reserved P99XX test band (not a " "real employee)", event.event_type, event.user_id, ) continue # Excluded termination reason (人员分类转换 / 跨DC调动 / 兼岗 # 结束): the person keeps working — record that we saw the # event and deliberately kept the account open, then move on. # First-class ``offboard_skipped`` ledger row (not a log # line) so both sides can audit the non-action later. if event.event_type == "offboard" and excluded_reason_codes: _reason_code = ( event.termination_reason or webhook_reason_by_uid.get(event.user_id) or "" ).strip().upper() if _reason_code and _reason_code in excluded_reason_codes: try: await self._record_offboard_skipped( event=event, reason_code=_reason_code, ) await self.db.commit() except Exception: # noqa: BLE001 # The SKIP verdict stands even if the audit row # fails — never let a ledger hiccup escalate a # transfer into an account closure. await self.db.rollback() logger.exception( "process_webhook_events: failed to record " "offboard_skipped ledger row for user_id=%s " "(reason %s) — skip verdict unaffected", event.user_id, _reason_code, ) summary.offboards_skipped_reason_excluded += 1 logger.info( "process_webhook_events: NOT offboarding " "user_id=%s — termination reason %s is on the " "no-account-closure exclusion list (internal " "transfer / reclassification)", event.user_id, _reason_code, ) continue # ``personIdExternal`` is the SF *Person* key that PerPhone / # PerEmail / PerPersonal are keyed by — it's how ``_apply_event`` # attaches a new hire's mobile, personal email, birthday & gender # (see the ``person_id_external`` gate in ``_upsert_person``). The # SF webhook body carries only ``userId``, so resolve it here via # EmpEmployment on THIS instance's client when the event omits it. # Without it the person lands with empty mobile_phone / # personal_email — and an empty ``personal_email`` in turn blocks # the welcome-with-password email (which by design refuses to ship # the first-login password to a blank/invalid personal address). # Best-effort: a resolution hiccup just leaves it None (== the # pre-fix behaviour), so it never breaks the apply. # # 2026-08-25: "omits it" now also covers a **self-referential** # carry (``person_id_external == user_id``). SF's outbound iflow # maps that field off the assignment's own userId, so a 兼岗 push # arrives claiming to be its own person — non-empty, so the old # ``if not person_id_external`` skipped the very lookup that # would have returned the REAL person key, and the # concurrent-employment guard in ``_upsert_person`` (which can # only fire on ``personIdExternal != userId``) went blind. Live # consequence: Hiroki Shinagawa's 兼岗 (userId 80100063, # personIdExternal P01008) was created as a second person with # a second mailbox ``hiroki.shinagawa01@pingcap.com`` and a real # Google account. See ``person_id_external_is_usable``. person_id_external = await self._resolve_person_id_external( event.user_id, event.person_id_external, force=True, ) employment = { "userId": event.user_id, "personIdExternal": person_id_external, "lastModifiedDateTime": datetime.now(UTC).isoformat(), "startDate": event.effective_date.isoformat() if event.event_type == "onboard" else None, "endDate": event.effective_date.isoformat() if event.event_type == "offboard" else None, } # Capture PRE-apply person state so the create-vs-reactivate # vs update split is computed against the snapshot SF saw # BEFORE this event landed. Reading after ``_apply_event`` # would always see ``employment_status='active'`` for a new # hire (the upsert just wrote it that way) and the # coordinator chain would route every onboard webhook # through the no-op ``update`` branch — silently dropping # provisioning, the welcome email, Google account creation, # and the identity-sync push. pre_apply_status: str | None = None pre_apply_existed = False pre_apply_scheduled_leaver = False if event.event_type == "onboard": # Instance-scope lookup — mirror ``_upsert_person`` line 2954. # Without the ``hris_instance_id`` filter, sf-cn-prod and # sf-eu-prod sharing a ``userId`` numerically would # mis-attribute a brand-new CN hire to the existing EU # row → action='update' → silent skip of provisioning / # welcome / Google account. Legacy rows with NULL instance # are still adopted (matches ``_upsert_person`` adoption # branch). pre_stmt = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == event.user_id, ) if self.hris_instance_id is not None: pre_stmt = pre_stmt.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) pre = await self.db.scalar(pre_stmt) if pre is None and self.hris_instance_id is not None: # Cross-DC transfer (2026-07-30, HRIS-aligned): 调往 # 中国/德国数据中心 moves the SAME human to the other # SF tenant — the target tenant then pushes an # onboard whose userId matches a person row OWNED BY # THE SOURCE instance, which the scoped lookup above # deliberately misses. Without this fallback the # coordinator chain classifies the transfer as # ``create`` and re-fires the full hire suite # (second corporate mailbox + Google account) for # someone who already has both. Only adopt the # foreign row when the event's legal entity is # canonically OURS (``owned_company_codes`` — the # same verdict ``_upsert_person``'s apply-time # reclaim uses), so two genuinely distinct humans # sharing a numeric userId across tenants still # split correctly (the 2026-05-02 mis-attribution # class this filter was built for). cross_row = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == event.user_id, ) ) if cross_row is not None: _snap = self._select_job_snapshot( "onboard", employment, jobs_by_user.get(event.user_id, []), ) or {} _owner = await self._company_owner_for( _snap.get("company") ) if _owner == self.hris_instance_id: logger.info( "process_webhook_events: onboard for " "user_id=%s matches a person owned by " "hris_instance_id=%s and company %s is " "canonically OURS — treating as cross-DC " "transfer (update/reactivate), not a new " "hire; apply will reclaim ownership", event.user_id, cross_row.hris_instance_id, _snap.get("company"), ) pre = cross_row pre_apply_existed = pre is not None pre_apply_status = pre.employment_status if pre else None pre_apply_hire_date = pre.hire_date if pre else None # Snapshot the notice-period verdict too — after # ``_apply_event`` the row may look different, and the # predicate needs the person's own TZ. pre_apply_scheduled_leaver = ( self._is_inflight_scheduled_leaver(pre) ) try: ( event_created, person_upserted, employee_upserted, prior_trigger_source, ) = await self._apply_event( event_type=event.event_type, employment=employment, user=self._overlay_event_fields( event, users.get(event.user_id), ), jobs=jobs_by_user.get(event.user_id, []), trigger_source="webhook", # Same golden-record write as the confirm path — SF's # outbound event ships the value as ``emailName`` (mapped # from PerPersonal.customString4), so both ingest routes # persist it identically. email_name=event.email_name, ) except Exception: # Per-event isolation mirrors confirm_batch (service.py:873): # one user_id's apply failure must not corrupt the # session for the next event. Log + skip; the next # webhook resend (or the daily reconciliation sync) # will pick up the missed transition. await self.db.rollback() logger.exception( "process_webhook_events: _apply_event failed for " "user_id=%s event_type=%s — skipping; next sync " "tick will reconcile.", event.user_id, event.event_type, ) summary.errors += 1 continue summary.events_created += int(event_created) summary.people_upserted += int(person_upserted) summary.hr_employees_upserted += int(employee_upserted) # Commit per event so the lifecycle ledger row + person # upsert are durable before any coordinator hook runs. # OffboardingService.dispatch_post_commit_actions # specifically requires the offboarding row to be visible # to the worker session it spawns. await self.db.commit() # Resolve this hire's HRBP / HR Ops (SF EmpJobRelationships # 2448/2450) NOW, before the coordinator dispatch below — the # account-creation card (``_notify_admins_google_account_outcome``) # and both new-hire mails read # ``person.hris_hrbp_person_id`` / ``hris_hrops_person_id`` to # pick recipients, and on the webhook path those columns are # otherwise NULL until the next full-roster sync (so they would # reach only the IT contact, never the hire's HRBP/HR Ops — the # 2026-07-06 Camila gap). ``_dispatch_lifecycle_coordinators`` # makes the same call, but it is gated on ``event_created`` and # runs further down: keeping it here means the columns are also # filled for a replayed webhook that dispatches nothing. The # helper skips the SF pull when both columns already resolved, so # the second call on the dispatching path costs nothing. Onboard # only — that's the sole event that fires the card. if event.event_type == "onboard": await self._resolve_hire_hr_relationships(event.user_id) # A webhook can carry the same HRIS start-date correction as the # staged sync path. Keep the existing onboarding process aligned # without treating this active employee as a new hire. if event.event_type == "onboard" and pre_apply_hire_date is not None: current_hire_date = await self.db.scalar( select(IdentityPerson.hire_date).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == event.user_id, ) ) if current_hire_date != pre_apply_hire_date: from app.core.coordinators import hr_sync_coordinator await hr_sync_coordinator.refresh_onboarding_for_hire_date_change( self.db, event.user_id, ) await self.db.commit() # Dispatch when EITHER: (a) brand-new lifecycle row, or # (b) prior row was ``initial_seed`` (bootstrap, dispatch # suppressed by design). _upsert_lifecycle_event upgrades # the row's trigger_source to 'webhook' in-place so a # subsequent replay sees the post-dispatch state. upgraded_from_seed = prior_trigger_source == "initial_seed" # (c) the row exists but its coordinators never ran — same # committed-then-dispatched hole as confirm_batch (#2220). The # webhook path's own except handler logs "next sync tick will # reconcile", which was only true for the person upsert, never # for dispatch. pending_dispatch_event_id = self._consume_pending_dispatch() # (#2215) a refused stale offboard creates the audit ledger row # but must NOT start an offboarding process for someone who is # staying active. dispatch_suppressed = self._dispatch_is_suppressed() offboarding_dispatch_target: tuple[UUID, UUID] | None = None if not dispatch_suppressed and ( event_created or upgraded_from_seed or pending_dispatch_event_id is not None ): # Derive ``action`` from the PRE-apply snapshot: # * onboard, person didn't exist → ``create`` # * onboard, person was a leaver → ``reactivate`` # * onboard, person was already active → ``update`` # (shouldn't happen on a healthy SF tenant — Employee # Hire fires once per hire — coordinator no-ops if # it does) # * offboard → ``deactivate`` # # ``upgraded_from_seed=True`` overrides ``update`` to # ``create`` for onboard — the seeded person was # bootstrapped active but provisioning was intentionally # suppressed, so this webhook IS their first real # lifecycle dispatch. if event.event_type == "onboard": if not pre_apply_existed: action = "create" elif pre_apply_scheduled_leaver: # Confirmed future leaver still inside the notice # period — a re-pushed onboard is an attribute # refresh, not a rehire (2026-08-11). Firing the # hire chain here would provision accounts for a # departing employee. action = "update" elif pre_apply_status in ( "leaver_pending", "left", "terminated", ): action = "reactivate" elif upgraded_from_seed: # Bootstrap-seeded but never provisioned — treat # as a fresh ``create`` so all four hire hooks # fire. action = "create" else: action = "update" else: action = "deactivate" # Per-event isolation, mirroring the _apply_event guard # above and confirm_batch's dispatch isolation. The # offboard branch's start_offboarding_for deliberately # does NOT swallow (its docstring), so a raise here — # e.g. a FOR-UPDATE lock-wait timeout or an AppException # from blocker scanning — would otherwise propagate out # of the whole for-loop and silently drop every # remaining event in the webhook batch (SF then replays # the whole batch, wedging on the same event forever). # The person upsert is already committed (above); a # dispatch failure self-heals: ProcessAdvanceService # creates the missing offboarding process at termination # (advance_service._advance_single), and onboard hooks # swallow internally. try: offboarding_dispatch_target = ( await self._dispatch_lifecycle_coordinators( event_type=event.event_type, action=action, hris_user_id=event.user_id, employment=employment, email_name=event.email_name, ) ) # Coordinators ran: stamp the ledger row so a replay # doesn't re-dispatch, and so a FAILED dispatch (the # except below) leaves it un-stamped and re-armed # (#2220). await self._mark_lifecycle_dispatched( pending_dispatch_event_id, ) # Coordinators may have written their own rows # (OnboardingProcess, audit logs, etc.); commit so # the post-commit offboarding dispatch below — and # any concurrent reader — sees a consistent view. await self.db.commit() except Exception: await self.db.rollback() logger.exception( "process_webhook_events: lifecycle-coordinator " "dispatch failed for user_id=%s event_type=%s " "action=%s — person upsert already durable; " "next sync tick / T+0 advance will reconcile.", event.user_id, event.event_type, action, ) summary.errors += 1 offboarding_dispatch_target = None continue if offboarding_dispatch_target is not None: _person_id, _proc_id = offboarding_dispatch_target try: from app.domains.offboarding.service import ( # noqa: PLC0415 OffboardingService, ) _off_svc = OffboardingService(self.db) await _off_svc.dispatch_post_commit_actions( _person_id, _proc_id, trace_id=str(_person_id), actor_user_id=None, ) await self.db.commit() except Exception: # noqa: BLE001 # Mirror confirm_batch (service.py:831): process # row is durable; force-logout / notifications # may be partial. NO automatic retry — operator # must hit the offboarding console 'Re-notify'. logger.exception( "process_webhook_events: offboarding post-" "commit dispatch failed for user_id=%s " "person=%s process=%s — process is durable; " "force-logout / notifications may be partial.", event.user_id, _person_id, _proc_id, ) try: await self.db.rollback() except Exception: # noqa: BLE001 pass return summary async def list_offboarding_cases(self) -> list[dict]: result = await self.db.execute( select(IdentityPerson) .options(selectinload(IdentityPerson.manager)) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.termination_date.is_not(None), IdentityPerson.employment_status.in_(("leaver_pending", "left")), ) .order_by(IdentityPerson.termination_date.asc(), IdentityPerson.display_name.asc()) ) people = list(result.scalars().all()) if not people: return [] person_ids = [person.id for person in people] assets_by_user = await self._assets_by_user_id(person_ids) software_by_user = await self._software_by_user_ids(person_ids) latest_event_by_person = await self._latest_offboard_event_by_person_id(person_ids) return [ self._build_offboarding_case_payload( person=person, assets=assets_by_user.get(person.id, []), software_items=software_by_user.get(person.id, []), latest_event=latest_event_by_person.get(person.id), include_assets=False, ) for person in people ] async def get_offboarding_case(self, user_id: UUID) -> dict: person = await self.db.scalar( select(IdentityPerson) .options(selectinload(IdentityPerson.manager)) .where(IdentityPerson.id == user_id, IdentityPerson.deleted_at.is_(None)) ) if person is None: raise NotFoundError("User", str(user_id)) assets = list( ( await self.db.execute( select(AssetItem) .where(AssetItem.assigned_to == user_id) .order_by(AssetItem.asset_tag.asc()) ) ).scalars().all() ) software_items = await self._software_by_user_id(user_id) latest_event = await self.db.scalar( select(HrLifecycleEvent) .where( HrLifecycleEvent.person_id == user_id, HrLifecycleEvent.event_type == "offboard", ) .order_by(HrLifecycleEvent.effective_date.desc(), HrLifecycleEvent.created_at.desc()) ) return self._build_offboarding_case_payload( person=person, assets=assets, software_items=software_items, latest_event=latest_event, include_assets=True, ) async def update_offboarding_asset_status( self, *, user_id: UUID, asset_id: UUID, status: str, actor_user_id: UUID, note: str | None = None, ) -> dict: asset = await self.db.scalar( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.assigned_to == user_id) ) if asset is None: raise NotFoundError("Asset", str(asset_id)) custom_fields = dict(asset.custom_fields or {}) offboarding = dict(custom_fields.get("offboarding") or {}) offboarding.update( { "status": status, "note": note, "updated_at": datetime.now(UTC).isoformat(), "updated_by": str(actor_user_id), "user_id": str(user_id), } ) custom_fields["offboarding"] = offboarding asset.custom_fields = custom_fields await self.db.flush() return _build_asset_payload(asset) async def update_offboarding_software_status( self, *, user_id: UUID, software_id: UUID, status: str, actor_user_id: UUID, note: str | None = None, ) -> dict: software = await self.db.scalar( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.id == software_id, or_( AssetSoftwareCatalog.owner_person_id == user_id, AssetSoftwareCatalog.it_contact_person_id == user_id, ), ) ) if software is None: raise NotFoundError("Software", str(software_id)) custom_fields = dict(software.custom_fields or {}) offboarding = dict(custom_fields.get("offboarding") or {}) offboarding.update( { "status": status, "note": note, "updated_at": datetime.now(UTC).isoformat(), "updated_by": str(actor_user_id), "user_id": str(user_id), } ) custom_fields["offboarding"] = offboarding software.custom_fields = custom_fields await self.db.flush() return _build_software_payload(software, user_id) async def complete_offboarding( self, *, user_id: UUID, actor_user_id: UUID, note: str | None = None, ) -> dict: person = await self.db.scalar( select(IdentityPerson).where(IdentityPerson.id == user_id, IdentityPerson.deleted_at.is_(None)) ) if person is None: raise NotFoundError("User", str(user_id)) # Check blocking conditions: unresolved assets + owned software blocking_reasons: list[str] = [] assets = list( (await self.db.execute( select(AssetItem).where( AssetItem.deleted_at.is_(None), or_( AssetItem.assigned_to == user_id, AssetItem.borrowed_by == user_id, ), ) )).scalars().all() ) unresolved_asset_count = sum(0 if _is_asset_resolved(a) else 1 for a in assets) if unresolved_asset_count > 0: blocking_reasons.append(f"{unresolved_asset_count} unresolved physical asset(s)") software_items = await self._software_by_user_id(user_id) active_software = [s for s in software_items if s.status == "active"] if active_software: blocking_reasons.append(f"{len(active_software)} active software ownership(s)") # Pre-launch hardening 2026-05-04: also count active SoftwareSeatAllocation, # UserPermission grants, and ApiKey assignments. Catalog-ownership # (owner_person_id / it_contact_person_id) was already counted above # — but a user can hold a seat for software they don't OWN, and an # active permission grant or PAT they shouldn't carry into 'left'. from app.domains.asset.models import SoftwareSeatAllocation active_seat_count = ( await self.db.execute( select(func.count()) .select_from(SoftwareSeatAllocation) .where( SoftwareSeatAllocation.user_id == user_id, SoftwareSeatAllocation.status == "active", ) ) ).scalar() or 0 if active_seat_count: blocking_reasons.append( f"{active_seat_count} active software seat allocation(s)" ) from app.domains.permission.models import UserPermission active_perm_count = ( await self.db.execute( select(func.count()) .select_from(UserPermission) .where( UserPermission.user_id == user_id, UserPermission.status == "active", ) ) ).scalar() or 0 if active_perm_count: blocking_reasons.append( f"{active_perm_count} active permission grant(s)" ) from app.domains.apikey.models import ApiKeyAssignment active_apikey_count = ( await self.db.execute( select(func.count()) .select_from(ApiKeyAssignment) .where( ApiKeyAssignment.assigned_to == user_id, ApiKeyAssignment.status == "active", ) ) ).scalar() or 0 if active_apikey_count: blocking_reasons.append( f"{active_apikey_count} active API key assignment(s)" ) if blocking_reasons: raise ConflictError( f"Cannot complete offboarding: {'; '.join(blocking_reasons)}. " "Please resolve all assets and transfer software ownership before completing." ) # Round 2 P2 fix (CR2-004): mirror the SEC-002 gate from # ``OffboardingService.complete_process``. The legacy # ``hr.complete_offboarding`` route used to flip persons to # ``'left'`` directly without consulting # ``OffboardingProcess.sso_suspended_at``. Resource-level # checks above ensure no orphan ownership, but SSO suspension # is what ACTUALLY severs vendor sessions — without it, a # leaver completed via this route could retain active Google # / Feishu sessions until natural JWT TTL expiry. from app.domains.offboarding.models import ( # noqa: PLC0415 PROC_CANCELLED, PROC_COMPLETED, OffboardingProcess, ) proc = await self.db.scalar( select(OffboardingProcess).where( OffboardingProcess.person_id == user_id, ) ) if ( proc is not None and proc.status not in (PROC_CANCELLED, PROC_COMPLETED) and proc.sso_suspended_at is None ): from app.core.exceptions import AppException # noqa: PLC0415 raise AppException( code="PROCESS_NOT_READY", message=( "Cannot complete: SSO has not been suspended yet " "on the offboarding process. Run suspend_sso " "(or use the offboarding console's complete " "button which enforces the same invariant)." ), status_code=400, ) latest_event = await self.db.scalar( select(HrLifecycleEvent) .where( HrLifecycleEvent.person_id == user_id, HrLifecycleEvent.event_type == "offboard", ) .order_by(HrLifecycleEvent.effective_date.desc(), HrLifecycleEvent.created_at.desc()) ) # The asset / software conflict checks above already enforced # the "no orphan resources" invariant. Route the final flip # through LeaverBlockGuard so the audit row, listener, and # any newly-appeared blocker observations remain consistent # with the canonical path in ProcessAdvanceService. from app.domains.leaver_blocked.guard import ( LEAVER_VIA_ADMIN_API, LeaverBlockGuard, ) person.termination_date = person.termination_date or self.today guard = LeaverBlockGuard(self.db) await guard.flip_to_left( person, via=LEAVER_VIA_ADMIN_API, trace_id="hr.complete_offboarding", actor_user_id=actor_user_id, last_synced_at=datetime.now(UTC), ) if latest_event is not None: detail = dict(latest_event.detail or {}) detail["completion"] = { "completed_at": datetime.now(UTC).isoformat(), "completed_by": str(actor_user_id), "note": note, } latest_event.detail = detail await self.db.flush() return await self.get_offboarding_case(user_id) async def get_lifecycle_overview(self, *, limit: int = 5) -> dict: recent_hires = await self._list_recent_lifecycle_items(event_type="onboard", limit=limit) recent_leavers = await self._list_recent_lifecycle_items(event_type="offboard", limit=limit) return { "recent_hires": recent_hires, "recent_leavers": recent_leavers, } async def list_overdue_leavers(self) -> list[dict]: """List leaver_pending users whose ``termination_date`` has passed *in their own local timezone*. Mirrors the local-tz gate in :meth:`transition_overdue_leavers` so the admin "overdue leavers" dashboard does not surface a US employee whose UTC date has rolled over but who is still mid-last-day in PDT — that would prompt IT to manually shut down accounts the cron is intentionally holding back. """ result = await self.db.execute( select(IdentityPerson) .options(selectinload(IdentityPerson.manager)) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "leaver_pending", IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date <= self.today, ) .order_by(IdentityPerson.termination_date.asc()) ) people = list(result.scalars().all()) now_utc = datetime.now(UTC) out: list[dict] = [] for p in people: tz = resolve_timezone(p) local_today = now_utc.astimezone(tz).date() if local_today <= p.termination_date: continue out.append({ "user_id": str(p.id), "employee_id": p.employee_no, "display_name": p.display_name or p.hris_person_id, "email": p.work_email, "termination_date": p.termination_date.isoformat() if p.termination_date else None, "days_overdue": (local_today - p.termination_date).days, "manager_name": p.manager.display_name if p.manager else None, }) return out async def transition_overdue_leavers(self) -> dict: """Batch-transition leaver_pending → left for users past termination_date. Leaver-Blocked gate: a person is held in ``leaver_pending`` while their ``OffboardingProcess`` is in ``blocked`` or ``scanning``. Only ``ready`` / ``completed`` / ``cancelled`` — or the absence of any process — permits transition to ``left``. The manager-escalation worker (runs before this one in the same tick) is responsible for unblocking pending MUST_TRANSFER blockers by auto-transferring to the manager on leaver_date; it produces an escalation that then flips the process to ``ready`` once the manager accepts. **Local-timezone gate (2026-05-21)**: the comparison is done in the leaver's *own* timezone, resolved from ``IdentityPerson.country`` via :func:`app.core.working_hours.resolve_timezone`. The cron runs on UTC, but a US employee whose ``termination_date='2026-05-21'`` must stay active through 2026-05-21 23:59 PDT — not get cut off at 2026-05-21 00:00 UTC (which is 2026-05-20 17:00 PDT, *before* their last working day even starts locally). Without this, the leaver loses their email box hours before they can send a farewell mail and finish handover. The SQL still pre-filters at UTC `termination_date <= today_utc` to keep the scan cheap; the per-person Python check refines the borderline ±1 day window with the recipient's local date. Returns summary with count, transitioned IDs, held IDs (still blocked), and ``deferred_user_ids`` (past UTC termination_date but still on their local last day). The held list surfaces in admin dashboards as the "leavers past their date but still holding assets" warning. """ from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_SCANNING, OffboardingProcess, ) result = await self.db.execute( select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "leaver_pending", IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date <= self.today, ) ) overdue_people = list(result.scalars().all()) if not overdue_people: return { "transitioned_count": 0, "transitioned_user_ids": [], "held_user_ids": [], "deferred_user_ids": [], } # Filter out anyone whose local date has not yet rolled past # their ``termination_date``. The leaver remains fully active # (mailbox, SSO, SaaS seats) through their entire last working # day in their own time zone; the next hourly tick after their # local midnight rollover picks them up. now_utc = datetime.now(UTC) deferred_ids: list[str] = [] ready_to_transition: list[IdentityPerson] = [] for person in overdue_people: tz = resolve_timezone(person) local_today = now_utc.astimezone(tz).date() if local_today <= person.termination_date: deferred_ids.append(str(person.id)) continue ready_to_transition.append(person) if not ready_to_transition: return { "transitioned_count": 0, "transitioned_user_ids": [], "held_user_ids": [], "deferred_user_ids": deferred_ids, } overdue_people = ready_to_transition # Bulk-load offboarding processes for these people. proc_rows = await self.db.execute( select(OffboardingProcess).where( OffboardingProcess.person_id.in_([p.id for p in overdue_people]) ) ) proc_by_person = {p.person_id: p for p in proc_rows.scalars().all()} # Lazy-import LeaverBlockGuard to avoid the # hr.service ↔ leaver_blocked ↔ offboarding cycle. from app.core.exceptions import LeaverBlockedError from app.domains.leaver_blocked.guard import ( LEAVER_VIA_HRIS_SYNC, LeaverBlockGuard, ) guard = LeaverBlockGuard(self.db) transitioned_ids: list[str] = [] held_ids: list[str] = [] for person in overdue_people: proc = proc_by_person.get(person.id) if proc is not None and proc.status in (PROC_BLOCKED, PROC_SCANNING): held_ids.append(str(person.id)) continue try: await guard.flip_to_left( person, via=LEAVER_VIA_HRIS_SYNC, trace_id="hr.transition_overdue_leavers", actor_user_id=None, last_synced_at=datetime.now(UTC), ) transitioned_ids.append(str(person.id)) except LeaverBlockedError: # Caught by guard; the audit row was written there. # Keep the person in held_ids so the dashboard shows # the obstacle. held_ids.append(str(person.id)) if transitioned_ids: await self.db.flush() return { "transitioned_count": len(transitioned_ids), "transitioned_user_ids": transitioned_ids, "held_user_ids": held_ids, "deferred_user_ids": deferred_ids, } # NOTE: `_start_offboarding_for`, `_provision_for_new_hire`, and the # welcome-notification flow have been extracted to # ``app.core.coordinators.hr_sync_coordinator``. See there for the # current implementation. `confirm_batch` now calls into that module # instead of private methods. The move closes a cross-domain import # violation (hr → offboarding + provisioning). async def _apply_event( self, *, event_type: str, employment: dict, user: dict | None, jobs: list[dict], trigger_source: str = "sync_confirm", email_name: str | None = None, ) -> tuple[bool, bool, bool, str | None]: """Returns ``(event_created, person_upserted, employee_upserted, prior_trigger_source)``. ``prior_trigger_source`` is None on a brand-new ledger row OR when this apply was a benign no-op (e.g. person filtered by HR_SKIP_EMAIL_DOMAINS); when non-None it tells the caller what the previous lifecycle row was tagged with — load-bearing for the "upgrade initial_seed → webhook" coordinator-redispatch gate. """ user_id = employment.get("userId") effective_date = _effective_date(event_type, employment) if not user_id or effective_date is None: return False, False, False, None # Load companies/locations/departments/positions lookups once # per batch — they turn the SF codes on the job snapshot into # human-readable names and seed identity_org_nodes. await self._ensure_master_data_loaded() job_snapshot = self._select_job_snapshot(event_type, employment, jobs) person, person_created = await self._upsert_person( user_id, user, employment, event_type, email_name=email_name, ) if person is None: # ``_upsert_person`` returned None because the resolved # work_email landed in ``HR_SKIP_EMAIL_DOMAINS`` — typically # an SF-side duplicate userId whose canonical row already # owns the real email. Don't write anything else: no manager # link, no dept membership, no external identity, no # lifecycle event. Caller in ``confirm_batch`` will see # ``person_created=False, event_created=False`` and treat # this row as a benign no-op. return False, False, False, None await self._assign_manager(person, job_snapshot) await self._upsert_dept_membership(person, (job_snapshot or {}).get("department")) await self._upsert_external_identity(person, user_id, user) if person_created: # New hires + reactivations need the baseline ``employee`` # role or they land in a role-less limbo where every API # route guarded by RoleRoute returns 403 (including /me). await self._grant_default_employee_role(person) ( event_created, prior_trigger_source, event_row, ) = await self._upsert_lifecycle_event( person=person, event_type=event_type, effective_date=effective_date, employment=employment, job=job_snapshot, user=user, trigger_source=trigger_source, ) # #2220: publish "this ledger row has never had its coordinators # run" out-of-band rather than widening the return tuple (several # tests monkeypatch ``_apply_event`` with 4-tuple fakes). Callers # consume it via ``_consume_pending_dispatch``, which also resets # it, so a monkeypatched apply can never leave a stale True behind. self._pending_dispatch_event_id = ( event_row.id if event_row is not None and event_row.dispatched_at is None and not self._offboard_refused_stale else None ) # #2215: a refused stale offboard must not start an offboarding # process. The ledger row still lands (audit: "SF pushed this and # we refused it"), but the coordinators stay out of it. self._dispatch_suppressed = self._offboard_refused_stale return ( event_created, person_created or person is not None, False, prior_trigger_source, ) async def _resolve_hire_hr_relationships(self, hris_user_id: str) -> bool: """Fill this one hire's ``hris_hrbp_person_id`` / ``hris_hrops_person_id`` from SF ``EmpJobRelationships`` (2448 / 2450) before the onboarding hooks mail anyone. Returns True when a pull ran. Both new-hire mails pick their CC audience off those two columns — the credential mail through ``account_provisioner._resolve_welcome_cc_additions`` and the Day-1 mail through ``HireDayNotifyService._welcome_cc_emails``. A hire's 2448/2450 rows are future-dated in SF, so the tenant-wide as-of-today slice every roster pass reads is empty until the start date and both columns stay NULL. ``effective_as_of=hire_date`` reads the day-1 slice, which is the whole point of doing it here. Scoped to this one userId (~3-row pull) and memoised per service instance, so the webhook path — which calls this once itself and once more through ``_dispatch_lifecycle_coordinators`` — still costs a single pull. The memo is deliberately "already asked", NOT "both columns are already set": a rehire can carry a stale pointer at the HRBP who handled them last time, and skipping the pull on non-NULL would CC that person instead of the current one. Best-effort in the strongest sense: any failure is swallowed after a rollback and leaves the columns as they were — the mails then degrade to the IT contact, exactly as they did before this call existed. Never raises into ``confirm_batch`` / ``process_webhook_events``. """ if hris_user_id in self._hr_rel_resolved: return False try: from app.identity.successfactors_sync import ( # noqa: PLC0415 apply_hris_role_relationships, ) stmt = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == hris_user_id, ) if self.hris_instance_id is not None: stmt = stmt.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, IdentityPerson.hris_instance_id.is_(None), ) ) hire = await self.db.scalar(stmt) if hire is None: return False self._hr_rel_resolved.add(hris_user_id) client = await self._get_client() await apply_hris_role_relationships( self.db, client, [hire], scoped_pull=True, effective_as_of=hire.hire_date, ) await self.db.commit() return True except Exception: # noqa: BLE001 await self.db.rollback() logger.warning( "hr.sync: HRBP/HR Ops resolve failed for user_id=%s — the " "new-hire mails may CC only the IT contact; the next " "whole-roster refresh will backfill the columns.", hris_user_id, exc_info=True, ) return False async def _dispatch_lifecycle_coordinators( self, *, event_type: str, action: str, hris_user_id: str, employment: dict, is_initial_seed: bool = False, email_name: str | None = None, ) -> tuple[UUID, UUID] | None: """Fire onboarding / offboarding coordinators for a newly-applied lifecycle event. Returns ``(person_id, offboarding_process_id)`` when an offboarding process was started, ``None`` otherwise. Called from both ``confirm_batch`` (sync path) and ``process_webhook_events`` (event-driven path). The caller MUST gate the call on ``event_created=True`` so coordinator hooks fire exactly once per ``(person, event_type, effective_date)`` even when both webhook and a follow-up sync observe the same SF transition — this is the "sync = reconciliation only when webhook already handled it" property the architecture relies on. Caller responsibilities (intentionally NOT done here): * commit the per-event DB transaction BEFORE calling ``OffboardingService.dispatch_post_commit_actions`` for the returned target — force-logout + leaver/manager notifications must observe the durable offboarding row. * detect & dispatch role-change separately. Role-change is orthogonal to onboard/offboard: it fires on an ``update`` / ``reactivate`` action where dept/manager/title drifted. Encoding the before/after blob requires staged-row context (``_prior_role``) that the webhook path doesn't carry today. Role-change-via-webhook is tracked as a follow-up to land alongside the SF ``Job Information Change`` event subscription. ``initial_seed=True`` (full-bootstrap sync of an HRIS instance) suppresses all coordinator fires — initial seed is a historical backfill, not a real-time lifecycle signal. """ if is_initial_seed: return None from app.core.coordinators import hr_sync_coordinator if event_type == "onboard" and action in {"create", "reactivate"}: # Resolve HRBP / HR Ops BEFORE any onboarding hook runs. The very # first hook that mails anyone is ``provision_google_account`` → # ``account_provisioner._send_password_email``, whose CC audience # (``_resolve_welcome_cc_additions``) is read off # ``hris_hrbp_person_id`` / ``hris_hrops_person_id``. A hire whose # SF 2448/2450 rows are still future-dated has both columns NULL # right up to their start date, so the credential mail went out # CC'ing only the IT contact while the Day-1 mail — composed from # the same two columns moments later, after some other pass had # filled them — CC'd HR correctly (P10027 刘姝婕 Nikita, # 2026-09-01: HR Ops asked why they never saw her account mail). # Doing it here rather than inside the provisioner keeps the fix # in one place for every hook in this branch. await self._resolve_hire_hr_relationships(hris_user_id) # Reserve + write back the PLANNED corporate email immediately # (NOT deferred for future hires) so HRIS sees the address # before the hire date. ``email_name`` (when the inbound event # carried one) is the top-priority source for the local part. # The live Google account is still minted on hire_date by the # deferred hooks below; a rare collision at creation re-writes # the corrected value to SF. await hr_sync_coordinator.reserve_and_writeback_email( self.db, hris_user_id, email_name=email_name, ) await hr_sync_coordinator.provision_for_new_hire( self.db, hris_user_id, ) # Google account provisioning runs BEFORE the welcome email # so the welcome message references the just-minted # work_email rather than the SF placeholder. Both hooks are # gated on platform_settings switches + future-hire deferral # inside the coordinator; safe to call unconditionally. await hr_sync_coordinator.provision_google_account( self.db, hris_user_id, ) # Feishu account creation — the counterpart to Google. Own # dedicated hook (2026-07-13): onboarding previously had NO # Feishu 建号 step, so new hires outside the department-scoped # feishu-contact directory sync got no Helix-created Feishu # account. Scope-independent + reconcile-first; same gates as # Google. Fire-and-forget inside the coordinator. await hr_sync_coordinator.provision_feishu_account( self.db, hris_user_id, ) await hr_sync_coordinator.send_onboarding_it_info( self.db, hris_user_id, ) # Identity Sync Center event hook: pushes the new hire to # every enabled downstream (Feishu contact, Google contact, # …) within seconds rather than waiting on the daily # identity_sync cron. Coordinator gates on ``hire_date <= # today`` so future-dated hires don't surface in vendor # directories ahead of their start date. await hr_sync_coordinator.enqueue_identity_sync_for_hire( self.db, hris_user_id, ) # GitHub 组织自动加入 —— 按部门子树规则把新人加进该进的组织。 # 自己的协调器而不是 hr_sync_coordinator 的一个方法:它属于 # GitHub 域的生命周期,而 hr_sync_coordinator 已经是这条链路上 # 最长的文件之一。三道入职闸(未来入职 / 陈旧补录 / 墓碑)在 # 协调器里原样复用 hr_sync_coordinator 的判断,不重写。 from app.core.coordinators import ( # noqa: PLC0415 github_org_coordinator, ) await github_org_coordinator.apply_auto_join_for_new_hire( self.db, hris_user_id, ) return None if event_type == "offboard" and action == "deactivate": return await hr_sync_coordinator.start_offboarding_for( self.db, hris_user_id, employment, ) return None _PLACEHOLDER_EMAILS = frozenset({ "dummy@sap.com", "noreply@sap.com", "placeholder@sap.com", }) @classmethod def is_placeholder_email(cls, email: str | None) -> bool: """True when ``email`` is known to be a synthetic SF stub. Covers the exact matches in ``_PLACEHOLDER_EMAILS``, PingCAP's numbered stub pattern ``dummy@`` that SF fills in for employees without a corporate mailbox, and the platform's own ``@successfactors.local`` fallback. Used by both the UNIQUE-collision fallback and the config-gated preview skip. """ if not email: return True lowered = email.strip().lower() if not lowered: return True if lowered in cls._PLACEHOLDER_EMAILS: return True if lowered.endswith("@successfactors.local"): return True local = lowered.split("@", 1)[0] # ``dummy*`` — PingCAP SF fills this in for employees who have # no real mailbox yet. Covers ``dummy@pingcap.cn`` (the # numbered pattern) as well as plain ``dummy@qq.cn`` / # ``dummy@sap123.com`` variants that admin-style service rows use. if local.startswith("dummy"): return True return False @classmethod def _skip_user_id_set(cls) -> frozenset[str]: return frozenset( uid.strip().lower() for uid in (settings.HR_SKIP_USER_IDS or "").split(",") if uid.strip() ) @classmethod def _skip_email_domain_set(cls) -> frozenset[str]: """Domains from ``HR_SKIP_EMAIL_DOMAINS``, lower-cased and ``@``-stripped. ``"@gaiaworks.cn"`` and ``"gaiaworks.cn"`` both normalize to ``"gaiaworks.cn"`` so admins can't trip on the leading ``@``. """ return frozenset( d.strip().lower().lstrip("@") for d in (settings.HR_SKIP_EMAIL_DOMAINS or "").split(",") if d.strip() ) @classmethod def email_domain_in_skip_list(cls, email: str | None) -> bool: """True when ``email`` falls under a configured skip domain. Exact-match the domain plus any subdomain — so configuring ``gaiaworks.cn`` covers ``COMPadmin@partner.gaiaworks.cn`` without a separate entry. The check is intentionally conservative: domains must be configured explicitly (no built-in default for e.g. ``successfactors.local`` — that lives in the env var so the admin can audit the full skip set in one place). """ if not email: return False lowered = email.strip().lower() if "@" not in lowered: return False domain = lowered.rsplit("@", 1)[1] if not domain: return False skip = cls._skip_email_domain_set() if domain in skip: return True return any(domain.endswith("." + d) for d in skip) def should_skip_user(self, user_id: str | None, user: dict | None) -> bool: """True when this SF user should be dropped at preview time. Three non-employee cases the placeholder-email detector misses: (a) SF built-in service accounts (``APIadmin``, ``ONBadmin``, …) whose ``userId`` matches the configured skip list; (b) users whose email is a known placeholder when ``HR_SKIP_PLACEHOLDER_EMAILS`` is on; (c) users whose SF-returned email lives on a vendor domain configured via ``HR_SKIP_EMAIL_DOMAINS`` (SAP gaiaworks implementation-partner accounts on PingCAP 2026-05). Only consulted for onboard events — update/offboard paths must still process existing rows so terminations propagate. """ if user_id and user_id.strip().lower() in self._skip_user_id_set(): return True email = (user or {}).get("email") if settings.HR_SKIP_PLACEHOLDER_EMAILS: # Skip only on a POSITIVE placeholder match (``dummy*``, a # known stub, ``*@successfactors.local``) — NOT on a merely # absent/empty email. ``is_placeholder_email`` deliberately # returns True for None/"" so the UNIQUE-collision resolver # falls back to the synthetic address, but that must NOT mean # "drop this person from onboarding": a brand-new hire # legitimately has no corporate mailbox in SF yet (Helix mints # it from their name). Treating absent-email as a skip reason # silently dropped EU hire P10013 Robert Wood — real EmpJob + # future hireDate, email=null — from onboarding entirely # (2026-07-12). Genuine SF service accounts are caught by the # user_id skip list above and the domain skip below. stripped = (email or "").strip() if stripped and self.is_placeholder_email(stripped): return True if self.email_domain_in_skip_list(email): return True return False async def _resolve_unique_work_email( self, *, user_id: str, desired: str | None, current_person_id: str | None, current_email: str | None = None, ) -> str: """Pick a work_email that won't collide with the UNIQUE constraint. SAP SF commonly stuffs a shared placeholder like ``dummy@sap.com`` into the email field for records that have no real mailbox yet — a full-employee sync then crashes on the second duplicate insert. We fall back to a synthetic ``@successfactors.local`` whenever the desired value is empty, a known placeholder, or already owned by someone else. ``current_email`` is the existing ``person.work_email`` for an update; when supplied AND it's already a real (non-placeholder) value, we will keep it rather than downgrade to a placeholder if the SF-returned email collides or is empty. Without this guard, a transient SF dup-userId state observed once would persistently flip a real work_email to ``@successfactors.local``, leaving a real employee with a synthetic mailbox even after the SF-side dup is cleaned up. Confirmed in prod 2026-05-07: rows 100113 / 80100057 had real provider_email on their external_identity row but ``identity_persons.work_email`` had decayed to the placeholder. Every fallback to the placeholder logs a ``WARNING`` so the next identity-sync tick produces a visible signal for SRE / on-call rather than silently rotting in the DB. (Yes, the verbose log is intentional — placeholders should be rare and worth noticing.) """ # Normalize to lowercase: MySQL/TiDB collation on work_email is # case-insensitive (utf8mb4_general_ci), so a UNIQUE collision on # "John@x.com" vs "john@x.com" surfaces at flush time even though # the Python-level == check would treat them as different. Lower- # casing here keeps the application view in sync with the DB # uniqueness contract and prevents the swallowed-exception path # from leaving a leaver with a stale placeholder email. fallback = f"{user_id}@successfactors.local".lower() candidate = (desired or "").strip().lower() or fallback corporate_domains = await get_corporate_email_domains(self.db) def _keep_or_fallback(reason: str) -> str: """Either keep an already-real ``current_email`` (no downgrade) or warn-then-return the synthetic placeholder. ``reason`` is already-formatted human text describing why we got here.""" if current_email and not self.is_placeholder_email(current_email): logger.event( "HRIS_EMAIL_COLLISION_EXISTING_PRESERVED", { "reason": reason, "current_email": current_email, "user_id": user_id, }, level=logging.WARNING, ) return current_email.strip().lower() logger.event( "HRIS_PLACEHOLDER_EMAIL_ASSIGNED", { "reason": reason, "placeholder_email": fallback, "user_id": user_id, }, level=logging.WARNING, ) return fallback if self.is_placeholder_email(candidate): return _keep_or_fallback( f"SF returned no real email (desired={desired!r} is empty or placeholder)", ) # Corporate-mailbox protection: SF ``User.email`` is the person's # *primary* email, which stays a PERSONAL address (个人邮箱, e.g. # ``...@qq.com``) even after Helix has minted the real corporate # mailbox and written it to work_email. Without this guard the next # sync would flip work_email back to the personal address — # poisoning SSO matching / welcome-email / HRIS write-back / the # UNIQUE index. So an incoming NON-corporate email must never # OVERWRITE an existing real corporate work_email. (When there is # no corporate value to protect we keep the historical behaviour # and let the address flow through; the Google provisioner's own # corporate-domain guard in ``_local_part_from_hris_email`` still # derives the real mailbox from the name rather than the personal # local part.) See app.core.corporate_domains. if ( not email_is_corporate(candidate, corporate_domains) and current_email and not self.is_placeholder_email(current_email) and email_is_corporate(current_email, corporate_domains) ): logger.warning( "hr.sync: SF email %r is personal/non-corporate — keeping " "existing corporate work_email %r for user_id=%s " "(not overwriting)", candidate, current_email, user_id, ) return current_email.strip().lower() # Google-backed corporate-email protection (2026-07-12, the Julia # P01572 / roger.zhou03 lesson): when the incoming SF value and the # existing work_email are BOTH corporate but DIFFER, and the # existing one is backed by an ACTIVE Google login identity # (``provider_email`` matches — i.e. that mailbox really exists and # the person signs in with it), keep the existing address. Google # is authoritative for corporate mailboxes (Helix mints them and # writes them back to SF); an SF corporate value that disagrees # with a live, login-verified mailbox is an SF data-entry error # (Julia's SF row carried another hire's cloned roger.zhou03 for # weeks — every sync tick re-poisoned the fixed work_email and # broke her SSO match). A LEGITIMATE mailbox rename flows through # Google first, which updates ``provider_email`` — after which the # SF value agrees and this guard is a no-op. if ( current_email and email_is_corporate(current_email, corporate_domains) and candidate != current_email.strip().lower() and current_person_id ): google_backed = await self.db.scalar( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == current_person_id, IdentityExternalIdentity.provider == "google", IdentityExternalIdentity.identity_status == "active", IdentityExternalIdentity.deleted_at.is_(None), IdentityExternalIdentity.provider_email == current_email.strip().lower(), ) ) if google_backed is not None: logger.warning( "hr.sync: SF corporate email %r disagrees with the " "Google-login-backed work_email %r for user_id=%s — " "keeping the Google-verified address (fix the SF " "source; see Julia/roger.zhou03 2026-07-08)", candidate, current_email, user_id, ) return current_email.strip().lower() existing = await self.db.scalar( select(IdentityPerson).where(IdentityPerson.work_email == candidate) ) if existing is None or str(existing.id) == str(current_person_id): return candidate return _keep_or_fallback( f"SF email {candidate!r} collides with existing person {existing.id}", ) async def _apply_employee_no( self, person: IdentityPerson, new_empid: str | None, ) -> None: """Update ``person.employee_no`` unless another row already owns it. Cross-tenant shared empId: when two SF tenants both carry the same global empId (e.g., P0876 present in both CN and EU), the row created by the first-confirmed tenant locks the UNIQUE index. A subsequent confirm that found its person via ``hris_person_id`` would then crash on ``uq_identity_persons_employee_no`` as soon as autoflush tried to write the same empId onto the second row (seen 2026-04-21). Leave the second row's employee_no untouched and log a warning so admin can merge the two IdentityPerson rows. """ if not new_empid or new_empid == person.employee_no: return conflict = await self.db.scalar( select(IdentityPerson.id).where( IdentityPerson.source_system == "successfactors", IdentityPerson.employee_no == new_empid, IdentityPerson.id != person.id, ) ) if conflict is not None: logger.event( "HRIS_EMPLOYEE_NUMBER_CONFLICT", { "employee_number": new_empid, "conflicting_person_id": conflict, "person_id": person.id, "current_employee_number": person.employee_no, }, level=logging.WARNING, ) return person.employee_no = new_empid def _apply_job_snapshot_fields( self, person: IdentityPerson, job_snapshot: dict, ) -> None: """Apply the EmpJob-snapshot fields (everything derivable from one ``EmpJob`` row + the master-data caches) onto ``person``. Extracted from ``_upsert_person`` so the 24h-lead pre-start enrich path can apply the SAME field set from an effective-dated pull. Both callers first seed ``_ensure_master_data_loaded`` (company / location / position / job-function / job-family caches). Every write is write-when-present / cache-guarded exactly as the confirm path always was — an empty snapshot or empty cache never NULL-outs a good value, so calling this early (pre-start) then again on the hire date is a safe no-op-or-refine. Does NOT touch department membership or manager (``_upsert_dept_membership`` / ``_assign_manager``) or the contact/preference block (personIdExternal-keyed, seed-once) — those stay with their own call sites. """ position_code = job_snapshot.get("position") position_name = None if position_code and self._positions: position_name = (self._positions.get(position_code) or {}).get( "externalName_defaultValue" ) # The first three fallbacks are cache-free and shared with # ``_direct_job_title`` (staging's drift check); the last two need # ``self._positions``, which only exists at confirm time. Keeping # the shared prefix in one function is what stops the two sides # from silently drifting apart. new_title = ( self._direct_job_title(job_snapshot) or position_name or position_code ) if new_title: person.job_title = new_title company_code = job_snapshot.get("company") if company_code: person.company_code = company_code company_row = (self._companies or {}).get(company_code) or {} if self._companies: person.company_name = ( company_row.get("name") or person.company_name ) # country — the person's ACTUAL WORK country from # EmpJob.customString16 (HR-confirmed 2026-07-02: this is # the "实际工作国家" field, ISO-3 like CHN/JPN/USA). Takes # precedence over FOCompany.country (the company ENTITY's # country) so a cross-border hire — e.g. US-entity staff # working in China — gets their real work country on the # Feishu Region field + mobile dialing. Falls back to the # company country only when customString16 is empty, so a # coverage gap never blanks the field. (Was company-only.) work_country = ( job_snapshot.get("customString16") or "" ).strip().upper() if work_country: person.country = work_country elif company_row.get("country"): person.country = company_row["country"] # 是否中国主体(劳动合同主体). Previously this flag was written ONLY # by the dormant bootstrap sync — the live hire path left it at # its ``False`` default, so day-one consumers (welcome language + # 汇联易/Navan reimbursement tool, HR Ops 2026-07-15) saw a stale # value. Maintain it HERE too, from the FOCompany (劳动合同主体) # country with a CN work-location fallback, mirroring # ``successfactors_sync._is_china_employee``. # Guarded on a populated FOCompany cache like the adjacent # company_name/country writes (2026-07-16): with an EMPTY cache # every company_row is {}, so an unguarded write would flip a # correct True to False for every CN-entity employee on a sync # tick whose FOCompany fetch failed. Cache present but row # missing → still write (genuinely unknown company ⇒ derive # from work location alone, current behavior). if self._companies: person.is_china_employee = _is_china_employee( company_row, job_snapshot.get("location"), ) # ``cost_center`` ← SF ``EmpJob.costCenter`` (already in the # SELECT field set of both ``list_current_jobs`` and # ``get_jobs_for_users``). Required by the Identity Sync # Center → Navan target's ``enterprise.costCenter`` mapping # rule; without this write every employee renders an empty # value and finance loses the cost-center column in Navan's # T&E reports. NULL preserved on absent values so a finance- # team correction made directly in Helix doesn't get # overwritten by an empty SF response. cc = job_snapshot.get("costCenter") if cc: person.cost_center = cc # 区域邮件组/办公室 + 员工类型 — SF EmpJob.customString7 / # customString25 (migration 344). Stored RAW; the feishu_contact # target value-maps translate them (2659 → "SH Office", 1070 → # Feishu employee_type enum). NULL preserved on absent values so # a downstream correction isn't clobbered by an empty SF response. office_code = job_snapshot.get("customString7") if office_code: person.office_code = office_code # SF-verbatim office label (EmailGroup picklist), pushed to # Google Employee_Details.Office. 2659 → 'sh-office@pingcap.com'. # No Helix value-map — the string is whatever SF translates to. office_label = (self._email_group_labels or {}).get(office_code) if office_label: person.office_mail_group = office_label employee_type_code = job_snapshot.get("customString25") if employee_type_code: person.employee_type_code = employee_type_code # 是否推送到下游系统 — SF EmpJob.customString24 (migration 390). # Stored RAW for review; NOT yet used to gate the sync (that's a # deliberate follow-up once HR/SF confirm 2059=推送 / 2061=不推送). # NULL preserved on absent so an existing value isn't clobbered. downstream_push = job_snapshot.get("customString24") if downstream_push: person.sf_downstream_push_code = downstream_push # International assignment / 外派 — SF EmpJob custom fields (IDs # confirmed via live $metadata labels 2026-07-07, migration 392). # Supplementary context for HR reports / person detail; does NOT # repoint directory placement (office / company / dept stay as the # home job above). Write-when-present, never NULL-out. ISO-3 country # codes + status code stored RAW; dates parsed from SF /Date(ms)/. home_country = job_snapshot.get("customString44") # 外派前国家 if home_country: person.assignment_home_country = home_country host_country = job_snapshot.get("customString38") # 外派后国家 if host_country: person.assignment_host_country = host_country assignment_status = job_snapshot.get("customString39") # 外派状态 if assignment_status: person.assignment_status_code = assignment_status assignment_start = _parse_date(job_snapshot.get("customDate2")) if assignment_start: person.assignment_start_date = assignment_start assignment_end = _parse_date(job_snapshot.get("customDate4")) if assignment_end: person.assignment_end_date = assignment_end # GitHub handle ← SF cust_Other.cust_id, keyed by employee_no # (externalCode), from the master-data cache (migration 369). # Write-when-present, never NULL-out: a person without an entry # (the majority — sparse onboarding field) keeps whatever Feishu # already has. → Feishu ``GitHub ID`` custom attr. # ``github_id_locked_at`` (migration 579) is the self-service # override: non-NULL means an employee changed their handle in Helix # and it was approved, so Helix is authoritative until SF catches up. # Without this branch the write-when-present rule above SILENTLY # reverts that change on the next sync — no notification, no audit # trail the employee can see, they just find their handle back to the # onboarding-form value. Comparison is case-insensitive because GitHub # handles are; when SF agrees we take ITS casing and hand authority # back, so "SF converged" is observed rather than assumed. if self._github_ids and person.employee_no: gh = self._github_ids.get(person.employee_no) if gh: locked = person.github_id_locked_at is not None same = (gh or "").strip().lower() == ( person.github_id or "" ).strip().lower() if not locked or same: person.github_id = gh if locked: person.github_id_locked_at = None else: logger.info( "hr_sync: keeping self-service github_id for " "employee_no=%s (HRIS still says %r) — the HRIS " "write-back has not landed yet", person.employee_no, gh, ) # Job Function / Job Family ← SF EmpJob.customString19 / customString21 # FO externalCodes, resolved to the SF label verbatim (FOJobFunction # / cust_JobFamily). Pushed to Google Employee_Details. NULL preserved # on absent source so a value isn't clobbered empty. jf_code = job_snapshot.get("customString19") if jf_code: jf_label = (self._job_functions or {}).get(jf_code) if jf_label: person.job_function = jf_label jfm_code = job_snapshot.get("customString21") if jfm_code: jfm_label = (self._job_families or {}).get(jfm_code) if jfm_label: person.job_family = jfm_label # ``fte`` ← SF ``EmpJob.fte`` (migration 240). 100% populated # in prod; 99.6% are 1.0. Needed for 年假 / payroll / capacity # proration when a person is part-time. Stored as Decimal(3,2) # so 0.5 / 0.75 etc. round-trip cleanly. fte_raw = job_snapshot.get("fte") if fte_raw is not None and fte_raw != "": try: person.fte = Decimal(str(fte_raw)) except (InvalidOperation, ValueError): # SF returned something non-numeric — skip, keep prior # value rather than crash the whole upsert. logger.warning( "hr.sync: ignoring non-numeric EmpJob.fte %r for %s", fte_raw, person.hris_person_id, ) # ``probation_end_date`` ← SF ``EmpJob.probationPeriodEndDate`` # (migration 240). 72% populated; needed for 转正 流程 / 试用期 # 福利年假 起算 / IT 设备 长期分配 vs 临时分配 gating. Only # written when SF carries a value — never NULL-out a previous # value (HR may correct in SF after the field was first left # blank on a contractor, etc.). probation_end = _parse_date(job_snapshot.get("probationPeriodEndDate")) if probation_end: person.probation_end_date = probation_end location_code = job_snapshot.get("location") if location_code: person.work_location_code = location_code # residence_country ← ISO-3 prefix of the FOLocation code # (CHN0075 → CHN → CN), normalized to ISO-2. This is the geo # axis the g-all-china / g-all-abroad groups key on, and the # value Google Employee_Details.Country gets (99% matches the # legacy Authing-written value). Distinct from ``country`` # (FOCompany legal entity). resid = normalize_country_code(location_code[:3]) if resid: person.residence_country = resid if self._locations: location_row = self._locations.get(location_code) or {} location_name = location_row.get("name") if location_name: person.work_location_name = location_name # city (migration 048) — FOLocation.name is the # city-level label on PingCAP tenants (``Delhi``, # ``Beijing``, ``Mumbai``). Same string is already # used for work_location_name; splitting them # keeps the UI free to render either field. person.city = location_name async def _upsert_person( self, user_id: str, user: dict | None, employment: dict, event_type: str, email_name: str | None = None, ) -> tuple[IdentityPerson | None, bool]: # #2215: reset per call — set by the stale-endDate refusal so the # callers can skip coordinator dispatch for a termination we did # NOT apply (an OffboardingProcess for a person who stays # ``active`` is the exact inconsistent state the P0779 repro hit). self._offboard_refused_stale = False # Duplicate-employment short-circuit (2026-08-18) — MUST precede # the person lookup below. A registered alias means this userId is # a second employment of somebody Helix already has; the lookup # would otherwise (a) create a ghost person row with a synthetic # ``@successfactors.local`` email, or (b) find and # RESURRECT a previously-retired ghost, because the onboard branch # clears ``deleted_at`` / ``hris_record_state='purged'`` on touch. # That resurrection is why the 2026-05-07 ghost cleanup never # stuck. Mirrors the ``_stage_event`` guard so the webhook path # (which does not stage) is covered too. from app.identity.identity_reconcile import hris_user_alias_owner _alias_owner = await hris_user_alias_owner(self.db, user_id) if _alias_owner is not None: logger.info( "hr.sync: skipping person upsert for user_id=%s — duplicate " "employment of person %s (alias)", user_id, _alias_owner.id, ) return None, False # Scope the lookup to the owning integration when running under # multi-instance mode so CN and EU SF instances never collide on # a shared userId. In single-instance mode ``hris_instance_id`` # stays NULL and the scope falls back to the legacy "any # successfactors row" semantics. stmt = select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, ) if self.hris_instance_id is not None: stmt = stmt.where( or_( IdentityPerson.hris_instance_id == self.hris_instance_id, # Legacy rows imported before multi-instance was # introduced — adopt them into this instance on the # first touch so future syncs stay scoped. IdentityPerson.hris_instance_id.is_(None), ) ) person = await self.db.scalar(stmt) # Grave-collision guard (2026-09-04, P01412 战捷) — the apply-side # twin of the ``_stage_event`` check. Kept on BOTH sides because a # staged row can be confirmed manually hours later, and because # the webhook path never stages at all. if await self._claim_tombstoned_user_id( user_id, employment, matched=person, event_type=event_type, ): return None, False # Onboard-poach recovery, userId-keyed half (2026-07-16, Camila # P10009 follow-up). A mirrored pre-start hire that first # surfaced in the NON-owning tenant's incremental window was # created with that tenant's ``hris_instance_id`` — the # instance-scoped lookup above misses the row, and without this # block the create path below would either duplicate the person # or (via the empId fallback) hit the active-row ConflictError. # When the incoming job snapshot's company code says THIS # instance canonically owns the person, adopt the foreign row: # repoint ``hris_instance_id`` and continue as an update. The # preview-time twin in ``_stage_event`` lets exactly this shape # through (and defers the non-owner's create in the first place). _snap_company = ( ((employment.get("_jobs_snapshot") or [None])[0] or {}).get("company") ) if ( person is None and event_type == "onboard" and self.hris_instance_id is not None and _snap_company and await self._company_owner_for(_snap_company) == self.hris_instance_id ): foreign_row = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, IdentityPerson.hris_instance_id != self.hris_instance_id, ) ) if foreign_row is not None: logger.warning( "hr.sync: reclaiming person user_id=%s from " "hris_instance=%s to %s — company code %s is " "canonically owned by this instance (onboard-poach " "recovery)", user_id, foreign_row.hris_instance_id, self.hris_instance_id, _snap_company, ) foreign_row.hris_instance_id = self.hris_instance_id person = foreign_row # #2218: cross-tenant offboard adoption re-pins ``hris_instance_id`` # ONLY once the termination is actually applied. The endDate's # credibility is adjudicated hundreds of lines below (the stale- # endDate gate), and ``process_webhook_events`` commits per event # regardless of that verdict — an eager re-pin therefore committed # "person still active, but now pinned to the pushing tenant" for # every refused offboard, after which the owning tenant's nightly # sync skips the row on its cross-instance guard and nobody ever # closes the account. Adoption sites below set this to the previous # owner; the offboard apply branch performs the actual re-pin. deferred_repin_from: UUID | None = None # Cross-tenant offboard adoption, userId-keyed half (2026-07-31, # second prod repro of the P01550 cohort). The empId-keyed # adoption below requires ``user.empId`` — but the EU tenant's # User rows carry ``empId: null`` for mirrored people, so all # five of HRIS's first real offboard pushes slid past it into # the refuse-create guard and were dropped with a clean summary. # The userId itself matched a real person the whole time (the # ``_diagnose`` fallback resolves exactly this way); userId # alone is not proof across tenants (documented numeric-band # collisions), so require ONE corroborating identity fact — # global 工号, corporate email, or exact display name — before # adopting. Same substantiated-endDate bar and same load-bearing # re-pin rationale as the empId-keyed branch. if ( person is None and event_type == "offboard" and self.hris_instance_id is not None and employment.get("endDate") ): foreign_row = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == user_id, IdentityPerson.hris_instance_id != self.hris_instance_id, ) ) if foreign_row is not None: _u = user or {} _empid_match = bool( (empid := str(_u.get("empId") or "").strip()) and empid == (foreign_row.employee_no or "") ) _email_match = bool( (mail := str(_u.get("email") or "").strip().lower()) and mail == (foreign_row.work_email or "").lower() ) _name_match = bool( (dname := str(_u.get("displayName") or "").strip()) and dname == (foreign_row.display_name or "").strip() ) if _empid_match or _email_match or _name_match: logger.warning( "hr.sync: adopting cross-instance row user_id=%s " "from hris_instance=%s to %s on offboard " "(endDate=%r, corroborated by %s) — the " "termination was entered in THIS tenant while " "the person row was pinned to the other; without " "adoption the offboard is dropped and nothing " "ever closes the account", user_id, foreign_row.hris_instance_id, self.hris_instance_id, employment.get("endDate"), "empId" if _empid_match else ("email" if _email_match else "displayName"), ) # Re-pin deferred to the offboard apply branch (#2218). deferred_repin_from = foreign_row.hris_instance_id person = foreign_row else: logger.warning( "hr.sync: offboard for user_id=%s matches a row " "on hris_instance=%s by userId ONLY — no " "corroborating empId/email/displayName; refusing " "cross-tenant adoption (possible id collision), " "dropping the event", user_id, foreign_row.hris_instance_id, ) # #2219: actually drop it. Falling through leaves # ``person is None`` with no empId (this branch's own # precondition), so the create path below INSERTs a # second row on the ``(source_system, hris_person_id)`` # pair ``foreign_row`` already owns — guaranteed # IntegrityError on ``uq_identity_persons_source``. The # deliberate refusal then surfaces as an infrastructure # error (``summary.errors += 1`` + a misleading # "next sync tick will reconcile" traceback) and trips # error-rate alerting. Return the same skip shape the # skip-domain refusal below uses. return None, False # Cross-instance dedup: the same employee can exist in both # CN and EU SF tenants with different ``userId`` but the SAME # global ``empId``. Without this lookup, the second instance's # confirm would crash on ``uq_identity_persons_employee_no`` # (seen 2026-04-21: EU confirmed first → created P0876; CN # confirm then blew up trying to create its own P0876). Treat # equal empId as the same person, merge updates onto the row. desired_empid = ((user or {}).get("empId") or "").strip() if person is None and desired_empid: person = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.employee_no == desired_empid, ) ) # Pre-launch P1 fix: if we matched on empId across # tenants, refuse to overwrite the OTHER tenant's # ownership. Previously the second tenant's sync would # silently mutate job_title / manager / work_email on a # row whose ``hris_instance_id`` still pointed at the # FIRST tenant — the row would lie about which tenant is # authoritative until someone investigated. Surface a # ConflictError so the operator decides whether to keep # the dual-record (rare; require manual ``hris_instance_id`` # rewrite) or mark one tenant authoritative. if ( person is not None and self.hris_instance_id is not None and person.hris_instance_id is not None and person.hris_instance_id != self.hris_instance_id ): # Add-order race recovery (2026-07-06). CN + EU SF tenants # MIRROR one org and ops enters an overseas hire in BOTH. # When the hire is added via import / front-end (not the # onboarding module) the two tenants have no ordering # guarantee: the non-owning tenant's full sync can create # the row first — and if it misclassifies a future hire as # a leaver (see the ``status='f'`` guard in # ``preview_full_employee_sync``) the row ends up owned by # the WRONG tenant at ``status='left'``. The OWNING # tenant's authoritative ``onboard`` (webhook) must then be # able to RECLAIM it, or the hire is permanently stuck as a # leaver (Camila P10009). Allow reclaim ONLY when the # incoming event is an onboard AND the existing row is NOT # active — an active row on another tenant is a genuine # dual-ownership conflict that still needs manual review. # Onboard-poach extension (2026-07-16): also reclaim an # ACTIVE foreign row when the incoming job snapshot's # company code says THIS instance canonically owns the # person (``owned_company_codes``) — covers mirrored # hires whose userId differs across tenants (empId is # the shared key), where the userId-keyed reclaim above # can't match. Without a mapping verdict, an active row # on another tenant stays a manual-review conflict. _canonical_here = ( _snap_company is not None and await self._company_owner_for(_snap_company) == self.hris_instance_id ) # Cross-tenant termination adoption (2026-07-31, P01550 # / P0596 cohort). PingCAP enters overseas terminations # in the EU tenant at approval time, but MOST of those # employees' Helix rows are pinned to the CN instance # (the CN tenant mirrors them and its sync created the # rows first). The termination is only mirrored into the # CN tenant LATER — sometimes days after the last # working day — so the EU webhook push is the ONLY # timely signal, and this branch used to reject it with # the ConflictError below (offboard was not an allowed # reclaim shape). Adopt instead: the empId (= org-global # 工号, UNIQUE in identity_persons, unchanged across # tenants per the 2026-07-30 HRIS alignment) matching a # real person plus a substantiated endDate is the same # corroboration bar the onboard reclaim uses. Re-pinning # (not just flipping) is load-bearing: the CN tenant # still reports the person as ACTIVE until the mirror # lands, and the CN nightly full sync would stage a # reactivate for a leaver_pending row it owns — once the # row is ours, the CN sync's cross-instance guard skips # it, and THIS tenant's roster (which carries the # termination) stays the single authority. _offboard_adoptable = ( event_type == "offboard" and bool(employment.get("endDate")) ) if event_type == "onboard" and ( person.employment_status in ("left", "leaver_pending") or _canonical_here ): logger.warning( "hr.sync: reclaiming cross-instance row empId=%r " "from hris_instance=%s to %s on onboard (prev " "status=%s, canonical_owner=%s) — add-order race " "recovery; the owning tenant's onboard reclaims a " "row the other tenant mis-created", desired_empid, person.hris_instance_id, self.hris_instance_id, person.employment_status, _canonical_here, ) person.hris_instance_id = self.hris_instance_id # Fall through: ``person`` is now adopted by this # instance; the onboard branch below flips it back to # ``employment_status='active'`` and clears the stale # termination_date. elif _offboard_adoptable: logger.warning( "hr.sync: adopting cross-instance row empId=%r " "from hris_instance=%s to %s on offboard " "(endDate=%r) — the termination was entered in " "THIS tenant while the person row was pinned to " "the other; without adoption the offboard is " "dropped and nothing ever closes the account " "(the other tenant only mirrors the termination " "later, if at all)", desired_empid, person.hris_instance_id, self.hris_instance_id, employment.get("endDate"), ) # Re-pin deferred to the offboard apply branch (#2218): # if the endDate turns out to be a stale rehire # artifact, the row stays with its current owner. deferred_repin_from = person.hris_instance_id # Fall through: the offboard branch below flips the # person to ``leaver_pending`` with the pushed # endDate as termination_date. else: logger.warning( "hr.sync: cross-instance empId collision empId=%r " "owned_by=%s incoming=%s status=%s — refusing merge", desired_empid, person.hris_instance_id, self.hris_instance_id, person.employment_status, ) raise ConflictError( f"empId {desired_empid!r} already exists on " f"hris_instance {person.hris_instance_id} — cannot " f"merge from hris_instance {self.hris_instance_id}. " f"Either deduplicate the empId at the SF source, or " f"manually re-tag the canonical Helix row's " f"hris_instance_id and retry." ) created = False desired_email = (user or {}).get("email") if person is None: # Concurrent-employment detection, the AUTHORITATIVE half # (2026-08-18). ``personIdExternal`` is SF's *person* key and # ``userId`` its *employment* key; when the incoming # employment's personIdExternal is already held by a live # ACTIVE person, this userId is that human's second # employment — not a new hire. Register the alias and stop: # every later reference to this userId (managerId, # relUserId) now resolves to the canonical person, and no # ghost row is minted. Normally zero extra API calls — the # value rides along on the EmpEmployment row the event already # carries. # # Deliberately scoped to an ``active`` owner: a ``left`` / # ``leaver_pending`` owner with a new userId is a REHIRE, and # aliasing that would silently freeze the rehire's row at its # departed state. Rehire handling stays exactly as it was. # # The value is re-resolved from SF whenever what we hold is not # usable AS A PERSON KEY — empty, or self-referential # (== ``user_id``). This is the backstop that makes the guard # live on EVERY ingest path, not just the ones that happen to # carry a real ``EmpEmployment`` row: # * the **full-sync** synthetic employments used to hardcode # ``personIdExternal: user_id`` as a master-data cache key, # so ``str(pid) != user_id`` was *structurally* false there # — which is why 80100057 / 80100061 / 80100062 had to be # aliased by a one-off backfill script instead of by the # guard that was written for them. They now carry the real # value when an EmpEmployment row was fetched, and fall back # to ``user_id`` (→ re-resolved here) when none was; # * the **incremental** path's no-EmpEmployment fallback still # synthesizes ``user_id`` and relies on this backstop; # * the **webhook** path is already normalized upstream, and # re-checking here is free (the resolver memoizes). # Cost is bounded to one EmpEmployment call per userId per batch # and only on the CREATE branch — an update/reactivate never # reaches this block. Failure keeps the carried value, i.e. the # pre-change behaviour. _pid_ext = await self._resolve_person_id_external( user_id, (employment or {}).get("personIdExternal"), # Only the synthesis sites that made the value up ask for a # re-resolution; a value off a real EmpEmployment row is # authoritative even when it equals ``user_id``. force=bool( (employment or {}).get("_person_id_external_is_user_id") ), ) if self.person_id_external_is_usable(_pid_ext, user_id): canonical = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == str(_pid_ext), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) ) if canonical is not None: from app.identity.identity_reconcile import ( record_hris_user_alias, ) await record_hris_user_alias( self.db, person=canonical, alias_user_id=user_id, reason="concurrent_employment", person_id_external=str(_pid_ext), hris_instance_id=self.hris_instance_id, note="EmpEmployment.personIdExternal already held by " "an active person", ) # Skip ONLY if the mapping now actually exists. # ``record_hris_user_alias`` refuses a userId that a # LIVE person holds as their own primary key; in that # case there is no alias to resolve through later, so # dropping the event here would be a silent loss — # fall through to the pre-existing create/conflict # handling instead. (Re-reading is cheap and, unlike # the boolean return, distinguishes "already mapped" # from "refused".) from app.identity.identity_reconcile import ( hris_user_alias_owner, ) if await hris_user_alias_owner(self.db, user_id) is not None: logger.event( "HRIS_CONCURRENT_EMPLOYMENT_ALIASED", { "user_id": user_id, "person_id_external": str(_pid_ext), "canonical_hris_person_id": canonical.hris_person_id, "event_type": event_type, }, level=logging.WARNING, ) return None, False resolved_email = await self._resolve_unique_work_email( user_id=user_id, desired=desired_email, current_person_id=None, ) # Second checkpoint for ``HR_SKIP_EMAIL_DOMAINS``: catches # SF-side duplicate userIds. The preview-time check sees the # canonical SF ``email`` (real ``@pingcap.com``) and lets # the row through — but ``_resolve_unique_work_email`` # collides on the UNIQUE index because the canonical row # already owns that email, and falls back to # ``@successfactors.local``. Without this gate the # duplicate would land in ``identity_persons`` as a ghost # row with a synthetic email and orphan ``email_aliases``. # Confirmed in prod 2026-05-07: SF returned 100113 + # P01115 both as ``yun.zhu@pingcap.com`` for 朱赟 Claire Zhu; # P01115 won the email, 100113 was synthesized into a # broken duplicate. # # BUT (regression fix 2026-07-01, from #191): this gate must # distinguish WHY we're on a skip-domain address, or it eats # legitimate new hires: # (A) SF gave a REAL email that collided (→ synthetic # ``.local`` fallback) OR a configured vendor skip-domain # (gaiaworks) → genuine duplicate / non-employee → refuse. # (B) SF gave NO email at all (empty / placeholder) → the # ``.local`` fallback just means "no mailbox yet", NOT a # collision. HRIS never carries a corporate email — Helix # MINTS it: ``reserve_and_writeback_email`` derives # ``@`` from the name AFTER create and writes # it back to SF (exactly the flow that worked pre-#191 and # already wrote email names back to HRIS). Refusing here # silently DROPS real new hires so they never appear on the # onboarding page for the admin to action (assign laptop, # etc.). #191 introduced this drop; it ate P10007 (Victor # Schmidt) + P10010 (Yuang Zhou) on 2026-07-01. Create with # the placeholder and let the coordinator mint the address. if self.email_domain_in_skip_list(resolved_email): resolved_is_synthetic_placeholder = resolved_email.endswith( "@successfactors.local" ) desired_was_real_email = bool( (desired_email or "").strip() ) and not self.is_placeholder_email(desired_email) if resolved_is_synthetic_placeholder and not desired_was_real_email: # Case (B): brand-new hire, no SF email — create so the # onboarding case is visible; coordinator mints the # corporate address from the name and writes it to SF. logger.info( "hr.sync: new hire user_id=%s has no SF email — " "creating with placeholder %r so the onboarding case " "is visible; reserve_and_writeback_email will mint the " "corporate address from the name and write it back to " "SF", user_id, resolved_email, ) else: # Case (A): SF-side duplicate userId / vendor skip-domain # — canonical row owns the real email, don't ghost it. logger.event( "HRIS_CANONICAL_EMAIL_COLLISION_SKIPPED", { "user_id": user_id, "employee_number": (user or {}).get("empId"), "resolved_email": resolved_email, }, ) # Record WHOSE duplicate this is (2026-08-18). Refusing # to mint the ghost row was only half the job: every # HRIS reference to this userId — most importantly a # subordinate's ``EmpJob.managerId`` — resolved to # nothing, so four active people showed no manager at # all and the backfill re-missed them every sync (prod # 孙若曦: canonical P0050, duplicate 80100061). The # canonical row is the one that owns the real email. _dup_owner = await self.db.scalar( select(IdentityPerson).where( IdentityPerson.work_email == (desired_email or "").strip().lower(), IdentityPerson.deleted_at.is_(None), ) ) if _dup_owner is not None: from app.identity.identity_reconcile import ( record_hris_user_alias, ) await record_hris_user_alias( self.db, person=_dup_owner, alias_user_id=user_id, reason="email_collision", person_id_external=( str(_pid_ext) if _pid_ext else None ), hris_instance_id=self.hris_instance_id, note="canonical row owns the SF work_email", ) return None, False # Pre-compute the historical-leaver employment_status BEFORE # the IdentityPerson INSERT so the row lands at the right # state in a SINGLE write — not via a post-flush UPDATE. # # The leaver_blocked listener exempts ``session.new`` INSERTs # but blocks any unguarded UPDATE that transitions # ``employment_status`` → ``'left'`` (the listener intercepts # ``session.dirty``, see ``leaver_blocked/listener.py:86-89``). # ``self.db.add(person)`` followed by ``self.db.flush()`` # immediately moves ``person`` out of ``session.new`` into # the persistent set, so the offboard branch's later # assignment ``person.employment_status = "left"`` would be # a dirty UPDATE and trip the listener with # ``DENIED_PROCESS_NOT_FOUND`` for the historical-leaver # path. By computing the status here at construction time, # the INSERT carries the final value and the downstream # offboard branch never needs to flip ``'active'`` → ``'left'``. # # Reviewer pass 1 (2026-05-30) caught this — without the # pre-compute, the wangdi-PR full-sync forfeited path would # fail at confirm-batch time for every >30-day-old leaver, # exactly the 48 prod rows the PR was written to recover. initial_status = "active" initial_termination_date = None if event_type == "offboard": _end_date_for_insert = _parse_date(employment.get("endDate")) _user_status_for_insert = str( (user or {}).get("status") or "" ).strip().lower() if _user_status_for_insert == "t" and ( _end_date_for_insert is None or _end_date_for_insert < self.today ): # SF still flags active — stale historical endDate # on EmpEmployment from a prior re-hire. Keep # ``active`` so this person doesn't land as a leaver. # Downstream offboard branch will short-circuit too. # # Scoped to a PAST endDate to mirror the UPDATE branch # (#2207/#2215): SF keeps ``status='t'`` until the last # working day passes, so a today-or-future endDate on a # still-'t' person is a SCHEDULED termination and must # be honored — the unscoped form silently created every # such person as plain ``active``. No person row exists # yet here, so the Shanghai anchor is the only clock # available; the ±1-day edge only shifts a same-day # termination into the (conservative) leaver_pending # bucket. pass elif _end_date_for_insert: initial_termination_date = _end_date_for_insert # 30-day historical cutoff — see the more detailed # comment block in the offboard UPDATE branch below. if (self.today - _end_date_for_insert).days >= 30: initial_status = "left" else: initial_status = "leaver_pending" else: # Offboard event without an endDate (data integrity # issue on SF side). Conservative: leaver_pending. initial_status = "leaver_pending" person = IdentityPerson( source_system="successfactors", hris_instance_id=self.hris_instance_id, hris_person_id=user_id, work_email=resolved_email, employment_status=initial_status, termination_date=initial_termination_date, ) self.db.add(person) # Flush so subsequent same-batch dup-email checks see this row. # ``person`` moves from session.new → persistent after this # call; any further employment_status mutation is a dirty # UPDATE subject to the leaver_blocked listener. await self.db.flush() created = True elif self.hris_instance_id is not None and person.hris_instance_id is None: # Adopt a legacy unscoped row into this instance. person.hris_instance_id = self.hris_instance_id if user: if desired_email: # Pre-launch P2 fix: also gate update-path on the # email skip list. ``should_skip_user`` only fires at # preview-time for ONBOARD events — if SF later # mutates an existing real employee's email to a # ``HR_SKIP_EMAIL_DOMAINS`` value (e.g. typo'd to a # gaiaworks.cn alias mid-batch), the update path # would propagate the bad email onto the live row. # Skip the email rewrite when the desired value # matches the skip list; keep current. if self.email_domain_in_skip_list(desired_email): logger.info( "hr.sync: refusing to update person %s " "work_email to %r (matches " "HR_SKIP_EMAIL_DOMAINS); keeping current %r", person.hris_person_id, desired_email, person.work_email, ) else: # Thread ``current_email`` so a transient collision (e.g. # SF dup-userId state visible for one tick) does NOT # downgrade an already-real work_email to a placeholder # — see ``_resolve_unique_work_email`` docstring. Conversely, # when an existing row sits at a stale placeholder and SF # now returns a real uncontested email, the resolver # returns the real email and this assignment promotes # ``person.work_email`` automatically — that's the # backfill-on-every-sync path requested in the bug report. person.work_email = await self._resolve_unique_work_email( user_id=user_id, desired=desired_email, current_person_id=person.id, current_email=person.work_email, ) # SF composes displayName as " " and, when # the English half is absent, some configs serialise the gap as a # literal "null" — arriving as "张东晖 null". Strip that sentinel # here (the single persistence chokepoint) so it never leaks into # search/pinyin, feed snapshots, bill snapshots, or the latin_name # transform we push to Feishu. Whole-token match only — a real # name like "Knull Hu" is untouched. person.display_name = ( strip_sentinel_name_tokens(user.get("displayName")) or person.display_name or user_id ) # P1-D: keep ``name_pinyin`` / ``name_pinyin_abbr`` in sync # with the latest ``display_name``. The live webhook+confirm # pipeline (this path) historically didn't write these — only # the dormant full-roster path did — so directory search by # 拼音 / 拼音 abbr ("孟" → "meng" / "MQ") would silently miss # every employee onboarded since the multi-instance refactor. # Recompute on every touch so a legal name change in HR # propagates the same sync. from app.core.pinyin_utils import to_pinyin, to_pinyin_abbr person.name_pinyin = to_pinyin(person.display_name) person.name_pinyin_abbr = to_pinyin_abbr(person.display_name) # SF authoritative given/family split — drives the Google # email local-part order (migration 240). Updated on EVERY # sync touch so a downstream HR correction (e.g. legal # name change) propagates without manual fix-up. Empty # strings are coerced to None so the email builder's # ``both populated`` gate stays meaningful. _first = (user.get("firstName") or "").strip() or None _last = (user.get("lastName") or "").strip() or None if _first is not None: person.first_name = _first if _last is not None: person.last_name = _last await self._apply_employee_no( person, user.get("empId") or person.employee_no or user_id, ) person.source_updated_at = _parse_datetime(user.get("lastModifiedDateTime")) else: person.display_name = person.display_name or user_id await self._apply_employee_no(person, person.employee_no or user_id) # HR-assigned 邮箱名 → golden record (migration 582). Written on EVERY # apply, INCLUDING an onboard ``skip`` — those fall through to # ``_apply_event`` as a refresh-only pass, which is exactly the case # P10026 needed: HR filled ``customString4`` two days after the create # event, so all ~40 ticks in the nine days before his hire date carried # the right value under an action the coordinator gate refuses. # Persisting here decouples # "Helix learned the name" from "a lifecycle coordinator happened to # fire", and ``account_provisioner`` reads the column at mint time. # # Apply-only, deliberately: a non-empty value overwrites, blank/absent # never clears. Blank means "HR gave no instruction" — clearing would # let one partial SF read re-derive a person's address from pinyin. _email_name = (email_name or "").strip() if _email_name and person.hris_email_name != _email_name: logger.info( "hr.sync: hris_email_name for %s: %r → %r", user_id, person.hris_email_name, _email_name, ) person.hris_email_name = _email_name # Un-purge on re-appearance: any person SF returns in a sync is, # by definition, carried by the upstream HRIS again. Clear a # prior ``purged`` tombstone so the row stops being treated as # gone — critical now that ``is_effectively_departed`` blocks # login for ``hris_record_state='purged'`` (a rehire who was # purged during their gap must be able to authenticate again). # The reconciliation purge handler only ever SETS ``purged``; # this is its inverse, and the only place that clears it. # Captured BEFORE the un-purge block flips it — the onboard soft-delete # revive below must not write a SECOND ``hris_unpurged`` lifecycle row # when the purge block already wrote one (same (person,type,date) unique # key → 1062). _was_purged = person.hris_record_state == "purged" if person.hris_record_state == "purged": logger.info( "hr.sync: un-purging person %s (hris=%s) — reappeared in " "the HRIS roster", person.id, person.hris_person_id, ) person.hris_record_state = "present" person.hris_purged_at = None person.hris_purged_reason = None # Symmetric lifecycle trail (the purge handler writes # ``hris_purged`` to the same table). Un-purge re-enables # login (it flips ``is_effectively_departed`` False), so the # state change must leave a queryable, durable record — not # just a log line. self.db.add(HrLifecycleEvent( person_id=person.id, event_type="hris_unpurged", effective_date=self.today, detail={ "hris_person_id": person.hris_person_id, "work_email": person.work_email, "display_name": person.display_name, "reason": "reappeared_in_roster", "trigger_event_type": event_type, }, )) # Un-SOFT-DELETE on onboard re-appearance (2026-07-20, P10019 陈建华). # The cancelled-future-hire cleanup (scheduled_jobs.handlers, #1988) # tombstones a future hire that SF stopped returning with TWO markers: # hris_record_state='purged'/hris_purged_at AND ``deleted_at``. The # un-purge block above only reverses the PURGE half — ``deleted_at`` was # never cleared here, so a hire SF later re-sends (P10019 arrived # nameless → cleaned up → re-sent 3 days later WITH a real name) stayed # soft-deleted forever and was skipped by every ``deleted_at IS NULL`` # query, including the 24h-lead account/建号 flow → no Google/Feishu/ # welcome on the start date. NOT gated on the purge state (a prior # session already reset P10019's purge to 'present', so the block above # no longer fires for him) — SF actively returning an ONBOARD for this # userId is itself proof they are a live hire, so revive. Onboard-only: # an offboard for a soft-deleted person must never resurrect them. if event_type == "onboard" and person.deleted_at is not None: logger.info( "hr.sync: reviving soft-deleted person %s (hris=%s) — SF " "re-sent an onboard event; clearing deleted_at (was %s)", person.id, person.hris_person_id, person.deleted_at, ) person.deleted_at = None if person.hris_record_state == "purged": # Defence-in-depth: if it were still purged (block above didn't # run for some reason), clear it too so we never leave a # half-revived row. person.hris_record_state = "present" person.hris_purged_at = None person.hris_purged_reason = None # Only write the lifecycle row when the un-purge block above didn't # already (they share the (person, 'hris_unpurged', today) unique # key). For a delete-only tombstone (P10019: purge already reset) # this fires; for a purge+delete row the un-purge event covers it. if not _was_purged: self.db.add(HrLifecycleEvent( person_id=person.id, event_type="hris_unpurged", effective_date=self.today, detail={ "hris_person_id": person.hris_person_id, "work_email": person.work_email, "display_name": person.display_name, "reason": "revived_soft_deleted_on_onboard_reappearance", "trigger_event_type": event_type, }, )) person.hire_date = _parse_date(employment.get("startDate")) or person.hire_date # 参加工作日期 (社会工龄 N 起点) — SF EmpEmployment.firstDateWorked # is the field PingCAP HR populates with the person's first-ever # career start. PR ahead of 2026-06-02 originally pulled # ``originalStartDate`` here, which is actually SF's "first start # at THIS company across multiple stints" (司龄 anchor for # rehires) — diverges from startDate only for the 47/432 rehires # in prod, NOT the 380+ employees with genuine pre-PingCAP # careers. Confirmed against PingCAP's SF tenant 2026-06-02: # shuai.feng's firstDateWorked=2011-11-11 vs hireDate=2022-06-30 # gives the real 11-year career start. # # Fallback chain: firstDateWorked → originalStartDate → keep # whatever HR last entered manually. The 3-tier order means a # row always converges to the best available career-start signal: # genuine career data when SF carries it, rehire-aware "first # PingCAP start" when only that's present, and manual HR override # only when SF has neither. first_worked = _parse_date(employment.get("firstDateWorked")) original_start = _parse_date(employment.get("originalStartDate")) career_start = first_worked or original_start if career_start: person.work_start_date = career_start # 司龄 S 起点 — SF EmpEmployment.seniorityDate. Canonical SF # tenure anchor, distinct from ``hire_date`` in rehire / # employment-segment-reset cases (3/500 in prod 2026-06-02). # 99.8% populated. Added migration 240. seniority = _parse_date(employment.get("seniorityDate")) if seniority: person.seniority_date = seniority # Birthday — pulled from SF ``PerPersonal.customDate1`` (NOT # ``User.dateOfBirth``; that field is null in PingCAP's tenant — # verified 2026-06-01). PerPersonal is master-data, so the # map is built once per sync in ``_ensure_master_data_loaded`` # and keyed by ``personIdExternal``. # # WRITE ONLY when the existing column is NULL so a self-edit # via ``/me/settings`` is never overwritten by HRIS. The feed # 动态广场 cron uses month+day only — year never surfaces in # the UI (privacy by design — see # ``feed.birthday_visibility`` pref). if ( person.birthday is None and self._personal_birthdays and (pid_ext := employment.get("personIdExternal")) ): sf_birthday = _parse_date(self._personal_birthdays.get(pid_ext)) if sf_birthday is not None: person.birthday = sf_birthday # 性别 — SF PerPersonal.gender (migration 344). HRIS-authoritative # (no Helix self-edit path), so overwrite when SF carries a value; # never NULL-out on an empty response. Projected to Feishu's gender # enum by the feishu_contact target. if self._personal_genders and (pid_g := employment.get("personIdExternal")): sf_gender = self._personal_genders.get(pid_g) if sf_gender: person.gender = sf_gender person.last_synced_at = datetime.now(UTC) # Enrich from the selected job snapshot (staged during preview). # ``get_jobs_for_users`` now pulls company/department/location/ # position/managerId, so the confirm phase can finally populate # ``job_title`` + org codes instead of leaving them NULL. The # master-data caches (_companies/_locations/_positions) turn # the raw SF codes into the human-readable names the UI renders; # without them the employee list shows ``PO00001``/``DP00012`` # instead of ``Senior Engineer``/``A PingCAP_China``. Extracted so # the 24h-lead pre-start enrich path # (``HireDayNotifyService._enrich_prestart_hire_department``) applies # the SAME field set from an effective-dated pull — not just # department + manager (the #2013/#2015 gap that shipped English # welcome emails with ``你的团队`` / ``(待定)`` and mis-detected # ``is_china_employee`` for pre-start NA/EMEA hires, 2026-08-03). job_snapshot = (employment.get("_jobs_snapshot") or [None])[0] or {} self._apply_job_snapshot_fields(person, job_snapshot) # Contact + preference fields (migration 048). # * preferred_language ← SF User.defaultLocale (SEED ONCE only) # * mobile_phone ← SF PerPhone.phoneNumber (phoneType='1043') # * personal_email ← SF PerEmail.emailAddress (emailType='1047') # PerPhone / PerEmail are keyed by personIdExternal (not userId), # so we look up via the master-data caches populated in # ``_ensure_master_data_loaded``. Only overwrite when SF returns # a non-empty value — empty/null means "not configured" not # "clear it". # # preferred_language is a USER-OWNED UI preference, NOT an # HRIS-authoritative field. Helix ships an in-app language switcher # (header globe + /me/settings ▸ ProfileBasicTab) that PATCHes # ``preferred_language`` directly, and the boot path reads it back # from /auth/me. SF's ``defaultLocale`` is therefore only a SEED # for a brand-new person: once the row already has any value # (seeded from SF earlier OR chosen by the user) the sync must NOT # overwrite it. Without this guard every full HRIS sync re-stamped # SF's locale (en_US for most employees) and silently snapped the # user back to English — users reported their zh-CN choice # reverting after each sync/release (regression fixed 2026-06-29). # # The raw SF value is also normalised: ``defaultLocale`` arrives as # free-text (``english`` / ``chinese`` / ``中文``), never BCP-47, so # we canonicalise to the same closed set the PATCH endpoint # enforces (``zh_CN`` / ``en_US`` / ``ja_JP``) and skip seeding when # SF's value is unrecognisable rather than poisoning the column. if not person.preferred_language: # ``(user or {})`` — ``_upsert_person`` explicitly supports # ``user is None`` (a webhook offboard/onboard whose overlay # payload carried no fields → _overlay_event_fields returns # None). Every other ``user`` deref here is ``if user:``-guarded; # this one was not, so a NEW person with no preferred_language # and ``user is None`` hit ``NoneType.get`` and the whole event # was rolled back + dropped until the next reconciliation sync. seeded_locale = _normalize_preferred_language( (user or {}).get("defaultLocale") ) if seeded_locale: person.preferred_language = seeded_locale person_id_external = employment.get("personIdExternal") if person_id_external: if self._personal_mobiles: new_mobile = self._personal_mobiles.get(person_id_external) if new_mobile: person.mobile_phone = new_mobile # Cross-instance fallback: owning instance carried no mobile # but another enabled tenant has this person's PerPhone (guarded # by same-person email match). Fills ONLY when still empty — # never overrides the owning instance's own value. if not person.mobile_phone: supp = self._crossinstance_mobile_for( person.work_email, person_id_external, self._crossinstance_phones, ) if supp: person.mobile_phone = supp["number"] if supp.get("cc") and not person.mobile_country_code: person.mobile_country_code = supp["cc"] if self._personal_mobile_countries: # ISO-2 of the phone's OWN country (host country for an # international assignee). Write-when-present, never NULL-out — # a person whose PerPhone has no countryCode keeps any prior # value and the phone formatter falls back to home country. new_mobile_cc = self._personal_mobile_countries.get( person_id_external, ) if new_mobile_cc: person.mobile_country_code = new_mobile_cc if self._personal_emails: new_pers_email = self._personal_emails.get(person_id_external) if new_pers_email: person.personal_email = new_pers_email # Event-sourced fallbacks (webhook carried the values directly via # ``_overlay_event_fields``), so a hire is fully created before the SF # data sync runs. Fill ONLY when the pull above (PerEmail / PerPhone / # company master data — all absent pre-sync) left the field empty; # never overwrite a real synced value. ``country`` matters for the # E.164 recovery phone; personal_email/mobile for the welcome-with- # password mail + Google home/recovery slots. if user: if not person.personal_email and user.get("_event_personal_email"): person.personal_email = user["_event_personal_email"] if not person.mobile_phone and user.get("_event_mobile_phone"): person.mobile_phone = user["_event_mobile_phone"] if not person.country and user.get("_event_country"): person.country = user["_event_country"] if event_type == "onboard": if not created and self._is_inflight_scheduled_leaver(person): # In-flight scheduled exit — keep the confirmed leaver # state. See ``_is_inflight_scheduled_leaver`` for the # full rationale (2026-08-11). Every other attribute the # onboard event carries (name / dept / manager / title) # still flows through above; only these two fields are # preserved. logger.info( "hr.sync: refusing to revert scheduled leaver " "user_id=%s to active on onboard — employment_status=" "'leaver_pending', termination_date=%s is not in the " "past (SF keeps status='t' through the notice period)", person.hris_person_id, person.termination_date, ) else: person.employment_status = "active" # Clear any stale termination_date from a prior offboard. # Without this, a reactivate/onboard leaves # ``termination_date`` untouched → the person ends up in a # corrupt ``active + termination_date <= today`` state, and # the next tick that *any* sweep observes (incremental sync # picking an old employment row with endDate, # process_advance, etc.) flips them back to ``left``. This # was the "孟凡生 keeps going left on its own" bug that # reproduced twice in prod. person.termination_date = None else: termination_date = _parse_date(employment.get("endDate")) if ( termination_date is not None and person.hire_date is not None and termination_date < person.hire_date ): # Universal impossible-state backstop (2026-07-16, Robert Wood # P10013 class): a termination_date before hire_date is # logically impossible. This is the APPLY-time half of the # invariant and is what actually covers the WEBHOOK push path # (``process_webhook_events`` → ``_apply_event`` → here), which # bypasses ``_stage_event``'s preview-time guard entirely. # Leave the person untouched; the daily # ``hris_status_divergence`` check surfaces the bad SF data. logger.error( "hr.sync: IGNORING impossible offboard for user_id=%s — " "termination_date %s precedes hire_date %s; leaving " "employment unchanged (data-quality signal, not a real " "leaver).", person.hris_person_id, termination_date, person.hire_date, ) elif termination_date: # Safety gate: SF's User.status is the authoritative # current-employment flag. If SF still reports the # person as active (``'t'``), don't honor a stale # endDate coming off an old EmpEmployment row. This is # the second half of the "孟凡生 keeps going left on # its own" fix — re-hires and termination-then-reactivate # flows leave orphaned EmpEmployment rows that the # incremental sync's ``any with endDate → offboard`` # heuristic misreads as a fresh termination. user_status = str((user or {}).get("status") or "").strip().lower() # Scope refinement (2026-07-31, P0779/P0596 cohort): only # a PAST endDate is the stale-rehire-artifact shape this # guard exists for — SF keeps ``status='t'`` until the # last working day passes, so an approval-time offboard # push (the event-driven model: HRIS pushes weeks before # the LWD) ALWAYS arrives while the person is still # ``'t'``. The unscoped guard silently reverted every # such scheduled termination to active/None: ledger + # offboarding process existed, but the person never # turned ``leaver_pending`` — invisible in the # offboarding cockpit (filters on person status) and # outside every leaver-aware downstream gate. A today-or # -future endDate on a still-active person is a # SCHEDULED termination, not a stale one; apply it. # Timezone: compare against the PERSON's local today, not # the Shanghai business day (#2215) — see # ``_local_today_for``. An EU termination entered on the # last working day is "today", not "past", even while # Shanghai has already flipped the date. if user_status == "t" and termination_date < self._local_today_for(person): # #2215: this gate REFUSES TO APPLY a stale endDate; it # must never REVERT a person who is already through the # leaver pipeline. SF Integration Center is # at-least-once, so a replayed offboard arriving after # the LWD (while SF still says ``status='t'``) used to # flip ``leaver_pending``/``left`` back to ``active`` # with ``termination_date=None`` — after which # ``hr.process_advance`` can never transition them # (nothing left to advance) and every # ``employment_status``-based gate (Leaver Blocked, # cockpit, manager-binding exclusion) counts a departed # person as employed. The #1744 anti-regression guard # below only lives on the else-branch, and # LeaverBlockGuard only polices writes INTO 'left', so # this branch needs its own explicit guard. self._offboard_refused_stale = True if not created and person.employment_status in ( "leaver_pending", "left", ): logger.info( "hr.sync: ignoring stale endDate=%s for user_id=%s " "(SF User.status='t', date already past) — person " "is already %s; refusing to apply, NOT reverting " "to active (#2215)", termination_date, person.hris_person_id, person.employment_status, ) else: logger.info( "hr.sync: ignoring stale endDate=%s for user_id=%s " "because SF User.status='t' (active) and the date " "is already past — rehire-history artifact", termination_date, person.hris_person_id, ) person.employment_status = "active" person.termination_date = None else: # #2218: the endDate survived every credibility gate, # so the cross-tenant adoption decided far above is now # justified — re-pin here (and only here), so a refused # offboard leaves ownership with the current tenant and # its nightly sync keeps the row in scope. if deferred_repin_from is not None: logger.warning( "hr.sync: re-pinning adopted row user_id=%s from " "hris_instance=%s to %s — offboard endDate=%s " "confirmed applicable (#2218)", person.hris_person_id, deferred_repin_from, self.hris_instance_id, termination_date, ) person.hris_instance_id = self.hris_instance_id deferred_repin_from = None person.termination_date = termination_date # Pre-launch hardening 2026-05-04: never write 'left' # directly from the upsert path on UPDATE. The # leaver-block listener requires an OffboardingProcess # row to accompany every transition to 'left' — a # same-day termination would otherwise trip # ``DENIED_PROCESS_NOT_FOUND`` mid-flush, abort the # event, and mark the staged event ``failed`` so # same-day terminations could never be confirmed. # The dedicated ``hr.process_advance`` cron flips # past-due rows to ``'left'`` through the guarded # path that creates the matching offboarding process. # # 2026-05-30 (wangdi PR) — INSERT-only carve-out for # historical leavers lives in the ``if person is # None:`` block above, BEFORE the flush, so the # session.new exemption on the leaver_blocked # listener applies. When ``created=True``, the # initial INSERT already wrote the correct value # (``'left'`` for ≥30-day-old endDate, ``'leaver_ # pending'`` for recent). We MUST NOT overwrite it # here — the row is now persistent (session.dirty), # so any UPDATE to ``'left'`` would trip the # listener with DENIED_PROCESS_NOT_FOUND. Skip the # write for newly-created rows; for existing rows # use the conservative ``leaver_pending`` value. # # #1744 idempotency guard: never regress an already- # ``left`` person back to ``leaver_pending``. SF # Integration Center is at-least-once (the 504-era # replay storms are on record), and the offboard UPDATE # branch is reached on every replayed offboard event. # The sync-confirm path already guards this at stage # time (``_stage_event`` marks action='skip' when the # existing person is 'left', service.py ~2600); the # webhook path (``process_webhook_events`` → ``_apply_event`` # → here) had no symmetric guard, so a replay flipped a # departed employee back to ``leaver_pending`` — re- # surfacing them on the overdue-leaver board, re-entering # the ``employment_status != 'left'`` exclusion windows # (a still-active subordinate sync could then bind a # departed person as manager), and only self-healing on # the next ``transition_overdue_leavers`` cron. The # LeaverBlockGuard only blocks writes flowing INTO # 'left', never 'left' → 'leaver_pending', so this guard # must be explicit. ``termination_date`` is still # refreshed above, so a corrected endDate on the replay # is honored; only the status regression is suppressed. if not created and person.employment_status == "left": logger.info( "hr.sync: offboard replay for user_id=%s ignored " "— person already 'left'; kept status, refreshed " "termination_date=%s (#1744)", person.hris_person_id, termination_date, ) elif not created: person.employment_status = "leaver_pending" return person, created async def _ensure_master_data_loaded(self) -> None: """Populate the ``_companies/_locations/_departments/_positions`` caches on first use. Also upserts ``identity_org_nodes`` for every active department so the UI's dept join lights up. Called lazily from ``_apply_event`` so a single confirm batch pays the ~4 SF round-trips once regardless of how many events it holds. Missing codes fall back to ``{}`` so downstream lookups ``.get()`` without exploding. """ if self._companies is not None: return client = await self._get_client() async def _safe(op_name: str, key: str): """Run a list endpoint, swallowing both transport failures and missing methods (the lifecycle-test ``_FakeLifecycleClient`` only stubs the endpoints it cares about).""" method = getattr(client, op_name, None) if method is None: return [] try: return await method() except Exception as exc: # noqa: BLE001 logger.warning("hr.sync: %s failed: %s", op_name, exc) return [] companies = await _safe("list_active_companies", "externalCode") locations = await _safe("list_active_locations", "externalCode") departments = await _safe("list_active_departments", "externalCode") positions = await _safe("list_active_positions", "code") # Personal contact (migration 048) — populated alongside master # data so confirm can resolve personIdExternal → mobile / personal # email without a round-trip per person. personal_phones = await _safe("list_personal_phones", "personIdExternal") personal_emails = await _safe("list_personal_emails", "personIdExternal") # Birthday — PingCAP SF stores DoB on PerPersonal.customDate1 # (NOT on User.dateOfBirth, which is null for all users in # this tenant; verified 2026-06-01). Map keyed by # personIdExternal, value is the raw SF date string so the # caller passes it through the standard ``_parse_date`` helper. personal_birthdays = await _safe( "list_personal_birthdays", "personIdExternal", ) # SF-verbatim label master data for Google Employee_Details. Each # is the string SF translates the EmpJob code into, stored on the # person and pushed verbatim to Google (no Helix value-map). job_functions = await _safe("list_job_functions", "externalCode") job_families = await _safe("list_job_families", "externalCode") # GitHub handles — SF cust_Other.cust_id keyed by externalCode # (employee_no), migration 369. Sparse onboarding-form field. github_others = await _safe("list_github_ids", "externalCode") try: email_group_labels = await client.list_email_group_labels() except Exception as exc: # noqa: BLE001 logger.warning("hr.sync: list_email_group_labels failed: %s", exc) email_group_labels = {} self._companies = {c["externalCode"]: c for c in companies if c.get("externalCode")} self._locations = {loc["externalCode"]: loc for loc in locations if loc.get("externalCode")} self._departments = {d["externalCode"]: d for d in departments if d.get("externalCode")} self._positions = {p["code"]: p for p in positions if p.get("code")} # office_code (picklist option id) → EmailGroup label (str→str). self._email_group_labels = dict(email_group_labels or {}) # FO externalCode → label (FOJobFunction.name / cust_JobFamily.externalName). self._job_functions = { f["externalCode"]: f.get("name") for f in job_functions if f.get("externalCode") } self._job_families = { f["externalCode"]: f.get("externalName") for f in job_families if f.get("externalCode") } # Build personIdExternal → first non-empty value maps. Prefer # ``isPrimary`` rows; fall back to any row so tenants that don't # flag primary still get a value. A person can have multiple # rows (old + current) — picking primary-then-first keeps this # deterministic. def _pick(rows: list[dict], value_key: str) -> dict[str, str]: by_pid: dict[str, str] = {} # First pass: primary rows for r in rows: pid = (r or {}).get("personIdExternal") val = (r or {}).get(value_key) if pid and val and r.get("isPrimary") and pid not in by_pid: by_pid[pid] = val # Second pass: fill in non-primary for pids still missing for r in rows: pid = (r or {}).get("personIdExternal") val = (r or {}).get(value_key) if pid and val and pid not in by_pid: by_pid[pid] = val return by_pid self._personal_mobiles = _pick(personal_phones, "phoneNumber") # Resolve each mobile's OWN country: SF ``PerPhone.countryCode`` is a # ``CountryCode`` picklist OPTION id; the picklist's ``externalCode`` # is the ISO-2 (option 1734 → ``SG``). ``_pick`` uses the same # primary-then-first row selection as the number above, so the number # and its country come from the SAME PerPhone row. Tolerant: a # picklist-fetch failure degrades to "no phone country" (the phone # formatter falls back to the person's home country), never blocks the # sync. try: _country_opt_to_iso2 = await client.list_country_code_picklist() except Exception as exc: # noqa: BLE001 logger.warning( "hr.sync: list_country_code_picklist failed: %s", exc, ) _country_opt_to_iso2 = {} _raw_phone_country = _pick(personal_phones, "countryCode") self._personal_mobile_countries = { pid: _country_opt_to_iso2[str(opt)] for pid, opt in _raw_phone_country.items() if str(opt) in _country_opt_to_iso2 } self._personal_emails = _pick(personal_emails, "emailAddress") # Cross-instance mobile supplement: for a person whose owning # instance carries no PerPhone but whose number lives in ANOTHER # enabled tenant (see ``_load_crossinstance_phones`` and the # 2026-08-03 root-cause). Best-effort; never blocks the sync. self._crossinstance_phones = await self._load_crossinstance_phones() self._personal_birthdays = _pick(personal_birthdays, "customDate1") # Gender shares the PerPersonal master pull (migration 344) — no # extra SF round-trip. Reuse the same row list as birthdays. self._personal_genders = _pick(personal_birthdays, "gender") # GitHub handle (migration 369) — cust_Other is keyed by # externalCode (employee_no), NOT personIdExternal, so it doesn't # go through ``_pick``. Filter the onboarding-form junk # placeholders ("无"/"none"/"N/A"/"-") so they never land as a # bogus handle; write-when-present preserves Feishu's existing # value for everyone without a real entry. gh_junk = {"无", "none", "n/a", "na", "-", "—", "null"} self._github_ids = { ec: v for r in github_others if (ec := (r or {}).get("externalCode")) # str() coerces a non-string cust_id (a numeric-typed handle # would otherwise AttributeError on .strip()) before trimming. and (v := str((r or {}).get("cust_id") or "").strip()) and v.lower() not in gh_junk } if self._departments: await self._upsert_org_nodes_from_departments() async def _upsert_org_nodes_from_departments(self) -> None: """Mirror SF FODepartment rows into ``identity_org_nodes`` so the UI's dept dropdowns / org-tree view have labels to render. Mirrors the legacy ``SuccessFactorsSync._upsert_org_nodes`` but scoped to this service so the batch sync is self-contained.""" if not self._departments: return codes = list(self._departments.keys()) existing = { node.org_node_source_id: node for node in ( await self.db.scalars( select(IdentityOrgNode).where( IdentityOrgNode.source_system == "successfactors", IdentityOrgNode.org_node_type == "department", IdentityOrgNode.org_node_source_id.in_(codes), ) ) ).all() } now = datetime.now(UTC) for code, dept in self._departments.items(): node = existing.get(code) if node is None: node = IdentityOrgNode( source_system="successfactors", org_node_type="department", org_node_source_id=code, ) self.db.add(node) node.org_node_code = code node.org_node_name = _department_name(dept, code) node.parent_org_node_source_id = dept.get("parent") node.org_path = _build_org_path(dept, self._departments) node.org_level = _parse_org_level(dept) node.node_status = "active" node.last_synced_at = now node.deleted_at = None await self.db.flush() # Deactivation reconcile: SF's org pull is active-only # (``status eq 'A'``), so a department deactivated in SF just # disappears from ``self._departments`` and its node would keep # its stale ``active`` status forever. Flip every active node # absent from this run's full active set to ``inactive``. Safe: # this method only runs when ``self._departments`` is non-empty # (see ``_load_master_data``), so a transient empty pull can't # tombstone the tree. from app.identity.identity_reconcile import deactivate_absent_org_nodes await deactivate_absent_org_nodes(self.db, self._departments.keys()) async def _grant_default_employee_role(self, person: IdentityPerson) -> None: """Make sure every synced person can actually use the system. Without a role the JWT-issued user lands on a blank /me view and every API guarded by ``RoleRoute`` 403s. New hires need at least the ``employee`` baseline to see their own assets / requests. Idempotent: no-op if they already have it.""" role_id = await self.db.scalar( select(AuthRole.id).where(AuthRole.name == "employee") ) if role_id is None: return already = await self.db.scalar( select(AuthUserRole.role_id).where( AuthUserRole.user_id == person.id, AuthUserRole.role_id == role_id, ) ) if already is not None: return self.db.add(AuthUserRole(user_id=person.id, role_id=role_id)) async def _upsert_dept_membership( self, person: IdentityPerson, dept_code: str | None, ) -> None: """Ensure the person's primary department membership matches ``dept_code`` with the SF-sourced name + parent + org path. ``_ensure_master_data_loaded`` seeds ``self._departments`` with the FODepartment response before this runs, so the membership gets the real zh/en name and full org_path instead of falling back to the raw code (``1809807``) which the UI was rendering as a meaningless numeric label.""" if not dept_code: return dept = (self._departments or {}).get(dept_code) dept_name = _department_name(dept, dept_code) membership = await self.db.scalar( select(IdentityOrgMembership).where( IdentityOrgMembership.person_id == person.id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ) if membership is None: membership = IdentityOrgMembership(person_id=person.id) self.db.add(membership) membership.source_system = "successfactors" membership.org_node_type = "department" membership.org_node_source_id = dept_code membership.org_node_code = dept_code membership.org_node_name = dept_name membership.parent_org_node_source_id = (dept or {}).get("parent") membership.org_path = _build_org_path(dept, self._departments or {}) membership.is_primary = True membership.membership_status = "active" membership.last_synced_at = datetime.now(UTC) async def _assign_manager(self, person: IdentityPerson, job: dict | None) -> None: # Stale-pointer rule (Round-20 audit, finding M1): the # incremental-sync path used to ``return`` without clearing # ``manager_person_id`` whenever the new HRIS payload omitted # managerId or pointed at someone we hadn't seen yet. That left # the previous manager linked forever — RBAC kept granting the # old manager view over the subordinate's assets/permissions, # and Leaver-Blocked dual-sign still accepted them as a valid # secondary signer. Mirror the FULL-sync path # (``identity/successfactors_sync.py``: every branch sets # ``person.manager_person_id = None`` when no usable manager # resolves) so incremental and full sync converge on the same # owner-of-truth: a brief Null gap is safer than a stale pointer. from app.identity.identity_reconcile import ( normalize_manager_external_id, resolve_person_by_hris_user_id, ) manager_external_id = normalize_manager_external_id((job or {}).get("managerId")) # Persist the declared id (migration 270) so a late-arriving # manager can be re-linked by ``backfill_manager_links`` after # this subordinate's last event, and so the data-quality scanner # can tell a genuine HRIS "no manager" (NULL — incl. SF's # ``NO_MANAGER`` sentinel) from a Helix link failure (set, but # unresolved). person.manager_external_id = manager_external_id if not manager_external_id: if person.manager_person_id is not None: logger.info( "hr.sync: clearing manager_person_id for person %s " "(hris_person_id=%s) — HRIS event no longer reports managerId", person.id, person.hris_person_id, ) person.manager_person_id = None return # Multi-instance scoping (Round-20 audit, A3F1): without the # hris_instance_id filter, two SF tenants reusing the same # ``userId`` string (common when both tenants are seeded from a # shared directory) can wire one tenant's employee to the other # tenant's manager. RBAC then grants that wrong-tenant person # Manager view over the subordinate's assets / permissions, and # dual-sign overrides accept them as a valid secondary signer. # Mirror the ``(hris_instance_id == self.hris_instance_id) OR # IS NULL`` pattern this service uses everywhere else (lines # 1030, 1937) so legacy single-instance rows still resolve. # Round-49: refuse to bind a leaver / soft-deleted person as # someone's manager. HRIS update lag is common (the leaver's # manager pointer takes a tick to update); without these # filters Helix would wire subordinates to a tombstoned # manager whose mailbox is suspended and whose downstream # notifications vanish silently. The clear-on-miss branch # below then handles the resulting "no match" case the same # way it does for a literally-missing manager. # Alias-aware since 2026-08-18: ``managerId`` is an EMPLOYMENT key, # so a manager who holds two concurrent employments is referenced by # whichever userId owns the team — which may be a userId Helix keeps # no person row for. ``resolve_person_by_hris_user_id`` tries the # primary column (instance-scoped, as before) and then the alias # table; the live-row filters are unchanged. manager = await resolve_person_by_hris_user_id( self.db, manager_external_id, hris_instance_id=self.hris_instance_id, ) if manager is None: # Manager declared by HRIS but no IdentityPerson row exists yet # (manager not synced into Helix, or hard-deleted, OR they're # a leaver — Round-49). Same rationale as the FULL-sync path's # clear-on-miss (``successfactors_sync.py:340-353``): a stale # pointer to a previous manager is worse than a brief gap # that the next sync closes once the real manager onboards. if person.manager_person_id is not None: logger.info( "hr.sync: clearing manager_person_id for person %s — " "HRIS-declared manager %s has no live IdentityPerson row", person.id, manager_external_id, ) person.manager_person_id = None return # Self-reference guard: a corrupt OData feed can list the # employee as their own manager. Mirroring successfactors_sync # which has the same check at line ~322. Do NOT clear here — # an existing valid pointer should survive a corrupt event. if manager.id == person.id: return person.manager_person_id = manager.id async def _upsert_external_identity( self, person: IdentityPerson, user_id: str, user: dict | None, ) -> None: # NOTE on deleted_at: `_upsert_person` above is intentionally # soft-delete-tolerant (it adopts a previously-soft-deleted # person on re-import), so the ``person.id`` we receive here # may already have a soft-deleted ``identity_external_identities`` # row keyed on (person_id, provider='successfactors'). # Filtering by ``deleted_at IS NULL`` here would hide that row # → INSERT path → ``Duplicate entry for key # uq_identity_external_person_provider`` (1062) on flush, # because the UNIQUE constraint spans soft-deletes. Look up # without the deleted_at filter and revive whatever we find, # so re-importing a previously-cleaned-up employee Just Works. # Prod incident 2026-05-26: 438-event batch crashed on confirm # for sf-cn-prod because 6 SAP admin test accounts # (ECadmin / 100xxx / COMPadmin) and 6 terminated PingCAP # employees had soft-deleted SF identities that SF still # returns from EmpJob queries. identity = await self.db.scalar( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == person.id, IdentityExternalIdentity.provider == "successfactors", ) ) if identity is None: identity = IdentityExternalIdentity( person_id=person.id, provider="successfactors", provider_subject=user_id, ) self.db.add(identity) elif identity.deleted_at is not None: # Revive the soft-deleted row: clear the tombstone, refresh # provider_subject (may have changed if SF reassigned the # internal userId — rare but documented), let the field # updates below overwrite the rest. identity.deleted_at = None identity.provider_subject = user_id identity.provider_username = user_id identity.provider_email = (user or {}).get("email") # Identity status follows the person's employment status: forfeited # (status='f') SF users land their IdentityPerson at 'left' or # 'leaver_pending' via _upsert_person; tagging the external_identity # row 'active' for those persons would lie about the upstream # state and trip identity-sync targets (Navan / SCIM) that # propagate active=true to downstream systems. Mirror the # employment lifecycle: 'left' / 'leaver_pending' / 'terminated' # → inactive identity, anything else → active. (Pre-PR contract # always wrote 'active', which was harmless when the live sync # only touched onboard events — but the wangdi PR's full-sync # branch now drives 'offboard' through this code path, surfacing # the divergence the backfill script already had to work around.) _person_status = (person.employment_status or "").lower() identity.identity_status = ( "inactive" if _person_status in {"left", "leaver_pending", "terminated"} else "active" ) identity.is_login_identity = False identity.source_updated_at = _parse_datetime((user or {}).get("lastModifiedDateTime")) identity.last_synced_at = datetime.now(UTC) async def _upsert_lifecycle_event( self, *, person: IdentityPerson, event_type: str, effective_date: date, employment: dict, job: dict | None, user: dict | None, trigger_source: str = "sync_confirm", extra_detail: dict | None = None, ) -> tuple[bool, str | None, HrLifecycleEvent | None]: """Insert-if-missing for the lifecycle ledger row. Returns ``(event_created, prior_trigger_source, row)`` — ``row`` is the live ledger row (created or pre-existing, ``None`` only when a concurrent racer's row could not be re-read), which the caller uses to read/stamp ``dispatched_at`` (#2220). * ``event_created=True, prior_trigger_source=None`` — brand-new row. * ``event_created=False, prior_trigger_source=`` — row already existed; the caller uses ``prior_trigger_source`` to decide whether to re-fire coordinators (e.g. when the prior row was an ``initial_seed`` bootstrap that intentionally skipped dispatch and now a real webhook needs to upgrade it). When upgrading from ``initial_seed`` to a real trigger source (``webhook`` / ``sync_confirm``), the existing row's ``trigger_source`` is updated in-place so subsequent reads see the post-upgrade state. The row keeps its original ``created_at`` so the audit trail of "when SF first told us about this person" survives. """ existing = await self.db.scalar( select(HrLifecycleEvent).where( HrLifecycleEvent.person_id == person.id, HrLifecycleEvent.event_type == event_type, HrLifecycleEvent.effective_date == effective_date, ) ) if existing is not None: prior_source = existing.trigger_source # Upgrade ``initial_seed`` → real source so a future replay # sees the post-dispatch state and the coordinator-skip # gate stays accurate. if prior_source == "initial_seed" and trigger_source != "initial_seed": existing.trigger_source = trigger_source return False, prior_source, existing detail = { "source_system": "successfactors", "user_id": employment.get("userId"), "person_id_external": employment.get("personIdExternal"), "event_code": (job or {}).get("event"), "event_reason": (job or {}).get("eventReason"), "employment_status_code": (job or {}).get("emplStatus"), "employment_start_date": employment.get("startDate"), "employment_end_date": employment.get("endDate"), "last_modified_at": employment.get("lastModifiedDateTime"), "display_name": (user or {}).get("displayName"), "email": (user or {}).get("email"), } if extra_detail: detail.update(extra_detail) # Insert under a SAVEPOINT guarded by the (person_id, event_type, # effective_date) UNIQUE constraint. The select-then-insert above has # a TOCTOU window: two concurrent SuccessFactors webhook redeliveries # of the same event both see ``existing is None`` and both try to # INSERT. The unique constraint makes the race-loser's flush raise # IntegrityError; we roll back to the savepoint, re-read the winner's # row, and return ``(False, prior_source)`` so the caller does NOT # fire onboarding/offboarding coordinators a second time. begin_nested # keeps the parent transaction usable for the rest of the batch. new_event = HrLifecycleEvent( person_id=person.id, event_type=event_type, effective_date=effective_date, detail=detail, trigger_source=trigger_source, ) try: async with self.db.begin_nested(): self.db.add(new_event) await self.db.flush() except IntegrityError: winner = await self.db.scalar( select(HrLifecycleEvent).where( HrLifecycleEvent.person_id == person.id, HrLifecycleEvent.event_type == event_type, HrLifecycleEvent.effective_date == effective_date, ) ) return ( False, (winner.trigger_source if winner is not None else None), winner, ) return True, None, new_event async def _jobs_by_user_id(self, user_ids: list[str]) -> dict[str, list[dict]]: client = await self._get_client() jobs = await client.get_jobs_for_users(user_ids) grouped: dict[str, list[dict]] = {} for job in jobs: user_id = job.get("userId") if not user_id: continue grouped.setdefault(user_id, []).append(job) return grouped async def _software_by_user_id(self, user_id: UUID) -> list[AssetSoftwareCatalog]: """Get software catalog items where user is owner or IT contact.""" result = await self.db.execute( select(AssetSoftwareCatalog) .where( or_( AssetSoftwareCatalog.owner_person_id == user_id, AssetSoftwareCatalog.it_contact_person_id == user_id, ) ) .order_by(AssetSoftwareCatalog.name.asc()) ) return list(result.scalars().all()) async def _software_by_user_ids(self, user_ids: list[UUID]) -> dict[UUID, list[AssetSoftwareCatalog]]: """Batch: get software catalog items where users are owner or IT contact.""" if not user_ids: return {} result = await self.db.execute( select(AssetSoftwareCatalog) .where( or_( AssetSoftwareCatalog.owner_person_id.in_(user_ids), AssetSoftwareCatalog.it_contact_person_id.in_(user_ids), ) ) .order_by(AssetSoftwareCatalog.name.asc()) ) grouped: dict[UUID, list[AssetSoftwareCatalog]] = {} for item in result.scalars().all(): if item.owner_person_id in user_ids: grouped.setdefault(item.owner_person_id, []).append(item) if item.it_contact_person_id in user_ids and item.it_contact_person_id != item.owner_person_id: grouped.setdefault(item.it_contact_person_id, []).append(item) return grouped async def _assets_by_user_id(self, user_ids: list[UUID]) -> dict[UUID, list[AssetItem]]: if not user_ids: return {} result = await self.db.execute( select(AssetItem) .where(AssetItem.assigned_to.in_(user_ids)) .order_by(AssetItem.asset_tag.asc()) ) grouped: dict[UUID, list[AssetItem]] = {} for asset in result.scalars().all(): if asset.assigned_to is None: continue grouped.setdefault(asset.assigned_to, []).append(asset) return grouped async def _latest_offboard_event_by_person_id( self, person_ids: list[UUID], ) -> dict[UUID, HrLifecycleEvent]: if not person_ids: return {} result = await self.db.execute( select(HrLifecycleEvent) .where( HrLifecycleEvent.person_id.in_(person_ids), HrLifecycleEvent.event_type == "offboard", ) .order_by(HrLifecycleEvent.person_id.asc(), HrLifecycleEvent.effective_date.desc(), HrLifecycleEvent.created_at.desc()) ) latest: dict[UUID, HrLifecycleEvent] = {} for event in result.scalars().all(): latest.setdefault(event.person_id, event) return latest async def _list_recent_lifecycle_items(self, *, event_type: str, limit: int) -> list[dict]: result = await self.db.execute( select(HrLifecycleEvent, IdentityPerson) .join(IdentityPerson, HrLifecycleEvent.person_id == IdentityPerson.id) .options(selectinload(IdentityPerson.org_memberships)) .where(HrLifecycleEvent.event_type == event_type, IdentityPerson.deleted_at.is_(None)) .order_by(HrLifecycleEvent.effective_date.desc(), HrLifecycleEvent.created_at.desc()) .limit(limit) ) rows = result.all() unresolved_counts = await self._unresolved_asset_counts([person.id for _, person in rows]) items: list[dict] = [] for event, person in rows: items.append( { "user_id": str(person.id), "display_name": person.display_name or person.hris_person_id, "email": person.work_email, "status": person.employment_status, "effective_date": event.effective_date, "department": person.department, "unresolved_asset_count": unresolved_counts.get(person.id, 0) if event_type == "offboard" else 0, } ) return items async def _unresolved_asset_counts(self, user_ids: list[UUID]) -> dict[UUID, int]: assets_by_user = await self._assets_by_user_id(user_ids) counts: dict[UUID, int] = {} for user_id, assets in assets_by_user.items(): counts[user_id] = sum(0 if _is_asset_resolved(asset) else 1 for asset in assets) return counts def _build_offboarding_case_payload( self, *, person: IdentityPerson, assets: list[AssetItem], software_items: list[AssetSoftwareCatalog], latest_event: HrLifecycleEvent | None, include_assets: bool, ) -> dict: asset_payloads = [_build_asset_payload(asset) for asset in assets] software_payloads = [_build_software_payload(s, person.id) for s in software_items] completion = dict((latest_event.detail or {}).get("completion") or {}) if latest_event else {} unresolved_asset_count = sum(0 if asset["is_resolved"] else 1 for asset in asset_payloads) active_software_count = sum(0 if _is_software_resolved(s) else 1 for s in software_items) blocking_reasons: list[str] = [] if unresolved_asset_count > 0: blocking_reasons.append(f"{unresolved_asset_count} unresolved physical asset(s)") if active_software_count > 0: blocking_reasons.append(f"{active_software_count} active software ownership(s)") return { "user_id": str(person.id), "employee_id": person.employee_no, "display_name": person.display_name or person.hris_person_id, "email": person.work_email, "status": person.employment_status, "last_day": person.termination_date, "manager_name": person.manager.display_name if person.manager is not None else None, "unresolved_asset_count": unresolved_asset_count, "unresolved_software_count": active_software_count, # Already-``left`` persons are terminal: offboarding is done # whether or not a ``completion`` record was written (HRIS sync # can flip a leaver straight to 'left' without going through # this route). Don't offer "完成离职" again — completed⟹left, and # left ⟹ nothing left to complete. "can_complete": ( len(blocking_reasons) == 0 and completion.get("completed_at") is None and person.employment_status != "left" ), "blocking_reasons": blocking_reasons, "assets": asset_payloads if include_assets else [], "software_assets": software_payloads if include_assets else [], "completed_at": completion.get("completed_at"), "completed_by": completion.get("completed_by"), "completion_note": completion.get("note"), } def _select_job_snapshot( self, event_type: str, employment: dict, jobs: list[dict], ) -> dict | None: target_date = _effective_date(event_type, employment) if target_date is None or not jobs: return None for job in jobs: if _parse_date(job.get("startDate")) == target_date: return job if event_type == "offboard" and _parse_date(job.get("endDate")) == target_date: return job return max( jobs, key=lambda job: _parse_datetime(job.get("lastModifiedDateTime")) or datetime.min.replace(tzinfo=UTC), ) def _build_asset_payload(asset: AssetItem) -> dict: offboarding = dict((asset.custom_fields or {}).get("offboarding") or {}) status = offboarding.get("status") or "pending" return { "asset_id": str(asset.id), "asset_tag": asset.asset_tag, "asset_name": asset.asset_name, "category": asset.category, "asset_status": asset.status, "handling_status": status, "note": offboarding.get("note"), "updated_at": offboarding.get("updated_at"), "updated_by": offboarding.get("updated_by"), "is_resolved": status in RESOLVED_OFFBOARDING_STATUSES, } def _build_software_payload(software: AssetSoftwareCatalog, person_id: UUID) -> dict: role = [] if software.owner_person_id == person_id: role.append("owner") if software.it_contact_person_id == person_id: role.append("it_contact") offboarding = dict((software.custom_fields or {}).get("offboarding") or {}) handling_status = offboarding.get("status") or "pending" return { "software_id": str(software.id), "name": software.name, "asset_type": software.asset_type, "status": software.status, "role": ", ".join(role), "department": software.department, "handling_status": handling_status, "note": offboarding.get("note"), "updated_at": offboarding.get("updated_at"), "updated_by": offboarding.get("updated_by"), "is_resolved": handling_status in RESOLVED_OFFBOARDING_STATUSES, } def _is_software_resolved(software: AssetSoftwareCatalog) -> bool: offboarding = dict((software.custom_fields or {}).get("offboarding") or {}) return (offboarding.get("status") or "pending") in RESOLVED_OFFBOARDING_STATUSES def _is_asset_resolved(asset: AssetItem) -> bool: offboarding = dict((asset.custom_fields or {}).get("offboarding") or {}) return (offboarding.get("status") or "pending") in RESOLVED_OFFBOARDING_STATUSES def _effective_date(event_type: str, employment: dict) -> date | None: field = "startDate" if event_type == "onboard" else "endDate" return _parse_date(employment.get(field)) def _normalize_preferred_language(raw: str | None) -> str | None: """Map a SuccessFactors ``User.defaultLocale`` value to a canonical Helix locale code (``zh_CN`` / ``en_US`` / ``ja_JP``), or ``None`` if it can't be recognised. SF does NOT return BCP-47 codes here — in prod ``defaultLocale`` comes back as free-text display labels (observed: ``english`` ×313, ``chinese``, ``中文``, plus the occasional already-canonical ``en_US`` / ``zh_CN``). Writing those raw poisoned ``identity_persons. preferred_language`` with values the frontend's locale resolver couldn't parse (``chinese`` / ``中文`` → no match → silent fallback), and re-clobbered user-chosen values on every sync. Normalising at the write boundary keeps the column to the same closed set the self-service PATCH endpoint already enforces. """ if not raw: return None norm = raw.strip().replace("_", "-").replace("-", "").lower() # Chinese: zh / zh-cn / chinese / 中文 / 中 if norm.startswith("zh") or "chinese" in norm or "中" in raw: return "zh_CN" # Japanese: ja / japanese / 日本… if norm.startswith("ja") or "japanese" in norm or "日本" in raw: return "ja_JP" # English: en / en-us / english if norm.startswith("en") or "english" in norm: return "en_US" return None def _parse_date(value: str | date | datetime | None) -> date | None: # Delegate to the hardened SF parser (regex-based ``/Date(ms)/`` # handling that survives negative epochs, plus a try/except around # ``fromisoformat`` so a malformed string skips the field instead of # aborting event staging). The local implementation previously did # ``.split("-")[0]`` which silently turned any negative-epoch sentinel # (e.g. the 1900-01-01 null marker) into 1970-01-01, and let a bad # ISO string raise. Single source of truth — same helper the sync # path uses for hireDate / firstDateWorked / endDate. return _sf_parse_date(value) def _parse_datetime(value: str | datetime | None) -> datetime | None: if not value: return None if isinstance(value, datetime): return value if value.startswith("/Date("): inner = value.removeprefix("/Date(").removesuffix(")/") milliseconds = int(inner.split("+")[0].split("-")[0] or "0") return datetime.fromtimestamp(milliseconds / 1000, tz=UTC) return datetime.fromisoformat(value.replace("Z", "+00:00")) async def run_batch_apply_in_background( batch_id: UUID, confirmed_by: str, hris_instance_id: UUID | None = None, ) -> None: """Run ``_apply_staged_events`` for an ``applying`` batch on a fresh session — the coroutine the admin-confirm HTTP path hands to ``background.spawn`` after flipping the batch to ``applying``. Owns its own DB session (the request's session is closed once the ``202`` is returned) and raises the per-session lock-wait ceiling the same way the cron path does, since a full-bootstrap apply chains many ``identity_persons`` upserts. On an unexpected top-level failure the batch is intentionally LEFT in ``applying``: the per-event loop already isolates per-employee errors, so reaching here means an infra-level fault (lost DB connection, etc.). Marking it ``failed`` would strand a recoverable batch; instead the ``hris.resume_stuck_apply`` cron picks it up once the heartbeat goes stale and re-runs the idempotent apply. """ from sqlalchemy import text as _text from app.core.database import async_session_factory try: async with async_session_factory() as session: await session.execute(_text("SET SESSION innodb_lock_wait_timeout = 120")) service = SuccessFactorsLifecycleService( session, hris_instance_id=hris_instance_id ) result = await service._apply_staged_events(batch_id, confirmed_by) logger.info( "HRIS background apply finished for batch %s: %s", batch_id, result, ) except Exception: # noqa: BLE001 logger.exception( "HRIS background apply crashed for batch %s — batch left in " "'applying'; the hris.resume_stuck_apply cron will resume it " "once the heartbeat goes stale.", batch_id, ) async def run_webhook_events_in_background( events: list[SuccessFactorsWebhookEvent], integration_id: UUID | None = None, config: dict | None = None, ) -> None: """Apply an SF-pushed webhook batch on a fresh session, OFF the request thread — the coroutine ``_handle_sf_webhook`` hands to ``background.spawn`` after authenticating the request and returning ``202`` immediately. Why (504 fix): an SF onboard/offboard event is just a dated signal, but ``process_webhook_events`` turns each webhook into a mini full sync — a cold per-request master-data cache means ~12 SF OData list pulls (companies / departments / locations / positions / personal contact / job-function / job-family / email-labels) plus a ``get_users`` + ``get_jobs`` fetch, all before the person upsert and coordinator dispatch even start. Run inside the request that is tens of seconds against SF's EU datacenter — past the gateway proxy timeout, so SF's integration center saw ``504 Gateway Time-out`` and kept replaying. The record + email-writeback + onboarding-case work now runs here; the heavy account provisioning is still deferred to ``hire_date`` by the coordinators' future-hire gate. Owns its own DB session (the request's session is closed once the ``202`` is returned) and raises the per-session lock-wait ceiling the same way ``run_batch_apply_in_background`` does, since a burst of events chains many ``identity_persons`` upserts. Best-effort: on an unexpected crash the events are NOT lost — the scheduled incremental HRIS sync reconciles any onboard/offboard the webhook dropped (the same backstop the batch-apply path relies on). """ from sqlalchemy import text as _text from app.core.database import async_session_factory try: async with async_session_factory() as session: await session.execute( _text("SET SESSION innodb_lock_wait_timeout = 120") ) if integration_id is not None and config is not None: service = SuccessFactorsLifecycleService.for_instance( session, integration_id=integration_id, config=config, ) else: service = SuccessFactorsLifecycleService(session) summary = await service.process_webhook_events(events) logger.info( "HRIS webhook background apply finished: %s", summary.to_dict(), ) except Exception: # noqa: BLE001 logger.exception( "HRIS webhook background apply crashed — events not lost; the " "scheduled incremental HRIS sync will reconcile the missed " "transition(s).", ) _EMPTY_LEAVE_SUMMARY = { "accounts_upserted": 0, "postings_upserted": 0, "accounts_tombstoned": 0, "persons_touched": 0, "legacy_derived": 0, } # A manual one-person refresh performs several synchronous vendor calls. Keep # the lock comfortably beyond the normal request budget and renew it while the # worker is alive; otherwise an unusually slow SF response could let a second # request enter when the original Redis TTL lapses. LEAVE_REFRESH_LOCK_TTL_SECONDS = 120 LEAVE_REFRESH_FRESHNESS_SECONDS = 15 LEAVE_REFRESH_BUSY_RETRY_SECONDS = 2 def _leave_refresh_lock_key(person_id: object) -> str: return f"hr:leave_refresh:person:{person_id}" def _leave_refresh_freshness_key(person_id: object) -> str: return f"helix:hr:leave_refresh:fresh:{person_id}" def _leave_refresh_caller_lock_key(actor_id: object) -> str: return f"hr:leave_refresh:caller:{actor_id}" def _leave_refresh_caller_freshness_key(actor_id: object) -> str: return f"helix:hr:leave_refresh:caller_fresh:{actor_id}" def _leave_refresh_unavailable(message: str) -> AppException: return AppException( code="LEAVE_REFRESH_COORDINATION_UNAVAILABLE", message=message, status_code=503, headers={"Retry-After": str(LEAVE_REFRESH_BUSY_RETRY_SECONDS)}, ) _MARK_LEAVE_REFRESH_FRESHNESS_SCRIPT = """ redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) redis.call('set', KEYS[2], ARGV[1], 'EX', ARGV[2]) return 1 """ _CLEAR_FRESHNESS_IF_OWNED_SCRIPT = """ local removed = 0 for i = 1, #KEYS do if redis.call('get', KEYS[i]) == ARGV[1] then removed = removed + redis.call('del', KEYS[i]) end end return removed """ async def _clear_leave_refresh_freshness_if_owned( person_id: object, actor_id: object, owner_token: str | None, ) -> None: """Undo our success marker if the enclosing distributed lock was lost. The compare-and-delete is essential: after a lock loss, the new owner may already have completed and written its own marker. A plain DEL would erase that valid marker and reopen the vendor-call race. """ if owner_token is None: return try: redis = get_redis() await cast(Awaitable[Any], redis.eval( _CLEAR_FRESHNESS_IF_OWNED_SCRIPT, 2, _leave_refresh_freshness_key(person_id), _leave_refresh_caller_freshness_key(actor_id), owner_token, )) except Exception: # noqa: BLE001 — marker remains bounded by its TTL logger.warning( "could not clear leave-refresh freshness after lock loss " "(person=%s)", person_id, exc_info=True, ) async def _assert_leave_refresh_not_fresh( person_id: object, actor_id: object, ) -> None: """Raise a stable 429 while a successful refresh is still fresh. PTTL gives millisecond precision at the expiry boundary. Redis uses -2 for a missing key and -1 for a persistent key. The latter violates this gate's bounded-throttle invariant, so fail closed with a retryable 503 instead of throttling forever or silently calling SF forever. """ try: redis = get_redis() remaining_values = [ int(await redis.pttl(_leave_refresh_freshness_key(person_id))), int(await redis.pttl(_leave_refresh_caller_freshness_key(actor_id))), ] except Exception as exc: # noqa: BLE001 — Redis client/transport variants raise _leave_refresh_unavailable( "Leave refresh coordination is temporarily unavailable. Retry shortly.", ) from exc if -1 in remaining_values: raise _leave_refresh_unavailable( "Leave refresh coordination has an invalid freshness marker. Retry shortly.", ) remaining_ms = max(remaining_values) if remaining_ms > 0: retry_after = max(1, math.ceil(remaining_ms / 1000)) raise AppException( code="LEAVE_REFRESH_THROTTLED", message="Leave balances were refreshed recently. Retry shortly.", status_code=429, headers={"Retry-After": str(retry_after)}, ) async def _sync_one_person_leave_uncoordinated( db: AsyncSession, person: IdentityPerson, ) -> dict: """Vendor/database portion of one-person refresh; caller owns the gate.""" totals: dict = dict(_EMPTY_LEAVE_SUMMARY) services = await SuccessFactorsLifecycleService.fan_out(db) if not services: return {**totals, "skipped": "no_hris_instance"} pinned_instance_id = getattr(person, "hris_instance_id", None) if pinned_instance_id is not None: services = [ svc for svc in services if str(svc.hris_instance_id) == str(pinned_instance_id) ] if not services: # A pinned person whose owning integration is disabled/missing is # not a successful empty snapshot. Marking it fresh would suppress # retries and present old balances as vendor-confirmed. return {**totals, "skipped": "no_owning_hris_instance"} failed: list[str] = [] for svc in services: try: summary = await svc.sync_leave_data( user_ids=[person.hris_person_id], commit=False, ) except Exception: # noqa: BLE001 — isolate per region, keep going logger.exception( "per-person leave sync failed (instance=%s, person=%s)", svc.hris_instance_id or "primary", person.hris_person_id, ) failed.append(svc.hris_instance_id or "primary") continue for key in _EMPTY_LEAVE_SUMMARY: totals[key] += int(summary.get(key, 0) or 0) if failed: totals["failed_instances"] = failed return totals async def sync_one_person_leave_now( db: AsyncSession, person: IdentityPerson, *, actor_id: object | None = None, before_commit: Callable[[dict], Awaitable[None]] | None = None, ) -> dict: """Synchronously re-pull ONE person's SF EC Time-Off into ``hr_leave_*`` on the CALLER's session, so the HTTP response can hand back freshly-synced balances. This is the targeted counterpart to the daily ``hr.leave_account_sync`` cron / the admin "refresh everyone" background fan-out (:func:`run_leave_sync_in_background`): when an employee just edited a leave request in SF, HR wants *that* person current immediately rather than waiting for tomorrow's 06:30 tick. One person = a handful of SF calls, fast enough to run inline. Fans out across HRIS instances but only the instance that owns the person does any SF work — the others short-circuit on an empty roster (``sync_leave_data`` filters by ``user_ids`` + ``hris_instance_id``). Idempotent (re-keys on SF ``externalCode``); the refresh transaction is committed before a success freshness marker is published. Returns the accumulated summary, with ``skipped`` set when the person isn't SF-sourced (nothing to pull). """ if person.source_system != "successfactors" or not person.hris_person_id: totals = {**_EMPTY_LEAVE_SUMMARY, "skipped": "not_sf_sourced"} if before_commit is not None: try: await before_commit(totals) await db.commit() except Exception: await db.rollback() raise return totals # Fast rejection saves a Redis SETNX round-trip for an obvious repeat. # Re-check INSIDE the lock closes the race where two callers both saw # the old marker absent just before one of them completed. actor_id = actor_id or person.id await _assert_leave_refresh_not_fresh(person.id, actor_id) lock_key = _leave_refresh_lock_key(person.id) caller_lock_key = _leave_refresh_caller_lock_key(actor_id) freshness_token: str | None = None try: async with distributed_lock( caller_lock_key, ttl=LEAVE_REFRESH_LOCK_TTL_SECONDS, auto_renew=True, ): async with distributed_lock( lock_key, ttl=LEAVE_REFRESH_LOCK_TTL_SECONDS, auto_renew=True, ): await _assert_leave_refresh_not_fresh(person.id, actor_id) totals = await _sync_one_person_leave_uncoordinated(db, person) # The HTTP and MCP mutation surfaces supply their audit write # here. It must join the leave rows in the SAME transaction: # committing the vendor snapshot first left durable HR data # without evidence whenever AuditService.log failed (#2467). # The success marker must never precede the durable database # + audit commit. Otherwise a commit/audit failure leaves old # or unaudited balances behind while every retry is told the # stale cache was refreshed successfully. try: if before_commit is not None: await before_commit(totals) await db.commit() except Exception: await db.rollback() raise # A partial/empty result is deliberately NOT freshness-cached: # the next click must be able to retry SF immediately. if not totals.get("failed_instances") and not totals.get("skipped"): freshness_token = secrets.token_hex(16) marked = await cast(Awaitable[Any], get_redis().eval( _MARK_LEAVE_REFRESH_FRESHNESS_SCRIPT, 2, _leave_refresh_freshness_key(person.id), _leave_refresh_caller_freshness_key(actor_id), freshness_token, str(LEAVE_REFRESH_FRESHNESS_SECONDS), )) if not marked: raise _leave_refresh_unavailable( "Leave refresh completed but could not be safely recorded. Retry shortly.", ) return totals except LockBusy as exc: # Covers both acquire contention and auto-renew detecting a lost # lock. In either case another worker may own this person's run, so # the safe outcome is a bounded retry — never continued SF writes. await _clear_leave_refresh_freshness_if_owned( person.id, actor_id, freshness_token, ) raise AppException( code="LEAVE_REFRESH_IN_PROGRESS", message="Leave balances are already being refreshed. Retry shortly.", status_code=429, headers={"Retry-After": str(LEAVE_REFRESH_BUSY_RETRY_SECONDS)}, ) from exc except RedisError as exc: # Fail closed: Redis is the cross-worker correctness boundary. A # fail-open vendor call would reintroduce the quota/unique-key race. raise _leave_refresh_unavailable( "Leave refresh coordination is temporarily unavailable. Retry shortly.", ) from exc except AppException: raise except Exception as exc: # Redis clients do not consistently normalize transport failures to # RedisError (notably connection wrappers used in tests/deploys). # Once business data is committed, only the marker write can reach # here, so surface a safe retryable coordination failure. if freshness_token is not None: raise _leave_refresh_unavailable( "Leave refresh completed but could not be safely recorded. Retry shortly.", ) from exc raise async def run_leave_sync_in_background() -> None: """Fan-out leave-data sync across every enabled HRIS instance on a fresh session — the coroutine the admin "refresh now" HTTP path hands to ``background.spawn`` so the browser gets an immediate ``202`` instead of blocking on a full-roster SF pull. Idempotent (re-keys on SF ``externalCode``); a single region's failure is logged and the rest still complete. Mirrors the daily ``hr.leave_account_sync`` cron, just triggered on demand. """ from app.core.database import async_session_factory try: async with async_session_factory() as session: services = await SuccessFactorsLifecycleService.fan_out(session) for svc in services: try: summary = await svc.sync_leave_data() logger.info( "leave-data manual sync (instance=%s): %s", svc.hris_instance_id or "primary", summary, ) except Exception: # noqa: BLE001 — isolate per region logger.exception( "leave-data manual sync failed for instance %s", svc.hris_instance_id or "primary", ) except Exception: # noqa: BLE001 logger.exception("leave-data manual sync crashed at the top level")