"""Is this SF employment somebody's SECOND job? — the one place that answers. Twice now a 兼岗 (concurrent employment) has been minted into a real Google mailbox: * 2026-08-25 — Hiroki Shinagawa's 兼岗 became a second person plus ``hiroki.shinagawa01@pingcap.com`` (#3228); * 2026-09-04 — 战捷's EU 兼岗 landed on a purged leaver's row and was provisioned at **that dead colleague's** address, which Helix then wrote back into SF as 战捷's primary business email (#3434). Both fixes were correct and both were *ingest-side*. What neither addressed is that **nothing ever checked the question at the point of no return**: `create_account_for` minted the mailbox without once asking "is this person a second employment of somebody we already know?". An ingest guard can only cover the paths it is wired into; this module exists so the question can also be asked where the irreversible act happens, and so both sides ask it the same way. Three independent signals, cheapest first: 1. **the alias table** — a userId already registered as somebody's duplicate employment. Pure DB, no network. (A state reachable when ingest aliased the userId while the row was a tombstone and some later path revived the row.) 2. **``EmpEmployment.personIdExternal``** — SF's *person* key. When it names a DIFFERENT live person, this userId is that human's second employment. This is the authority, and it does not depend on HR having ticked anything. 3. **``EmpJob.customString15`` = 【是否主岗】** — the HR-facing flag, located 2026-09-04 once HR had actually ticked "No" on a 兼岗 (before that we had only falsified `isPrimaryAssignment` / `isPrimary` / `assignmentClass`, none of which exist in either tenant's OData dictionary). Live across both prod tenants it separates 22/22 rows: every one of the 10 known 兼岗 rows is ``2061``/No and every one of the 12 canonical rows is ``2059``/Yes. Signal 3 is deliberately LAST and deliberately one-directional: the picklist is nullable (prod has 兼岗 rows whose sibling flag is ``None``), so an empty value means "no evidence" and must never be read as "this is the primary job". ``2061`` is only meaningful as an explicit No. ⚠️ ``customString15`` (是否主岗) and ``customString24`` (是否推送下游) are two different fields sharing one Yes/No picklist, so the SAME option ids ``2059``/``2061`` appear in both. Seeing ``2061`` is not an answer until you know which field it came from. **Fail-open on failure, fail-closed on evidence.** An SF outage returns ``None`` (no verdict) rather than blocking provisioning — refusing a real new hire is how #191 silently ate two of them. Only an actual answer from SF stops the mint. **This module never writes.** It refuses a mailbox; it does not tidy the duplicate row, and in particular it does not register an alias — the row it is refusing holds that userId as its own primary key while being live, which is exactly the shape ``record_hris_user_alias`` refuses (aliasing somebody's primary key would stop syncing them altogether). Reorganising identity is the ingest side's job, where the row is absent or already a tombstone. Re-deriving the verdict next attempt costs one SF call and always answers the same. """ from __future__ import annotations import logging from collections.abc import Mapping from dataclasses import dataclass from typing import Any from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) #: SF ``EmpJob`` field carrying HR's 【是否主岗】 flag, and its picklist #: option ids. See the module docstring for how these were established #: and why the ids are shared with ``customString24``. IS_PRIMARY_ASSIGNMENT_FIELD = "customString15" IS_PRIMARY_YES = "2059" IS_PRIMARY_NO = "2061" 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 one **段雇佣** by ``userId``. Two shapes make the carried value unusable AS A PERSON KEY: empty, and self-referential (``pid == user_id``) — the latter is the employment key echoed back, emitted both by SF's outbound iflow and by Helix's own full-sync synthetic employment dicts. Note the asymmetry: for a single-employment person ``pid == userId`` legitimately, so "unusable" means **"not proof of anything"**, never "wrong". Canonical home for this predicate; ``SuccessFactorsLifecycleService.person_id_external_is_usable`` delegates here so the ingest guard and the account minter cannot drift apart. """ pid = (person_id_external or "").strip() if not pid: return False return pid != (user_id or "").strip() def job_says_not_primary(job: Mapping[str, Any] | None) -> bool: """True only when SF EXPLICITLY says 【是否主岗】= No. One-directional on purpose: ``None`` / missing / anything else is "no evidence", NOT "primary". The picklist is nullable in prod, so reading absence as a Yes would hand us back the exact failure mode this module exists to close. """ if not job: return False raw = job.get(IS_PRIMARY_ASSIGNMENT_FIELD) return str(raw).strip() == IS_PRIMARY_NO if raw is not None else False @dataclass(frozen=True, slots=True) class DuplicateEmployment: """Evidence that a person row is somebody else's second employment.""" #: Which signal fired — ``registered_alias`` / ``person_id_external`` #: / ``not_primary_assignment``. Recorded in the audit reason so a #: later reader knows how strong the verdict was. signal: str detail: str owner_id: UUID | None = None owner_hris_person_id: str | None = None person_id_external: str | None = None async def duplicate_employment_for( db: AsyncSession, person: IdentityPerson, ) -> DuplicateEmployment | None: """Is ``person``'s HRIS key a second employment of somebody else? Returns the evidence, or ``None`` for "no reason to think so" — which includes every failure path (see the module docstring on fail-open). Callers treat a non-``None`` result as "do not mint". Cost: one DB read always; up to two SF calls, and only for a SuccessFactors-sourced person. Account creation runs a handful of times a day, so unlike the sync path this can afford to ask. """ if person.source_system != "successfactors": return None user_id = (person.hris_person_id or "").strip() if not user_id: return None # ── 1. the alias table (no network) ─────────────────────────── from app.identity.identity_reconcile import ( # noqa: PLC0415 hris_user_alias_owner, ) alias_owner = await hris_user_alias_owner(db, user_id) if alias_owner is not None and alias_owner.id != person.id: return DuplicateEmployment( signal="registered_alias", detail=( f"userId {user_id} is registered in " f"identity_hris_user_aliases as a duplicate employment of " f"person {alias_owner.hris_person_id}" ), owner_id=alias_owner.id, owner_hris_person_id=alias_owner.hris_person_id, ) try: from app.identity.hris_client_routing import ( # noqa: PLC0415 sf_client_for_person, ) client = await sf_client_for_person(db, person) except Exception as exc: # noqa: BLE001 logger.warning( "sf_assignment: no SF client for person=%s (%s) — duplicate-" "employment check degrades to the alias table only", person.id, exc, ) return None # ── 2. personIdExternal: SF's own person key (the authority) ── try: pid_ext = await client.get_person_id_external(user_id) except Exception as exc: # noqa: BLE001 logger.warning( "sf_assignment: EmpEmployment lookup failed for userId=%s (%s) " "— failing OPEN so a real hire is never blocked by an SF outage", user_id, exc, ) pid_ext = None if person_id_external_is_usable(pid_ext, user_id): canonical = await db.scalar( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.hris_person_id == str(pid_ext), IdentityPerson.deleted_at.is_(None), IdentityPerson.id != person.id, ) ) if canonical is not None: # Deliberately does NOT register the alias. Reaching here means # a LIVE person row holds ``user_id`` as its own primary key — # the exact shape ``record_hris_user_alias`` refuses, because # aliasing somebody's primary key short-circuits their person # upsert and would stop syncing them entirely. This function's # job is to refuse the mailbox, not to reorganise identity; # tidying the duplicate row is the ingest side's business # (where the row is either absent or already a tombstone) or a # deliberate operator action. Re-deriving the verdict on the # next attempt costs one SF call and always answers the same. return DuplicateEmployment( signal="person_id_external", detail=( f"SF says EmpEmployment(userId={user_id}) belongs to " f"personIdExternal={pid_ext}, which is person " f"{canonical.hris_person_id} " f"({canonical.work_email}) — this row is that human's " f"second employment, not a new hire" ), owner_id=canonical.id, owner_hris_person_id=canonical.hris_person_id, person_id_external=str(pid_ext), ) # ── 3. HR's 【是否主岗】 flag (explicit No only) ─────────────── try: jobs = await client.get_jobs_for_users([user_id]) except Exception as exc: # noqa: BLE001 logger.warning( "sf_assignment: EmpJob lookup failed for userId=%s (%s) — " "skipping the 是否主岗 signal", user_id, exc, ) jobs = [] for job in jobs or []: if not isinstance(job, Mapping): continue if str(job.get("userId") or "").strip() != user_id: continue if job_says_not_primary(job): return DuplicateEmployment( signal="not_primary_assignment", detail=( f"SF EmpJob.{IS_PRIMARY_ASSIGNMENT_FIELD} = " f"{IS_PRIMARY_NO} (【是否主岗】= 否) for userId " f"{user_id} — HR marked this employment as NOT the " f"primary job, so it must never get its own mailbox" ), person_id_external=str(pid_ext) if pid_ext else None, ) return None