"""Scheduled-job handlers for Google Workspace sync tasks. * ``google.group_sync`` — periodic mirror of every Google mail group into ``email_groups``, with owners resolved back to :class:`IdentityPerson`. Google-side ownership is authoritative; Helix just mirrors. * ``google.seat_sync`` — Google user list mirror + orphan-account discovery + drift cleanup. Also reconciles the google login LINK for matched active staff (Phase A, 2026-06-26) on this same daily tenant sweep. It does NOT auto-suspend leaver accounts — retention is a human call via the orphan inventory. The legacy ``google.dept_sync`` handler used to live here. It was migrated to the Identity Sync Center in PR for #121 — the work is now done by an ``identity_sync_targets`` row of kind ``google_schema``, which uses the same Google driver but is configured (filter, mapping, schedule, on_offboard) through the Sync Center admin UI instead of ``platform_settings.google.dept_schema``. The two pure helpers from the old handler — :func:`_derive_dept_hierarchy` and :func:`_load_dept_schema_settings` — stay below for backwards compatibility with tests and as a hand-debugging aid for ops. The new code path computes the same results in ``identity_sync.mapping``. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.locale import Lang, resolve_lang from app.core.txn import best_effort from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, import_dedup_filter, raise_if_total_failure, record_seat_reconciliation, safe_to_drift_revoke, write_drift_event_audit, write_sync_audit, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # Per-user concurrency cap retained for the legacy helpers below — the # scheduled-job version of dept sync is gone, but ``_load_dept_schema_settings`` # may still be called by ad-hoc scripts that need to know the # previous behavior. The Identity Sync Center runner sets its own # concurrency directly. _DEPT_SYNC_CONCURRENCY = 20 # ``google.seat_sync.orphan_stale_days`` platform setting default: # how many days an orphan can go without being observed in the # Google tenant list before the seat-sync stamps ``deleted_at`` on # the row (filtering it out of admin list views by default). # Default 3 — short enough that admins don't see deleted-upstream # rows for more than ~72h, long enough to absorb weekend outages or # a single failed sync run. Set to 0 to disable the soft-delete # sweep entirely (escape hatch — rows never auto-mark deleted). _DEFAULT_ORPHAN_STALE_DAYS = 3 # ─── Legacy helpers (kept for tests + ops debugging) ─────────────── async def _load_google_driver(db: AsyncSession): """Resolve the primary Google Workspace integration driver. Mirrors EmailGroupService._get_driver so callers here don't have to import the email_group domain (avoids cross-domain import). Returns None (and logs) if no integration is configured — the job summary will reflect this without the whole task failing. """ from app.integrations.drivers.google_workspace import GoogleWorkspaceDriver from app.integrations.models import Integration from app.integrations.service import _resolve row = ( await db.execute( select(Integration) .where( Integration.kind == "software_provisioning", Integration.provider == "google_workspace", Integration.enabled.is_(True), ) .order_by(Integration.is_primary.desc(), Integration.created_at) ) ).scalars().first() if row is None: logger.warning("google_workspace integration not configured; skipping") return None resolved = await _resolve(db, row) return GoogleWorkspaceDriver( integration_id=str(row.id), config=resolved.config, ) async def _load_dept_schema_settings(db: AsyncSession) -> dict[str, Any]: """Read ``google.dept_schema`` from platform_settings. Default shape (matches the screenshot labels the user provided): { "schema_id": "Department_Information", "fields": { "l1": "L1_Dept", "l2": "L2_Dept", "l3": "L3_Dept", "l4": "L4_Dept", }, "skip_offboarded": True, } Ops can update the setting in admin UI without a redeploy if the real schema uses different keys. """ from app.integrations.service import get_platform_setting default = { "schema_id": "Department_Information", "fields": { "l1": "L1_Dept", "l2": "L2_Dept", "l3": "L3_Dept", "l4": "L4_Dept", }, "skip_offboarded": True, # Top-level segments to strip from ``org_path`` before mapping # to L1-L4. ``cust_L0`` from SuccessFactors is the legal entity # (e.g. "PingCAP"), which Google customers treat as the tenant # itself, not a department — including it pushes the real top # business unit down into L2. "drop_root_segments": ["PingCAP"], } raw = await get_platform_setting( "google.dept_schema", default=None, session=db, ) if not isinstance(raw, dict): return default # shallow merge so a partial override still picks up defaults merged = {**default, **{k: v for k, v in raw.items() if v is not None}} merged["fields"] = {**default["fields"], **(raw.get("fields") or {})} return merged def _derive_dept_hierarchy( org_path: str | None, level_max: int = 4, drop_root_segments: list[str] | None = None, ) -> list[str]: """Walk a `/`-delimited org_path and return the innermost ``level_max`` segments. ``drop_root_segments`` names top-level entries (case-insensitive) that should be skipped when they appear at position 0 — typically the legal entity from ``cust_L0`` (e.g. "PingCAP") which is the tenant itself, not a real department. After that strip: * 3 levels total → [L1, L2, L3] * 4 levels → [L1, L2, L3, L4] * 5+ levels → take the deepest 4 (skip remaining holding / legal entity segments — per user they're not useful) Empty / missing path returns an empty list. The caller fills L1/L2/... positions in order and sends None for any missing slot so Google clears the stale field. """ if not org_path: return [] parts = [p.strip() for p in org_path.split("/") if p.strip()] if not parts: return [] drop = {s.strip().lower() for s in (drop_root_segments or []) if s and s.strip()} while drop and parts and parts[0].lower() in drop: parts = parts[1:] if len(parts) <= level_max: return parts # If we still have >level_max after stripping the configured roots, # the top is likely another holding/parent segment; drop it. return parts[-level_max:] # ``google.dept_sync`` handler intentionally NOT registered here — # the work moved to the Identity Sync Center (``identity_sync_targets`` # row of kind=google_schema, slug=google-dept). Migration 121 deletes # the legacy ScheduledJob row from the DB, so re-registering a stub # handler "for safety" would be misleading: it would let an admin # recreate the row and have it appear to succeed (returning a fake- # success deprecated message) instead of failing loudly. # ─── Mail group sync ───────────────────────────────────────────── # A greenfield / post-restore tick where "every group is new" must not # write one audit row per group — that is a 760-row burst of pure noise # that would bury the human operations it sits next to in the 操作日志. # Past this many discoveries in a single tick we write ONE aggregate # row naming the count and a sample instead. _SYNC_DISCOVERY_BULK_THRESHOLD = 25 async def _log_group_sync_event( db: AsyncSession, *, action: str, group_id: uuid.UUID | None, group_email: str, detail: dict[str, Any], trace_id: str, errors: list[str] | None = None, ) -> None: """Write ONE ``email_group.sync_*`` audit row for a real change. Why the group sync audits at all: Google is the authoritative source for group membership / ownership / existence, so the most consequential things that happen to a mail group are not Helix mutations — a group gets deleted in the Google console, an owner is swapped, a group is renamed. Before this the 邮件组管理 operation log could only show Helix-originated writes, which made "nothing happened to this group" indistinguishable from "everything that happened to it happened somewhere else". **Only called on an actual diff.** A steady-state tick over ~760 unchanged groups writes zero rows, which is what keeps the log readable and the table from growing by 6,000 rows/day. Best-effort: an audit failure must never abort a reconcile that has already applied its Google-side truth locally. **The flush below is load-bearing, not tidiness.** ``best_effort`` opens a SAVEPOINT, and ``Session.begin_nested()`` does NOT flush first — while ``AuditService.log`` ends in a session-WIDE flush. So without this, the caller's still-pending ``UPDATE email_groups`` (the rename, the revival, the tombstone) would be emitted *inside* the savepoint, and a rolled-back audit write would silently revert the very change it was describing, while the run summary still counted it as applied. Flushing here means the savepoint contains only the audit INSERT. A swallowed failure is also appended to ``errors`` when the caller passes the list, so the tick cannot report a green ``groups_reconciled_ok`` for a group whose event was lost. """ from app.audit.service import AuditService # See the docstring: settle the caller's pending row mutations # OUTSIDE the savepoint that is about to be opened. await db.flush() payload: dict[str, Any] = {"source": "google.group_sync"} # Estate-wide events (the bulk-discovery aggregate) name no single # group. Omit the key rather than storing "" — the operation log # falls back to ``detail.group_email`` for rows whose group row is # gone, and an empty string there would render as a blank cell that # looks like a data bug instead of "not about one group". if group_email: payload["group_email"] = group_email payload.update(detail) async with best_effort(db, label=f"group_sync audit {action}") as step: await AuditService(db).log( trace_id=trace_id, actor_type="system", action=action, status="success", resource_type="email_group", resource_id=str(group_id) if group_id else None, detail=payload, ) if not step.ok: # ``step.ok`` is the only safe thing to read here — a rolled-back # savepoint expires the ORM objects touched inside it. msg = ( f"{group_email or '(estate-wide)'}: operation-log event " f"{action} could not be written" ) logger.warning("group_sync: %s (non-fatal)", msg) # Surface it in the run summary. Silently dropping it let a tick # report ``status='ok'`` while part of its operation log was # missing — and the operation log is the only place a # Google-side change is ever recorded. if errors is not None: errors.append(msg) @register_scheduled_job( key="google.group_sync", display_name="Google 邮件组同步", description=( "拉取 Google Workspace 租户的全部邮件组,镜像到 email_groups 表。" "Owner 以 Google 侧为准——解析为 Helix identity_persons 后覆盖" "email_group_owners。Google 上已删除但 Helix 还在的组软删为" "status='deleted'。每 6 小时一次,也可手动触发。" ), ) async def run_google_group_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Deadlock-victim / connection-drop restart on fresh sessions. The # 2026-06-17 failure was errno 2013 ("Lost connection to MySQL server # during query") on an autoflushed ``UPDATE email_groups`` mid-loop — # a TiDB Cloud node RST'd an in-flight query; ``pool_pre_ping`` only # probes at checkout so it can't catch a mid-query drop. The whole # reconcile re-lists from Google + re-scans the DB and upserts, so a # fresh-session re-run is idempotent (same incident class as # ``google.seat_sync`` / ``jira.seat_sync``). from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_google_group_sync_once(session, config), label="google.group_sync", ) async def _run_google_group_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Mirror Google Workspace groups → ``email_groups``; see the registered handler's docstring. Idempotent: caller re-runs the whole thing on a fresh session after a transient DB error. Owns its commit.""" from sqlalchemy import and_, delete from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.identity.person_models import IdentityPerson driver = await _load_google_driver(db) if driver is None: return { "groups_listed": 0, "created": 0, "updated": 0, "soft_deleted": 0, "new_owner_links": 0, "groups_reconciled_ok": 0, "errors": ["google_workspace integration not configured"], } # ``list_all_groups`` now raises :class:`GoogleApiError` when the # FIRST page fails (nothing collected) — previously the empty list # made "groups_listed: 0" look benign when the real cause was a # 403 on the service-account scope. Catch + surface so the ops page # reads "HTTP 403: Forbidden | (SA missing # admin.directory.group.readonly)" instead of silently listing zero. from app.integrations.drivers.google_workspace import GoogleApiError try: google_groups = await driver.list_all_groups() except GoogleApiError as e: # RAISE, don't return: a returned dict lands scheduled_job_runs # status='ok' and the failure-notifier never fires, so a total # group-sync outage (e.g. SA missing admin.directory.group.readonly) # would show perpetually green while zero groups sync. This also # bypassed the raise_if_total_failure gate at the bottom. The # driver-not-configured early-return above stays a clean no-op. raise RuntimeError( f"google.group_sync: list_all_groups failed: {e}" ) from e # Build lookup of existing Helix rows by email (case-insensitive — # Google treats "Ops@Pingcap.com" = "ops@pingcap.com"). existing_rows = (await db.execute(select(EmailGroup))).scalars().all() existing_by_email = {row.email.lower(): row for row in existing_rows} # Person lookup by email for owner resolution person_rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.work_email.isnot(None), IdentityPerson.deleted_at.is_(None), ) )).scalars().all() person_by_email = {p.work_email.lower(): p for p in person_rows if p.work_email} now = datetime.now(UTC) created = 0 updated = 0 owner_links = 0 # ``groups_reconciled_ok`` is a per-group "made it all the way through" # counter — increments only for groups whose property AND owner # reconcile both finished without an error/skip. It exists to keep the # ``raise_if_total_failure`` gate honest on a steady-state tenant: # PingCAP prod has 742 groups, and a stable hourly run has # created/updated/soft_deleted all zero. Without this counter, a # single ``owners fetch indeterminate`` skip (one group with a 403 on # the owners endpoint) would false-fire the gate as "total failure" # because zero mutations were recorded — even though 741 groups # reconciled cleanly. With the counter in ``progress_keys``, the # gate still fires correctly when every group fails (e.g. credentials # rotated and every owners-fetch returns None → reconciled_ok stays 0). groups_reconciled_ok = 0 errors: list[str] = [] # Cloud Identity dynamic-group detection (best-effort). Returns a set # of dynamic group emails, or None when the lookup couldn't run (SA # missing cloud-identity.groups.readonly scope, Cloud Identity not # enabled). ``None`` ⇒ indeterminate: we leave every row's existing # ``is_dynamic`` untouched rather than clearing it to non-dynamic, so # an unconfigured scope never flips a known-dynamic group back to # "self-service leave OK". # Cloud Identity dynamic-group detection now returns full detail # (email + membership query) so we can also mirror the query text and # its parsed structured rule into the row for the HR management UI + # drift detection. ``None`` ⇒ indeterminate (scope missing / API off). from app.domains.email_group import dynamic_cel as _cel dynamic_groups = await driver.list_dynamic_groups() if dynamic_groups is None: dynamic_emails = None dynamic_by_email: dict[str, dict] = {} errors.append( "dynamic-group detection unavailable (Cloud Identity " "groups.readonly scope likely missing) — is_dynamic left " "unchanged this run" ) else: dynamic_by_email = { g["email"]: g for g in dynamic_groups if g.get("email") } dynamic_emails = set(dynamic_by_email.keys()) seen_emails: set[str] = set() # ── Change-event audit bookkeeping (operation log) ── # # One trace_id for the whole tick so an admin can pull every event # this run produced with a single ``/audit/trace/{id}`` lookup, the # same convention ``write_sync_audit`` uses for the summary row. audit_trace_id = f"google.group_sync:{now.isoformat()}" # person_id → email, for rendering owner diffs as addresses rather # than UUIDs. Built from the same map the owner resolution uses, so # it cannot disagree with it. email_by_person_id = {p.id: e for e, p in person_by_email.items()} # Count discoveries UP FRONT so the very first group can already # know whether this tick is a normal trickle (per-group rows) or a # greenfield burst (one aggregate row). Deciding as we go would # emit 25 individual rows and then switch mode mid-run. discovered_emails = [ e for e in ( (g.get("email") or "").lower() for g in google_groups ) if e and e not in existing_by_email ] bulk_discovery = len(discovered_emails) > _SYNC_DISCOVERY_BULK_THRESHOLD # Same reasoning for REVIVALS. ``existing_by_email`` includes # tombstones, so a post-incident recovery tick (the 50% floor below # permits up to half the estate to have been soft-deleted) takes the # update branch for every one of them — which is the 760-row burst # the discovery threshold exists to prevent, arriving through the # other door. restored_emails = [ e for e in ( (g.get("email") or "").lower() for g in google_groups ) if e and (row := existing_by_email.get(e)) is not None and row.status == "deleted" ] bulk_restore = len(restored_emails) > _SYNC_DISCOVERY_BULK_THRESHOLD for g in google_groups: email = (g.get("email") or "").lower() if not email: continue # Snapshot of ``errors`` length at iteration start — used at the # bottom of the loop to decide whether this group counts toward # ``groups_reconciled_ok``. Catches the 50% owner-removal # defensive refusal at L380 (which appends to ``errors`` but # falls through instead of ``continue``-ing), so a 100% # paginator-truncation event would otherwise log N errors AND # increment reconciled_ok N times and silently bypass the gate. errors_before = len(errors) seen_emails.add(email) name = g.get("name") or email description = g.get("description") or None google_id = g.get("id") # Merge editable + non-editable aliases — both route to the # group from a user's perspective; the distinction only matters # when deleting aliases. Sort + dedupe so equality comparisons # below don't bounce on insertion-order noise. merged_aliases = sorted({ a.lower() for a in ( list(g.get("aliases") or []) + list(g.get("nonEditableAliases") or []) ) if isinstance(a, str) and a }) # Google returns directMembersCount as a string sometimes (it's # documented as a stringified int). Be permissive. raw_count = g.get("directMembersCount") try: members_count = int(raw_count) if raw_count is not None else None except (TypeError, ValueError): members_count = None # Dynamic flag from the Cloud Identity set. ``None`` (indeterminate # lookup) → leave is_dynamic unset/unchanged; a real set → True iff # this email is in it. is_dynamic_val = ( (email in dynamic_emails) if dynamic_emails is not None else None ) # Mirror the membership query + parsed structured rule when the # detection was determinate. For a non-dynamic group the query is # cleared; for a dynamic one we store both the raw CEL (source of # truth for un-parseable rules) and the parsed structured form. dyn_query: str | None = None dyn_rule: dict | None = None if dynamic_emails is not None: info = dynamic_by_email.get(email) if info is not None: dyn_query = info.get("query") or None dyn_rule = _cel.parse_membership_query(dyn_query).to_dict() row = existing_by_email.get(email) if row is None: row = EmailGroup( email=email, name=name, description=description, google_group_id=google_id, status="active", last_synced_at=now, aliases=merged_aliases, direct_members_count=members_count, is_dynamic=is_dynamic_val, membership_query=dyn_query, dynamic_rule=dyn_rule, ) db.add(row) # SAVEPOINT — if the 6-hour scheduler and a manual /sync-now race # on creating the same new group, the ``email`` unique key throws # IntegrityError here. Without the savepoint that poisoned the # shared session and the un-caught error propagated out of the # loop, discarding every group already reconciled this tick (the # single commit at the end never ran), and the lock-retry wrapper # does NOT retry IntegrityError. Roll back just this insert and # skip the group — the next tick sees the now-existing row and # updates it normally. Mirrors the owner-insert savepoint below. try: async with db.begin_nested(): await db.flush() # assign id for owner FK except IntegrityError: errors.append( f"group {email}: concurrent create race " "(skipped this tick, reconciles next run)" ) continue created += 1 if not bulk_discovery: await _log_group_sync_event( db, action="email_group.sync_discovered", group_id=row.id, group_email=email, detail={ "name": name, "direct_members_count": members_count, "is_dynamic": is_dynamic_val, }, trace_id=audit_trace_id, errors=errors, ) else: changed = False # Snapshot the fields whose changes are worth an operation-log # entry BEFORE mutating the row: a rename, a revival, and a # static↔dynamic flip each change what the group *is*, unlike # the member count (which moves on its own every day and # would drown the log). prev_name = row.name prev_description = row.description prev_status = row.status prev_is_dynamic = row.is_dynamic prev_aliases = ( sorted(row.aliases) if row.aliases is not None else None ) if row.name != name: row.name = name changed = True if row.description != description: row.description = description changed = True if row.google_group_id != google_id: row.google_group_id = google_id changed = True if row.status != "active": row.status = "active" changed = True # Distinguish "never synced" (``row.aliases is None``) from # "synced, has none" (``row.aliases == []``) explicitly — # ``None or []`` collapses both to ``[]`` and would leave # genuinely-unsynced rows stuck at ``None`` forever once a # group with no aliases passed through this loop. Sorting # the existing-side is for list-order noise from the JSON # column round-trip; ``merged_aliases`` is already sorted. existing_aliases = sorted(row.aliases) if row.aliases is not None else None if existing_aliases != merged_aliases: row.aliases = merged_aliases changed = True if row.direct_members_count != members_count: row.direct_members_count = members_count changed = True # Only update is_dynamic when the lookup was determinate # (set is not None) — otherwise leave the prior value intact. if is_dynamic_val is not None and row.is_dynamic != is_dynamic_val: row.is_dynamic = is_dynamic_val changed = True # Mirror query + parsed rule only on a determinate lookup. if dynamic_emails is not None: if row.membership_query != dyn_query: row.membership_query = dyn_query changed = True if row.dynamic_rule != dyn_rule: row.dynamic_rule = dyn_rule changed = True row.last_synced_at = now if changed: updated += 1 # ── Operation-log events for the identity-changing diffs ── if prev_status == "deleted" and not bulk_restore: await _log_group_sync_event( db, action="email_group.sync_restored", group_id=row.id, group_email=email, detail={ "note": ( "group re-appeared on Google; Helix tombstone " "revived" ), "name": name, }, trace_id=audit_trace_id, errors=errors, ) # NOTE — "renamed" here means the DISPLAY NAME or the # description changed, and the two are deliberately one # event: they are the same edit in the Google console and # splitting them would double the log for a single action. # A description-only edit therefore files as # ``sync_renamed``, so that filter slightly over-reports — # read ``detail.old_name`` / ``new_name`` to tell which. # # It does NOT mean the ADDRESS changed. Google treats a # primary-address change as a different group key, so it # surfaces as an unlinked ``sync_removed`` + # ``sync_discovered`` pair; nothing here bridges them. if prev_name != name or prev_description != description: await _log_group_sync_event( db, action="email_group.sync_renamed", group_id=row.id, group_email=email, detail={ "old_name": prev_name, "new_name": name, "old_description": prev_description, "new_description": description, }, trace_id=audit_trace_id, errors=errors, ) # An alias set that changed behind Helix's back is an # identity change, not member churn: a second live address # appeared on (or vanished from) this group. It is also # load-bearing for the cleanup workbench, where a real alias # is a BLOCKING signal — without this event the log can # answer "when did this group become undeletable?" only with # silence. The FIRST sync of a never-synced group # (``prev_aliases is None``) is a backfill, not a change, so # it is not an event. if prev_aliases is not None and prev_aliases != merged_aliases: await _log_group_sync_event( db, action="email_group.sync_alias_changed", group_id=row.id, group_email=email, detail={ "old_aliases": prev_aliases, "new_aliases": merged_aliases, "added": sorted( set(merged_aliases) - set(prev_aliases) ), "removed": sorted( set(prev_aliases) - set(merged_aliases) ), }, trace_id=audit_trace_id, errors=errors, ) # Only a DETERMINATE flip is an event. ``is_dynamic_val is # None`` means detection could not run this tick, and the # row was deliberately left untouched above — reporting that # as a change would invent a transition that never happened. if is_dynamic_val is not None and prev_is_dynamic != is_dynamic_val: await _log_group_sync_event( db, action="email_group.sync_dynamic_flag_changed", group_id=row.id, group_email=email, detail={ "old_is_dynamic": prev_is_dynamic, "new_is_dynamic": is_dynamic_val, "membership_query": dyn_query, }, trace_id=audit_trace_id, errors=errors, ) # Fetch Google-side owners → normalise → overwrite local join. # ``None`` signals indeterminate (403 / network error): skip the # owner reconciliation for this group rather than blanking it # — see `list_group_owners` docstring. google_owners = await driver.list_group_owners(email) if google_owners is None: errors.append(f"{email}: owners fetch indeterminate; keeping existing owner links") continue google_owner_emails = { (o.get("email") or "").lower() for o in google_owners if o.get("email") } resolved_person_ids = { person_by_email[e].id for e in google_owner_emails if e in person_by_email } # Replace pattern: delete rows not in the new set, insert rows # not already present. Keep ``created_at`` intact for unchanged # owners so audit timelines stay sensible. current_owner_rows = (await db.execute( select(EmailGroupOwner).where(EmailGroupOwner.email_group_id == row.id) )).scalars().all() current_person_ids = {r.person_id for r in current_owner_rows} to_remove = current_person_ids - resolved_person_ids to_add = resolved_person_ids - current_person_ids # Safety floor on the diff (P1 from LS2): without this, a # transient Google Admin SDK paginator that returns 1 of 5 # owners (HTTP 200 + truncated body) would mark 4 owners for # hard-delete in one tick — wiping the historical "who owned # this group" link that's the only handle for "why did Alice # have edit rights last March" forensic queries. The 50% # ratio matches what seat-sync handlers use; below it we # refuse the delete and surface as an error. if to_remove and current_person_ids: removal_ratio = len(to_remove) / len(current_person_ids) if removal_ratio > 0.5 and len(current_person_ids) >= 4: errors.append( f"group {row.email!r}: owner reconcile would remove " f"{len(to_remove)} of {len(current_person_ids)} " f"owners ({removal_ratio:.0%}) — refused as " "likely paginator truncation. Re-run after " "confirming Google returned the full owner list." ) to_remove = set() if to_remove: await db.execute( delete(EmailGroupOwner).where( and_( EmailGroupOwner.email_group_id == row.id, EmailGroupOwner.person_id.in_(to_remove), ) ) ) # Per-row savepoint on the insert — if the 6-hour scheduler and # a manual `/sync-now` race on the same (group_id, person_id), # the unique constraint throws IntegrityError. Swallow per-row # to keep the batch making progress rather than abort the # whole transaction on a duplicate someone else just wrote. for pid in to_add: try: async with db.begin_nested(): db.add(EmailGroupOwner(email_group_id=row.id, person_id=pid)) owner_links += 1 except IntegrityError: # concurrent insert won — the row now exists, our job is done pass # Ownership on Google is authoritative, so an owner swap done in # the Google console is invisible to Helix's own audit trail — # this is the event that makes "why did Alice have edit rights # in March" answerable from the operation log. Rendered as # addresses, not person UUIDs. Note ``to_remove`` may have been # emptied by the 50% truncation guard above, in which case # nothing was applied and nothing is logged. if to_add or to_remove: await _log_group_sync_event( db, action="email_group.sync_owner_changed", group_id=row.id, group_email=email, detail={ "added": sorted( email_by_person_id.get(pid, str(pid)) for pid in to_add ), "removed": sorted( email_by_person_id.get(pid, str(pid)) for pid in to_remove ), # Google's own answer, including owners that resolve # to nobody in Helix (externals, leavers whose person # row is gone) — those never become owner links, so # without this the log would claim a group has no # owner when Google says otherwise. "google_owner_emails": sorted(google_owner_emails), }, trace_id=audit_trace_id, errors=errors, ) # Reached the end of the iteration without ``continue`` AND # without appending to ``errors`` (the 50% owner-removal # defensive refusal falls through instead of continuing — so # the bare "did we reach this line" check isn't sufficient). if len(errors) == errors_before: groups_reconciled_ok += 1 # Commit THIS group before moving to the next one (#2129). # Production has ~742 groups and each iteration makes an # external ``list_group_owners`` call, so a single trailing # commit meant every row this loop touched — new # ``email_groups`` rows, and a delete+insert of each group's # ``email_group_owners`` — stayed locked across the whole # ~742-call fan-out. One Google API slowdown or a driver retry # stretches that to minutes (#1837's shape: locks held across # slow external I/O). The ``begin_nested`` savepoints above are # per-row error isolation; a savepoint releases nothing. # # Per-group commit is also what makes a mid-run crash useful: # the groups already reconciled stay reconciled instead of # rolling back to nothing. await db.commit() # Soft-delete groups that existed in Helix but no longer on Google. # Defense-in-depth: ``list_all_groups`` raises on any pagination # failure (round-17 hardening) so a truncated list shouldn't reach # us — but mirror the seat-sync handlers' 50% ratio floor anyway, # so a future change that re-introduces partial-fetch tolerance # cannot mass-soft-delete groups in a single tick. candidates = { email for email, row in existing_by_email.items() if email not in seen_emails and row.status != "deleted" } soft_deleted = 0 # Ratio computed against ACTIVE-only rows so the floor doesn't # silently weaken as historical ``status='deleted'`` rows # accumulate (over months/years the inflated denominator would # let a true mass-truncation slip through). Mirrors seat-sync # handlers' active-only denominator. (Round-17 audit follow-up.) active_total = sum( 1 for r in existing_by_email.values() if r.status != "deleted" ) if candidates and active_total: ratio = len(candidates) / active_total if ratio > 0.5 and active_total >= 4: errors.append( f"group sync: would soft-delete {len(candidates)} of " f"{active_total} active groups ({ratio:.0%}) — " "refused as likely Google API truncation. Re-run " "after confirming the full group list is returned." ) candidates = set() for email in candidates: row = existing_by_email[email] # CONFIRM the absence before tombstoning. ``list_all_groups`` # completed BEFORE the DB snapshot was taken (several paginated # pages plus a read), so a group created in that window is # absent from ``seen_emails`` while existing perfectly well on # Google — and this loop now also writes an audit row asserting # "absent from Google", so both the tombstone and the log entry # would be false. One ``groups.get`` settles it, and unlike a # timestamp comparison it does not depend on whole-second # ``created_at`` granularity or on a naive-vs-aware datetime. # Steady state has zero candidates, and the 50% floor above # already bounds how many can reach this loop. exists = await driver.group_exists(email=email) if exists is not False: errors.append( f"{email}: absent from the group listing but " + ( "still present on Google" if exists else "Google could not confirm its absence" ) + " — skipped soft-delete (reconciles next run)" ) continue row.status = "deleted" row.last_synced_at = now soft_deleted += 1 # The single most consequential thing that can happen to a mail # group — it stopped existing — and it happens entirely outside # Helix. Always per-group (never aggregated): the whole point is # that the 操作日志 can answer "when did this address die". await _log_group_sync_event( db, action="email_group.sync_removed", group_id=row.id, group_email=email, detail={ "note": "absent from Google; Helix row soft-deleted", "name": row.name, "last_known_members_count": row.direct_members_count, }, trace_id=audit_trace_id, errors=errors, ) if bulk_restore and restored_emails: await _log_group_sync_event( db, action="email_group.sync_restored_bulk", group_id=None, group_email="", detail={ "restored_count": len(restored_emails), "sample_emails": sorted(restored_emails)[:20], "note": ( "bulk revival — per-group rows suppressed above " f"{_SYNC_DISCOVERY_BULK_THRESHOLD} revived groups in " "one run" ), }, trace_id=audit_trace_id, errors=errors, ) # Greenfield / post-restore tick: one aggregate row instead of the # 760 individual ones the per-group branch skipped. if bulk_discovery and created: await _log_group_sync_event( db, action="email_group.sync_discovered_bulk", group_id=None, group_email="", detail={ "discovered_count": created, "sample_emails": sorted(discovered_emails)[:20], "note": ( "bulk discovery — per-group rows suppressed above " f"{_SYNC_DISCOVERY_BULK_THRESHOLD} new groups in one run" ), }, trace_id=audit_trace_id, errors=errors, ) summary = { "groups_listed": len(google_groups), "created": created, "updated": updated, "soft_deleted": soft_deleted, "new_owner_links": owner_links, "groups_reconciled_ok": groups_reconciled_ok, "errors": errors, } await write_sync_audit(db, job_key="google.group_sync", summary=summary) await db.commit() # Round-42 audit: ``groups_listed`` is a FETCH counter (size of # Google's reply, not work done by this handler). If every write # path failed AND the fetch succeeded, the handler would silently # green-light a totally-failed run. Drop fetch counters from the # progress set so a non-zero workload requires at least one actual # mutation to advance. # # Round-44 (2026-05-19): Round-42 conflated "work attempted" with # "work succeeded" — ``created`` and ``updated`` count property # mutations that happen BEFORE the owners-fetch, so a partial # greenfield with a credentials rotation in flight (e.g. 5 new # groups created + all 742 owners-fetches returning ``None``) # would silently green-light. The right invariant is "at least # one group reached the end of its iteration without an error" — # captured by the new ``groups_reconciled_ok`` per-group counter, # gated on ``errors_before == len(errors)`` so the 50% owner- # removal defensive refusal (which appends to ``errors`` but # falls through instead of ``continue``-ing) cannot inflate it. # ``soft_deleted`` stays in the set because it's OUT-OF-LOOP # work — a tick that does nothing but soft-delete N orphaned # groups is still legitimate progress. raise_if_total_failure( job_key="google.group_sync", summary=summary, progress_keys=("groups_reconciled_ok", "soft_deleted"), ) return summary # ─── Seat sync (PR 2026-05-02) ──────────────────────────────────── def _parse_google_timestamp(value: Any) -> datetime | None: """Google Admin SDK returns ISO 8601 strings with 'Z'. Parse to a timezone-aware datetime; return None on any oddity (we don't want one bad timestamp aborting a 5000-user reconciliation pass). """ if not value: return None if isinstance(value, datetime): return value if not isinstance(value, str): return None try: normalized = value.replace("Z", "+00:00") return datetime.fromisoformat(normalized) except Exception: # noqa: BLE001 return None # ── Leaver-data guard for the base-seat drift cleanup (2026-08-13) ── # # See the long comment at the drift branch itself. In short: the Google # Workspace seat is what the offboarding blocker cascade works THROUGH to # transfer Drive/Calendar, and this sweep used to remove it the morning # after the account went dark in the tenant — which is routinely BEFORE # Helix knows the person is leaving. async def _classify_drift_revoke_candidates( db: AsyncSession, user_ids: list[uuid.UUID], ) -> dict[str, set[uuid.UUID]]: """Split drift-revoke candidates into the two cases that need care. Returns ``{"protected": {...}, "active_employees": {...}}``: * ``protected`` — a live offboarding process still needs this seat for the data handover. Never drift-revoke these. * ``active_employees`` — Helix still lists the person as active. Their seat IS drift (the account really is dark) so the revoke proceeds, but the situation is an anomaly worth naming. A savepoint-local classification failure uses the pre-2026-08-13 default (revoke, warn nobody) rather than skipping cleanup. Transaction-aborting database errors propagate to the scheduled-job retry boundary. """ empty: dict[str, set[uuid.UUID]] = {"protected": set(), "active_employees": set()} if not user_ids: return empty async with best_effort( db, label="classify Google drift-revoke candidates", log=logger, ) as _outcome: from app.domains.offboarding.models import ( PROC_CANCELLED, PROC_COMPLETED, OffboardingProcess, ) from app.domains.offboarding.service import OffboardingService from app.identity.person_models import IdentityPerson persons = list((await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(user_ids)), )).scalars().all()) active_ids = { p.id for p in persons if p.deleted_at is None and (p.employment_status or "").lower() == "active" } procs = list((await db.execute( select(OffboardingProcess).where( OffboardingProcess.person_id.in_(user_ids), OffboardingProcess.status != PROC_CANCELLED, ), )).scalars().all()) svc = OffboardingService(db) protected: set[uuid.UUID] = set() for proc in procs: # Reuse the same predicate the handover reconciler uses, so # "the handover still matters" has exactly one definition: # open processes, and completed ones whose terminal vendor # deletes have not all landed. if proc.status != PROC_COMPLETED or svc._handover_still_matters(proc): protected.add(proc.person_id) return {"protected": protected, "active_employees": active_ids - protected} return empty async def _warn_dark_accounts_of_active_employees( db: AsyncSession, person_ids: set[uuid.UUID], ) -> None: """Tell IT once per person that an ACTIVE employee's Google account went dark in the tenant. This is the early-warning the alison.u@ incident had no channel for: her account was suspended by hand on 07-31 while HRIS still showed her employed, the seat was silently revoked that morning, and the consequence (a hard delete with no data handover) landed on 08-12. Ten days of silence. Deduped per person via ``platform_settings``-free bookkeeping: the audit row IS the dedupe key — one alert per person per 30 days, so a long-running mismatch does not become a daily nag. """ if not person_ids: return from app.notifications import deliver_enqueued # Enqueue inside the savepoint, push after it — ``deliver()`` commits, # and a commit inside ``begin_nested()`` closes the block's transaction # and strands the row in ``delivering`` → ``delivery_uncertain`` (never # sent, never auto-resent). Both alerts this job ever raised (2026-08-14 # and 08-18) were lost that way. See ``deliver_enqueued``. queued: list[tuple[Any, Any]] = [] async with best_effort( db, label="warn dark Google accounts of active employees", log=logger, ) as outcome: from app.audit.models import AuditLog from app.audit.service import AuditService from app.auth.models import AuthRole, AuthUserRole from app.identity.person_models import IdentityPerson from app.notifications import NotificationEnvelope, NotificationService cutoff = datetime.now(UTC) - timedelta(days=30) recent = set((await db.execute( select(AuditLog.resource_id).where( AuditLog.action == "google.seat_sync.active_employee_account_dark", AuditLog.created_at >= cutoff, ), )).scalars().all()) fresh = [pid for pid in sorted(person_ids, key=str) if str(pid) not in recent] if not fresh: return persons = list((await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(fresh)), )).scalars().all()) if not persons: return audit = AuditService(db) for person in persons: await audit.log( trace_id="system:cron:google.seat_sync", actor_type="system", actor_user_id=None, action="google.seat_sync.active_employee_account_dark", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={ "work_email": person.work_email, "employment_status": person.employment_status, "note": ( "Google account is not active in the tenant while " "Helix still lists the person as an active employee" ), }, ) admin_ids = list((await db.execute( select(AuthUserRole.user_id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthRole.name.in_(("global_admin", "it_admin"))) .distinct(), )).scalars().all()) admins = [] for aid in admin_ids: admin = await db.get(IdentityPerson, aid) if ( admin is not None and admin.deleted_at is None and (admin.employment_status or "").lower() == "active" and (admin.work_email or "").strip() ): admins.append(admin) if not admins: return # Language-neutral detail block (name + address only): the prose that # frames it lives in the ``google.active_employee_account_dark`` # template in all three languages, so the same block can be dropped # into a Japanese admin's card without leaving a Chinese fragment. listed = "\n".join( f" · {p.display_name or p.work_email} <{p.work_email}>" for p in persons[:20] ) if len(persons) > 20: listed += f"\n · ... (+{len(persons) - 20} more)" notify = NotificationService(db) for admin in admins: env = NotificationEnvelope( recipient=admin, title=f"[账号异常] {len(persons)} 位在职员工的 Google 账号已不活跃", body_text="(see template)", channels=["feishu", "email"], template_key="google.active_employee_account_dark", context={"count": len(persons), "person_list": listed}, url="/admin/offboarding", related_resource_type="identity_person", related_resource_id=persons[0].id, ) queued.append((await notify.enqueue(env), env)) if outcome.error is None: await deliver_enqueued( db, queued, trace_id="system:cron:google.seat_sync", log=logger, ) @register_scheduled_job( key="google.seat_sync", display_name="Google Workspace · 坐席同步 + 孤儿账号扫描", description=( "每天 03:30 UTC:(a) 拉 Google Workspace 全量用户,把 active 用户镜像" "到 Google Workspace 基础坐席;(b) 任何 Google-rooted SoftwareLicensePlan" "的 group_member binding,按 group 成员同步;(c) 不在 identity_persons" "里的 Google 账号入 google_orphan_accounts(共用账号 / 服务账号)。" "drift cleanup:Google 上已 suspend 但 Helix 还 active 的坐席自动 revoke。" ), ) async def run_google_seat_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Deadlock-victim restart on fresh sessions — same incident class # as jira.seat_sync 2026-06-12 (worker restart double-run racing # INSERTs). The reconcile is idempotent: import has the 25h-grace # dedup filter, revoke has safe_to_drift_revoke, orphan inventory # is an upsert. from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_google_seat_sync_once(session, config), label="google.seat_sync", ) async def _run_google_seat_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Three-phase tenant reconciliation; see docstring above. Returns a summary suitable for the scheduled_jobs admin page — counts per phase + any per-user errors. Caller commits. """ from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareExternalHolder, SoftwareLicensePlan, SoftwareSeatAllocation, SoftwareSeatSyncStat, ) from app.domains.google_workspace.orphan_service import OrphanAccountService from app.identity.email_lookup import build_person_by_email_index from app.identity.person_models import IdentityExternalIdentity from app.integrations.drivers.google_workspace import GoogleApiError from app.integrations.models import Integration summary: dict[str, Any] = { "tenant_users_listed": 0, "base_seats_imported": 0, "base_seats_revoked": 0, "group_plans_synced": 0, "group_seats_imported": 0, "group_seats_revoked": 0, "orphans_observed": 0, "orphans_new": 0, "orphans_soft_deleted": 0, # 2026-06-26 — google login-LINK reconcile folded into this same # daily tenant sweep (Phase A). Leaver-account disposition is NOT # automated here — see the note before Phase B. "identity_links_created": 0, # Leaver-data guard (2026-08-13). ``kept_for_offboarding`` = seats # NOT drift-revoked because a live offboarding process still needs # them to run the Drive/Calendar handover. # ``revoked_while_active`` = accounts that went dark in the tenant # while Helix still lists the person as an active employee — a # manual suspension outside Helix or an HRIS lag, and the earliest # warning that a leaver is about to be processed with no seat. "base_seats_kept_for_offboarding": 0, "base_seats_revoked_while_active": 0, "errors": [], } driver = await _load_google_driver(db) if driver is None: summary["errors"].append("google_workspace integration not configured") return summary try: all_users = await driver.list_all_users(include_suspended=True) except GoogleApiError as e: # Pre-launch hardening 2026-05-05: Phase A (base seats) needs # the full tenant roster, but Phase B (per-group seat sync) # is independently scoped — each group fetch resolves on its # own. A transient mid-stream 5xx on the tenant listing # previously aborted the whole sync; now we capture the # error in the summary, mark Phase A as skipped, and fall # through so per-group bindings still reconcile. Phase C # (orphan inventory) also needs the tenant list and is # skipped consistently with Phase A. summary["errors"].append( f"Phase A/C skipped: tenant list failed: {e}", ) all_users = [] summary["tenant_users_listed"] = len(all_users) now = datetime.now(UTC) # ── Person lookup (work_email + aliases) ──────────────────── # Leavers (``left`` / ``leaver_pending``) are excluded so Phase A # doesn't re-import a base seat for someone who left but whose # Google account hasn't been suspended yet. Their existing seats # naturally fall through to drift cleanup ("user_id not in # active_user_ids" → revoked). Phase C's orphan detection benefits # too: a leaver's still-existing Google account becomes an orphan # record so admin can clean it up explicitly. # ``leaver_pending`` within the notice period is NOT excluded (the # helper's ``leaver_notice_grace`` default): those people are still # working, and dropping them churned the seat row → duplicate # offboarding cancel events (2026-08-11). user_emails = [ (u.get("primaryEmail") or "").strip().lower() for u in all_users if u.get("primaryEmail") ] person_index = await build_person_by_email_index( db, user_emails, exclude_employment_statuses=("left", "leaver_pending"), ) # ── Phase A — base "Google Workspace" plan reconciliation ─── base_catalog = (await db.execute( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.name == "Google Workspace", ).limit(1), )).scalar_one_or_none() base_plan: SoftwareLicensePlan | None = None if base_catalog is not None: base_plan = (await db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.software_id == base_catalog.id, SoftwareLicensePlan.status == "active", ).order_by(SoftwareLicensePlan.created_at.asc()).limit(1), )).scalar_one_or_none() if base_plan is None: summary["errors"].append( "base 'Google Workspace' license plan missing — " "migration 103 likely not applied", ) elif not all_users: # Tenant list failed earlier — skip Phase A entirely so we # don't import nothing AND drift-revoke everyone. Surfaced via # the existing summary["errors"] message from the catch above. summary["errors"].append( "Phase A skipped: empty tenant list (see prior error)", ) else: # Active OR pending OR recently-revoked — see zoom_handlers # comment for the rationale (migration 119's partial unique # index for active+pending; round 12 hardening 2026-05-08 # added the recently-revoked arm so the import branch doesn't # race a recent admin-revoke / leaver_blocked flip). existing_all = list((await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == base_plan.id, import_dedup_filter(SoftwareSeatAllocation, now=now), ), )).scalars().all()) existing_by_user_id = {s.user_id: s for s in existing_all} existing_seats = [s for s in existing_all if s.status == "active"] # Person ids that already carry an ACTIVE google login link, so the # loop below only creates the link for the ones missing it. A # Feishu-login employee has a Google mailbox + (soon) a base seat # but no google IdentityExternalIdentity until they Google-SSO — # leaving the Identity Hub + Google-SSO match incomplete. We # reconcile the LINK here in the same sweep that imports the seat # (seat stays seat_sync's job; this only adds the identity row). linked_google_person_ids: set = set((await db.execute( select(IdentityExternalIdentity.person_id).where( IdentityExternalIdentity.provider == "google", IdentityExternalIdentity.identity_status == "active", IdentityExternalIdentity.deleted_at.is_(None), ), )).scalars().all()) # Persons whose Google account was minted but left SUSPENDED because # the welcome email failed (identity_status='pending_welcome' — see # google_workspace.account_provisioner). Their mailbox is suspended, # so the tenant sweep above skips it (``if u.get("suspended")``) and # their base seat would fall into drift cleanup and be revoked after # 2h — stripping the seat of an employee still awaiting re-provision. # Protect those seats: the account_provisioner re-provision path owns # promoting them back to active (or an admin resolves manually); until # then the seat must survive so it isn't lost the moment they heal. pending_welcome_person_ids: set = set((await db.execute( select(IdentityExternalIdentity.person_id).where( IdentityExternalIdentity.provider == "google", IdentityExternalIdentity.identity_status == "pending_welcome", IdentityExternalIdentity.deleted_at.is_(None), ), )).scalars().all()) active_user_ids: set = set() # Reconciliation: non-suspended tenant accounts with email are the # base-plan license holders; the leftover (no Helix person) is the # external set persisted for the drawer. base_holders_with_email = 0 external_observations: list[ExternalHolderObservation] = [] # Per-row begin_nested so a single bad insert (FK violation, # unique-index collision after the migration 119 partial index # lands, malformed timestamp) doesn't abort the whole Phase A # flush and corrupt the in-memory counters. Mirrors the # email_group sync pattern. (P1 from CR2.) for u in all_users: email = (u.get("primaryEmail") or "").strip().lower() if not email: continue if u.get("suspended"): # Suspended users do NOT occupy a base seat. continue base_holders_with_email += 1 person = person_index.get(email) if person is None: _name = u.get("name") or {} external_observations.append(ExternalHolderObservation( email=email, vendor_account_id=u.get("id"), account_status="active", display_name=(_name.get("fullName") or None), )) continue active_user_ids.add(person.id) # Reconcile the google login LINK (idempotent; seat handled below). if person.id not in linked_google_person_ids: try: async with db.begin_nested(): db.add(IdentityExternalIdentity( person_id=person.id, provider="google", provider_subject=u.get("id") or email, provider_email=email, provider_username=email, is_login_identity=True, identity_status="active", )) linked_google_person_ids.add(person.id) summary["identity_links_created"] += 1 except OperationalError: raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"identity link create failed for {email}: " f"{type(e).__name__}: {e}", ) if person.id in existing_by_user_id: continue try: async with db.begin_nested(): db.add(SoftwareSeatAllocation( plan_id=base_plan.id, user_id=person.id, status="active", allocated_at=now, reason=( f"auto-imported by google.seat_sync from Google " f"Workspace tenant ({email})" ), )) summary["base_seats_imported"] += 1 except OperationalError: # Deadlock (1213) killed the whole transaction — # propagate so the fresh-session lock-retry wrapper # restarts the run. raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"base seat insert failed for {email}: " f"{type(e).__name__}: {e}", ) # Drift cleanup — seats Helix thinks are active but Google # either deleted, suspended, or never had. Skip seats <2h old # to avoid racing the create flow. Safety guard via shared # helper enforces a percentage floor (50%) on matched-vs- # active so a transient paginator truncation can't mass-revoke. safe, reason = safe_to_drift_revoke( matched_count=len(active_user_ids), existing_count=len(existing_seats), label="Google Workspace base seats", want_count=len(active_user_ids), ) if not safe: summary["errors"].append(reason or "drift cleanup refused") else: # ── Leaver-data guard (2026-08-13) ─────────────────────── # This drift branch is what silently removed the evidence in # the alison.u@ incident: her Google account went inactive in # the tenant on 07-31 (suspended by hand, before HRIS said she # was leaving), this loop revoked her Helix seat that morning, # and when the offboarding process opened on 08-10 the scan # found no Google Workspace seat — so no blocker, so the # ``user_suspend`` revoke that transfers Drive/Calendar never # ran, and the account was hard-deleted with no handover. # # Two narrow cases now behave differently: # * an offboarding process is live and its handover still # matters → KEEP the seat, so the blocker cascade retains # ownership of the data transfer; # * Helix still considers the person an ACTIVE employee → # revoke as before (their seat genuinely is drift), but # this is an anomaly worth naming: an employee on the # books whose account went dark in the tenant is either a # manual suspension outside Helix or an HRIS lag, and it # is the earliest possible warning of exactly this # incident. Ten days early, in alison's case. leaver_guard = await _classify_drift_revoke_candidates( db, [s.user_id for s in existing_seats if s.user_id not in active_user_ids], ) for seat in existing_seats: if seat.user_id in active_user_ids: continue if seat.user_id in pending_welcome_person_ids: # Minted-but-suspended, welcome-email pending re-provision. # The account_provisioner self-heal owns this seat; do NOT # drift-revoke it out from under the pending employee. continue if seat.user_id in leaver_guard["protected"]: # An offboarding process owns this person's data # handover right now — do not take away the seat it # works through. summary["base_seats_kept_for_offboarding"] += 1 continue allocated = seat.allocated_at if allocated is not None: if allocated.tzinfo is None: allocated = allocated.replace(tzinfo=UTC) if (now - allocated).total_seconds() < 7200: continue # rollback expires modified instances; capture first seat_id, seat_user_id, seat_plan_id = ( seat.id, seat.user_id, seat.plan_id, ) try: # Per-seat SAVEPOINT — one bad flush must not poison # the session and take the rest of the run with it. async with db.begin_nested(): seat.status = "revoked" seat.revoked_at = now seat.reason = ( (seat.reason or "") + "\n[auto-revoked by google.seat_sync: account " "no longer active in Google Workspace tenant]" ).strip() await write_drift_event_audit( db, job_key="google.seat_sync", seat_id=seat_id, user_id=seat_user_id, plan_id=seat_plan_id, event="revoked", detail={ "reason": ( "account no longer active in Google " "Workspace tenant" ), }, ) await db.flush() summary["base_seats_revoked"] += 1 if seat_user_id in leaver_guard["active_employees"]: summary["base_seats_revoked_while_active"] += 1 except OperationalError: # Whole-transaction rollback (deadlock) — propagate # to the fresh-session lock-retry wrapper. raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"base seat revoke failed for seat {seat_id}: " f"{type(e).__name__}: {e}", ) await _warn_dark_accounts_of_active_employees( db, leaver_guard["active_employees"], ) await db.flush() await record_seat_reconciliation( db, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=uuid.UUID(driver.integration_id), software_id=base_plan.software_id, plan_id=base_plan.id, provider="google", product_key="google-workspace", site=None, vendor_license_holders=base_holders_with_email, holders_matched=base_holders_with_email - len(external_observations), holders_no_email=0, externals=external_observations, now=now, ) # NOTE: Helix does NOT auto-suspend a leaver's still-live Google # account here. A ``left`` person's account legitimately stays open # when it's intentionally retained — the orphan inventory surfaces it # as a ``former_employee`` row and the admin marks it kept (e.g. # ``purpose='CBG保留 …使用'``). Auto-suspending would override exactly # those retain decisions ("这些是不能关闭的"). Leaver-account disposition # stays a HUMAN call through the orphan feature (which only tombstones # a remnant AFTER the admin deletes it in Google); the offboarding flow # owns the real-time suspend-on-leave for accounts that should close. # ── Phase B — per-plan group_member reconciliation ───────── # Mirrors the Jira pattern: every Google-rooted SoftwareLicensePlan # whose connector binding selects ``operation=group_member`` gets # its group enumerated and reconciled. integration_row = (await db.execute( select(Integration).where( Integration.kind == "software_provisioning", Integration.provider == "google_workspace", Integration.enabled.is_(True), ).order_by(Integration.is_primary.desc(), Integration.created_at), )).scalars().first() if integration_row is not None: from app.domains.provisioning.models import SoftwareConnector connector_rows = list((await db.execute( select(SoftwareConnector).where( SoftwareConnector.integration_id == integration_row.id, SoftwareConnector.is_active.is_(True), ), )).scalars().all()) for connector in connector_rows: cfg = connector.binding_config or {} if cfg.get("operation") != "group_member": continue group_email = (cfg.get("group_email") or "").strip() if not group_email: summary["errors"].append( f"connector {connector.id}: group_member binding " "missing group_email; skipped", ) continue plan = (await db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.software_id == connector.software_id, SoftwareLicensePlan.status == "active", ).order_by(SoftwareLicensePlan.created_at.asc()).limit(1), )).scalar_one_or_none() if plan is None: summary["errors"].append( f"software {connector.software_id}: no active license plan; " "group sync skipped", ) continue try: members = await driver.list_group_members(group_email) except GoogleApiError as e: summary["errors"].append( f"group {group_email}: {e}", ) continue summary["group_plans_synced"] += 1 member_emails = { (m.get("email") or "").strip().lower() for m in members if m.get("email") } # Leavers excluded — see jira/zoom/feishu/slack for the # same rationale (notice-period ``leaver_pending`` stays in # scope). Their existing seats fall through to drift # cleanup via "user_id not in active_user_ids". group_index = await build_person_by_email_index( db, member_emails, exclude_employment_statuses=("left", "leaver_pending"), ) existing_all = list((await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, import_dedup_filter(SoftwareSeatAllocation, now=now), ), )).scalars().all()) existing_by_user_id = {s.user_id: s for s in existing_all} existing_seats = [s for s in existing_all if s.status == "active"] active_user_ids = set() for email in member_emails: person = group_index.get(email) if person is None: continue active_user_ids.add(person.id) if person.id in existing_by_user_id: continue try: async with db.begin_nested(): db.add(SoftwareSeatAllocation( plan_id=plan.id, user_id=person.id, status="active", allocated_at=now, reason=( f"auto-imported from Google Group {group_email} " "via google.seat_sync" ), )) summary["group_seats_imported"] += 1 except OperationalError: # Whole-transaction rollback (deadlock) — propagate # to the fresh-session lock-retry wrapper. raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"group seat insert failed for {email} " f"({group_email}): {type(e).__name__}: {e}", ) # Safety guard via shared helper — see slack/jira/zoom for # the same percentage-floor protection. safe, reason = safe_to_drift_revoke( matched_count=len(active_user_ids), existing_count=len(existing_seats), label=f"Google Group {group_email!r}", want_count=len(active_user_ids), ) if not safe: summary["errors"].append(reason or "drift cleanup refused") else: for seat in existing_seats: if seat.user_id in active_user_ids: continue allocated = seat.allocated_at if allocated is not None: if allocated.tzinfo is None: allocated = allocated.replace(tzinfo=UTC) if (now - allocated).total_seconds() < 7200: continue # rollback expires modified instances; capture first seat_id, seat_user_id, seat_plan_id = ( seat.id, seat.user_id, seat.plan_id, ) try: # Per-seat SAVEPOINT — one bad flush must not # poison the session and kill the rest of the run. async with db.begin_nested(): seat.status = "revoked" seat.revoked_at = now seat.reason = ( (seat.reason or "") + f"\n[auto-revoked by google.seat_sync: " f"user no longer in group {group_email}]" ).strip() await write_drift_event_audit( db, job_key="google.seat_sync", seat_id=seat_id, user_id=seat_user_id, plan_id=seat_plan_id, event="revoked", detail={ "reason": ( f"user no longer in Google Group " f"{group_email}" ), }, ) await db.flush() summary["group_seats_revoked"] += 1 except OperationalError: # Whole-transaction rollback (deadlock) — # propagate to the lock-retry wrapper. raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"group seat revoke failed for seat {seat_id} " f"({group_email}): {type(e).__name__}: {e}", ) await db.flush() # ── Phase C — orphan-account inventory ───────────────────── # Every Google account NOT matched to an IdentityPerson lands in # google_orphan_accounts. Owners are NOT inferred — admin assigns # via the UI. The leaver scan picks orphans up automatically once # an owner is set. Skipped when ``all_users`` is empty (tenant # listing failed earlier) so we don't blank ``last_seen_at`` on # legitimate orphans during a transient outage. if not all_users: summary["errors"].append( "Phase C skipped: empty tenant list (see prior error)", ) else: orphan_svc = OrphanAccountService(db) # Per-row try so a single malformed Google row (e.g. missing # primaryEmail mid-pagination, FK violation) doesn't abort the # whole Phase C orphan import. (P1 from CR2.) for u in all_users: email = (u.get("primaryEmail") or "").strip().lower() if not email: continue if person_index.get(email) is not None: # This account belongs to a known Helix person — not an # orphan, skip. continue name_obj = u.get("name") or {} display = name_obj.get("fullName") or email try: async with db.begin_nested(): _row, created = await orphan_svc.upsert_from_google( primary_email=email, google_user_id=u.get("id"), display_name=display, org_unit_path=u.get("orgUnitPath"), suspended=bool(u.get("suspended")), is_admin=bool(u.get("isAdmin")), creation_time=_parse_google_timestamp(u.get("creationTime")), last_login_time=_parse_google_timestamp(u.get("lastLoginTime")), observed_at=now, ) summary["orphans_observed"] += 1 if created: summary["orphans_new"] += 1 except OperationalError: # Whole-transaction rollback (deadlock) — propagate # to the fresh-session lock-retry wrapper. raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"orphan upsert failed for {email}: " f"{type(e).__name__}: {e}", ) # ── Phase D — soft-delete stale orphans ──────────────────── # Phase C only stamps ``last_seen_at`` on rows we just observed. # Anything whose ``last_seen_at`` is older than the grace window # has been deleted (or otherwise removed) from Google upstream. # We mark ``deleted_at = now`` so list endpoints hide them by # default, but never hard-delete — owner + purpose + audit trail # are preserved for explicit ``include_deleted=true`` review. # # Skipped when ``all_users`` is empty (a tenant-wide outage upstream) # so a transient API failure doesn't synthetically wipe every # orphan in one tick. Same defensive check the orphan-import phase # uses above. grace_days = _DEFAULT_ORPHAN_STALE_DAYS if all_users: from app.integrations.service import get_platform_setting raw_grace = await get_platform_setting( "google.seat_sync.orphan_stale_days", default=_DEFAULT_ORPHAN_STALE_DAYS, session=db, ) try: grace_days = max(int(raw_grace), 0) except (TypeError, ValueError): # Surface the bad config — silent fallback would let an # admin's typo (e.g. ``three`` instead of ``3``) sit # undetected for weeks. We still apply the default rather # than failing the sync because the soft-delete sweep is a # defensive secondary feature: skipping it doesn't break # base-seat / orphan-import correctness. logger.warning( "google.seat_sync: platform setting " "'google.seat_sync.orphan_stale_days' has invalid " "value %r; falling back to default %d. Fix at " "/admin/settings.", raw_grace, _DEFAULT_ORPHAN_STALE_DAYS, ) summary["errors"].append( f"orphan_stale_days bad value {raw_grace!r}; " f"used default {_DEFAULT_ORPHAN_STALE_DAYS}", ) grace_days = _DEFAULT_ORPHAN_STALE_DAYS soft_deleted_rows = await orphan_svc.soft_delete_stale( now=now, grace_days=grace_days, ) summary["orphans_soft_deleted"] = len(soft_deleted_rows) if soft_deleted_rows: logger.info( "google.seat_sync: soft-deleted %d stale orphan(s) " "(last_seen_at < now - %d days): %s", len(soft_deleted_rows), grace_days, ", ".join(r.primary_email for r in soft_deleted_rows[:20]) + (" …" if len(soft_deleted_rows) > 20 else ""), ) summary["orphan_stale_grace_days"] = grace_days await write_sync_audit(db, job_key="google.seat_sync", summary=summary) await db.commit() # Round-42 audit: ``tenant_users_listed`` is a FETCH counter — the # size of Google Admin SDK's listing reply, not work this handler # accomplished. If every per-plan reconcile failed AND the tenant # listing succeeded, the prior progress set still hit non-zero and # the failure-notifier stayed silent. Keep only true mutation # counters in the progress set. raise_if_total_failure( job_key="google.seat_sync", summary=summary, progress_keys=( "base_seats_imported", "base_seats_revoked", "group_seats_imported", "group_seats_revoked", ), ) return summary # Per-language labels for the drift digest's detail lines. The digest's # frame lives in the ``dynamic_email_group_drift`` notification template # (``app/notifications/templates/email_group.py``); only these two labels # are produced here, because the values they introduce are per-group data # the template has no way to iterate over. _DRIFT_LABELS: dict[Lang, tuple[str, str]] = { "zh": ("失效值:", "未覆盖新值:"), "en": ("Dead values: ", "Uncovered new values: "), "ja": ("無効な値:", "未カバーの新しい値:"), } # Value separator inside one label's list — full-width for CJK, comma for # English, matching each language's list punctuation. _DRIFT_JOINERS: dict[Lang, tuple[str, str]] = { "zh": ("、", ";"), "en": (", ", "; "), "ja": ("、", ";"), } def _format_drift_lines(entries: list[dict[str, Any]], lang: Lang) -> str: """Render the drift digest's per-group bullet lines in ``lang``. One line per drifting group:: • grp@corp (Group name) — Dead values: eng, sre; Uncovered new values: ml Called once per recipient (not once per run) so each IT admin reads the substance in their own language. """ dead_label, missing_label = _DRIFT_LABELS[lang] value_join, part_join = _DRIFT_JOINERS[lang] lines: list[str] = [] for e in entries: parts: list[str] = [] if e["dead"]: parts.append(dead_label + value_join.join(e["dead"])) if e["missing"]: parts.append(missing_label + value_join.join(e["missing"])) lines.append(f"• {e['email']}({e['name']}) — " + part_join.join(parts)) return "\n".join(lines) @register_scheduled_job( key="google.dynamic_group_drift", display_name="Google 动态邮件组 · 规则漂移检测", description=( "每天检查每个动态邮件组的成员规则:引用了已 0 人的部门/属性值" "(『死值』——HR 改了部门名却没更新组规则,会导致成员静默清空)" "→ 告警 IT/HR;仅当组开启『缺失值提醒』时,才额外检查是否有新出现" "的属性值未被规则覆盖(默认关,避免对故意排除的值误报)。去重:每个" "组仅在漂移集合变化时才再次推送,长期未修的漂移不会每天打扰。" "只推送给 IT/HR,绝不打扰普通员工/经理。" ), ) async def run_dynamic_group_drift_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Detect membership-rule drift on dynamic email groups. Dead values (rule references an attribute value matched by zero active users) are always flagged — this is the g-all/EMEA incident: HR renames a department in SuccessFactors, Helix pushes the new value, but the group query still names the old one, silently emptying the group. Missing values (a live attribute value the rule doesn't cover) are only flagged for groups that opted in, because most groups intentionally exclude some values. """ from app.domains.email_group import dynamic_cel as _cel from app.domains.email_group.dynamic_service import compute_drift_signature from app.domains.email_group.models import EmailGroup from app.notifications.service import ( NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import list_admin_recipients summary: dict[str, Any] = { "groups_checked": 0, "groups_with_dead_values": 0, "groups_with_missing_values": 0, "digest_notified": 0, } driver = await _load_google_driver(db) if driver is None: summary["skipped"] = "google_workspace integration not configured" return summary users = await driver.scan_user_attributes() if users is None: # Real read failure (not "no dynamic groups") — raise so the run # is marked failed and the failure-notifier fires, rather than # silently reporting zero drift. raise RuntimeError( "scan_user_attributes returned None — cannot evaluate drift " "(Google Directory read failed)" ) value_counts: dict[str, dict[str, int]] = {} for u in users: for fk, fv in (u.get("fields") or {}).items(): value_counts.setdefault(fk, {}) value_counts[fk][fv] = value_counts[fk].get(fv, 0) + 1 rows = ( await db.execute( select(EmailGroup).where( EmailGroup.is_dynamic.is_(True), EmailGroup.status == "active", ) ) ).scalars().all() # Groups whose drift set changed this run → trigger one digest. changed_groups: list[dict[str, Any]] = [] current_drift: list[dict[str, Any]] = [] for row in rows: summary["groups_checked"] += 1 rule = _cel.parse_membership_query(row.membership_query) if row.dynamic_rule and isinstance(row.dynamic_rule, dict): # Prefer the stored parsed rule (keeps parseable flag stable). rule = _cel.StructuredRule( field=row.dynamic_rule.get("field"), match_values=list(row.dynamic_rule.get("match_values") or []), include_emails=list(row.dynamic_rule.get("include_emails") or []), negate=bool(row.dynamic_rule.get("negate")), parseable=bool(row.dynamic_rule.get("parseable", True)), raw_query=row.dynamic_rule.get("raw_query") or (row.membership_query or ""), ) dead = _cel.dead_values(rule, value_counts) missing: list[str] = [] if row.alert_on_missing_values and rule.parseable and rule.field and not rule.negate: live_vals = set(value_counts.get(rule.field, {}).keys()) covered = set(rule.match_values) missing = sorted(live_vals - covered) # Persist the freshly-computed dead set for the UI badge. row.dynamic_dead_values = dead if dead: summary["groups_with_dead_values"] += 1 if missing: summary["groups_with_missing_values"] += 1 sig = compute_drift_signature(dead, missing) has_drift = bool(dead or missing) if not has_drift: # Drift cleared → reset dedup so a future recurrence re-alerts. if row.drift_signature is not None: row.drift_signature = None continue entry = { "email": row.email, "name": row.name, "dead": dead, "missing": missing, } current_drift.append(entry) if row.drift_signature != sig: changed_groups.append(entry) row.drift_signature = sig # Only fan out a digest when at least one group's drift set is new or # changed since last run — standing unfixed drift doesn't ping daily. if changed_groups: notify = NotificationService(db) admins = await list_admin_recipients(db) for admin in admins: # The per-group detail lines carry their own labels ("失效值" / # "未覆盖新值"), so they are composed PER RECIPIENT in that # recipient's language rather than once for the whole fan-out. # Composing them once would have pinned the digest's payload to # Chinese no matter how well the surrounding template is # translated — the frame would read English and the substance # Chinese. drift_lines = _format_drift_lines(current_drift, resolve_lang(admin)) env = NotificationEnvelope( recipient=admin, title=f"动态邮件组规则漂移:{len(current_drift)} 个组需核对", body_text=( "以下动态邮件组的成员规则出现漂移,可能导致成员静默缺失," "请在 Helix「动态邮件组」管理页核对:\n\n" + _format_drift_lines(current_drift, "zh") ), template_key="dynamic_email_group_drift", context={ "person_name": admin.display_name or admin.work_email or "", "drift_count": str(len(current_drift)), "drift_lines": drift_lines, }, ) try: await notify.dispatch(env, trace_id=str(uuid.uuid4())) summary["digest_notified"] += 1 except Exception: logger.exception( "dynamic-group drift digest dispatch failed for %s", admin.work_email, ) await db.commit() return summary # ───────────────────────────────────────────────────────────────── # Leaver account end-state audit (Google half of 账号终态保障) # ───────────────────────────────────────────────────────────────── # Leavers whose Google account is still loginable this many days after # termination are surfaced. Offboarding suspends at T+0 and the terminal # delete fires at ``delete_lead_days`` (default 2), so anything past a # few days is a real gap rather than a race. _GOOGLE_LEAVER_GRACE_DAYS = 3 # Bound on the in-memory join, same rationale as the Jira audit: the # vendor side is one paginated pull regardless. _GOOGLE_LEAVER_AUDIT_MAX = 5000 def _config_int(config: dict, key: str, default: int) -> int: """``config[key]`` as an int, honouring an explicitly configured ``0``. ``config.get(key) or default`` — the obvious spelling, and the one this file used first — silently throws away a deliberate ``0``, because ``0 or 3`` is ``3``. For a grace window that is exactly backwards: ``0`` is the value an operator picks when they want the check to have no grace at all, and it is the one setting guaranteed to be ignored. Caught by the grace-window test, kept as a helper so the next threshold added here cannot reintroduce it. """ raw = config.get(key) if raw is None or (isinstance(raw, str) and not raw.strip()): return default try: return int(raw) except (TypeError, ValueError): return default @register_scheduled_job( key="google.leaver_account_audit", display_name="Google · 离职账号终态对账", description=( "拉 Google Workspace 全量名册,与 Helix 标记为 left 的人对账,回答" "「这个离职的人还能不能登录公司 Google 账号」,结果写进『账号终态』。" "\n判据只取厂商此刻的真话(账号在不在、是不是 suspended),不取 Helix " "以为自己做过什么 —— 离职台账里写着 suspended 而账号其实活着,正是" "这个对账要抓的东西。\n" "已知会命中的三类分别显示、不混成一个数:①离职流程从没跑过(早于" "该流程上线);②跑了但 suspend 没确认(终态删除闸因此一直不放行);" "③被 external_account_registry 的『保留』分类豁免掉 —— 而那条豁免" "可能既没有负责人也没有理由,那种豁免和「被忘了」在效果上没有区别。" ), ) async def run_google_leaver_account_audit( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Refresh the Google half of the account-assurance table. Raises on a hard vendor read failure so the run lands FAILED rather than quietly recording "nobody is at risk" against an empty roster — the same fail-loud rule the Jira audit follows, and the reason a broken credential can't masquerade as a clean bill of health. """ from app.core.coordinators import account_lifecycle_coordinator from app.domains.offboarding import assurance_service as assurance from app.domains.offboarding.models import ( ASSURANCE_SYSTEM_GOOGLE, VERDICT_AT_RISK, VERDICT_EXEMPT, OffboardingProcess, ) from app.identity.person_models import IdentityPerson from app.integrations.drivers.google_workspace import ( GoogleApiError, GoogleWorkspaceDriver, ) from app.integrations.service import _resolve grace_days = _config_int( config, "grace_days", _GOOGLE_LEAVER_GRACE_DAYS, ) # Reuse the platform's own resolver rather than re-deriving the # predicate. The first version of this handler invented # ``kind == "identity_provider"`` and shipped: prod's Google # Workspace row is ``kind='software_provisioning'``, so the sweep # reported "integration not configured" and the Google half of the # assurance table stayed permanently empty — a governance page # silently missing one of its two systems, which looks exactly like # "nothing is wrong there". Two places computing the same lookup # differently is the shape; one shared resolver is the fix. integ = await account_lifecycle_coordinator._primary_google_integration(db) if integ is None: return { "leavers_matched": 0, "at_risk": 0, "exempt": 0, "ok": 0, "errors": [], "skipped": ["Google Workspace integration not configured"], } resolved = await _resolve(db, integ) driver = GoogleWorkspaceDriver( integration_id=str(integ.id), config=resolved.config, ) try: tenant = await driver.list_all_users(include_suspended=True) except GoogleApiError as exc: raise RuntimeError( f"google.leaver_account_audit: tenant roster read failed — {exc}" ) from exc # Index by primary email AND aliases: an offboarded address commonly # survives as an alias on somebody else's account, and matching only # on primaryEmail would read that as "this leaver still has an # account" — a false positive that would put a colleague's mailbox in # the at-risk list. by_email: dict[str, dict[str, Any]] = {} for u in tenant: if not isinstance(u, dict): continue primary = str(u.get("primaryEmail") or "").strip().lower() if primary: by_email[primary] = u alias_owner: dict[str, str] = {} for u in tenant: if not isinstance(u, dict): continue primary = str(u.get("primaryEmail") or "").strip().lower() for al in (u.get("aliases") or []): alias_owner.setdefault(str(al).strip().lower(), primary) leavers = list((await db.execute( select(IdentityPerson) .where( IdentityPerson.employment_status == "left", IdentityPerson.work_email.is_not(None), IdentityPerson.deleted_at.is_(None), ) .order_by(IdentityPerson.termination_date.desc()) .limit(_GOOGLE_LEAVER_AUDIT_MAX) )).scalars().all()) emails = [(p.work_email or "").strip().lower() for p in leavers] exemptions = await assurance.registry_exemptions(db, emails) # Scoped to the leavers this run actually looked at. The unfiltered # ``select(OffboardingProcess)`` this replaced pulled every process # ever recorded into memory to read one JSON column off a fraction # of them — fine at today's size, and exactly the kind of full-table # load that stops being fine quietly. leaver_ids = [p.id for p in leavers] procs: dict[Any, Any] = {} if leaver_ids: procs = { proc.person_id: proc for proc in ( await db.execute( select(OffboardingProcess).where( OffboardingProcess.person_id.in_(leaver_ids), ) ) ).scalars().all() } today = datetime.now(UTC).date() rows: list[dict[str, Any]] = [] at_risk_samples: list[dict[str, Any]] = [] for person in leavers: email = (person.work_email or "").strip().lower() if not email: continue days = ( (today - person.termination_date).days if person.termination_date else None ) if days is not None and days < grace_days: continue # still inside the normal suspend→delete window gu = by_email.get(email) state, evidence = assurance.google_vendor_state(gu) if gu is None and email in alias_owner: evidence = { **evidence, "alias_of": alias_owner[email], "note": "address survives as an alias on another account", } exemption = exemptions.get(email) verdict, exempt_reason = assurance.verdict_for( state, exemption=exemption, ) proc = procs.get(person.id) detail = { **evidence, "days_since_termination": days, "work_email": email, "employee_no": person.employee_no, } if exemption: detail["exemption"] = exemption rows.append({ "person_id": person.id, "vendor_state": state, "vendor_account_id": (gu or {}).get("id"), "helix_action": assurance.helix_action_for_google( getattr(proc, "session_enforcement", None), ), "verdict": verdict, "exempt_reason": exempt_reason, "detail": detail, }) if verdict in (VERDICT_AT_RISK, VERDICT_EXEMPT): at_risk_samples.append({ "email": email, "name": person.display_name, "verdict": verdict, "days_since_termination": days, "last_login_at": evidence.get("last_login_at"), "exempt_quality": (exemption or {}).get("quality"), }) counts = await assurance.upsert_rows( db, system=ASSURANCE_SYSTEM_GOOGLE, rows=rows, ) summary = { "leavers_matched": len(rows), "tenant_users_listed": len(tenant), "at_risk": counts.get(VERDICT_AT_RISK, 0), "exempt": counts.get(VERDICT_EXEMPT, 0), "ok": counts.get("ok", 0), "unknown": counts.get("unknown", 0), "rows_removed": counts.get("removed", 0), "samples": sorted( at_risk_samples, key=lambda r: r.get("days_since_termination") or 0, reverse=True, )[:30], "errors": [], } await write_sync_audit( db, job_key="google.leaver_account_audit", summary=summary, ) if summary["at_risk"]: logger.warning( "google.leaver_account_audit: %d leaver(s) can still sign in to " "their company Google account: %s", summary["at_risk"], ", ".join( s["email"] for s in at_risk_samples if s["verdict"] == VERDICT_AT_RISK )[:400], ) await db.commit() return summary