"""One-shot: retire person rows minted for a 兼岗 (concurrent employment). Context (prod 2026-08-25). SF keys a *person* by ``personIdExternal`` and one *段雇佣* by ``userId``; a 兼岗 is a second ``userId`` under the SAME ``personIdExternal``. Helix keys a person by ``userId``, and the guard that stops the second employment becoming a second person went blind because both the SF outbound iflow and Helix's own full-sync synthesis echo ``userId`` back into the ``personIdExternal`` field (fixed in the same PR as this script). Live damage this repairs: Hiroki Shinagawa's 兼岗 (``userId 80100063`` → ``personIdExternal P01008``) became a second identity_persons row with a second corporate mailbox ``hiroki.shinagawa01@pingcap.com``, a real Google Workspace account, an open onboarding process and an org membership under the 兼岗's department. The 08-18 sibling script (``fix_sf_duplicate_userid_aliases_2026_08_18``) cannot find this shape: it discovers ghosts by their synthetic ``@successfactors.local`` email, and this one carries a REAL corporate address. So discovery here is a full-roster sweep instead: every live SF person → SF EmpEmployment(userId) → personIdExternal → is that person key owned by a DIFFERENT live ACTIVE person? Nothing is hardcoded and nothing is trusted: every mapping is confirmed against live SF per tenant, and the script refuses anything SF does not corroborate. Safe to re-run (idempotent; already-retired rows drop out of the candidate set). Google deletion is behind its OWN flag on purpose — the DB repair is reversible, ``users().delete()`` is not (Google keeps a ~20-day undelete window and nothing more). Three hard assertions guard it: the address must not be the canonical person's, must not belong to any other live person row, and the account must have never been logged into. In the prod backend container (it holds the real credential key; the local dev key cannot decrypt prod integration secrets). ``PYTHONPATH=/app`` is load-bearing — running ``scripts/x.py`` by path puts ``/app/scripts`` on ``sys.path[0]``, not ``/app``, so ``import app`` fails:: docker exec -w /app -e PYTHONPATH=/app compose-backend-1 \ /app/.venv/bin/python \ scripts/fix_kengang_ghost_accounts_2026_08_25.py # dry-run docker exec -w /app -e PYTHONPATH=/app compose-backend-1 \ /app/.venv/bin/python \ scripts/fix_kengang_ghost_accounts_2026_08_25.py --commit docker exec -w /app -e PYTHONPATH=/app compose-backend-1 \ /app/.venv/bin/python \ scripts/fix_kengang_ghost_accounts_2026_08_25.py --commit --delete-google """ from __future__ import annotations import argparse import asyncio import json from datetime import UTC, datetime from uuid import uuid4 from sqlalchemy import select import app.audit.models # noqa: F401 import app.auth.models # noqa: F401 import app.domains.hr.models # noqa: F401 import app.domains.onboarding.models # noqa: F401 import app.identity.person_models # noqa: F401 from app.audit.models import AuditLog from app.core.database import async_session_factory from app.domains.onboarding.models import ( PROC_CANCELLED, PROC_OPEN_STATUSES, OnboardingProcess, ) from app.identity.identity_reconcile import record_hris_user_alias from app.identity.person_models import ( IdentityExternalIdentity, IdentityOrgMembership, IdentityPerson, ) from app.identity.successfactors_auth import build_auth_from_config from app.identity.successfactors_client import SuccessFactorsClient from app.integrations.compat import list_hris_instances async def _pid_map(session) -> dict[str, str]: """``userId → personIdExternal`` for every live SF person, from SF. One chunked ``get_employments_for_users`` pass per tenant (chunks of 50) rather than a lookup per person — ~11 requests for a 500-person roster. A userId lives in exactly one tenant, so first hit wins. """ user_ids = [ u for u in (await session.execute( select(IdentityPerson.hris_person_id).where( IdentityPerson.source_system == "successfactors", IdentityPerson.deleted_at.is_(None), IdentityPerson.hris_person_id.is_not(None), ) )).scalars().all() if u ] mapping: dict[str, str] = {} for row, cfg in await list_hris_instances(session=session): base_url = (cfg.get("base_url") or "").strip() if not base_url: continue client = SuccessFactorsClient( base_url=base_url, auth=build_auth_from_config(cfg), ) missing = [u for u in user_ids if u not in mapping] if not missing: break try: rows = await client.get_employments_for_users(sorted(missing)) except Exception as exc: # noqa: BLE001 — try the next tenant print(f" ! {row.name}: EmpEmployment pull failed " f"({type(exc).__name__}) — skipping this tenant") continue for emp in rows: uid, pid = emp.get("userId"), emp.get("personIdExternal") if uid and pid: mapping.setdefault(str(uid), str(pid)) print(f" {row.name}: resolved {len(mapping)}/{len(user_ids)}") return mapping async def run(*, dry_run: bool, delete_google: bool) -> int: async with async_session_factory() as session: print("Resolving personIdExternal for the live roster from SF…") pid_by_uid = await _pid_map(session) if not pid_by_uid: print("SF returned no mapping at all — refusing to act.") return 2 people = (await session.execute( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.deleted_at.is_(None), ) )).scalars().all() by_hris = {p.hris_person_id: p for p in people if p.hris_person_id} # ── Discover ───────────────────────────────────────────────── # A ghost is a live person row whose userId SF says belongs to a # DIFFERENT live ACTIVE person. ``pid == uid`` is the normal # single-employment shape and is skipped; an unknown/unresolved # userId is skipped (SF did not corroborate). planned: list[tuple[IdentityPerson, IdentityPerson, str]] = [] for person in people: uid = person.hris_person_id pid = pid_by_uid.get(uid or "") if not uid or not pid or pid == uid: continue canonical = by_hris.get(pid) if canonical is None: print(f" {uid}: SF says person key {pid}, but Helix has no " f"live row for it — REFUSING (not a duplicate we can " f"resolve; the canonical row must exist first)") continue if canonical.id == person.id: continue if canonical.employment_status != "active": # A ``left`` owner + a new userId is a REHIRE, not a 兼岗 — # aliasing it would freeze the rehire at its departed state. print(f" {uid}: owner {pid} is " f"{canonical.employment_status!r}, not active — " f"REFUSING (rehire shape, not a concurrent employment)") continue planned.append((person, canonical, pid)) if not planned: print("\nNo 兼岗-minted person rows found. Nothing to do.") return 0 print(f"\nGhost rows to retire: {len(planned)}") google_targets: list[tuple[IdentityPerson, IdentityExternalIdentity]] = [] for ghost, canonical, pid in planned: print(f"\n ghost userId={ghost.hris_person_id} " f"person_id={ghost.id}") print(f" → canonical userId={canonical.hris_person_id} " f"(personIdExternal={pid}) person_id={canonical.id}") print(f" ghost work_email={ghost.work_email!r} " f"hire_date={ghost.hire_date} " f"status={ghost.employment_status!r}") exts = (await session.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == ghost.id, IdentityExternalIdentity.deleted_at.is_(None), ) )).scalars().all() for ext in exts: print(f" external identity: {ext.provider} " f"{ext.provider_email or ext.provider_subject} " f"status={ext.identity_status!r}") if ext.provider == "google" and ext.provider_email: google_targets.append((ghost, ext)) procs = (await session.execute( select(OnboardingProcess).where( OnboardingProcess.person_id == ghost.id, OnboardingProcess.status.in_(PROC_OPEN_STATUSES), ) )).scalars().all() for proc in procs: print(f" open onboarding process {proc.id} " f"status={proc.status!r} → will cancel") memberships = (await session.execute( select(IdentityOrgMembership).where( IdentityOrgMembership.person_id == ghost.id, IdentityOrgMembership.deleted_at.is_(None), ) )).scalars().all() for m in memberships: print(f" org membership {m.org_path!r} → will soft-delete") if dry_run: print("\nDRY RUN — no changes written. Re-run with --commit" + (" --delete-google" if delete_google else "") + ".") return 0 # ── Google first, while the person row still names the account ─ # Deleting BEFORE the DB retirement means a failure here leaves a # fully-consistent (if still wrong) state rather than an orphaned # live mailbox nothing in Helix points at any more. deleted_google: list[str] = [] if delete_google and google_targets: from app.domains.google_workspace.account_provisioner import ( _get_google_integration, ) from app.integrations.driver_base import ProvisioningContext from app.integrations.drivers.google_workspace import ( GoogleWorkspaceDriver, ) from app.integrations.service import _resolve integration = await _get_google_integration(session) if integration is None: print(" !! no enabled Google Workspace integration — " "skipping account deletion (DB repair continues)") google_targets = [] else: resolved = await _resolve(session, integration) driver = GoogleWorkspaceDriver( integration_id=str(integration.id), config=resolved.config, ) live_emails = { (p.work_email or "").lower() for p in people } - { (g.work_email or "").lower() for g, _, _ in planned } for ghost, ext in google_targets: email = (ext.provider_email or "").lower() canonical = next( c for g, c, _ in planned if g.id == ghost.id ) # Three refusals, in order of how bad the mistake would be. if email == (canonical.work_email or "").lower(): print(f" !! {email} IS the canonical person's mailbox — " f"REFUSING to delete") continue if email in live_emails: print(f" !! {email} is owned by another live person row " f"— REFUSING to delete") continue profile = await driver.get_user_profile(email) if profile is None: print(f" {email}: already absent from Google") else: last_login = str(profile.get("lastLoginTime") or "") print(f" {email}: suspended={profile.get('suspended')} " f"lastLoginTime={last_login!r}") # Google reports the epoch for "never signed in". An # account somebody actually used is not a ghost we get # to delete unattended. if last_login and not last_login.startswith("1970-"): print(f" !! {email} HAS been signed into " f"({last_login}) — REFUSING to delete; " f"escalate to IT") continue result = await driver._delete_user( ProvisioningContext(user_email=email), ) if not result.ok: print(f" !! delete failed: {result.error_message}") continue print(f" deleted Google user {email} " f"({result.raw_response})") deleted_google.append(email) # ── Retire the ghost rows, then alias ──────────────────────── now = datetime.now(UTC) for ghost, canonical, pid in planned: ghost_uid = ghost.hris_person_id for ext in (await session.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == ghost.id, IdentityExternalIdentity.deleted_at.is_(None), ) )).scalars().all(): ext.deleted_at = now ext.is_login_identity = False for m in (await session.execute( select(IdentityOrgMembership).where( IdentityOrgMembership.person_id == ghost.id, IdentityOrgMembership.deleted_at.is_(None), ) )).scalars().all(): m.deleted_at = now m.membership_status = "inactive" for proc in (await session.execute( select(OnboardingProcess).where( OnboardingProcess.person_id == ghost.id, OnboardingProcess.status.in_(PROC_OPEN_STATUSES), ) )).scalars().all(): proc.status = PROC_CANCELLED proc.cancelled_at = now # Retire BEFORE aliasing: ``record_hris_user_alias`` refuses a # userId held by a LIVE person as its own primary key (the # guard that stops an alias hiding a real employee), and a # soft-deleted row does not trip it. ghost.deleted_at = now ghost.hris_record_state = "purged" ghost.hris_purged_at = now # VARCHAR(64) — keep the two ids, put the narrative in the audit # row (the 08-18 script lost a whole run to a 1406 here). ghost.hris_purged_reason = ( f"兼岗 userId {ghost_uid} → canonical {canonical.employee_no}" )[:64] await session.flush() record = await record_hris_user_alias( session, person=canonical, alias_user_id=ghost_uid, reason="concurrent_employment", person_id_external=pid, hris_instance_id=ghost.hris_instance_id, note="兼岗 row retired; confirmed against live SF EmpEmployment", ) session.add(AuditLog( trace_id=str(uuid4()), actor_user_id=None, actor_type="system", action="hr.sync.retire_concurrent_employment_person", resource_type="identity_person", resource_id=str(canonical.id), status="success", detail=json.dumps({ "alias_user_id": ghost_uid, "person_id_external": pid, "canonical_employee_no": canonical.employee_no, "retired_ghost_person_id": str(ghost.id), "retired_ghost_work_email": ghost.work_email, "alias_recorded": bool(record), "google_deleted": [ e for e in deleted_google if e == (ghost.work_email or "").lower() ], "script": "fix_kengang_ghost_accounts_2026_08_25.py", }, ensure_ascii=False), ip_address=None, user_agent="cleanup-script", )) print(f" retired {ghost_uid} → aliased to " f"{canonical.hris_person_id}") await session.commit() print(f"\nDone. Retired {len(planned)} ghost row(s); " f"deleted {len(deleted_google)} Google account(s).") return 0 def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--commit", action="store_true", help="write changes (default: dry run)") ap.add_argument("--delete-google", action="store_true", help="also hard-delete the ghost's Google Workspace " "account (irreversible beyond Google's ~20-day " "undelete window)") args = ap.parse_args() return asyncio.run(run( dry_run=not args.commit, delete_google=args.delete_google, )) if __name__ == "__main__": raise SystemExit(main())