"""Shared helpers for software-seat sync handlers. Five vendors (Zoom, Slack, Google, Jira, Feishu) each run a daily seat-sync handler that imports new seats and (most of them) revokes seats whose vendor-side membership has gone away. Three pre-launch hardening rules apply uniformly across them and live here: 1. ``safe_to_drift_revoke`` — refuses mass-revoke when the vendor's matched count drops below a configurable percentage of the local active seat count. The previous binary "0 matched" guard caught the empty-tenant outage but missed the more insidious "transient API truncation that returns a small subset" failure mode where a misconfigured paginator silently returns 5 of 500 users and the sync would mass-revoke 495 paid seats. The ratio guard (default 50%) holds the line until two consecutive runs agree. 2. ``write_sync_audit`` — emits one ``audit_logs`` row per run with the per-vendor summary so "why was Jane's Zoom seat revoked on April 17?" is answerable from the audit log alone, not by stitching together ``seat.reason`` and ``scheduled_job_runs``. Without this, CLAUDE.md's audit-by-design invariant (#6) is violated by every seat-sync handler simultaneously. 3. ``write_drift_event_audit`` — per-seat audit row when drift cleanup actually flips a seat to revoked. Keeps the per-row trace even when the bulk run-level summary is too coarse for forensic work. All helpers are import-safe with no domain dependencies; they must not pull from ``app.domains.*`` to keep the cross-domain rule clean. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import date, datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import and_, or_ from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import is_transaction_aborting_db_error logger = logging.getLogger(__name__) # Default ratio for the drift-revoke safety guard. Operations can # override per-handler by passing ``min_ratio=`` explicitly. The # guard is "matched_count must be >= min_ratio * existing_count" to # proceed with drift cleanup — anything below means refuse the mass # revoke and surface as an error in the run summary. _DEFAULT_MIN_RATIO = 0.5 # Per-tick floor for the "want set is empty" guard (round 7 hardening # 2026-05-05). Independent of ``_DEFAULT_MIN_RATIO`` because it answers # a different question: "did the predicate that builds the want set # legitimately empty it, or did the upstream signal that feeds the # predicate fail?". For Zoom add-on plans whose predicate is a # feature-flag check sourced from ``/users/{id}/settings``, a silent # 401 / scope-downgrade on that endpoint produces an empty want set # while the BASE roster still looks healthy — and the percentage-floor # guard above happily passes because matched_count uses base roster # size. Result: every paid add-on seat flips to revoked. Floor below # is the size of ``existing`` we'll tolerate seeing wiped without an # extra signal — anything above it triggers the empty-want guard. _DEFAULT_EMPTY_WANT_FLOOR = 5 def safe_to_drift_revoke( *, matched_count: int, existing_count: int, label: str, min_ratio: float = _DEFAULT_MIN_RATIO, want_count: int | None = None, empty_want_floor: int = _DEFAULT_EMPTY_WANT_FLOOR, ) -> tuple[bool, str | None]: """Decide whether drift cleanup is safe for this run. Returns ``(ok, reason)``. When ``ok=False`` the caller appends ``reason`` to the per-run errors list and skips the revoke loop. Three independent safety nets, in order of evaluation: 1. **Empty-vendor guard** — ``matched_count == 0`` with ``existing_count > 0`` means the vendor side returned a genuinely empty roster (credentials revoked, scope downgraded, full outage). Refuse mass-revoke. 2. **Ratio guard** — ``matched_count / existing_count`` below ``min_ratio`` means a transient paginator truncation returned a small subset; refusing protects paid seats. 3. **Empty-want guard** (when ``want_count`` is supplied) — even if the BASE vendor roster is healthy, a per-plan predicate that empties the want set while ``existing`` is large is almost always a predicate-feed failure, not a legitimate "everyone lost the feature flag" event. Refuse to mass-revoke when ``want_count == 0`` and ``existing_count > empty_want_floor``. Pass ``want_count=None`` to skip this check (legacy callers that don't track the want set). """ if existing_count <= 0: # No seats to revoke against — the question doesn't apply, # caller can run the (no-op) revoke loop without risk. return True, None if matched_count == 0: return False, ( f"{label}: vendor returned 0 matched users but Helix has " f"{existing_count} active seat(s) — refusing to mass-revoke. " "Likely causes: integration credentials revoked, API scope " "downgraded, or a transient vendor-side outage. Investigate " "before re-running." ) ratio = matched_count / max(existing_count, 1) if ratio < min_ratio: return False, ( f"{label}: vendor returned {matched_count} matched user(s) " f"vs {existing_count} active Helix seat(s) — match ratio " f"{ratio:.0%} is below the safety floor {min_ratio:.0%}. " "Refusing drift cleanup so a transient API truncation can't " "mass-revoke paid seats. Re-run after confirming the vendor " "side is fully populated." ) if ( want_count is not None and want_count == 0 and existing_count > empty_want_floor ): return False, ( f"{label}: predicate emptied the want set ({want_count} of " f"{matched_count} matched user(s) belong to this plan) while " f"{existing_count} active Helix seat(s) exist — refusing to " "mass-revoke. Likely causes: the per-user feed that feeds " "the predicate (Zoom user_settings, Google org-unit lookup, " "Feishu feature flag) failed silently. Re-run after " "confirming the predicate's data source is healthy." ) return True, None async def write_sync_audit( db: AsyncSession, *, job_key: str, summary: dict[str, Any], ) -> None: """Emit one ``audit_logs`` row summarising a seat-sync run. ``actor_type='system'`` because cron-driven; ``trace_id`` is the job_key + run timestamp so an admin can correlate a specific run via ``GET /api/v1/audit?trace_id=...`` and pull both the summary row and any per-seat drift events written by :func:`write_drift_event_audit` from the same run. The ``detail`` JSON is the same shape returned to the scheduled_jobs admin page (``imported`` / ``revoked`` / ``errors`` etc) so a future query like "every revoke spike on Slack in May" runs against this single column. """ from datetime import UTC, datetime from app.audit.service import AuditService trace_id = f"{job_key}:{datetime.now(UTC).isoformat()}" # Slim the detail down — drop unboundedly long lists (per-plan # summary, errors). Keep counts + first 5 errors so the row stays # under the audit_logs.detail size budget while preserving the # signal a future audit query needs. slim_detail: dict[str, Any] = {} for k, v in summary.items(): if isinstance(v, list): slim_detail[k] = v[:5] if len(v) > 5: slim_detail[f"{k}_truncated_total"] = len(v) else: slim_detail[k] = v audit = AuditService(db) try: await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action=f"{job_key}.run_completed", status="success" if not summary.get("errors") else "partial", resource_type="scheduled_job", resource_id=job_key, detail=slim_detail, ) except Exception as e: # noqa: BLE001 # Audit failure must never break the sync run — log and # continue. The next run will still record a summary; we # don't want a transient audit-table issue to abort a 500- # employee daily reconciliation. # # …but a transaction-aborting errno (1205/1213/1305) is NOT # survivable by swallowing: the session is left pending-rollback # and every caller runs this one line before ``await db.commit()``, # so the commit raises and the run's entire seat imports / revokes / # drift stamps are discarded under a misleading error. Re-raise those # so the caller restarts on a fresh transaction — same contract the # sibling ``record_seat_reconciliation`` in this file already honours # (2026-08-06; this helper feeds 11 vendor sync crons). if is_transaction_aborting_db_error(e): raise logger.exception("write_sync_audit failed for job_key=%s", job_key) async def write_drift_event_audit( db: AsyncSession, *, job_key: str, seat_id: UUID, user_id: UUID, plan_id: UUID, event: str, detail: dict[str, Any] | None = None, ) -> None: """Per-seat audit row for a sync-driven state change. Use sparingly — one row per imported / revoked seat is fine for a 500-employee tenant (peak ~50 events/day per vendor) but pay attention to it if the company scales 10×. ``event`` is one of ``imported`` / ``revoked``. """ from app.audit.service import AuditService trace_id = f"{job_key}:seat:{seat_id}" payload: dict[str, Any] = { "user_id": str(user_id), "plan_id": str(plan_id), "event": event, } if detail: payload.update(detail) audit = AuditService(db) try: await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action=f"software_seat.sync_{event}", status="success", resource_type="software_seat_allocation", resource_id=str(seat_id), detail=payload, ) except Exception as e: # noqa: BLE001 # Same contract as ``write_sync_audit`` above — a transaction-aborting # errno must propagate rather than leave the caller's session # pending-rollback for the rest of the sweep. if is_transaction_aborting_db_error(e): raise logger.exception( "write_drift_event_audit failed seat=%s event=%s job=%s", seat_id, event, job_key, ) # Per-tick max_seats cap-change safety floor. Refuse to lower # max_seats by more than this fraction in one sync tick — Zoom's # /accounts/me/plans is normally sticky, so a 50% drop in one tick # is almost always a transient API misread (eventual-consistency # during a billing cycle change) rather than a genuine plan # downgrade. Surface it to ops instead of silently shrinking the # plan and triggering eviction loops on the coordinator. MAX_SEATS_DROP_FLOOR = 0.25 def is_safe_max_seats_change( *, current: int | None, new: int | None, ) -> tuple[bool, str | None]: """Decide whether to accept an auto-tune of ``max_seats``. Returns ``(ok, reason)``. When ``ok=False`` caller leaves the existing value alone and surfaces ``reason`` in the run summary. Allows: any increase; first-time set (current is None); decrease that's within ``MAX_SEATS_DROP_FLOOR`` of current. """ if new is None: return False, "auto-tune received None for max_seats; ignored" if new < 0: return False, f"auto-tune received negative max_seats={new}; ignored" if current is None: return True, None if new >= current: return True, None drop = (current - new) / max(current, 1) if drop > MAX_SEATS_DROP_FLOOR: return False, ( f"auto-tune wanted to drop max_seats from {current} to " f"{new} ({drop:.0%}, > floor {MAX_SEATS_DROP_FLOOR:.0%}); " "leaving current value alone. Adjust manually if this is " "intentional or wait for the next tick to confirm." ) return True, None # Pre-launch hardening 2026-05-08 (round 12): grace window for the # "just-revoked" half of the seat-sync race that was the root cause # of the duplicate active-vs-revoked rows on Kevin Lu's IT 宇宙 # screenshot. Symmetric to the existing 2h fresh-grant grace on the # drift-revoke side (zoom_handlers.py:472-477, # google_handlers.py:581-582): # # • drift-revoke side: skip seats <2h old so a sync running mid- # grant doesn't see a fresh row, miss the vendor-side mirror, # and revoke the seat we just granted. # # • import side (this constant): skip user_ids whose latest seat # row was revoked recently. Without this, the sync sees the # vendor still has the user (the vendor revoke task hasn't run # yet — workers poll every minute, drivers can take 10s, retries # can stretch to 30s+) and creates a brand-new active row, # undoing the revoke. The next sync cycle would then drift- # revoke the duplicate, but by then admin sees "I revoked this, # why is it back as active?" and a phantom revoked row # accumulates in history. # # Bumped 2026-05-28 from 2h → 25h after the shengbo.ma / sunxiaoguang # bounce: their revokes happened at 16:24 UTC and 07:28 UTC; the # Atlassian seat-sync runs at 03:00 UTC daily. With a 2h window both # revokes were past grace by the time sync ran, and the import branch # re-created the allocations (the loop the user wrote about). 25h # covers any revoke that happened in the last full daily cycle — i.e. # every revoke between two consecutive seat-sync runs is protected, # regardless of clock alignment. The bound is intentionally a bit over # 24h (vs exactly 24h) so a revoke that lands a few minutes after sync # starts is still covered when the NEXT sync runs 24h later. SEAT_SYNC_REVOKE_GRACE_HOURS = 25 def import_dedup_filter( model_cls: Any, *, now: datetime, grace_hours: int = SEAT_SYNC_REVOKE_GRACE_HOURS, ) -> Any: """SQLAlchemy WHERE clause for the seat-sync's import-dedup scope. Pass it as one of the ``where(...)`` args of the existing ``select(SoftwareSeatAllocation).where(SoftwareSeatAllocation.plan_id == plan.id, ...)`` pattern in each handler. ``model_cls`` is the ``SoftwareSeatAllocation`` class — passed in to avoid importing ``app.domains.asset.models`` here (cross-domain rule). Matches: * ``status IN ('active', 'pending')`` — the unique-index scope from migration 119; everything that's currently a live seat. * ``status = 'revoked'`` AND ``revoked_at >= now - grace_hours`` — recently-revoked rows whose vendor side may not have propagated yet. Including them in the dedup map means ``existing_by_user_id[person.id]`` is truthy and the import branch's ``continue`` skips re-importing the seat. The recipe in each handler is: 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} # Drift cleanup still iterates only currently-active seats. existing = [s for s in existing_all if s.status == "active"] """ cutoff = now - timedelta(hours=grace_hours) return or_( model_cls.status.in_(("active", "pending")), and_( model_cls.status == "revoked", model_cls.revoked_at >= cutoff, ), ) # Defensive-refusal error fragments emitted by ``safe_to_drift_revoke`` # / ``is_safe_max_seats_change``. These represent the handler doing its # job correctly (refusing a risky mass action) — not a real failure. We # strip them from the all-failed gate so a healthy vendor that happens # to be in an "empty roster transient" or "predicate-feed silent miss" # state doesn't false-fire the failure notifier every tick. # # Substring match (case-sensitive) — every fragment must appear verbatim # in the matching helper's RuntimeError-like string. Keep this list # in sync if either helper's wording changes. _DEFENSIVE_REFUSAL_FRAGMENTS: tuple[str, ...] = ( "refusing to mass-revoke", "below the safety floor", # jamf.device_sync retirement floor: the roster looked truncated or # empty, so the run declined to mark anything gone. Surfaced in # ``errors`` so it reaches the job row an operator reads, but it is a # correct safety behaviour, not a failed unit of work. "[retire-refused]", "predicate emptied the want set", "leaving current value alone", # is_safe_max_seats_change drop refuse "refusing drift cleanup", # Feishu tenant_users mode: partial 通讯录范围 drift. The driver # already isolated the failure to specific departments and walked # the rest of the tenant — surfaced for admin visibility, but it's # not a "total failure" outcome (it's an informational warning). # Without this fragment, a run with partial scope drift + zero # progress would silently raise. Incident 2026-05-28. "departments unreachable", # Feishu group_member mode: partial users/batch failure. Symmetric # to "departments unreachable" above — a single failing 50-id # chunk no longer aborts the whole group sync, but the warning # gets surfaced to admins. The driver already raises if EVERY # chunk fails (global condition), so this fragment safely # whitelists the partial case. "batches unreachable", # Feishu misconfigured binding (round-55, 2026-05-30). A connector # row with missing / empty / unknown ``operation`` is surfaced # by ``get_feishu_misconfigured_seat_sync_plans`` so admins see it # on the scheduled-jobs page — but it's a static configuration # signal, not a dispatch failure. Without this whitelist, a tenant # in steady-state mirror (Feishu is import-only by design — once # everyone is imported, every subsequent run has 0 new imports # AND 0 revokes = 0 progress) would crash the cron on every tick # if any single Feishu binding is misconfigured. The misconfig # stays in ``summary['errors']`` for visibility either way. "misconfigured Feishu seat-sync binding", # jira.seat_sync per-plan reconciliation identity mismatch # (Round-61, 2026-05-31). When the org directory holds two managed # Atlassian account_id rows with the SAME email (a real, if rare, # duplicate-email case), the handler's per-account_id bucket count # disagrees with the recorder's email-deduped observations and the # snapshot for that one plan is skipped (the jira handler already # isolated it — import/drift seat actions still ran). This is a # persistent directory-data condition, not a dispatch failure: on # a stable tenant where the daily run routinely has 0 imports AND # 0 revokes, leaving it off the whitelist makes ``raise_if_total_ # failure`` flip every steady-state run to ``failed`` and re-page # on-call on each ack-then-resurge cycle until a human edits the # directory. The mismatch stays in ``summary['errors']`` (partial # status, admin-visible) either way; a GENUINE systemic miscount # violates the identity for EVERY plan and still surfaces there. "vendor/observation identity mismatch", ) def _is_defensive_refusal(err: str) -> bool: """True when an error string was emitted by a safety-helper that intentionally refused a risky action. These are not failures.""" msg = str(err) return any(frag in msg for frag in _DEFENSIVE_REFUSAL_FRAGMENTS) def raise_if_total_failure( *, job_key: str, summary: dict[str, Any], progress_keys: tuple[str, ...], ) -> None: """Raise ``RuntimeError`` when every unit of work failed. Pre-launch P1 fix (2026-05-16). Seat-sync handlers historically accumulated per-integration / per-plan errors into a ``summary[ "errors"]`` list and returned normally. The scheduler-service success-path then wrote ``scheduled_job_runs.status='ok'``, which means the dedup-aware failure-notifier never fired — so a totally- failed run (e.g. credentials rotated on every Feishu plan) showed green in the admin console for days. The memory rule ``feedback_handler_status_must_raise.md`` codified the fix; this helper centralises it. Convention: ``progress_keys`` is the list of summary keys whose sum represents "something successful happened" (e.g. ``("seats_imported", "seats_revoked")``). If every (non-defensive) error mention fired AND none of those keys advanced the counter, the handler raises and ``_execute_job_locked`` flips the run to ``failed``. Returns silently when: * ``summary["errors"]`` is empty (nothing to escalate), * every error is a defensive refusal from ``safe_to_drift_revoke`` / ``is_safe_max_seats_change`` (those are correct safety behaviours, not failures), OR * at least one progress counter is non-zero (partial success — per-row error counts still in summary["errors"] for ops). """ errors = summary.get("errors") or [] if not errors: return # Filter out defensive-helper refusals — those are not failures. real_errors = [e for e in errors if not _is_defensive_refusal(e)] if not real_errors: return progress = 0 for k in progress_keys: try: progress += int(summary.get(k) or 0) except (TypeError, ValueError): # Non-numeric progress fields shouldn't exist, but if a # caller passes one we treat it as "no progress" and let # the raise fire on a total failure. Better to over-alarm # than to silently swallow. pass if progress > 0: return # Surface the first 3 error strings in the raised message so # failure_notifier's preview / ops Slack-DM is actionable without # a separate DB lookup. Cap each at 250 chars to keep the joined # message tidy. preview = " | ".join(str(e)[:250] for e in real_errors[:3]) raise RuntimeError( f"{job_key}: all {len(real_errors)} unit(s) of work failed and no " f"progress recorded ({', '.join(progress_keys)} all zero). " f"First errors: {preview}" ) @dataclass(slots=True) class ExternalHolderObservation: """One vendor license holder that deviates from "employee using their seat". The minimal normalised shape every seat-sync handler maps its own vendor account object into, so :func:`record_seat_reconciliation` can persist the per-holder anomaly uniformly. ``email`` is the dedup key (one row per ``(plan, email)``); ``category`` discriminates among the four row-persisted buckets (``external`` / ``leaver_residual`` / ``vendor_inactive`` / ``off_paid_site``). ``email`` is lowercased by the recorder, callers needn't pre-normalise. Name kept as ``ExternalHolderObservation`` for backward compatibility with the original 2026-05-28 design (it used to store ONLY external holders); migration 219 generalised it to all four row-persisted categories. ``site`` carries the holder's vendor site host when relevant (``off_paid_site``: the non-paid site host so the drawer can show "Cheese Wong on hibernate.atlassian.net"; ``external`` on Atlassian: the production site). """ email: str vendor_account_id: str | None = None account_status: str | None = None last_active_on: date | None = None display_name: str | None = None # Discriminator — one of ``external`` / ``leaver_residual`` / # ``vendor_inactive`` / ``off_paid_site`` / ``no_email``. Defaults # to ``external`` for back-compat with pre-migration-217 callers # (Zoom / Google / Slack at the time of writing only emit external # observations; Jira emits four). The 5th bucket ``no_email`` was # historically count-only; since 2026-07-02 a handler MAY persist # no-email holders too by emitting ``category='no_email'`` # observations with a vendor-side ``holder_key`` (there is no email # to key on — Feishu uses ``feishu-user:``) so the account # is visible/annotatable in the external-accounts directory instead # of being an anonymous counter. category: str = "external" # The Helix person this holder matched, if any. NULL for ``external`` # (definitionally no match); populated for the three matched-person # categories so the drawer can deep-link to the offboarding view. person_id: UUID | None = None # Per-observation site (overrides the per-call ``site`` arg of # :func:`record_seat_reconciliation` when present — useful for # ``off_paid_site`` where each holder may be on a different # non-paid host). site: str | None = None # Optional dedup key that OVERRIDES ``email`` as the per-(plan) # uniqueness/dedup key for this observation. Per-USER vendors leave # this None → dedup keys on ``email`` exactly as before. Per-DEVICE # vendors (Jamf MDM, where one billable seat == one managed device # and a single employee may hold several devices) set it to the # device-unique key (serial / inventory id) so two devices of the # same person stay TWO distinct anomaly rows + count as two seats, # instead of collapsing on a shared email. ``email`` is still # carried for display/person-linking. Lowercased by the recorder. holder_key: str | None = None class SeatReconciliationIdentityError(ValueError): """Raised when a caller's counts violate the reconciliation identity. The UI displays the equation ``vendor_license_holders = holders_matched + holders_external + holders_no_email`` verbatim (``web/src/components/SeatReconciliationDrawer`` — ``t('seatRecon.equation')``). If a handler computes these counts inconsistently (e.g. ``holders_matched`` as ``len(set_of_person_ids)`` while ``holders_external`` is unique-email count — the alias-collapse bug pre-2026-05-29 that produced Jira's "厂商许可 365 = 318 + 46 + 0" off-by-1), the UI shows an equation that doesn't add up and the operator loses trust in the whole reconciliation surface. Fail loudly at write time instead of persisting a snapshot that lies. The handler's surrounding loop has no try-block, so the run lands ``status='failed'`` and the failure-notifier pages ops on the next tick. """ async def record_seat_reconciliation( db: AsyncSession, *, holder_model: Any, stat_model: Any, integration_id: UUID, software_id: UUID, plan_id: UUID, provider: str, product_key: str | None, site: str | None, vendor_license_holders: int, holders_matched: int, holders_no_email: int, externals: list[ExternalHolderObservation], now: datetime, run_id: UUID | None = None, # ── Migration 217 buckets — default 0 so legacy callers (Zoom / # Google / Slack pre-2026-05-29) keep working until they migrate to # the six-bucket model. The identity check tolerates these defaults # only as long as the totals balance: ``vendor = matched + external # + no_email`` when all three new buckets are 0 (which is the # legacy-behaviour identity). holders_leaver_residual: int = 0, holders_vendor_inactive: int = 0, holders_off_paid_site: int = 0, ) -> int: """Persist this plan's reconciliation snapshot + external-holder rows. Called once per plan by every ``*.seat_sync`` handler, right after it has computed the matched / external split. Two writes: 1. **External-holder rows** (``software_external_holder``) — upsert one row per (plan, external email): bump ``last_observed_at``, refresh detail, clear ``resolved``. ``first_observed_at`` is preserved on update (set only on first insert). Then any of *this plan's* rows whose ``last_observed_at`` is older than this run is flipped to ``resolved=True`` — a holder that fell off the vendor side OR finally matched an employee. Resolved rows are kept for history. 2. **Snapshot row** (``software_seat_sync_stat``) — upsert the single per-plan headline (vendor total / matched / external / no-email) + ``synced_at`` provenance. Occupancy and scheduled-cancel counts are deliberately NOT frozen here — they're queried live from ``software_seat_allocations`` because an employee can self-cancel between syncs and a frozen number would lie. **Reconciliation identity (NON-NEGOTIABLE, since 2026-05-30).** Callers must satisfy:: vendor_license_holders == holders_matched + holders_external + holders_no_email + holders_leaver_residual + holders_vendor_inactive ``holders_off_paid_site`` is recorded as anomaly observations but is **excluded from the licensed-seat total** — sandbox + personal sites are not part of the plan's paid subscription, so the vendor doesn't bill them and the workbench must not show them in the "real seat" count. Before this rule, JSM showed 85 on Helix vs 71 in Atlassian admin because 15 sandbox/personal-site holders were being counted toward the headline. The helper enforces this identity at write time and raises :class:`SeatReconciliationIdentityError` on violation — see the class docstring for the rationale. Caller MUST classify EVERY billable holder into exactly one of the six mutually-exclusive buckets — see :func:`app.integrations.atlassian_activity.classify_holder` for the canonical precedence rule (no_email > vendor_inactive > off_paid_site > leaver_residual > external > employees_active). Count by EMAIL on the matched side (not unique persons via ``len(set_of_person_ids)``) so an employee with multiple vendor accounts (e.g. ``taichi.kusaka@`` + ``japan_tidbcloud_taichi.kusaka@`` both holding jira-software) doesn't collapse to one in the snapshot while contributing two raw vendor accounts. ``vendor_license_holders`` is the **paid-site post-email-dedupe billable-identity count** — the number the vendor's admin console shows in its license-used column, with off-paid-site holders excluded. Helix's idle-reclaim and drift-revoke loops still operate on the filtered subset (active vendor accounts on paid sites only); the count is for display, the filter is for state-changing actions. Five of the six buckets are row-persistable as ``software_external_holder`` entries (one row per ``(plan_id, holder_key)``) with a ``category`` discriminator — passed in as observations in ``externals``. ``holders_matched`` stays unpersisted (the healthy baseline, no anomaly to drill into). ``no_email`` persistence is opt-in per handler: it requires a vendor-side ``holder_key`` since there is no email to key on (Feishu emits ``feishu-user:`` rows since 2026-07-02; the other vendors remain count-only until they opt in). ``holder_model`` / ``stat_model`` are the ``SoftwareExternalHolder`` / ``SoftwareSeatSyncStat`` classes, passed in (not imported) to keep this module free of ``app.domains.*`` — same pattern as :func:`import_dedup_filter`. Returns the number of distinct external holders recorded this run (so the caller can fold it into its ``result_summary`` if it wants). DB-write failures are swallowed + logged: reconciliation bookkeeping must never abort a 500-employee daily sync. **Writes are wrapped in a SAVEPOINT via ``db.begin_nested()``** so a FK violation / unique-key collision rolls back just this block — the parent transaction (the handler's primary import/revoke work + the ``write_sync_audit`` row + the final commit) stays clean. Without the savepoint, a single bad row would poison the session and silently fail the rest of the sync. Identity violations are NOT swallowed — they're a caller-side bug and must surface immediately, not corrupt the next day's snapshot. """ # Dedup within the run (a holder can surface on multiple sites of # the same plan, or the caller may emit them through multiple code # paths). Keyed on ``email`` alone — one row per (plan, email), # category transitions in-place across runs. Built BEFORE the # identity check + savepoint so we can compare against # ``len(by_email)`` (the value the stat row will actually persist). # # Per-category counters tally how many distinct emails we observed # in each row-persisted category, so the helper can enforce the # identity column-by-column (not just totals). # ``by_email`` is keyed by the DEDUP key, which is ``holder_key`` when # the caller supplied one (per-device vendors), else ``email`` (the # historical per-user behaviour — for those vendors holder_key is None # so the key, the ``holder_key`` column, and ``email`` all coincide). by_email: dict[str, ExternalHolderObservation] = {} by_category_counts: dict[str, int] = {} for obs in externals: raw_key = obs.holder_key or obs.email if not raw_key: continue key = raw_key.strip().lower() if not key: continue prev = by_email.get(key) if prev is None: by_email[key] = obs by_category_counts[obs.category] = ( by_category_counts.get(obs.category, 0) + 1 ) else: # Email seen twice in the same run for this plan. Take the # higher-severity category (leaver_residual > vendor_inactive # > off_paid_site > external — matches the precedence order # in classify_holder) and prefer richer detail fields. cat_priority = { "no_email": 5, "leaver_residual": 4, "vendor_inactive": 3, "off_paid_site": 2, "external": 1, } winner_cat = ( obs.category if cat_priority.get(obs.category, 0) > cat_priority.get(prev.category, 0) else prev.category ) # Decrement the loser, increment the winner — only the # winning category contributes to the deduped counts. if winner_cat != prev.category: by_category_counts[prev.category] = ( by_category_counts.get(prev.category, 0) - 1 ) by_category_counts[winner_cat] = ( by_category_counts.get(winner_cat, 0) + 1 ) merged = ExternalHolderObservation( email=obs.email or prev.email, vendor_account_id=obs.vendor_account_id or prev.vendor_account_id, account_status=obs.account_status or prev.account_status, last_active_on=obs.last_active_on or prev.last_active_on, display_name=obs.display_name or prev.display_name, category=winner_cat, person_id=obs.person_id or prev.person_id, site=obs.site or prev.site, holder_key=obs.holder_key or prev.holder_key, ) by_email[key] = merged # Per-category dedup totals — what will actually be persisted as # the stat row's ``holders_external`` (since that bucket maps 1:1 # to ``category='external'`` rows). The other three new buckets # are passed in by the caller and validated separately. dedup_external = by_category_counts.get("external", 0) dedup_leaver_residual = by_category_counts.get("leaver_residual", 0) dedup_vendor_inactive = by_category_counts.get("vendor_inactive", 0) dedup_off_paid_site = by_category_counts.get("off_paid_site", 0) dedup_no_email = by_category_counts.get("no_email", 0) # Identity check — ``vendor_license_holders`` may come from one of # two sources (caller's choice): # # 1. **Vendor's own authoritative endpoint** (e.g. Atlassian's # ``/rest/api/3/license/approximateLicenseCount``) — preferred # when available. In this mode, ``vendor_license_holders`` may # legitimately diverge from the 5-bucket paid-site sum by a few # holders because the vendor's internal license_assigned # accounting and the directory-derived product_access may not # be in perfect sync (Atlassian observed drift: 1–5 holders). # # 2. **Helix-computed 5-bucket paid-site sum** (matched + external # + no_email + leaver_residual + vendor_inactive) — fallback for # providers without an authoritative license endpoint (Slack, # Zoom, Google, Feishu, Confluence) or when the vendor endpoint # fails. In this mode, ``vendor_license_holders`` MUST equal the # 5-bucket sum exactly. # # ``holders_off_paid_site`` is always recorded as anomaly observations # for the drawer's audit tab but is NEVER part of the licensed-seat # total — sandbox + personal sites are not part of the plan's paid # subscription, so neither the vendor nor Helix bills for them. # # We can't tell modes apart from the wire, so we accept either: # ``vendor_license_holders`` must satisfy ONE of # (a) == 5-bucket paid-site sum (mode 2), or # (b) close to 5-bucket sum but allowed to drift (mode 1) — # specifically, ``vendor_license_holders ≥ holders_matched`` # (a vendor can't claim FEWER licenses than Helix has actively # matched employees to) and not wildly off (|delta| ≤ 50 OR # ≤ 30% of the 5-bucket sum, whichever is larger). Beyond that # reject as caller-side bug. # # Each row-persisted bucket count passed by the caller MUST match # the deduped observation total in that category — otherwise the # stat row would disagree with the observation rows (the drawer's # per-tab count would diverge from the headline number). That part # of the identity check is unchanged. expected_paid_site_sum = ( holders_matched + dedup_external + holders_no_email + dedup_leaver_residual + dedup_vendor_inactive ) if vendor_license_holders < holders_matched: raise SeatReconciliationIdentityError( f"seat reconciliation identity violated for " f"provider={provider} plan_id={plan_id} product_key={product_key}: " f"vendor_license_holders={vendor_license_holders} < " f"holders_matched({holders_matched}). The vendor cannot bill " "for fewer licenses than Helix has actively matched employees " "to — caller-side miscount." ) drift = abs(vendor_license_holders - expected_paid_site_sum) drift_limit = max(50, int(expected_paid_site_sum * 0.30)) if drift > drift_limit: raise SeatReconciliationIdentityError( f"seat reconciliation identity violated for " f"provider={provider} plan_id={plan_id} product_key={product_key}: " f"vendor_license_holders={vendor_license_holders} differs from " f"5-bucket paid-site sum ({expected_paid_site_sum}) by " f"{drift} holders (limit {drift_limit}). " f"5-bucket = holders_matched({holders_matched}) + " f"holders_external({dedup_external}) + " f"holders_no_email({holders_no_email}) + " f"holders_leaver_residual({dedup_leaver_residual}) + " f"holders_vendor_inactive({dedup_vendor_inactive}). " f"(holders_off_paid_site={dedup_off_paid_site} excluded as " "anomaly-only.) Small vendor-vs-Helix drift is expected when " "the vendor's authoritative endpoint is the source; gross " "divergence indicates caller-side bug." ) # Caller-passed counts MUST equal the deduped observation totals for # the three new row-persisted buckets — otherwise the stat headline # would lie about what the drawer's per-tab list will display. bucket_mismatches: list[str] = [] if holders_leaver_residual != dedup_leaver_residual: bucket_mismatches.append( f"holders_leaver_residual={holders_leaver_residual} != " f"observations[category=leaver_residual]={dedup_leaver_residual}" ) if holders_vendor_inactive != dedup_vendor_inactive: bucket_mismatches.append( f"holders_vendor_inactive={holders_vendor_inactive} != " f"observations[category=vendor_inactive]={dedup_vendor_inactive}" ) if holders_off_paid_site != dedup_off_paid_site: bucket_mismatches.append( f"holders_off_paid_site={holders_off_paid_site} != " f"observations[category=off_paid_site]={dedup_off_paid_site}" ) # no_email row-persistence is OPT-IN per handler (Feishu since # 2026-07-02; the other vendors stay count-only), so the check is # one-directional: a handler may persist FEWER no-email rows than # it counted (a member with no usable vendor key can't be keyed) # but never more — that would mean the drawer tab shows rows the # headline denies exist. if dedup_no_email > holders_no_email: bucket_mismatches.append( f"holders_no_email={holders_no_email} < " f"observations[category=no_email]={dedup_no_email}" ) if bucket_mismatches: raise SeatReconciliationIdentityError( f"seat reconciliation per-bucket counts disagree with " f"persisted observations for provider={provider} " f"plan_id={plan_id}: " + " ; ".join(bucket_mismatches) + ". " "Every row-persisted observation MUST contribute to its " "category's headline counter and vice versa." ) try: # SAVEPOINT — a FK violation / unique collision / type-bind error # rolls back just this block; the parent transaction (handler's # import/revoke work + write_sync_audit + final commit) stays # clean. Without it, a single bad row poisons the session and # every subsequent statement silently fails. async with db.begin_nested(): if by_email: rows = [ { "id": str(uuid.uuid4()), "integration_id": integration_id, "software_id": software_id, "plan_id": plan_id, "provider": provider, "product_key": product_key, # Prefer per-observation site (set by the handler # for off_paid_site holders so the drawer shows # the actual non-paid host) over the per-call # site arg (the plan's site_urls summary). "site": obs.site or site, # ``holder_key`` is the dedup/uniqueness key (== # email for per-user vendors, == device key for # per-device vendors). ``email`` stays the real # (lowercased) holder email for display/linking — # may coincide across rows for a multi-device # employee, which is exactly why holder_key exists. "holder_key": dkey, "email": (obs.email or "").strip().lower() or None, "category": obs.category, "person_id": obs.person_id, "display_name": obs.display_name, "vendor_account_id": obs.vendor_account_id, "account_status": obs.account_status, "last_active_on": obs.last_active_on, "first_observed_at": now, "last_observed_at": now, "resolved": False, "resolved_at": None, "created_at": now, "updated_at": now, } for dkey, obs in by_email.items() ] ins = mysql_insert(holder_model).values(rows) ins = ins.on_duplicate_key_update( provider=ins.inserted.provider, product_key=ins.inserted.product_key, site=ins.inserted.site, email=ins.inserted.email, category=ins.inserted.category, person_id=ins.inserted.person_id, display_name=ins.inserted.display_name, vendor_account_id=ins.inserted.vendor_account_id, account_status=ins.inserted.account_status, last_active_on=ins.inserted.last_active_on, last_observed_at=ins.inserted.last_observed_at, resolved=ins.inserted.resolved, resolved_at=ins.inserted.resolved_at, updated_at=ins.inserted.updated_at, ) await db.execute(ins) # Mark previously-anomalous holders NOT seen this run as # resolved (they fell off the vendor side, were finally # offboarded, or category transitioned). Keyed on the # seen-email set rather than a ``last_observed_at < now`` # comparison: MySQL ``DATETIME`` truncates to whole seconds, # so a freshly-upserted row's stored timestamp would compare # strictly-less-than the full-precision ``now`` and wrongly # resolve itself. The email set is precision-immune. # # Category transitions (external → leaver_residual, etc.) # are NOT considered "resolved" because the email is still # in this run's observation set; the upsert above flipped # ``category`` in place. Only true disappearances (email # absent from this run entirely) flip ``resolved=True``. resolve_q = ( holder_model.__table__.update() .where( holder_model.plan_id == plan_id, holder_model.resolved == False, # noqa: E712 ) .values(resolved=True, resolved_at=now, updated_at=now) ) if by_email: # Keyed on ``holder_key`` (the dedup key) not ``email`` — # ``by_email`` keys are holder_keys, and holder_key is the # uniqueness column. For per-user vendors holder_key == email # so this is identical to the historical behaviour. resolve_q = resolve_q.where( holder_model.holder_key.notin_(list(by_email.keys())), ) await db.execute(resolve_q) # Upsert the per-plan headline snapshot. # ``holders_external`` uses the deduped count to stay in # sync with what the persisted ``category='external'`` rows # show in the drawer — the identity check above already # verified this matches the caller's intent. stat_row = { "id": str(uuid.uuid4()), "integration_id": integration_id, "software_id": software_id, "plan_id": plan_id, "provider": provider, "product_key": product_key, "vendor_license_holders": int(vendor_license_holders), "holders_matched": int(holders_matched), "holders_external": dedup_external, "holders_no_email": int(holders_no_email), "holders_leaver_residual": int(holders_leaver_residual), "holders_vendor_inactive": int(holders_vendor_inactive), "holders_off_paid_site": int(holders_off_paid_site), "synced_at": now, "run_id": run_id, "created_at": now, "updated_at": now, } sins = mysql_insert(stat_model).values([stat_row]) sins = sins.on_duplicate_key_update( provider=sins.inserted.provider, product_key=sins.inserted.product_key, vendor_license_holders=sins.inserted.vendor_license_holders, holders_matched=sins.inserted.holders_matched, holders_external=sins.inserted.holders_external, holders_no_email=sins.inserted.holders_no_email, holders_leaver_residual=sins.inserted.holders_leaver_residual, holders_vendor_inactive=sins.inserted.holders_vendor_inactive, holders_off_paid_site=sins.inserted.holders_off_paid_site, synced_at=sins.inserted.synced_at, run_id=sins.inserted.run_id, updated_at=sins.inserted.updated_at, ) await db.execute(sins) return len(by_email) except Exception as e: # noqa: BLE001 # A lock-wait/deadlock/orphaned-savepoint errno (1205/1213/1305) # aborts the WHOLE transaction server-side — the SAVEPOINT rollback # can't rescue it, and swallowing it here leaves the handler's # session pending-rollback so its later import/revoke writes + final # commit silently fail. Re-raise those so the caller restarts on a # fresh transaction. if is_transaction_aborting_db_error(e): raise # Otherwise best-effort by design: the writes above run inside a # SAVEPOINT (begin_nested) that rolls back just this reconciliation- # snapshot block on failure, leaving the handler's primary import/ # revoke work + final commit clean. The reconciliation snapshot is # reporting-only (the load-bearing grant/revoke is a separate, # raise-on-failure path), so a snapshot-write miss is logged and # swallowed rather than failing the whole seat sync. logger.exception( "record_seat_reconciliation failed provider=%s plan_id=%s", provider, plan_id, ) return 0 __all__ = [ "MAX_SEATS_DROP_FLOOR", "SEAT_SYNC_REVOKE_GRACE_HOURS", "ExternalHolderObservation", "SeatReconciliationIdentityError", "import_dedup_filter", "is_safe_max_seats_change", "raise_if_total_failure", "record_seat_reconciliation", "safe_to_drift_revoke", "write_drift_event_audit", "write_sync_audit", ]