"""Scheduled-job handler for Jira / Confluence seat sync. ``jira.seat_sync`` walks every Atlassian-rooted SoftwareLicensePlan once a day and reconciles ``software_seat_allocations`` against the **real product licenses** reported by the Atlassian Org Admin API (``product_access``) — NOT against the membership of some default group. Why product_access, not group membership (the bug this fixes) ───────────────────────────────────────────────────────────── Counting "members of jira-software-users / jira-servicedesk-users" was wrong in three different directions, cross-verified against Atlassian's own billing (``/rest/api/3/applicationrole`` userCount): * Jira Software — the default group (287 members) is clean but incomplete: ~81 license holders sit in OTHER conferring groups (jira-software-users-pingcap-ticket, jira-administrators, …) and were undercounted. * JSM — ``jira-servicedesk-users`` (140 members) is the WRONG group entirely: ~108 of them have no JSM license, ~40 real agents are in other groups, only ~32 overlap → both over- AND under-counted. * Confluence — never synced at all (0). A "seat" is therefore an **active Atlassian account that holds the product on the plan's paid-production site(s)**, taken straight from ``product_access`` (see ``app.integrations.atlassian_activity`` — ``AtlassianActivityDirectory.license_holders``). This is the single source of truth the offboarding ("who still has a license") and idle-reclaim ("who's a holder") features also build on. Per-plan scope comes from the connector binding's ``binding_config``: * ``product_key`` — jira-software / jira-service-desk / confluence (falls back to deriving it from ``group_name`` when absent). * ``site_urls`` — the paid-production host(s) this plan counts (e.g. ``["tidb.atlassian.net"]`` for Jira; the two Confluence subscriptions pin ``tidb`` vs ``pingcap-us`` separately). Falls back to the resolved production-site allowlist for the product. * ``group_name`` — only used by grant/revoke (offboarding / idle-reclaim), NOT by counting. Retained in the summary for ops. * ``seat_revocable`` — an idle-reclaim-only opt-out flag (a tier-billed add-on plan sets it ``false``). Counting NEVER reads it: a plugin plan that reuses ``jira-software-users`` purely as a billing proxy still counts its per-head seats by ``product_key``. So statistics口径 is independent of whether idle-reclaim is allowed to touch the plan. Reconcile, per plan: * **Seat import** — every active license holder whose email matches an ``identity_persons`` row (work_email or alias) gets an active SoftwareSeatAllocation if it doesn't already have one. Holders that don't match an employee (external / service accounts) are counted separately and never turned into employee seats. * **Drift cleanup** — Helix-side active seats whose person no longer holds the license flip to ``status=revoked`` (Helix bookkeeping only — the actual Atlassian removal is the offboarding / idle-reclaim line's job). Seats younger than 2h are skipped to avoid racing an in-flight grant, and a ratio safety guard refuses a mass-revoke when the directory looks suspiciously empty/truncated. Requires the Atlassian Org Admin API (``org_id`` + ``org_admin_api_key`` / Atlassian Guard) — a missing key raises, failing the run loudly rather than counting against an empty directory. Idempotent: re-running yields zero diffs in steady state. Per-plan failures are collected; one bad plan doesn't abort the rest. Benign config skips (unbound plan, unclassifiable product) land in ``skipped``, never ``errors``, so a steady-state run doesn't false-page the failure notifier. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import format_exception_for_log from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, SeatReconciliationIdentityError, 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__) # Cap on per-plan unmatched-holder emails carried in the run summary — # keeps the run-log row bounded while still making the bill-vs-seat gap # explainable (who the external / service accounts are). _UNMATCHED_SAMPLE_CAP = 50 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 def _resolve_plan_scope( binding: dict[str, Any], production_sites: dict[str, frozenset[str]], ) -> tuple[str | None, frozenset[str], str]: """Resolve (product_family, site_hosts, group_name) for a plan. Product-access seat-counting is **opt-in via the binding's explicit ``product_key``** — a plan with no ``product_key`` returns ``(None, …)`` and the caller skips it as a benign config gap. This is deliberate: the one Jira integration backs not just the three counted products but also Marketplace add-on catalogs (Tempo, ScriptRunner, eazyBI, …) whose connectors reuse ``group_name='jira-software-users'``. Deriving the family from ``group_name`` would mis-count every add-on as a duplicate Jira Software seat, so counting requires the operator (or migration 206) to declare the product explicitly. ``site_hosts`` is the binding's explicit ``site_urls`` (normalized) when present, else the resolved production-site allowlist for the family. """ from app.integrations.atlassian_activity import ( normalize_site_host, product_family_for_access_key, ) group_name = str(binding.get("group_name") or "").strip() raw_key = binding.get("product_key") family = ( product_family_for_access_key(str(raw_key)) if raw_key else None ) if family is None: return None, frozenset(), group_name raw_sites = binding.get("site_urls") # ``site_urls`` is a JSON array when seeded by migration 206, but an # admin editing the binding in the UI may type a comma-separated # string — accept both. Empty / absent → fall back to the resolved # production-site allowlist for the family. if isinstance(raw_sites, str): raw_sites = [s for s in raw_sites.split(",")] if isinstance(raw_sites, list) and raw_sites: hosts = { normalize_site_host(s) for s in raw_sites if isinstance(s, str) } hosts.discard("") sites = frozenset(hosts) or production_sites.get(family, frozenset()) else: sites = production_sites.get(family, frozenset()) return family, sites, group_name @register_scheduled_job( key="jira.seat_sync", display_name="Jira / Confluence · 产品许可同步", description=( "每天通过 Atlassian Org Admin API 的 product_access 核对真实产品许可" "(谁真有 Jira Software / JSM / Confluence license,按付费生产站点)," "把匹配到 Helix 员工的持有者镜像到 software_seat_allocations,并把已不" "再持有许可的 Helix 坐席标为 revoked。坐席口径 = product_access(真有" "license),不再数某个默认组的成员。需要 org_id + org_admin_api_key" "(Atlassian Guard)。每个 plan 的 product_key / site_urls 由 connector " "binding 决定;Confluence 的 tidb / pingcap-us 各算各的订阅。" ), ) async def run_jira_seat_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # 2026-06-12 incident: a worker restart double-ran this job; the # rerun lost an INSERT deadlock (1213) on software_seat_allocations # and the whole run failed. The reconcile is idempotent (import has # the 25h-grace dedup filter, revoke has safe_to_drift_revoke), so # the deadlock victim simply restarts on a fresh session. from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_jira_seat_sync_once(session, config), label="jira.seat_sync", ) async def _run_jira_seat_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.jira_seat_coordinator import ( get_jira_plans, get_plan_binding, load_jira_driver, ) from app.domains.asset.models import ( SoftwareExternalHolder, SoftwareSeatAllocation, SoftwareSeatSyncStat, ) from app.identity.email_lookup import build_person_by_email_index from app.identity.leaver_notice import is_within_leaver_notice_period from app.integrations.atlassian_activity import ( ALL_HOLDER_BUCKETS, HOLDER_EMPLOYEE_ACTIVE, HOLDER_EXTERNAL, HOLDER_LEAVER_RESIDUAL, HOLDER_NO_EMAIL, HOLDER_OFF_PAID_SITE, HOLDER_VENDOR_INACTIVE, _load_production_sites, classify_holder, fetch_atlassian_activity, ) driver = await load_jira_driver(db) if driver is None: # Benign config state, not a failure: ``errors`` is the list # ``raise_if_total_failure`` escalates, so "not configured" goes # in ``skipped`` (informational) and the run stays green. return { "plans_synced": 0, "seats_imported": 0, "seats_revoked": 0, "errors": [], "skipped": ["Jira integration not configured or not enabled"], } plans = await get_jira_plans(db) if not plans: return { "plans_synced": 0, "seats_imported": 0, "seats_revoked": 0, "errors": [], "skipped": [ "No active Jira/Confluence license plans found. Run " "migrations 077 + 206 or create a plan whose parent catalog " "name starts with 'Jira' or 'Confluence'." ], } # Org-wide pull ONCE up front. Raises on hard failure (missing # org_admin_api_key, auth, 5xx) — correct: we must NOT reconcile # against an empty/partial directory (that would mass-revoke every # paid seat). The framework turns the raise into status='failed' + a # deduped failure notification. directory = await fetch_atlassian_activity(db, driver=driver) production_sites = await _load_production_sites(db) total_imported = 0 total_revoked = 0 errors: list[str] = [] skipped: list[str] = [] per_plan_summary: list[dict[str, Any]] = [] now = datetime.now(UTC) for plan in plans: plan_name = plan.name binding = await get_plan_binding(db, plan) if binding is None: skipped.append( f"{plan_name}: no active software_connectors row; skipped", ) per_plan_summary.append({ "plan": plan_name, "skipped": "no_connector", }) continue family, sites, group_name = _resolve_plan_scope( binding, production_sites, ) if family is None: skipped.append( f"{plan_name}: binding has no product_key — not configured " "for product_access seat-counting (e.g. a Marketplace add-on " "catalog reusing the Jira connector); skipped", ) per_plan_summary.append({ "plan": plan_name, "skipped": "no_product_key", }) continue if not sites: skipped.append( f"{plan_name} ({family}): no production site_urls resolved; " "skipped", ) per_plan_summary.append({ "plan": plan_name, "skipped": "no_sites", }) continue # ── COUNTING side: org-wide, all-status, all-sites ───────── # The displayed ``vendor_license_holders`` must match what # Atlassian's Admin / billing UI shows — that means EVERY # account holding the product family across the org, regardless # of ``account_status`` or site. Helix's old filter # (``license_holders(family, sites)``) gave a subset that # silently hid 7 holders on prod 2026-05-29 (1 inactive + # 6 sandbox / personal-site) — user-facing reconciliation # number diverged from the vendor's billing total. all_holders = directory.all_license_holders(family) # Resolve emails to Helix persons ONCE — include leavers so # the leaver_residual bucket can pick them up. The downstream # import branch still excludes leavers (only active employees # become Helix seats). all_emails = {a.email for a in all_holders if a.email} person_idx_all = await build_person_by_email_index(db, all_emails) # Notice-period exception (2026-08-11): a confirmed future # leaver (``leaver_pending``, last working day not yet passed) # is still working and still holds their Jira/Confluence # licence. Excluding them drift-revoked the seat row the night # IT confirmed the termination and re-imported a NEW row once # they re-matched — new ids defeated offboarding's per-seat # revoke-task dedup and duplicated the cancel event. person_idx_active_only = { email: p for email, p in person_idx_all.items() if p.employment_status == "active" or is_within_leaver_notice_period(p) } # ── IMPORT / DRIFT-REVOKE side: filtered subset (unchanged) ── # The import / drift-revoke loops still operate on # active-employee + paid-site holders only — sandbox seats # and inactive vendor accounts must NOT auto-create Helix # seats, and we should NOT auto-revoke for sandbox-only or # inactive-vendor states. holders_active_paid = directory.license_holders(family, sites) holders_by_email_active_paid: dict[str, Any] = {} for acct in holders_active_paid: if acct.email: holders_by_email_active_paid[acct.email] = acct # ── CLASSIFY every org-wide holder into one of six buckets ── # The bucket precedence is documented at # ``app.integrations.atlassian_activity.classify_holder``. bucket_counts: dict[str, int] = {b: 0 for b in ALL_HOLDER_BUCKETS} external_observations: list[ExternalHolderObservation] = [] external_unmatched_samples: list[str] = [] # The active-employee person id set drives the per-person # drift-revoke loop further down — one revoke per person, # not per email. active_employee_user_ids: set = set() for acct in all_holders: email_key = acct.email or None person = person_idx_all.get(email_key) if email_key else None helix_status = person.employment_status if person else None bucket = classify_holder( acct, family, sites, helix_employment_status=helix_status, ) bucket_counts[bucket] += 1 # For the import side, track only active-employee bucket if bucket == HOLDER_EMPLOYEE_ACTIVE and person is not None: active_employee_user_ids.add(person.id) # Persist anomaly buckets (everything except healthy # active-employee and the count-only no_email bucket). if bucket in ( HOLDER_EMPLOYEE_ACTIVE, HOLDER_NO_EMAIL, ): continue # Site to record on the observation: the holder's actual # site(s) for this family (joined for readability when # the holder is on multiple sites). holder_sites = sorted( acct.access_sites_by_product.get(family, frozenset()), ) obs_site = ", ".join(holder_sites) if holder_sites else None external_observations.append(ExternalHolderObservation( email=email_key or "", vendor_account_id=acct.account_id, account_status=acct.account_status, last_active_on=acct.last_active_for(family), display_name=( person.display_name if person else None ), category=bucket, person_id=person.id if person else None, site=obs_site, )) # Keep a capped sample of external emails for the run-log # JSON (same convention as before — explains the gap). if ( bucket == HOLDER_EXTERNAL and len(external_unmatched_samples) < _UNMATCHED_SAMPLE_CAP and email_key ): external_unmatched_samples.append(email_key) holders_matched = bucket_counts[HOLDER_EMPLOYEE_ACTIVE] holders_external = bucket_counts[HOLDER_EXTERNAL] holders_no_email = bucket_counts[HOLDER_NO_EMAIL] holders_leaver_residual = bucket_counts[HOLDER_LEAVER_RESIDUAL] holders_vendor_inactive = bucket_counts[HOLDER_VENDOR_INACTIVE] holders_off_paid_site = bucket_counts[HOLDER_OFF_PAID_SITE] # ``vendor_license_holders`` is the "真实坐席" number surfaced in # the workbench top stat and the seat-reconciliation drawer. It # must mirror what Atlassian admin's Apps / Subscriptions page # bills for. Layered sourcing — first match wins: # # 1. **Atlassian's official ``approximateLicenseCount`` endpoint** # (``GET /rest/api/3/license/approximateLicenseCount/product/{key}``, # Basic auth on the paid site). This is Atlassian's own # "users assigned a license for this product" number, the # closest public-API match to the Admin UI. Prod 2026-05-30: # jira-software → 367 (Atlassian Apps page 372, delta 5), # jira-servicedesk → 73 (Apps page 71, delta 2). Supports # ``jira-*`` only — Atlassian provides no equivalent for # ``confluence``. # # 2. **Helix-computed paid-site bucket sum** for confluence and # any jira call that fails. Excludes ``holders_off_paid_site`` # because sandbox + personal sites do NOT consume the plan's # licenses (Atlassian doesn't bill them). Includes the four # paid-site anomaly buckets + ``no_email`` because each of # those still occupies a license seat. # # Prod 2026-05-30 motivation: pre-fix JSM showed 85 on Helix vs # 71 in Atlassian admin (off_paid_site=15 sandbox+personal agents); # Confluence 134 vs ~105 (off_paid_site=29 PingCAP-US holders). # Switching to Atlassian's own number trims the residual delta # from 11 holders (paid-site product_access) to 2-5 holders # (Atlassian's own subsystem internal drift). atlassian_license_count: int | None = None try: atlassian_license_count = ( await driver.get_approximate_license_count(family) ) except Exception: # pragma: no cover — defensive atlassian_license_count = None paid_site_bucket_sum = ( holders_matched + holders_external + holders_no_email + holders_leaver_residual + holders_vendor_inactive ) if ( atlassian_license_count is not None and atlassian_license_count > 0 ): vendor_license_holders = atlassian_license_count vendor_license_source = "atlassian_approximate" else: vendor_license_holders = paid_site_bucket_sum vendor_license_source = "helix_paid_site_buckets" # Sanity: every org-wide holder lands in exactly one bucket. # ``vendor_license_holders`` may diverge from ``paid_site_bucket_sum`` # by Atlassian's own internal subsystem drift (1–5 holders) when # the authoritative endpoint was used — that's expected and not # an error. assert len(all_holders) == ( holders_matched + holders_external + holders_no_email + holders_leaver_residual + holders_vendor_inactive + holders_off_paid_site ), ( f"jira.seat_sync identity broken pre-write for plan={plan_name}: " f"{len(all_holders)} != sum(buckets)" ) # Active OR pending OR recently-revoked — see _seat_sync_helpers # import_dedup_filter. Avoids IntegrityError under migration 119's # partial unique index AND prevents racing a recent admin-revoke / # leaver_blocked flip with a re-import. 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"] # Import seats ONLY for active-employee holders on the paid # site (sandbox holders / vendor-inactive / leavers must NOT # become Helix allocations — those are anomaly buckets, # surfaced in the drawer but not back-filled as active seats). seats_imported = 0 for email, acct in holders_by_email_active_paid.items(): person = person_idx_active_only.get(email) if person is None: continue # not an active employee (external / leaver / inactive) if person.id in existing_by_user_id: continue try: # Per-row SAVEPOINT — a single bad INSERT (FK violation, # unique-index collision) rolls back just this row, not # the whole shared session (per-process SAVEPOINT # isolation, same rule as the other cron loops). 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 by jira.seat_sync: holds {family} " f"license on {sorted(sites)}" ), )) await db.flush() seats_imported += 1 except OperationalError: # Deadlock (1213) rolls back the WHOLE transaction # server-side — no savepoint can save the session. # Propagate so the fresh-session lock-retry wrapper # restarts the (idempotent) run cleanly. raise except Exception as e: # noqa: BLE001 errors.append( f"{plan_name}/{email}: import failed: " f"{format_exception_for_log(e)}" ) if len(errors) > 30: errors.append("...truncated") break seats_revoked = 0 # SAFETY GUARD via shared helper — see _seat_sync_helpers.py. # ``matched_count`` here is the active-employee-on-paid-site set # (what the drift-revoke loop operates on), NOT the org-wide # vendor total — the ratio guard's "do we have enough live # users to safely revoke" semantic is about the import set, not # billed-but-inactive accounts. safe, reason = safe_to_drift_revoke( matched_count=len(active_employee_user_ids), existing_count=len(existing_seats), label=f"{plan_name} ({family} @ {sorted(sites)})", want_count=len(active_employee_user_ids), ) if not safe: errors.append(reason or "drift cleanup refused") else: for seat in existing_seats: if seat.user_id in active_employee_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 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 "") + f"\n[auto-revoked by jira.seat_sync: no longer " f"holds {family} license on {sorted(sites)}]" ).strip() await write_drift_event_audit( db, job_key="jira.seat_sync", seat_id=seat_id, user_id=seat_user_id, plan_id=seat_plan_id, event="revoked", detail={ "reason": ( f"no longer holds {family} license " f"on {sorted(sites)}" ), }, ) await db.flush() seats_revoked += 1 except OperationalError: # Whole-transaction rollback (deadlock) — see the # import loop above. raise except Exception as e: # noqa: BLE001 errors.append( f"{plan_name}/seat {seat_id}: drift revoke failed: " f"{format_exception_for_log(e)}" ) total_imported += seats_imported total_revoked += seats_revoked per_plan_summary.append({ "plan": plan_name, "product": family, "sites": sorted(sites), "group_name": group_name, # The Atlassian billable-seat total — preferred source is # Atlassian's own approximate-license-count endpoint # (``vendor_license_source='atlassian_approximate'``); the # 5-bucket paid-site sum is the fallback for confluence and # for failed endpoint calls. "license_holders": vendor_license_holders, "license_count_source": vendor_license_source, "paid_site_bucket_sum": paid_site_bucket_sum, # Five paid-site buckets sum to paid_site_bucket_sum; # holders_off_paid_site is reported separately for audit # but is NOT part of the licensed-seat total. When # ``license_count_source='atlassian_approximate'``, # ``license_holders`` may differ from ``paid_site_bucket_sum`` # by a few holders (Atlassian's own subsystem drift). "holders_matched_to_employee": holders_matched, "holders_external_unmatched": holders_external, "holders_external_unmatched_samples": external_unmatched_samples, "holders_without_email": holders_no_email, "holders_leaver_residual": holders_leaver_residual, "holders_vendor_inactive": holders_vendor_inactive, "holders_off_paid_site": holders_off_paid_site, "seats_imported": seats_imported, "seats_revoked": seats_revoked, }) try: await record_seat_reconciliation( db, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=uuid.UUID(driver.integration_id), software_id=plan.software_id, plan_id=plan.id, provider="jira", product_key=family, site=", ".join(sorted(sites)) if sites else None, vendor_license_holders=vendor_license_holders, holders_matched=holders_matched, holders_no_email=holders_no_email, holders_leaver_residual=holders_leaver_residual, holders_vendor_inactive=holders_vendor_inactive, holders_off_paid_site=holders_off_paid_site, externals=external_observations, now=now, ) except SeatReconciliationIdentityError as e: # The handler counts ``vendor_license_holders`` and the buckets # per Atlassian ``account_id``; the recorder re-validates the # identity against EMAIL-deduped observations. When the org's # directory legitimately holds two managed account_id rows with # the same email (a real, if rare, case — see the explicit # "duplicate-email collision" handling in # ``atlassian_activity.fetch_atlassian_activity``), and at least # one falls in an external-side bucket, the two axes disagree and # the recorder raises. That must NOT abort the whole daily sync: # this raise sits before ``write_sync_audit`` / ``db.commit()`` and # outside any per-plan guard, so an unguarded raise would skip the # audit row AND every plan after this one in the loop, landing the # run ``status='failed'`` and paging on-call every tick until a # human edits the Atlassian directory. Isolate it to THIS plan: # record the per-plan error (surfaced on the scheduled-jobs admin # page) and continue so the other plans still reconcile. Only the # snapshot for this one plan goes stale (the import/drift seat # actions for it already ran above). NB: a genuine caller-side # miscount would violate the identity for EVERY plan, and every # plan's error lands in ``summary['errors']`` (surfaced on the # scheduled-jobs admin page); ``raise_if_total_failure`` below # additionally pages on-call only when the run made zero progress # (no seat imported or revoked), so a systemic bug is always # visible even when it doesn't escalate to a page. err = ( f"{plan_name} ({family}): seat reconciliation snapshot skipped " f"— vendor/observation identity mismatch (likely duplicate-email " f"Atlassian accounts in the org directory): " f"{format_exception_for_log(e)}" ) logger.warning("jira.seat_sync: %s", err) errors.append(err) per_plan_summary.append({ "plan": plan_name, "reconciliation_error": "identity_mismatch", }) summary = { "plans_synced": len(plans), "atlassian_accounts": directory.total_accounts, "seats_imported": total_imported, "seats_revoked": total_revoked, "per_plan": per_plan_summary, "skipped": skipped, "errors": errors, } await write_sync_audit(db, job_key="jira.seat_sync", summary=summary) await db.commit() raise_if_total_failure( job_key="jira.seat_sync", summary=summary, progress_keys=("seats_imported", "seats_revoked"), ) return summary # Cap on the number of Helix leavers matched per run (in-memory join # against the org directory — bounded so the summary row stays sane as # the company's leaver history grows; the Atlassian side is ONE paginated # call regardless). _LEAVER_AUDIT_MAX = 5000 # How many prior audit-driven remediation tasks (within the backoff # window) we allow for the same offender+target before we STOP # auto-revoking and escalate to admin instead. Rationale: # * Pass 1-2: catches transient slips — a race between offboarding's # "remove all groups" and a slow SCIM tick, an Atlassian replication # lag, etc. Idempotent retry fixes it. # * Pass 3 in a week = the source side (SCIM directory, default-group # config, an admin manually re-adding the user) is actively # re-asserting access. No amount of Helix-side revoke will stick # until a human fixes the source — so further auto-revokes are pure # noise, and the admin alert IS the next correct action. _AUDIT_REMEDIATION_BACKOFF_THRESHOLD = 3 _AUDIT_REMEDIATION_BACKOFF_WINDOW_DAYS = 7 # ── Leaver ACCOUNT-lifecycle watchdog ──────────────────────────── # # Helix never deactivates an Atlassian account: offboarding strips every # group plus org-level product access and deliberately leaves the # account alive so issue history stays attributed (see # ``JiraSoftwareDriver.revoke_all_access``). What actually flips a # leaver's account to ``inactive`` is the ORG's IdP directory sync — # Atlassian Guard's Google Workspace / SCIM user provisioning noticing # the upstream account is gone. # # That makes the sync a silent single point of failure, and on # 2026-08-13 it failed: one save of the Google-sync settings form turned # user provisioning off, and for 16 days every leaver kept a live # Atlassian account. This audit still reported clean runs the whole # time, because it only ever looked at ``product_access`` — never at # ``account_status``. Nothing in Helix noticed, because nothing in Helix # was looking. # # A leaver whose account is STILL ``active`` this many days after # termination is therefore evidence that the upstream deactivation path # is broken, independent of whether any product access remains. Grace # days cover the normal chain: Helix suspends the Google account at # T+0, deletes it at T+``delete_lead_days`` (default 2), and the # Atlassian sync then runs on its own (4-hourly) cadence. # Overridable per-deploy via the job's ``config.stale_active_grace_days``. _STALE_ACTIVE_ACCOUNT_GRACE_DAYS = 7 # Cap on stale-active rows carried in the run summary — the alert names # the count, the summary carries a bounded, actionable sample. _STALE_ACTIVE_SAMPLE_CAP = 50 # A non-group residual (Trello / org default access / a lagging # ``product_access`` entry) is re-stripped via remove-access on every # run, and a 2xx is treated as "done — it may reappear for <=24h until # Atlassian's listing refreshes". For some products that eventual # consistency never arrives: Trello access is *workspace* membership, # which ``DELETE directory/users`` cannot touch, so the call 204s # forever while the product stays put and the row never alerts. Two # leavers sat in exactly that silent loop from March to August. # Past this many days it is not lag — stop calling it self-healing and # put it in front of a human. # Overridable via the job's ``config.non_group_residual_stale_days``. _NON_GROUP_RESIDUAL_STALE_DAYS = 30 async def _recent_audit_remediation_tasks( db: AsyncSession, *, user_id: uuid.UUID, target_type: str, target_id: uuid.UUID, product_key: str | None, ) -> list[Any]: """Audit-source remediation tasks for this (user, target, product) within the backoff window. Returns the rows (any status — pending / in_progress / completed / failed all count). The caller uses ``len()`` to drive the "escalate after N consecutive bounces" rule and the rows themselves to dedup against an already-in-flight remediation. ``product_key`` filtering is Python-side because ``context_data`` is a JSON column and the per-DB JSON-path syntax doesn't survive the MySQL-8 / TiDB / SQLite-in-tests matrix. The candidate set is bounded by ``(user_id, target_type, target_id, source=audit, backoff_window)`` so the in-memory filter is cheap. """ from datetime import timedelta from app.domains.provisioning.models import ( ACTION_REVOKE, SOURCE_AUDIT, ProvisioningTask, ) cutoff = datetime.now(UTC) - timedelta( days=_AUDIT_REMEDIATION_BACKOFF_WINDOW_DAYS, ) rows = list((await db.execute( select(ProvisioningTask).where( ProvisioningTask.user_id == user_id, ProvisioningTask.source == SOURCE_AUDIT, ProvisioningTask.action == ACTION_REVOKE, ProvisioningTask.target_type == target_type, ProvisioningTask.target_id == target_id, ProvisioningTask.created_at >= cutoff, ) )).scalars().all()) # A revoke task whose ``external_ref`` ends in ``:no_account`` did # NOTHING — site user-search couldn't resolve the account by email, so # the driver returned a no-op "success" without stripping any group. # These must not count toward the 3-strike backoff: counting them lets # a GDPR-hidden leaver (email invisible to /user/search) accrue three # phantom "bounces" and fire a false "source keeps re-adding them" # alert when no removal was ever attempted. Now that ``revoke_all_access`` # falls back to the audit-supplied account_id, real attempts carry the # account_id as ``external_ref``; only genuine attempts gate the backoff. rows = [ r for r in rows if not str(r.external_ref or "").endswith(":no_account") ] if product_key is None: return rows pk_norm = product_key.strip().lower() return [ r for r in rows if str((r.context_data or {}).get("product_key") or "") .strip().lower() == pk_norm ] def _has_in_flight_audit_task(rows: list[Any]) -> bool: """True if any of the candidate rows is still pending / in_progress / notified — i.e. the dispatcher hasn't run it yet (or it's been handed to a human via manual_notify). Enqueueing another would just stack duplicates; let the existing one finish first.""" from app.domains.provisioning.models import ( STATUS_IN_PROGRESS, STATUS_NOTIFIED, STATUS_PENDING, ) non_terminal = {STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED} return any(r.status in non_terminal for r in rows) def _extract_managed_account(u: dict[str, Any]) -> dict[str, Any]: """Normalise one ``list_org_users_with_activity`` record. The Admin API returns either a flat dict or one nested under ``attributes`` depending on version — flatten to ``{email, account_id, account_status, product_access}``. """ attrs = u.get("attributes") if isinstance(u.get("attributes"), dict) else {} def _pick(*keys: str) -> Any: for k in keys: if u.get(k) is not None: return u.get(k) if attrs.get(k) is not None: return attrs.get(k) return None pa = _pick("product_access") or [] return { "email": str(_pick("email") or "").strip().lower(), "account_id": _pick("account_id", "id"), "account_status": _pick("account_status", "status"), "product_access": pa if isinstance(pa, list) else [], } async def _account_site_groups( driver: Any, account_id: str | None, ) -> list[str] | None: """The account's current group memberships on the production site, or ``None`` if the lookup couldn't be performed (missing id / API error). This is the linchpin that tells a genuinely group-revocable residual apart from one a group revoke can never clear. The Atlassian Org Admin API's ``product_access`` (what both audits read to decide "still has access") does NOT track group membership in real time: * **Trello** access is board/workspace membership, not group- conferred — ``product_access`` shows ``trello`` while ``/user/groups`` is empty (observed: chujie / wangdi / yudingyi). * **product_access lags** group removal — after a revoke strips every group, ``product_access`` can keep listing the products for hours/days (observed: liangzhiyuan — 11 groups removed on pass 1, then zero groups yet 9 products still reported on passes 2-3). * org-level **default access** / a direct product-role grant confer a product with zero group membership. In all of those the account has **no groups**, so re-running the group revoke is a no-op (``removed: []``) and the next audit re-reads the same lingering ``product_access`` and false-reports a perpetual "bounce" — burning the 3-strike backoff and firing a misleading "SCIM keeps re-adding them" alert when nothing is being re-added. An account with ≥1 group is group-revocable; HOW each caller uses the list differs by what its revoke removes: * ``revoke_all_access`` (leaver path) strips **every** group, so any group at all means the residual is revocable — do NOT narrow by product family here, or a leaver whose only access is via a group ``product_family_for_group`` can't classify (``bitbucket-users``, ``jira-administrators``, a ``jira-product-discovery`` group, …) would be wrongly treated as non-revocable and never cleaned. * ``revoke_product_access`` (active-employee path) strips only the target product's conferring groups, so that caller intersects by family. Returns ``None`` (not ``[]``) when the id is missing or the API call raises, so the caller stays conservative (still remediate) instead of wrongly concluding "no groups". ``get_user_groups`` maps a 404 (account unknown to the production site — the same site the revoke operates on) to ``[]``, which correctly reads as "group revoke can't reach this account here". """ if not account_id: return None try: return await driver.get_user_groups(str(account_id)) except Exception as exc: # noqa: BLE001 logger.warning( "jira audit: get_user_groups(%s) failed: %s — treating residual " "as group-revocable (conservative).", account_id, exc, ) return None @register_scheduled_job( key="jira.leaver_account_audit", display_name="Jira · 离职残留访问自动清理 + 持续反弹兜底", description=( "拉一次 Atlassian Org Admin API 的托管账号清单,跟 Helix 标记为 left" "(已离职)的员工对账。离职但仍有 product_access 的人,**先查他在不在" "任何 group 里**(revoke_all_access 会删光所有 group):" "在 → 自动入队 revoke_all_access 重新移出,无需人工;7 天内已重 revoke " "3 次 group 仍被加回 → 升级 IT 告警(真·源头反弹:SCIM/default-group/" "误操作)。一个 group 都不在(Trello 看板成员 / 默认访问 / " "删组后 product_access 尚未回落)→ 删组清不掉,单独 fire-once 告警," "不再空转重撤、不再误报反弹。只在确有需要时才打扰人。" "另外单独看一个与访问无关的信号:离职满 N 天(默认 7)而 Atlassian " "**账号本身**仍是 active —— Helix 按策略从不停用账号,做这件事的是" "组织侧的身份目录同步(Guard 的 Google Workspace / SCIM 用户供给)," "所以这一条为真 = 那条链路断了,告警给 IT。" ), ) async def run_jira_leaver_account_audit( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Auto-remediate leavers who still have ANY Atlassian product access. Self-healing safety net for the offboarding full-access-revoke flow: a properly-offboarded leaver has been removed from every group and so has zero ``product_access``; anyone marked ``left`` in Helix who STILL shows product access on Atlassian had their offboarding revoke slip past at least one group. The handler now ENQUEUES a fresh ``target_type='account'`` provisioning task per offender — same code path the offboarding flow uses, idempotent (``revoke_all_access``) — instead of pinging an admin to do it by hand. Admin only gets a notification when the same leaver has been auto-revoked ≥ ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in ``_AUDIT_REMEDIATION_BACKOFF_WINDOW_DAYS``: at that point the source side (SCIM / default-group / manual re-add) is actively bouncing the user back and a human has to fix the root cause. """ from sqlalchemy import select as _select from app.core.coordinators.jira_seat_coordinator import load_jira_driver from app.identity.person_models import IdentityPerson driver = await load_jira_driver(db) if driver is None: # Benign config state (Jira not wired up) — a ``skipped`` note, not # an ``errors`` entry, so the audit run doesn't surface as failed. return { "leavers_matched": 0, "still_have_access": 0, "offenders": [], "errors": [], "skipped": ["Jira integration not configured or not enabled"], } if not driver._org_id() or not driver._org_admin_api_key(): return { "leavers_matched": 0, "still_have_access": 0, "offenders": [], "errors": [], "skipped": [ "Jira integration missing org_id / org_admin_api_key — the " "managed-account roster is read via the Org Admin API." ], } # ONE paginated pull of the whole org's managed accounts. Raises on a # hard failure (auth/connectivity) → the run lands FAILED rather than # a misleading green "0 offenders". raw_accounts = await driver.list_org_users_with_activity() access_by_email: dict[str, dict[str, Any]] = {} for u in raw_accounts: if not isinstance(u, dict): continue rec = _extract_managed_account(u) if rec["email"]: access_by_email[rec["email"]] = rec leavers = list((await db.execute( _select(IdentityPerson) .where( IdentityPerson.employment_status == "left", IdentityPerson.work_email.is_not(None), ) .order_by(IdentityPerson.termination_date.desc()) .limit(_LEAVER_AUDIT_MAX) )).scalars().all()) stale_grace_days = _config_int( config, "stale_active_grace_days", _STALE_ACTIVE_ACCOUNT_GRACE_DAYS, ) today = datetime.now(UTC).date() offenders: list[dict[str, Any]] = [] non_group_residual: list[dict[str, Any]] = [] stale_active: list[dict[str, Any]] = [] for person in leavers: email = (person.work_email or "").strip().lower() rec = access_by_email.get(email) if rec is None: continue # no managed account → nothing to revoke days_since_termination = ( (today - person.termination_date).days if person.termination_date is not None else None ) # Account-lifecycle watchdog — deliberately BEFORE the # product_access early-return: the 2026-08-13 outage looked # exactly like a clean run down this path (every leaver had zero # product_access because Helix's own revoke worked) while their # ACCOUNTS stayed active because the org's IdP directory sync had # been switched off. Account status is a separate signal from # access, and only this check reads it. # See _STALE_ACTIVE_ACCOUNT_GRACE_DAYS. if ( str(rec["account_status"] or "").strip().lower() == "active" and days_since_termination is not None and days_since_termination >= stale_grace_days ): stale_active.append({ "email": email, "name": person.display_name, "account_id": rec["account_id"], # ``days_since_termination is not None`` above is only # true when ``termination_date`` is set — it is derived # from it. Stating that here rather than relying on the # correlation keeps both mypy and the next reader from # having to re-derive it. "termination_date": ( person.termination_date.isoformat() if person.termination_date else None ), "days_since_termination": days_since_termination, }) if not rec["product_access"]: continue # already stripped of all product access — clean row = { "email": email, "name": person.display_name, "account_id": rec["account_id"], "account_status": rec["account_status"], "days_since_termination": days_since_termination, "products": [ str(p.get("key") or p.get("name")) for p in rec["product_access"] if isinstance(p, dict) ][:10], "termination_date": ( person.termination_date.isoformat() if person.termination_date else None ), } # ``revoke_all_access`` removes the leaver from EVERY group, so the # residual is group-revocable iff the account is in ANY group at # all (do NOT narrow by product family — that would wrongly exempt # leavers whose access comes via bitbucket / jira-administrators / # product-discovery groups). Zero groups = Trello / default-access # / a not-yet-dropped product_access entry that group removal can't # clear — route to a fire-once bucket instead of the auto-revoke + # 3-strike loop. ``None`` (missing id / lookup error) → stay # conservative: enqueue the revoke (it resolves the user by email # anyway) and let the backoff cap retries. site_groups = await _account_site_groups(driver, rec["account_id"]) row["site_group_count"] = ( len(site_groups) if site_groups is not None else None ) if site_groups is None or len(site_groups) > 0: offenders.append(row) else: non_group_residual.append(row) # ── Account end-state assurance (Atlassian half) ───────────── # # Written from the roster this run already pulled, so the governance # page is a plain DB read instead of a live fan-out. Scope is EVERY # matched leaver, not just the offenders — the whole lesson of # 2026-08-13 is that the rows which look clean by access are exactly # where a dead deactivation chain hides. from app.domains.offboarding import assurance_service as _assurance from app.domains.offboarding.models import ( ASSURANCE_SYSTEM_ATLASSIAN as _SYS_ATLASSIAN, ) _exemptions = await _assurance.registry_exemptions( db, [(p.work_email or "").strip().lower() for p in leavers], ) _assurance_rows: list[dict[str, Any]] = [] for person in leavers: email = (person.work_email or "").strip().lower() if not email: continue # Same grace the stale-active watchdog uses, and for the same # reason: offboarding revokes at T+0 and the org's directory # sync then runs on its own 4-hourly cadence, so a leaver in # their first days is a process in flight, not a finding. # Without this the page would light up red for every fresh # leaver and go green a day later — and a governance number that # churns on its own is one people learn to wait out. _age = ( (today - person.termination_date).days if person.termination_date else None ) if _age is not None and _age < stale_grace_days: continue # Optional on purpose: no managed account is a real answer here # (``absent`` → the person cannot sign in → ``ok``), and # ``atlassian_vendor_state`` takes ``None`` for exactly that. assurance_rec: dict[str, Any] | None = access_by_email.get(email) state, evidence = _assurance.atlassian_vendor_state(assurance_rec) exemption = _exemptions.get(email) verdict, exempt_reason = _assurance.verdict_for( state, exemption=exemption, ) detail = { **evidence, "work_email": email, "employee_no": person.employee_no, "days_since_termination": _age, } if exemption: detail["exemption"] = exemption _assurance_rows.append({ "person_id": person.id, "vendor_state": state, "vendor_account_id": (assurance_rec or {}).get("account_id"), # Helix's own action on the Atlassian side is always the same # one: strip every group. Naming it explicitly keeps the # column meaningful next to Google's richer ledger. "helix_action": "groups_revoked" if assurance_rec else None, "verdict": verdict, "exempt_reason": exempt_reason, "detail": detail, }) assurance_counts = await _assurance.upsert_rows( db, system=_SYS_ATLASSIAN, rows=_assurance_rows, ) # Attach each stale-active row's retention decision (if any) before # deciding what to do with it — the exemption is what separates # "Helix should close this" from "a person must settle this", and # that call has to be made with the decision in hand, not inferred. for _row in stale_active: _ex = _exemptions.get(_row["email"]) if _ex: _row["exemption"] = _ex # Helix can close these itself now. Two buckets, and the split is a # deliberate safety property, not a nicety: # # * no retention decision on file → suspend it. Leaving a leaver # able to authenticate because an org-side sync is unwell is # exactly the failure this whole change exists to end. # * a retention decision IS on file → NEVER auto-suspend, even when # the decision looks thin. Somebody chose to keep that account # open; the three accounts that motivated this work were all # still being logged into daily, months after termination. # Quietly locking a person out because their exemption row lacks # a note would be the platform making a people decision on its # own. It goes to the page instead, flagged as an unowned # exemption, for a human to settle. stale_suspended: list[dict[str, Any]] = [] stale_needs_decision: list[dict[str, Any]] = [] if stale_active: stale_suspended, stale_needs_decision = ( await _auto_suspend_stale_accounts(driver, stale_active) ) auto_queued: list[dict[str, Any]] = [] persistent: list[dict[str, Any]] = [] remediation_errors: list[dict[str, str]] = [] if offenders: await _auto_remediate_leavers( db, driver_integration_id=uuid.UUID(driver.integration_id), offenders=offenders, auto_queued=auto_queued, persistent=persistent, errors=remediation_errors, ) # Non-group residual (Trello workspace / org default access / direct # product grants): group removal can't clear these, but the # Organizations remove-access API (DELETE directory/users, UNSCOPED # org key) can. Auto-strip org access here so future leavers don't # linger; only rows we genuinely can't auto-fix (no unscoped key / # API refused) fall through to the admin notification. residual_auto_removed: list[dict[str, Any]] = [] residual_manual: list[dict[str, Any]] = list(non_group_residual) if non_group_residual: residual_auto_removed, residual_manual = ( await _auto_remove_non_group_residual(driver, non_group_residual) ) # A 2xx from remove-access only means "the org directory accepted # the call". For access that isn't org-directory-managed (Trello # workspace membership above all) it changes nothing, so the row # comes back next run, gets another 2xx, and — because a 2xx # routes to the silent ``auto_removed`` bucket — never reaches a # human. Long enough past termination, "eventual consistency" is # not a credible explanation any more. # See _NON_GROUP_RESIDUAL_STALE_DAYS. stale_days = _config_int( config, "non_group_residual_stale_days", _NON_GROUP_RESIDUAL_STALE_DAYS, ) self_healing: list[dict[str, Any]] = [] never_healing: list[dict[str, Any]] = [] for row in residual_auto_removed: age = row.get("days_since_termination") if age is not None and age >= stale_days: row["auto_remove_reason"] = ( f"remove-access returned {row.get('auto_remove_status')} " f"but this access has persisted {age} days past " "termination — that is not sync lag. The access is not " "org-directory-managed (typically Trello workspace " "membership) and has to be removed at the product itself." ) never_healing.append(row) else: self_healing.append(row) residual_auto_removed = self_healing residual_manual = [*residual_manual, *never_healing] summary = { "org_accounts_scanned": len(access_by_email), "leavers_matched": len(leavers), # Every leaver who still holds product access — group-revocable # AND non-group alike. This used to count ``offenders`` only, so # a run with six leavers sitting in the non-group bucket still # printed ``still_have_access: 0``; the number that operators and # audits read was reporting the opposite of the truth. "still_have_access": len(offenders) + len(non_group_residual), "still_have_access_group_revocable": len(offenders), "offenders": offenders, # Auto-remediation buckets: ``auto_queued`` = fresh revoke task # enqueued this run; ``persistent`` = a conferring group is being # genuinely re-added every run, escalated to admin notification; # ``in_flight_skipped`` = an earlier remediation task is still # in the dispatcher queue, no new task created. "auto_queued": auto_queued, "persistent": persistent, # Leavers whose residual product_access is NOT backed by any # conferring group (Trello / default-access / lagging # product_access). Group revoke can't touch these; Helix now # auto-strips them via the org remove-access API. # ``non_group_residual`` = all detected; ``..._auto_removed`` = # cleared via API this run (no alert — eventual-consistency means # they may reappear for ≤24h and get removed again, idempotent); # ``..._manual`` = still need a human (no unscoped key / API # refused) → the only ones that reach the admin bell. "non_group_residual": non_group_residual, "non_group_residual_auto_removed": residual_auto_removed, "non_group_residual_manual": residual_manual, # Leavers whose Atlassian ACCOUNT is still active well past # termination — the org's IdP directory sync is not deactivating # them. Independent of product access: this is the only bucket # that would have caught the 2026-08-13 provisioning outage. "stale_active_accounts": stale_active[:_STALE_ACTIVE_SAMPLE_CAP], "stale_active_account_count": len(stale_active), # Split of the above: what Helix closed on its own vs what a # person still has to settle (retention decision on file, missing # credential, or the API refused). "stale_active_auto_suspended": ( stale_suspended[:_STALE_ACTIVE_SAMPLE_CAP] ), "stale_active_needs_decision": ( stale_needs_decision[:_STALE_ACTIVE_SAMPLE_CAP] ), # Account end-state assurance, refreshed from this run's roster. "assurance": assurance_counts, "remediation_errors": remediation_errors, "errors": remediation_errors, } await write_sync_audit(db, job_key="jira.leaver_account_audit", summary=summary) # WARNING only when something still needs attention (offenders, or # residual we could NOT auto-strip). A run where every non-group # residual was auto-removed is success → INFO, so alert-on-WARNING # tooling isn't paged on a clean self-healing run. if offenders or residual_manual or stale_active: logger.warning( "jira.leaver_account_audit: %d group-revocable leaver(s) " "(auto-queued=%d, persistent=%d); %d non-group residual " "(auto-removed=%d, manual=%d): %s", len(offenders), len(auto_queued), len(persistent), len(non_group_residual), len(residual_auto_removed), len(residual_manual), ", ".join(o["email"] for o in offenders[:20]), ) elif residual_auto_removed: logger.info( "jira.leaver_account_audit: auto-removed non-group residual " "access for %d leaver(s), nothing needs attention.", len(residual_auto_removed), ) if persistent: await _notify_admins_of_persistent_leavers(db, persistent) # Only the rows Helix could NOT auto-strip reach the admin — a fully # auto-removed run is silent (the whole point of this fix). if residual_manual: await _notify_admins_of_non_group_residual(db, residual_manual) if stale_active: logger.warning( "jira.leaver_account_audit: %d leaver account(s) still ACTIVE " ">= %d days after termination — the org's IdP directory sync " "is not deactivating them: %s", len(stale_active), stale_grace_days, ", ".join(r["email"] for r in stale_active[:20]), ) if stale_needs_decision: await _notify_admins_of_stale_active_accounts( db, stale_needs_decision, grace_days=stale_grace_days, ) await db.commit() # Surface a hard failure only when the remediation enqueue itself # threw — a persistent-bouncer notification IS the success path # for that offender bucket, so it must not trigger the cron's # failure notifier. Mirrors the # ``feedback_handler_status_must_raise`` rule: silently swallowing # a real enqueue error would leave leavers carrying product access # and no operator the wiser. if remediation_errors: raise RuntimeError( "jira.leaver_account_audit: " f"{len(remediation_errors)} remediation enqueue(s) failed — " f"sample: {remediation_errors[0]['error'][:200]}" ) return summary async def _auto_remediate_leavers( db: AsyncSession, *, driver_integration_id: uuid.UUID, offenders: list[dict[str, Any]], auto_queued: list[dict[str, Any]], persistent: list[dict[str, Any]], errors: list[dict[str, str]], ) -> None: """For each offender, enqueue a ``revoke_all_access`` task UNLESS the same leaver has already been auto-revoked ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in the backoff window — in which case the source is bouncing and we escalate to admin instead. ``offenders`` rows are mutated in-place with a ``remediation`` field describing what happened ('queued' / 'in_flight_skipped' / 'persistent' / 'error'); ``auto_queued`` / ``persistent`` / ``errors`` are OUT parameters the caller surfaces in the run summary (same shape as ``recently_revoked_audit.notification_failures``). """ from app.domains.provisioning.models import SOURCE_AUDIT as _SRC_AUDIT from app.domains.provisioning.schemas import ProvisioningTaskCreate from app.domains.provisioning.service import ProvisioningService from app.identity.person_models import IdentityPerson # Find person rows for every offender — we need the UUID to enqueue. emails = [o["email"] for o in offenders] persons = list((await db.execute( select(IdentityPerson).where( IdentityPerson.work_email.in_(emails), IdentityPerson.employment_status == "left", ) )).scalars().all()) by_email = {(p.work_email or "").strip().lower(): p for p in persons} svc = ProvisioningService(db) for off in offenders: email = off["email"] person = by_email.get(email) if person is None: # Shouldn't happen — the same query produced these # offenders. Skip safely. off["remediation"] = "skipped_no_person" continue prior = await _recent_audit_remediation_tasks( db, user_id=person.id, target_type="account", target_id=driver_integration_id, product_key=None, ) if len(prior) >= _AUDIT_REMEDIATION_BACKOFF_THRESHOLD: off["remediation"] = "persistent" off["prior_remediation_count"] = len(prior) persistent.append({**off}) continue if _has_in_flight_audit_task(prior): # An earlier remediation is still pending / in_progress — # the dispatcher will execute it on the next tick. Don't # stack duplicates. off["remediation"] = "in_flight_skipped" off["prior_remediation_count"] = len(prior) continue try: # Per-offender SAVEPOINT — create_task does db.add + db.flush # on the shared session; one bad flush would leave the # session in PendingRollbackError and silently drop the # tail of the batch (those leavers keep access). The # savepoint rolls back just this offender on failure # (per-process SAVEPOINT isolation, same rule as the # seat-sync loops above). async with db.begin_nested(): task = await svc.create_task(ProvisioningTaskCreate( user_id=person.id, software_id=None, action="revoke", target_type="account", target_id=driver_integration_id, source=_SRC_AUDIT, context_data={ "audit_job": "jira.leaver_account_audit", "audit_attempt": len(prior) + 1, "revoke_reason": ( "auto-remediation: leaver still in Atlassian " "groups after offboarding completed" ), "account_id": off.get("account_id"), "products_at_detection": off.get("products") or [], }, )) except OperationalError: # Deadlock (1213) rolls back the WHOLE transaction # server-side — no savepoint can save the session. # Propagate so the fresh-session lock-retry wrapper # restarts the (idempotent) run cleanly. raise except Exception as exc: # noqa: BLE001 # Don't fail the whole audit run on one enqueue glitch — # the savepoint rolled back this offender; collect the # error and continue so other offenders still get processed. err = f"{type(exc).__name__}: {exc}" logger.exception( "jira.leaver_account_audit: enqueue failed for %s: %s", email, err, ) off["remediation"] = "error" off["remediation_error"] = err errors.append({"email": email, "error": err}) continue off["remediation"] = "queued" off["remediation_task_id"] = str(task.id) off["prior_remediation_count"] = len(prior) auto_queued.append({**off}) async def _notify_admins_of_persistent_leavers( db: AsyncSession, offenders: list[dict[str, Any]], ) -> None: """In-app push to IT admins listing leavers whose Atlassian access has been auto-revoked ≥ ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in the backoff window and STILL bounces back. This is the rare path now — the regular case (leaver detected → auto-revoke enqueued) is silent. Only persistent bouncers reach the admin's bell, because those need a human to fix the source (SCIM directory still has the user, default-group auto-includes them on next sign-in, an admin re-added them, …). Deduped weekly per admin via ``NotificationOutbox.template_key``. """ from datetime import UTC, datetime, timedelta from sqlalchemy import and_, select from app.notifications.models import NotificationOutbox from app.notifications.service import NotificationEnvelope, NotificationService from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(db) if not admins: logger.info( "jira.leaver_account_audit: %d persistent offender(s) but no " "eligible it_admin recipient — skipping notification.", len(offenders), ) return week_ago = datetime.now(UTC) - timedelta(days=7) # Language-neutral detail lines: email, products, and the retry count # as a bare "N×". The surrounding prose is the template's job, so the # block can be dropped verbatim into the zh / en / ja body without # leaving a Chinese fragment inside a Japanese card. sample = "\n".join( f" · {o['email']} " f"({', '.join(o.get('products') or []) or 'product access'}) — " f"{o.get('prior_remediation_count', '?')}×" for o in offenders[:10] ) extra = ( f"\n · ... ({len(offenders) - 10} more)" if len(offenders) > 10 else "" ) sample_block = f"{sample}{extra}" body = ( f"过去 {_AUDIT_REMEDIATION_BACKOFF_WINDOW_DAYS} 天内," f"{len(offenders)} 名已离职员工被 Helix 从 Atlassian 授权 " f"group 中重复移除 ≥ {_AUDIT_REMEDIATION_BACKOFF_THRESHOLD} 次," f"该 group 仍每次被重新加回 —— 这是真正的源头反弹" f"(SCIM directory 还没删人 / default-group 自动包含 / 管理员手动" f"加回)。Helix 已停止继续自动 revoke,需要人工排根因:\n\n" f"{sample_block}\n\n" "处理:去 Atlassian Admin UI 检查这些账号被反复加回的 group 来源。\n" "(注:仅列出确实还在某个授权 group 里、删了又被加回的人;Trello / " "默认访问 这类非 group 授予的残留单独告警。)" ) notif = NotificationService(db) for admin in admins: already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "jira.leaver_account_audit", NotificationOutbox.created_at >= week_ago, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: continue try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=( f"[治理] {len(offenders)} 名离职员工 Atlassian 访问" f"持续反弹(自动 revoke ≥ " f"{_AUDIT_REMEDIATION_BACKOFF_THRESHOLD} 次仍未清干净)" ), body_text=body, template_key="jira.leaver_account_audit", related_resource_type="offboarding", related_resource_id=None, context={ "offender_count": len(offenders), "samples": [o["email"] for o in offenders[:20]], # Template placeholders. ``sample_block`` is the # rendered detail list (kept language-neutral # above) — the template supplies the prose around # it in the admin's own language. "threshold": _AUDIT_REMEDIATION_BACKOFF_THRESHOLD, "window_days": ( _AUDIT_REMEDIATION_BACKOFF_WINDOW_DAYS ), "sample_block": sample_block, }, ), ) except Exception as exc: # noqa: BLE001 — never fail the audit run logger.warning( "jira.leaver_account_audit: notify dispatch failed for " "admin %s: %s", admin.id, exc, ) async def _auto_remove_non_group_residual( driver: Any, residual: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Strip org-level product access for each non-group residual leaver via the Organizations remove-access API (needs the UNSCOPED org key). This is what finally clears Trello workspace membership / org default-access / direct product grants that group removal can't — the access this audit used to be able only to *report*. Each row is annotated in-place with ``auto_removed`` (bool), ``auto_remove_status`` and (on failure) ``auto_remove_reason``. Returns ``(removed, manual)``: * ``removed`` — remove-access returned 2xx/404 (idempotent; rows may reappear for ≤24h until Atlassian's listing refreshes and will simply be removed again). These do NOT alert. * ``manual`` — no unscoped key configured, missing account_id, or the API refused (401/403 = key still scoped/revoked, or 5xx). These are the only rows that reach the admin notification. Gating on ``_org_unscoped_api_key()`` (not the scoped admin key): remove-access 401s with a scoped key, so without an unscoped key we don't fire pointless calls — we route straight to the manual bucket with actionable guidance. """ removed: list[dict[str, Any]] = [] manual: list[dict[str, Any]] = [] has_unscoped = driver._org_unscoped_api_key() is not None for row in residual: account_id = row.get("account_id") if not has_unscoped or not account_id: row["auto_removed"] = False row["auto_remove_reason"] = ( "org_unscoped_api_key not configured" if not has_unscoped else "missing account_id" ) manual.append(row) continue try: res = await driver.remove_org_user_access(account_id) except Exception as exc: # noqa: BLE001 — never fail the audit run logger.warning( "jira.leaver_account_audit: remove-access raised for %s: " "%s: %s", row.get("email"), type(exc).__name__, exc, ) row["auto_removed"] = False row["auto_remove_reason"] = f"{type(exc).__name__}: {exc}" manual.append(row) continue row["auto_remove_status"] = res.get("status") if res.get("ok"): row["auto_removed"] = True logger.info( "jira.leaver_account_audit: auto-removed org access for " "%s (status=%s, already_absent=%s)", row.get("email"), res.get("status"), res.get("already_absent"), ) removed.append(row) else: row["auto_removed"] = False row["auto_remove_reason"] = ( res.get("error") or f"HTTP {res.get('status')}" ) manual.append(row) return removed, manual async def _notify_admins_of_non_group_residual( db: AsyncSession, offenders: list[dict[str, Any]], ) -> None: """In-app push for leavers whose Atlassian ``product_access`` is NOT backed by any conferring group, so ``revoke_all_access`` (group removal) can never clear it. This is the bucket that used to masquerade as a "persistent bounce": the old audit re-queued a group revoke 3× (each a ``removed: []`` no-op), burned the backoff, then fired the misleading "SCIM keeps re-adding them" alert. There is nothing to re-add — the access simply isn't group-conferred: * **Trello** — board/workspace membership, revoked in Trello, not via Atlassian groups. * org-level **default access** / a direct product-role grant. * a ``product_access`` entry that just hasn't dropped yet after the groups were already stripped (eventual consistency). So we report it ONCE (deduped weekly, distinct ``template_key`` from the genuine-bounce alert) with accurate framing and the right manual action — never the auto-revoke loop. """ from datetime import UTC, datetime, timedelta from sqlalchemy import and_, select from app.notifications.models import NotificationOutbox from app.notifications.service import NotificationEnvelope, NotificationService from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(db) if not admins: logger.info( "jira.leaver_account_audit: %d non-group-residual leaver(s) but " "no eligible admin recipient — skipping notification.", len(offenders), ) return week_ago = datetime.now(UTC) - timedelta(days=7) # Language-neutral detail lines (see the sibling audit above): the # per-row reason comes straight off the Atlassian API response, and # the fallback is English rather than Chinese so a Japanese or # English admin isn't handed a stray Chinese fragment. sample = "\n".join( f" · {o['email']} " f"({', '.join(o.get('products') or []) or 'product access'}) — " f"{o.get('auto_remove_reason') or 'auto-revoke did not succeed'}" for o in offenders[:10] ) extra = ( f"\n · ... ({len(offenders) - 10} more)" if len(offenders) > 10 else "" ) sample_block = f"{sample}{extra}" body = ( f"{len(offenders)} 名已离职员工在 Atlassian 上仍报有产品访问,且这些" f"访问[b]不由任何 group 授予[/b](Trello / 组织默认访问 / 直接授权)。" f"Helix 未能自动清除以下这些,需要人工跟进(每行附原因):\n\n" f"{sample_block}\n\n" "最常见原因:[b]未配置 org_unscoped_api_key[/b](此时 Helix 不会尝试" "调用 API)—— 撤访问的端点不在 granular scope 体系里,带 scope 的 " "key 会 401,必须在 admin.atlassian.com/o/{orgId}/api-keys 新建一把" "[b]不选任何 scope[/b] 的 org API key 填进 Jira integration 的 " "org_unscoped_api_key。配好后下个对账周期会自动撤干净、不再打扰。\n" "若 key 已配但 API 仍拒绝(403/401)或 Trello 工作区成员制清不掉:" "在 trello.com / Atlassian Admin UI 里手动移出工作区或停用账号。" ) notif = NotificationService(db) for admin in admins: already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "jira.leaver_account_audit.non_group_residual", NotificationOutbox.created_at >= week_ago, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: continue try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=( f"[治理] {len(offenders)} 名离职员工 Atlassian 残留" f"访问非 group 授予(需人工处理,Helix 无法自动清)" ), body_text=body, template_key="jira.leaver_account_audit.non_group_residual", related_resource_type="offboarding", related_resource_id=None, context={ "offender_count": len(offenders), "samples": [o["email"] for o in offenders[:20]], "sample_block": sample_block, }, ), ) except Exception as exc: # noqa: BLE001 — never fail the audit run logger.warning( "jira.leaver_account_audit: non-group-residual notify " "dispatch failed for admin %s: %s", admin.id, exc, ) async def _auto_suspend_stale_accounts( driver: Any, rows: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Suspend sign-in for leavers nobody has decided to keep. Returns ``(suspended, needs_decision)``. A row lands in ``needs_decision`` — and therefore in front of a person — when: * a retention decision covers the account (``exemption`` set); * the integration has no unscoped org key, so Helix physically cannot perform the write (naming the missing credential beats a generic failure); * or the API refused. Never raises: the audit's job is to report the truth about every leaver, and one account's write failing must not cost the run. """ suspended: list[dict[str, Any]] = [] needs_decision: list[dict[str, Any]] = [] has_key = bool(driver._org_unscoped_api_key() and driver._org_id()) for row in rows: if row.get("exemption"): row["suspend_skipped_reason"] = "retention_decision_on_file" needs_decision.append(row) continue if not has_key: row["suspend_skipped_reason"] = "org_unscoped_api_key not configured" needs_decision.append(row) continue account_id = row.get("account_id") if not account_id: row["suspend_skipped_reason"] = "missing account_id" needs_decision.append(row) continue try: res = await driver.set_org_user_access_suspended( str(account_id), suspended=True, ) except Exception as exc: # noqa: BLE001 — never fail the audit run logger.warning( "jira.leaver_account_audit: suspend-access raised for %s: " "%s: %s", row.get("email"), type(exc).__name__, exc, ) row["suspend_skipped_reason"] = f"{type(exc).__name__}: {exc}" needs_decision.append(row) continue if res.get("ok"): row["suspended"] = True row["suspend_status"] = res.get("status") logger.info( "jira.leaver_account_audit: suspended leaver account %s " "(status=%s)", row.get("email"), res.get("status"), ) suspended.append(row) else: row["suspend_skipped_reason"] = ( res.get("error") or f"HTTP {res.get('status')}" ) needs_decision.append(row) return suspended, needs_decision async def _notify_admins_of_stale_active_accounts( db: AsyncSession, offenders: list[dict[str, Any]], *, grace_days: int, ) -> None: """In-app push when leavers' Atlassian ACCOUNTS are still active long after termination — i.e. the org's IdP directory sync has stopped deactivating them. Distinct from every other alert in this file, and deliberately so. The others are about *access Helix can revoke*; this one is about a lifecycle step Helix does not own and by policy never performs (it keeps the account so issue history stays attributed). The only thing Helix can do is notice, and until this alert existed it could not even do that: the 2026-08-13 Google-Workspace provisioning outage ran 16 days while this cron reported clean runs, because a leaver with zero ``product_access`` and a live account looked identical to a properly-offboarded one. Deduped weekly on its own ``template_key`` so a genuinely broken sync nags at a survivable cadence instead of daily. """ from datetime import UTC, datetime, timedelta from sqlalchemy import and_, select from app.notifications.models import NotificationOutbox from app.notifications.service import NotificationEnvelope, NotificationService from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(db) if not admins: logger.info( "jira.leaver_account_audit: %d stale-active leaver account(s) but " "no eligible admin recipient — skipping notification.", len(offenders), ) return week_ago = datetime.now(UTC) - timedelta(days=7) # Language-neutral detail lines (same reasoning as the sibling # alerts): the payload is emails and day counts, so an English or # Japanese admin isn't handed a stray Chinese fragment mid-list. sample = "\n".join( f" · {o['email']} — 离职 {o['termination_date']}" f" ({o['days_since_termination']}d)" for o in offenders[:10] ) extra = ( f"\n · ... ({len(offenders) - 10} more)" if len(offenders) > 10 else "" ) sample_block = f"{sample}{extra}" oldest = max( (o.get("days_since_termination") or 0) for o in offenders ) body = ( f"{len(offenders)} 名已离职员工的 Atlassian [b]账号本身[/b]在离职 " f"{grace_days} 天后仍是 active(最久的一个已 {oldest} 天):\n\n" f"{sample_block}\n\n" "[b]这不是 Helix 的撤权失败[/b] —— Helix 按既定策略只移除 group 与" "产品访问,从不停用 Atlassian 账号(保留 issue 历史归属)。真正把离职" "者账号置为 inactive 的是组织侧的[b]身份目录同步[/b]" "(Atlassian Guard 的 Google Workspace / SCIM 用户供给)。它连续多天" "没有生效,说明那条链路断了。\n\n" "排查:admin.atlassian.com → Security → User security → Identity " "providers → 对应目录,确认 [b]Provisioning 是否为 Disabled[/b]、" "Last sync 是否停在某一天、以及是否有同步报错。" "⚠️ 该设置页的『保存』会一次性提交整个表单(同步开关、邀请邮件文案、" "登录策略),改任何一项都可能连带改掉同步开关 —— 保存前逐项确认。" ) notif = NotificationService(db) for admin in admins: already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "jira.leaver_account_audit.stale_active_account", NotificationOutbox.created_at >= week_ago, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: continue try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=( f"[治理] {len(offenders)} 名离职员工的 Atlassian 账号" f"仍未停用 —— 身份目录同步可能已断" ), body_text=body, template_key=( "jira.leaver_account_audit.stale_active_account" ), # Land the reader on the page that shows every leaver's # end state, not just the ones in this alert — the # count in a notification ages, the page does not. url="/admin/offboarding?tab=assurance", related_resource_type="offboarding", related_resource_id=None, context={ "offender_count": len(offenders), "grace_days": grace_days, "oldest_days_since_termination": oldest, "samples": [o["email"] for o in offenders[:20]], "sample_block": sample_block, }, ), ) except Exception as exc: # noqa: BLE001 — never fail the audit run logger.warning( "jira.leaver_account_audit: stale-active notify dispatch " "failed for admin %s: %s", admin.id, exc, ) # Cap on how many recently-revoked task rows we'll examine per run. # A normal company processes <50 revoke tasks/day; 1000 is safe headroom. _RECENTLY_REVOKED_AUDIT_MAX = 1000 # How far back to look. 30 days covers a normal sprint of admin # actions; long enough to catch a slow SCIM-bounce reinstatement, # short enough to keep the join table size sane. Per-tick configurable # via job ``config['lookback_days']`` if an admin needs more. _RECENTLY_REVOKED_AUDIT_LOOKBACK_DAYS = 30 @register_scheduled_job( key="jira.recently_revoked_audit", display_name="Jira · 在职员工撤销反弹自动重撤 + 持续反弹兜底", description=( "active-employee 兜底的 jira.leaver_account_audit:扫最近 N 天" "admin-revoke / 自助取消已完成的在职员工,逐个查 Atlassian " "product_access。SCIM / default-group 反弹 / 管理员误操作导致" "「已撤销但仍占座席」的,**自动入队 revoke_product_access 重新撤销**——" "状态已经确定(原任务已 completed),无需人工确认。同一 (员工, 产品族) " "在 7 天内已被自动重撤 3 次仍反弹的,才升级为 IT 管理员告警(说明源头" "持续把人加回来,需要排根因)。只在持续反弹时才打扰人。" ), ) async def run_jira_recently_revoked_audit( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Auto-remediate recent admin-revoke / self-cancel actions that didn't stick. The 2026-05-28 product-access revoke PR ensures Atlassian-side group removal succeeds at task-completion time. Edge cases that could re-instate access AFTER the task completed: * SCIM user-provisioning (Okta / Google IdP) reasserting the user's group set on next sync; * Atlassian "default groups" auto-add on next user sign-in; * Atlassian Access seamless integration policies; * An admin manually re-adding the user to a conferring group; * Direct ``products=[...]`` invite via the Org Admin API that bypasses group membership (the 2026-05-28 audit's paranoid case). Detection strategy: pull the same Atlassian managed-account roster ``jira.leaver_account_audit`` uses; cross-reference it against the set of (user_id, product_key) pairs that the dispatcher recently completed a ``revoke`` task for. Anyone still in ``product_access`` for a product the platform thought was revoked triggers an automatic ``target_type='product_access'`` revoke task (idempotent ``revoke_product_access``) — same code path the self-cancel UI uses. Only ACTIVE employees are included (leavers are covered by ``jira.leaver_account_audit``). Admin only gets a notification when the same (user, product family) has been auto-revoked ≥ ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in the backoff window: at that point the source side is actively bouncing and further auto-revokes are wasted churn — a human has to fix SCIM / default-group config. """ from datetime import timedelta from sqlalchemy import select as _select from app.core.coordinators.jira_seat_coordinator import load_jira_driver from app.domains.provisioning.models import ( ACTION_GRANT, ACTION_REVOKE, STATUS_COMPLETED, ProvisioningTask, ) from app.identity.person_models import IdentityPerson from app.integrations.atlassian_activity import ( product_family_for_access_key, product_family_for_group, ) driver = await load_jira_driver(db) if driver is None: return { "users_scanned": 0, "still_holding_access": 0, "offenders": [], "errors": [], "skipped": ["Jira integration not configured or not enabled"], } if not driver._org_id() or not driver._org_admin_api_key(): return { "users_scanned": 0, "still_holding_access": 0, "offenders": [], "errors": [], "skipped": [ "Jira integration missing org_id / org_admin_api_key — the " "managed-account roster is read via the Org Admin API." ], } # Configurable lookback (defaults to 30 days). lookback_days = config.get("lookback_days") or _RECENTLY_REVOKED_AUDIT_LOOKBACK_DAYS try: lookback_days = int(lookback_days) except (TypeError, ValueError): lookback_days = _RECENTLY_REVOKED_AUDIT_LOOKBACK_DAYS cutoff = datetime.now(UTC) - timedelta(days=lookback_days) # Collect recent revoke task → expected absent (user_id, product_key) # pairs. We look at BOTH target_type='product_access' (new path) AND # target_type='account' (Jira leaver scorched-earth) — the latter # because if a leaver gets re-hired and re-cancelled, the account # task path may produce the same bounce class. task_rows = list((await db.execute( _select(ProvisioningTask).where( ProvisioningTask.action == ACTION_REVOKE, ProvisioningTask.target_type.in_(("product_access", "account")), ProvisioningTask.status == STATUS_COMPLETED, ProvisioningTask.completed_at.is_not(None), ProvisioningTask.completed_at >= cutoff, ) .order_by(ProvisioningTask.completed_at.desc()) .limit(_RECENTLY_REVOKED_AUDIT_MAX) )).scalars().all()) # Map user_id → set of canonical product **families** their recent # task(s) targeted. We NORMALISE through ``product_family_for_access_key`` # on both sides (task context_data → family AND Atlassian # ``product_access[].key`` → family) so an admin who types a future- # Atlassian access_key like ``"jira-service-management"`` (vs the # current ``"jira-servicedesk"``) still intersects correctly — the # family resolves to the same canonical string. Without the # normalisation, the audit would silently miss false negatives on # any non-canonical binding (Round 2 P2 finding). # # For ``target_type='account'`` (whole-provider revoke), the # expected state is ZERO product_access — represented by sentinel # ``*``. The sentinel can't collide with a real Atlassian key # because ``product_family_for_access_key`` only returns # ``jira-software`` / ``jira-servicedesk`` / ``confluence`` / # ``None``. by_user: dict[uuid.UUID, set[str]] = {} # Track the LATEST revoke completion per (user, family) so we can # cross-check against a possibly-later grant. Without this, a # legitimate re-hire / re-onboarding sequence (revoke → grant in # the same lookback window) flares as a "SCIM bounce" — every # rehired employee within 30 days would page IT admins on the # next audit tick. The user explicitly called out # "no 疯狂告警" / "false-positive alerts on rehires" as the failure # mode to prevent in Round-55. latest_revoke_at: dict[tuple[uuid.UUID, str], datetime] = {} for t in task_rows: ctx = t.context_data or {} completed = t.completed_at # NOT NULL — query filter ensures this if t.target_type == "product_access": pk = str(ctx.get("product_key") or "").strip().lower() if pk: # Normalise through the family helper so the audit # intersection is on canonical families on BOTH sides. family = product_family_for_access_key(pk) or pk by_user.setdefault(t.user_id, set()).add(family) key = (t.user_id, family) prior = latest_revoke_at.get(key) if prior is None or completed > prior: latest_revoke_at[key] = completed elif t.target_type == "account": # Whole-provider revoke — every product matters. by_user.setdefault(t.user_id, set()).add("*") key = (t.user_id, "*") prior = latest_revoke_at.get(key) if prior is None or completed > prior: latest_revoke_at[key] = completed if not by_user: return { "users_scanned": 0, "still_holding_access": 0, "offenders": [], "errors": [], } # Look up COMPLETED grant tasks since the earliest revoke for any # of these users — these are the legitimate re-onboarding / # re-approval events that must NOT be reported as bounces. The # query is bounded to the same lookback window because # `latest_revoke_at`'s timestamps all sit in [cutoff, now], so any # superseding grant must also sit in that range. grant_rows = list((await db.execute( _select(ProvisioningTask).where( ProvisioningTask.action == ACTION_GRANT, ProvisioningTask.target_type.in_(("product_access", "account")), ProvisioningTask.status == STATUS_COMPLETED, ProvisioningTask.completed_at.is_not(None), ProvisioningTask.completed_at >= cutoff, ProvisioningTask.user_id.in_(by_user.keys()), ) )).scalars().all()) # For each (user, family), track the LATEST grant completion. A # whole-provider grant (target_type=account) supersedes every # family for that user — captured as the "*" key, which the # downstream filter respects via the same sentinel as revokes. latest_grant_at: dict[tuple[uuid.UUID, str], datetime] = {} for g in grant_rows: ctx = g.context_data or {} completed = g.completed_at if g.target_type == "product_access": pk = str(ctx.get("product_key") or "").strip().lower() if pk: family = product_family_for_access_key(pk) or pk key = (g.user_id, family) prior = latest_grant_at.get(key) if prior is None or completed > prior: latest_grant_at[key] = completed elif g.target_type == "account": key = (g.user_id, "*") prior = latest_grant_at.get(key) if prior is None or completed > prior: latest_grant_at[key] = completed def _grant_supersedes(user_id: uuid.UUID, family: str) -> bool: """Return True if a successful grant for (user, family) ran AFTER the latest revoke for the same pair — i.e. the user was legitimately re-granted access, not bounced by SCIM. Also treats a whole-provider grant (account-level) as superseding every family for that user.""" revoked_at = latest_revoke_at.get((user_id, family)) if revoked_at is None: return False granted_at = latest_grant_at.get((user_id, family)) if granted_at is not None and granted_at > revoked_at: return True whole_provider_grant = latest_grant_at.get((user_id, "*")) if whole_provider_grant is not None and whole_provider_grant > revoked_at: return True return False # Filter to ACTIVE employees only (leavers covered by the other audit). person_rows = list((await db.execute( _select(IdentityPerson) .where( IdentityPerson.id.in_(by_user.keys()), IdentityPerson.employment_status == "active", IdentityPerson.work_email.is_not(None), ) )).scalars().all()) persons_by_id = {p.id: p for p in person_rows} # ONE paginated pull of the managed accounts. Raises on hard # failure (auth / 5xx) → run lands FAILED, dedup'd notification. raw_accounts = await driver.list_org_users_with_activity() access_by_email: dict[str, dict[str, Any]] = {} for u in raw_accounts: if not isinstance(u, dict): continue rec = _extract_managed_account(u) if rec["email"]: access_by_email[rec["email"]] = rec # Families a connector binding has declared NOT individually # revocable (``seat_revocable=false``): org-wide entitlements (e.g. # Confluence conferred via ``employee_in_china`` / ``All users from # G Suite``) and shared-billing add-ons. A per-seat revoke against # these can never clear ``product_access`` — the org-wide identity # group keeps conferring the product and SCIM re-adds it — so a # "bounce" is the permanent steady state, not an incident. Chasing # it re-revoked 13 active employees ≥3× and paged IT every week # (2026-06-15 incident). Resolve revocability exactly as the # idle-reclaim job does (plan → binding → family) so the two agree. non_revocable_families: set[str] = set() try: from app.core.coordinators.jira_seat_coordinator import ( get_jira_plans, get_plan_binding, ) from app.scheduled_jobs.idle_reclaim_handlers import _seat_revocable for _plan in await get_jira_plans(db): _binding = await get_plan_binding(db, _plan) if not _binding: continue _fam = product_family_for_group( str(_binding.get("group_name") or ""), ) if _fam and not _seat_revocable(_binding): non_revocable_families.add(_fam) except Exception: # noqa: BLE001 # A resolution failure must not blind the audit to genuine # bounces on revocable products — log and proceed without the # gate (degrades to pre-fix behaviour for this run only). logger.exception( "jira.recently_revoked_audit: could not resolve seat_revocable " "families; proceeding without the org-wide-entitlement gate", ) offenders: list[dict[str, Any]] = [] for user_id, expected_absent in by_user.items(): person = persons_by_id.get(user_id) if person is None: # Either the user is a leaver (covered elsewhere) or the # user was deleted. Skip — not this job's concern. continue email = (person.work_email or "").strip().lower() rec = access_by_email.get(email) if rec is None: # No Atlassian account → trivially absent. The revoke # succeeded at the account level (no account = no # product_access). Skip. continue if not rec["product_access"]: # All product_access entries gone — the revoke really stuck. continue # Bounce candidates: product_access entries whose family matches # something the user's recent task expected to be absent. held_products = set() for p in rec["product_access"]: if not isinstance(p, dict): continue family = product_family_for_access_key(p.get("key")) if family: held_products.add(family) if "*" in expected_absent: # Whole-provider revoke (leaver re-hire / account task) — # ANY remaining product_access is a bounce, UNLESS a later # grant superseded the whole-provider revoke (legitimate # rehire flow). if _grant_supersedes(user_id, "*"): continue bounce_products = held_products else: # Drop any family the user was legitimately re-granted # after the revoke. Pre-Round-55 this set was uncorrected # and would alert-storm on every rehired employee whose # revoke + grant pair sat in the lookback window. true_bounce = { f for f in (held_products & expected_absent) if not _grant_supersedes(user_id, f) } bounce_products = true_bounce # Drop families declared not individually revocable # (``seat_revocable=false`` → org-wide entitlement / shared- # billing add-on). A per-seat revoke can't clear their # ``product_access``, so their "bounce" is expected, not an # incident — re-revoking churns and the weekly persistent-bouncer # page is pure noise for an unfixable condition. if non_revocable_families: bounce_products = { f for f in bounce_products if f not in non_revocable_families } if not bounce_products: continue # Confirm the bounce is REAL before re-revoking / alerting: the # user must be back in a conferring group for the family. # ``revoke_product_access`` only removes the family's conferring # groups (named *-software-* / *-servicedesk-* / *-confluence-*), # so classifying the user's CURRENT groups via # ``product_family_for_group`` is the right test — a family that # shows in product_access with no matching conferring group is a # stale/lagging entry or non-group (admin/default) access that # re-revoking can't clear anyway, not a re-grant. ``None`` (lookup # failed) → stay conservative and keep the candidates. site_groups = await _account_site_groups(driver, rec["account_id"]) if site_groups is not None: conferring_fams = { fam for g in site_groups if (fam := product_family_for_group(g)) is not None } bounce_products = {f for f in bounce_products if f in conferring_fams} if not bounce_products: continue offenders.append({ "user_id": str(user_id), "email": email, "name": person.display_name, "account_id": rec["account_id"], "account_status": rec["account_status"], "expected_absent_products": sorted(expected_absent), "still_holding_products": sorted(bounce_products), }) notification_failures: list[dict[str, str]] = [] was_capped = len(task_rows) >= _RECENTLY_REVOKED_AUDIT_MAX auto_queued: list[dict[str, Any]] = [] persistent: list[dict[str, Any]] = [] remediation_errors: list[dict[str, str]] = [] if offenders: await _auto_remediate_revoke_bouncers( db, driver_integration_id=uuid.UUID(driver.integration_id), offenders=offenders, persons_by_id=persons_by_id, auto_queued=auto_queued, persistent=persistent, errors=remediation_errors, ) summary: dict[str, Any] = { "users_scanned": len(by_user), "active_users_with_recent_revoke": len(persons_by_id), "still_holding_access": len(offenders), "offenders": offenders, "lookback_days": lookback_days, # Auto-remediation buckets — same shape as the leaver audit's # summary so dashboards can render both with one template. "auto_queued": auto_queued, "persistent": persistent, "remediation_errors": remediation_errors, # P2 (Round 3 audit): the task scan capped at # ``_RECENTLY_REVOKED_AUDIT_MAX``. Surface ``was_capped`` so a # real incident that overshoots the cap shows up in the # run-log row rather than silently truncating to the latest N # tasks. The handler raises below if capped, so the run lands # ``status='failed'`` and the failure-notifier pages ops — # capping is a configuration footgun, not a benign skip. "was_capped": was_capped, "task_cap": _RECENTLY_REVOKED_AUDIT_MAX, "errors": remediation_errors, "notification_failures": notification_failures, } if offenders: logger.warning( "jira.recently_revoked_audit: %d active employee(s) re-bounced " "to product_access after admin-revoke " "(auto-queued=%d, persistent=%d): %s", len(offenders), len(auto_queued), len(persistent), ", ".join(o["email"] for o in offenders[:20]), ) if persistent: await _notify_admins_of_persistent_revoke_bouncers( db, persistent, failures=notification_failures, ) await write_sync_audit( db, job_key="jira.recently_revoked_audit", summary=summary, ) await db.commit() # P1 (Round 3 audit): raise on capped scan OR dispatched-but-undelivered # notifications. Both are "the cron technically returned but the # operator never learned about something they need to know" — # exactly the silent-degrade pattern memory # ``feedback_handler_status_must_raise`` codifies. By raising we # turn the run ``status='failed'`` which trips the standard # cron-failure notifier (different surface than the # admin-bouncer-pings dedup-aware push, so the platform team gets # paged even if individual admins didn't). Auto-remediation # enqueue failures (new) are surfaced the same way — a silent # enqueue drop = persistent bouncer unnoticed. blocker_msgs: list[str] = [] if was_capped: blocker_msgs.append( f"task scan capped at {_RECENTLY_REVOKED_AUDIT_MAX} — " f"older bounces in last {lookback_days}d may be missed; " "lower lookback_days or raise the cap" ) if notification_failures: blocker_msgs.append( f"{len(notification_failures)} admin notification(s) failed " f"to dispatch — sample: " f"{notification_failures[0]['error'][:120]}" ) if remediation_errors: blocker_msgs.append( f"{len(remediation_errors)} remediation enqueue(s) failed " f"— sample: {remediation_errors[0]['error'][:120]}" ) if blocker_msgs: raise RuntimeError( "jira.recently_revoked_audit: " + " ; ".join(blocker_msgs) ) return summary async def _auto_remediate_revoke_bouncers( db: AsyncSession, *, driver_integration_id: uuid.UUID, offenders: list[dict[str, Any]], persons_by_id: dict[uuid.UUID, Any], auto_queued: list[dict[str, Any]], persistent: list[dict[str, Any]], errors: list[dict[str, str]], ) -> None: """Per-(offender, product-family) auto-remediation via the provisioning pipeline. Each bouncing product family becomes ONE ``target_type='product_access'`` task whose dispatcher call lands on ``revoke_product_access`` — same code path the active-employee self-cancel UI uses, idempotent. A family that has already been auto-revoked ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in the backoff window is bucketed as 'persistent' (admin notification) instead; one with a still-in-flight audit task is bucketed as 'in_flight_skipped' (let the existing one finish). The audit may surface a "*" sentinel family (whole-provider revoke bouncer — e.g. a leaver who got re-hired and re-cancelled). For that case we enqueue a ``target_type='account'`` task instead, landing on ``revoke_all_access`` (the leaver path), since "everything must go" can't be expressed as one product_key. """ from app.domains.provisioning.models import SOURCE_AUDIT as _SRC_AUDIT from app.domains.provisioning.schemas import ProvisioningTaskCreate from app.domains.provisioning.service import ProvisioningService svc = ProvisioningService(db) for off in offenders: try: user_uuid = uuid.UUID(off["user_id"]) except (KeyError, ValueError) as exc: err = f"bad user_id on offender: {type(exc).__name__}: {exc}" errors.append({"email": off.get("email", "?"), "error": err}) off["remediation"] = "error" off["remediation_error"] = err continue person = persons_by_id.get(user_uuid) if person is None: off["remediation"] = "skipped_no_person" continue families = list(off.get("still_holding_products") or []) if not families: # Defensive: a bounce candidate with no products to revoke # would loop on the next audit run — surface and skip. off["remediation"] = "skipped_no_family" continue per_family_results: list[dict[str, Any]] = [] any_queued = False any_persistent = False for family in families: if family == "*": target_type = "account" product_key_for_dedup: str | None = None else: target_type = "product_access" product_key_for_dedup = family prior = await _recent_audit_remediation_tasks( db, user_id=user_uuid, target_type=target_type, target_id=driver_integration_id, product_key=product_key_for_dedup, ) if len(prior) >= _AUDIT_REMEDIATION_BACKOFF_THRESHOLD: per_family_results.append({ "family": family, "outcome": "persistent", "prior_remediation_count": len(prior), }) any_persistent = True continue if _has_in_flight_audit_task(prior): per_family_results.append({ "family": family, "outcome": "in_flight_skipped", "prior_remediation_count": len(prior), }) continue ctx_data: dict[str, Any] = { "audit_job": "jira.recently_revoked_audit", "audit_attempt": len(prior) + 1, "revoke_reason": ( f"auto-remediation: active-employee {family} access " "bounced back after admin-revoke / self-cancel" ), "account_id": off.get("account_id"), } if target_type == "product_access": # Dispatcher routes to ``revoke_product_access`` which # reads ``ctx.extra['product_key']`` to scope the # group removal. ctx_data["product_key"] = family try: # Per-family SAVEPOINT — create_task does db.add + db.flush # on the shared session; one bad flush would leave the # session in PendingRollbackError and silently drop the # remaining families/offenders (their bounced access is # never re-revoked). The savepoint rolls back just this # family on failure. async with db.begin_nested(): task = await svc.create_task(ProvisioningTaskCreate( user_id=user_uuid, software_id=None, action="revoke", target_type=target_type, target_id=driver_integration_id, source=_SRC_AUDIT, context_data=ctx_data, )) except OperationalError: # Deadlock (1213) rolls back the WHOLE transaction # server-side — propagate for the fresh-session # lock-retry wrapper. raise except Exception as exc: # noqa: BLE001 err = f"{type(exc).__name__}: {exc}" logger.exception( "jira.recently_revoked_audit: enqueue failed for " "%s/%s: %s", off.get("email"), family, err, ) per_family_results.append({ "family": family, "outcome": "error", "error": err, }) errors.append({ "email": off.get("email", "?"), "family": family, "error": err, }) continue per_family_results.append({ "family": family, "outcome": "queued", "task_id": str(task.id), "prior_remediation_count": len(prior), }) any_queued = True off["remediation_by_family"] = per_family_results # Bucketing rule: # * any family escalated to persistent → user lands in # ``persistent`` (the bell wakes a human for THIS user). # * else any family got queued → user lands in # ``auto_queued`` (quiet success). # * else (all families in_flight_skipped / error) → already # counted in errors or ignored. if any_persistent: off["remediation"] = "persistent" persistent.append({**off}) elif any_queued: off["remediation"] = "queued" auto_queued.append({**off}) else: off["remediation"] = "in_flight_skipped" async def _notify_admins_of_persistent_revoke_bouncers( db: AsyncSession, offenders: list[dict[str, Any]], *, failures: list[dict[str, str]], ) -> None: """In-app push to IT admins listing active employees whose Atlassian access has been auto-revoked ≥ ``_AUDIT_REMEDIATION_BACKOFF_THRESHOLD`` times in the backoff window and STILL bounces back. This is the rare path — the regular case (bounce detected → auto-revoke task enqueued) is silent. Only persistent bouncers reach the admin's bell because at that point Helix can't fix them: the source side (SCIM directory, default-group config, an admin re-adding the user) needs human attention. Deduped weekly per admin via ``NotificationOutbox.template_key``. ``failures`` is an OUT parameter — the caller passes a list and every dispatch failure is appended to it as ``{admin_id, error}``. The caller surfaces these in the run summary AND raises so the cron lands ``status='failed'``. A silently-swallowed dispatch failure used to leave persistent bouncers undetected (Round 3 audit P1). """ from datetime import timedelta as _td from sqlalchemy import and_ as _and from sqlalchemy import select as _select from app.notifications.models import NotificationOutbox from app.notifications.service import ( NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(db) if not admins: logger.info( "jira.recently_revoked_audit: %d persistent offender(s) but " "no eligible it_admin recipient — skipping notification.", len(offenders), ) return week_ago = datetime.now(UTC) - _td(days=7) notif = NotificationService(db) sample_lines = [] for o in offenders[:10]: held = ", ".join(o.get("still_holding_products") or []) # Stitch together per-family attempt counts so the operator # sees WHICH product is sticking and HOW MANY times Helix has # already tried to revoke it. attempts = ", ".join( f"{r.get('family')}={r.get('prior_remediation_count', '?')}×" for r in (o.get("remediation_by_family") or []) if r.get("outcome") == "persistent" ) # Language-neutral: identifiers + a bare per-family "family=N×" # attempt tally. The prose that explains what the tally MEANS is # the template's job, so this block reads the same inside the # Chinese, English and Japanese bodies. sample_lines.append( f" · {o.get('email')} {o.get('name') or ''} ({held})" + (f" [{attempts}]" if attempts else "") ) if len(offenders) > 10: sample_lines.append(f" · ... ({len(offenders) - 10} more)") sample_block = "\n".join(sample_lines) body = ( f"{len(offenders)} 名在职员工 admin-revoke / 自助取消后,Atlassian " f"product_access 被 Helix 自动重撤 ≥ " f"{_AUDIT_REMEDIATION_BACKOFF_THRESHOLD} 次后仍持续反弹" f"(SCIM / default-group / 误操作把人在源头反复加回)。Helix 已" f"停止对这些 (员工, 产品) 继续自动 revoke,需要人工排根因:\n\n" + sample_block + "\n\n处理:去 Atlassian Admin UI 复查 group 来源 + 联系 SCIM / " "IdP 同步源(Okta / Google IdP / 钉钉 SCIM)定位反弹源头。" ) for admin in admins: already = ( await db.execute( _select(NotificationOutbox.id).where( _and( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "jira.recently_revoked_audit", NotificationOutbox.created_at >= week_ago, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: continue try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=( f"[治理] {len(offenders)} 名员工 Atlassian 撤销" f"持续反弹(自动重撤 ≥ " f"{_AUDIT_REMEDIATION_BACKOFF_THRESHOLD} 次仍未清干净)" ), body_text=body, template_key="jira.recently_revoked_audit", related_resource_type="provisioning_task", related_resource_id=None, context={ "offender_count": len(offenders), "samples": [o["email"] for o in offenders[:20]], "threshold": _AUDIT_REMEDIATION_BACKOFF_THRESHOLD, "sample_block": sample_block, }, ), ) except Exception as exc: # noqa: BLE001 # Round 3 audit P1: collect into ``failures`` so the # caller can include this in the run summary AND raise. # A silently-swallowed dispatch glitch = a persistent # SCIM bouncer never reaching the operator's eyes. logger.warning( "jira.recently_revoked_audit: notify dispatch failed for " "admin %s: %s", admin.id, exc, ) failures.append({ "admin_id": str(admin.id), "error": f"{type(exc).__name__}: {exc}", }) # ───────────────────────────────────────────────────────────────── # IdP directory-sync heartbeat # ───────────────────────────────────────────────────────────────── # # The dependency this watches is invisible by construction: Atlassian's # Google Workspace / SCIM user provisioning runs on Atlassian's side, on # Atlassian's schedule, and the only evidence it is alive is that # something keeps happening. Nobody notices an absence — which is why # 2026-08-13's silent shutdown ran 16 days. # # The fix is to make the absence into a positive signal with a deadline. # ``scim_auto_sync_complete`` lands every 4h when the sync is healthy, so # a beat older than the expected period means the chain is down — # whether or not anyone happens to be leaving that week. That # independence from leaver traffic is the whole point: the existing # stale-account watchdog can only fire once a victim has aged past its # grace window, and in August there would have been none for the first # several days. _IDP_SYNC_EXPECTED_PERIOD_MINUTES = 4 * 60 # Grace on top of the vendor's own cadence — one missed tick is a blip # (Atlassian-side retry, clock skew); two in a row is an outage. _IDP_SYNC_STALE_MULTIPLIER = 2 @register_scheduled_job( key="jira.idp_sync_heartbeat", display_name="Atlassian · 身份目录同步心跳", description=( "盯住 Atlassian 侧的 Google Workspace / SCIM 用户供给还在不在跑。" "把离职者账号置为 inactive 的一直是这条同步,不是 Helix —— Helix 按" "既定策略只撤 group 与产品访问、从不删账号(保 issue 历史归属)。" "2026-08-13 这条同步被一次设置页保存静默关掉,16 天没人发现,期间" "所有离职者账号全活着、8-16 之后入职的人一个账号都没建。\n" "做法:读组织审计事件里的 scim_auto_sync_complete(健康时 4 小时" "一条),超过期望周期没有新的一条 → 心跳置 stale + 告警 IT,并顺带" "报出最近一次 scim_sync_disabled 是谁、什么时候关的。" "它不依赖「恰好这周有人离职」—— 链路一断就叫,这正是上次缺的那半。" ), ) async def run_jira_idp_sync_heartbeat( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Record one beat of the Atlassian↔Google directory sync.""" from app.core.coordinators.jira_seat_coordinator import load_jira_driver from app.domains.offboarding import assurance_service as assurance period = _config_int( config, "expected_period_minutes", _IDP_SYNC_EXPECTED_PERIOD_MINUTES, ) stale_after = period * _config_int( config, "stale_multiplier", _IDP_SYNC_STALE_MULTIPLIER, ) driver = await load_jira_driver(db) if driver is None or not driver._org_id() or not driver._org_admin_api_key(): # Benign config state — record ``unknown`` (never ``ok``) so the # page cannot show a green chain Helix has never actually read. res = await assurance.record_heartbeat( db, chain_key=assurance.CHAIN_ATLASSIAN_IDP_SYNC, last_beat_at=None, expected_period_minutes=stale_after, detail={"reason": "jira integration or org admin key missing"}, read_ok=False, ) await db.commit() return { **res, "errors": [], "skipped": ["Jira integration / org_admin_api_key not configured"], } beats = await driver.list_org_events( action="scim_auto_sync_complete", limit=5, ) disabled = await driver.list_org_events( action="scim_sync_disabled", limit=3, ) enabled = await driver.list_org_events( action="scim_sync_enabled", limit=3, ) last_beat = _parse_atlassian_ts( (beats[0].get("time") if beats else None), ) # An empty result is ambiguous — no beats ever, or the read failed. # ``list_org_events`` already logs failures and returns []; treat a # total absence of BOTH beats and lifecycle events as "could not # read" rather than "definitely stale", so a broken credential does # not masquerade as an outage (and vice versa). read_ok = bool(beats or disabled or enabled) detail: dict[str, Any] = { "expected_period_minutes": period, "stale_after_minutes": stale_after, "recent_beats": [b.get("time") for b in beats[:5]], "last_disabled": _describe_sync_event(disabled[0]) if disabled else None, "last_enabled": _describe_sync_event(enabled[0]) if enabled else None, } res = await assurance.record_heartbeat( db, chain_key=assurance.CHAIN_ATLASSIAN_IDP_SYNC, last_beat_at=last_beat, expected_period_minutes=stale_after, detail=detail, read_ok=read_ok, ) summary = {**res, "detail": detail, "errors": []} if res["status"] == "stale": logger.warning( "jira.idp_sync_heartbeat: Atlassian↔Google directory sync has no " "beat since %s (stale after %d min). Last disabled: %s", res.get("last_beat_at"), stale_after, detail.get("last_disabled"), ) await _notify_admins_of_dead_idp_sync(db, res, detail) await db.commit() return summary def _parse_atlassian_ts(raw: Any) -> datetime | None: """Atlassian event timestamps are ISO-8601 with a ``Z`` suffix.""" if not raw: return None text = str(raw).strip() if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) def _describe_sync_event(attrs: dict[str, Any]) -> dict[str, Any]: raw_actor = attrs.get("actor") actor: dict[str, Any] = raw_actor if isinstance(raw_actor, dict) else {} return { "at": attrs.get("time"), "actor": actor.get("email") or actor.get("name"), } async def _notify_admins_of_dead_idp_sync( db: AsyncSession, beat: dict[str, Any], detail: dict[str, Any], ) -> None: """Page IT when the directory sync has stopped. Deduped **daily**, not weekly like the other alerts in this file. Every day this chain is down is another day of leavers keeping live accounts and new hires getting none; a weekly reminder would have fired twice during the 16-day outage. """ from datetime import timedelta from sqlalchemy import and_ from app.notifications.models import NotificationOutbox from app.notifications.service import NotificationEnvelope, NotificationService from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(db) if not admins: logger.info( "jira.idp_sync_heartbeat: sync is stale but no eligible admin " "recipient — skipping notification.", ) return day_ago = datetime.now(UTC) - timedelta(days=1) disabled = detail.get("last_disabled") or {} who = disabled.get("actor") or "—" when = disabled.get("at") or "—" body = ( "Atlassian 侧的 [b]Google Workspace / SCIM 用户供给[/b]已经停止:\n\n" f" · 最后一次成功同步:{beat.get('last_beat_at') or '(从未观察到)'}\n" f" · 期望节奏:每 {detail.get('expected_period_minutes')} 分钟一次\n" f" · 最近一次被关闭:{when}(操作人 {who})\n\n" "[b]为什么这条要紧[/b]:把离职者的 Atlassian 账号置为 inactive 的" "一直是这条同步,不是 Helix —— Helix 按既定策略只撤 group 与产品" "访问、从不删账号。这条链路停着的每一天,离职的人都保留着可登录的" "账号,新入职的人则一个账号都建不出来。\n\n" "[b]怎么修[/b]:admin.atlassian.com → Security → User security → " "Identity providers → 对应目录,确认 Provisioning 不是 Disabled;" "页面上的 Last sync 会告诉你停在哪一天。\n" "⚠️ 那个设置页的『保存』会一次性提交整个表单(同步开关 / 新人邀请" "邮件文案 / 登录策略)—— 2026-08-13 那次就是去改邀请邮件文案," "保存时把同步开关一起关掉的,两条审计事件相隔 74 毫秒。" "改任何一项之前,先看清同步开关处在什么状态。\n\n" "Helix 侧的全貌在 离职管理 → 账号终态。" ) notif = NotificationService(db) for admin in admins: already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "identity_chain.idp_sync_stale", NotificationOutbox.created_at >= day_ago, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: continue try: await notif.dispatch( NotificationEnvelope( recipient=admin, title="[治理] Atlassian 身份目录同步已停止 —— 离职账号不再被自动停用", body_text=body, template_key="identity_chain.idp_sync_stale", url="/admin/offboarding?tab=assurance", related_resource_type="offboarding", related_resource_id=None, context={ # The registered template renders from these, so # every placeholder needs a printable value — # a bare ``None`` would reach the reader as the # word "None" where a date belongs. "last_beat_at": ( beat.get("last_beat_at") or "(从未观察到)" ), "disabled_at": when, "disabled_by": who, }, ), ) except Exception as exc: # noqa: BLE001 — never fail the sweep logger.warning( "jira.idp_sync_heartbeat: notify dispatch failed for admin " "%s: %s", admin.id, exc, )