"""Idle software-seat reclaim — ``software_seat.idle_reclaim``. Daily job that recovers paid Atlassian (Jira Software / Jira Service Management / Confluence) seats that nobody is actually using, with a warn → grace-buffer → self-service-keep → reclaim lifecycle. Scope ───── Only Helix-tracked **active** ``software_seat_allocations`` for the Jira three-piece set — i.e. seats Helix itself provisioned / can revoke. Service accounts and external consultants that aren't in HR never get a Helix seat, so they're out of scope by construction (this is the "not in HR system → never reclaim" guardrail, satisfied by the scope itself). The "is it idle?" signal ──────────────────────── Cross-referenced against the production-site activity directory built by ``app.integrations.atlassian_activity`` — which already does the hard part right: for each (person, product) it takes the MAX ``last_active`` across the *paid production* sites only, ignoring sandbox / external sites. See that module for why (the previous attempt mistook a sandbox's frozen ``last_active`` for the real one and flagged daily users as two-years-idle). Per-seat state machine (only ``active`` employees holding an ``active`` seat are considered): * holder has no matching Atlassian account / account not ``active`` / ``last_active`` for this product is blank → **undetermined**, never reclaim (record for human visibility). * holder is an Atlassian admin (org/site admin, or member of a configured admin group) → **skip** (hard guardrail). * ``idle_kept_until`` set & in the future → **skip** (self-service keep opt-out). * idle_days ≤ threshold → **in use**; reset any stale warn/manual-review state. * idle_days > threshold, not yet warned → **warn**: stamp ``idle_warned_at`` (start of buffer) + notify holder. * warned, buffer not yet elapsed → **wait**. * warned, buffer elapsed, still idle, not kept → **reclaim**: enqueue a ``target_type='product_access'`` provisioning task via the coordinator's ``enqueue_revoke_for_seat`` — the SAME path the self-cancel UI and admin-revoke use. The driver's ``revoke_product_access`` strips the holder from EVERY group that confers the product (applicationrole ∪ admin-declared), exactly like Atlassian's own "Remove access" button, so multi-group holders (``jira-dev-engineers`` / ``RD Leader`` alongside the bound ``jira-software-users``) are genuinely reclaimed instead of punted to manual review (the pre-2026-06-11 behaviour that left seats idle forever). Identity groups and other products stay intact; the Atlassian account is never disabled (that's the offboarding line, not this one). The pipeline brings retries, dead-letter ops, sibling add-on cascade, and ``revert_seat_on_failure`` (a failed vendor revoke flips the seat back to active and DMs admins) for free. All day-math is computed in the holder's local timezone (``resolve_timezone``), not a naive UTC day boundary. Threshold / buffer / keep window / admin groups / production sites are all ``platform_settings``-configurable (with hard-coded fallbacks). Failures are collected per-seat; the handler raises only when every unit of work failed (so the dedup-aware failure notifier fires) — never returns a silent ``status='failed'``. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import UTC, date, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import lazyload from app.core.exceptions import AppException from app.core.txn import best_effort from app.core.working_hours import resolve_timezone from app.scheduled_jobs._seat_sync_helpers import ( raise_if_total_failure, write_sync_audit, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # Hard-coded fallbacks — mirrored by the seeded platform_settings in # migration 204. Used when a setting row is absent / malformed. _DEFAULT_IDLE_THRESHOLD_DAYS = 90 _DEFAULT_BUFFER_DAYS = 14 _DEFAULT_KEEP_DAYS = 180 _DEFAULT_ADMIN_GROUPS = ( "jira-administrators", "site-admins", "confluence-administrators", "org-admins", ) _SETTING_THRESHOLD = "software_seat.idle_reclaim.idle_threshold_days" _SETTING_BUFFER = "software_seat.idle_reclaim.buffer_days" _SETTING_ADMIN_GROUPS = "software_seat.idle_reclaim.admin_group_names" # Cap on per-seat errors carried in the run summary (keeps the run-log # row + failure-DM preview bounded). Mirrors jira.seat_sync. _MAX_ERRORS = 30 # Deep link the warn/reclaim cards point at — the canonical IT 宇宙 # self-service page where the holder's seats + the 保留坐席 action live. _SELF_SERVICE_URL = "/me" @dataclass(slots=True) class _Config: idle_threshold_days: int buffer_days: int admin_group_names: list[str] dry_run: bool def _coerce_int(value: Any, default: int) -> int: if isinstance(value, bool) or value is None: return default if isinstance(value, (int, float)): return int(value) if isinstance(value, str): try: return int(value.strip()) except ValueError: return default return default async def _resolve_config(db: AsyncSession, config: dict[str, Any]) -> _Config: """Merge job-config overrides over platform settings over defaults.""" from app.integrations.models import PlatformSetting async def _setting(key: str) -> Any: row = await db.get(PlatformSetting, key) return row.value if row is not None else None threshold = _coerce_int( config.get("idle_threshold_days"), _coerce_int(await _setting(_SETTING_THRESHOLD), _DEFAULT_IDLE_THRESHOLD_DAYS), ) buffer = _coerce_int( config.get("buffer_days"), _coerce_int(await _setting(_SETTING_BUFFER), _DEFAULT_BUFFER_DAYS), ) admin_groups_raw = config.get("admin_group_names") if not isinstance(admin_groups_raw, list): admin_groups_raw = await _setting(_SETTING_ADMIN_GROUPS) if isinstance(admin_groups_raw, list): admin_groups = [ str(g).strip() for g in admin_groups_raw if g and str(g).strip() ] else: admin_groups = list(_DEFAULT_ADMIN_GROUPS) # Guard against a fat-fingered setting that would defeat the buffer # entirely (0 days = warn-and-reclaim-same-run). Clamp to sane floors. threshold = max(1, threshold) buffer = max(1, buffer) return _Config( idle_threshold_days=threshold, buffer_days=buffer, admin_group_names=admin_groups, dry_run=bool(config.get("dry_run", False)), ) class _EnqueueReturnedNoTask(Exception): """Raised inside the reclaim savepoint when the coordinator resolved no plan/software/connector (returned ``None``) — propagating it rolls the optimistic seat flip back with the savepoint.""" def _as_utc(dt: datetime | None) -> datetime | None: """Treat a DB-loaded naive datetime as UTC (asyncmy round-trip strips tzinfo — see memory ``mysql_datetime_naive_on_fresh_load``).""" if dt is None: return None return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) def _local_date(now_utc: datetime, tz: ZoneInfo) -> date: return now_utc.astimezone(tz).date() def _seat_revocable(binding: dict[str, Any]) -> bool: """Whether idle_reclaim may remove a single holder from this plan's bound group. Defaults to ``True`` — a real product plan (Jira Software / JSM / Confluence) whose bound group *is* that product's conferring group, so removing a holder genuinely reclaims the product. A binding sets ``seat_revocable: false`` to opt OUT: shared-billing Marketplace add-on catalogs (Tempo / ScriptRunner / eazyBI) reuse the Jira Software conferring group ``jira-software-users`` purely as a per-head billing proxy. Removing a holder from it would revoke their *actual Jira Software* access — never valid; and a tier-licensed app has no per-person seat to reclaim anyway. Such plans are skipped wholesale here, while ``jira.seat_sync`` (which never reads this field) keeps counting them by ``product_key``. Tolerates the JSON/string forms a ``binding_config`` may carry: a real bool from the driver-decoded path, a ``false``/``0`` string an admin typed in the binding form, or a JSON ``0`` from a ``JSON_SET(..., false)`` data fix. Blank / absent → the revocable default. """ raw = binding.get("seat_revocable") if raw is None: return True if isinstance(raw, bool): return raw if isinstance(raw, str): s = raw.strip().lower() if not s: return True return s not in ("false", "0", "no", "off") if isinstance(raw, (int, float)): return bool(raw) return True @register_scheduled_job( key="software_seat.idle_reclaim", display_name="软件坐席 — 闲置坐席自动回收 (Jira 三件套)", description=( "每天扫描 Helix 跟踪的 Jira Software / JSM / Confluence active 坐席," "交叉 Atlassian 生产站 last_active(每人每产品取生产站 MAX,排除 " "sandbox / 外部站点)。idle_days>阈值(默认90)且 last_active 非空 → " "警告 + 缓冲(默认14天)+ 自助『保留坐席』;缓冲后仍闲置且未保留 → " "入队 product_access 撤销任务(与自助退订/管理员撤销同一管道):" "从该产品的全部授权组移除(applicationrole ∪ admin-declared," "等价 Atlassian 后台 Remove access,多组持有者一并真实回收;" "绝不停账号、不动身份组和其他产品)+ 兄弟 add-on 座席级联 + " "失败自动回滚座席并 DM 管理员。" "护栏:管理员组成员 / Atlassian org/site admin、离职或非在职员工、" "last_active 为空一律不自动回收;binding " "标了 seat_revocable=false 的 plan(按 Jira 用户分摊计费、复用 " "jira-software-users 组的 Marketplace 插件 Tempo/ScriptRunner/eazyBI)" "整体跳过(计入 skipped),绝不把人移出 Jira 本体组。" "天数按员工本地时区算。幂等。config 支持 dry_run / idle_threshold_days " "/ buffer_days 覆盖。" ), ) async def run_software_seat_idle_reclaim( 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 SoftwareSeatAllocation from app.identity.person_models import IdentityPerson from app.integrations.atlassian_activity import ( fetch_admin_account_ids, fetch_atlassian_activity, product_family_for_group, ) cfg = await _resolve_config(db, config) now = datetime.now(UTC) summary: dict[str, Any] = { "warned": 0, "reclaimed": 0, "kept_skipped": 0, "in_buffer": 0, "reset_active": 0, "undetermined": 0, "admin_skipped": 0, "leaver_skipped": 0, "plans_scanned": 0, "seats_scanned": 0, "dry_run": cfg.dry_run, "idle_threshold_days": cfg.idle_threshold_days, "buffer_days": cfg.buffer_days, # ``errors`` = genuine API / revoke / unexpected failures only — # this is the list ``raise_if_total_failure`` reads, so it must # NOT carry benign config skips (a plan with no binding, an # unclassifiable group, a holder with no Atlassian account). Those # go in ``skipped``: informational, never failure-worthy. Mixing # them would false-page admins on a steady-state run where nobody # is idle (all progress counters 0) but one plan is unbound. "errors": [], "skipped": [], } errors: list[str] = summary["errors"] skipped: list[str] = summary["skipped"] driver = await load_jira_driver(db) if driver is None: # Benign config state — Jira simply isn't wired up here. Per this # function's ``errors``-vs-``skipped`` contract (see summary init # above) this is a ``skipped`` note, NOT an ``errors`` entry: # ``errors`` is the escalation list ``raise_if_total_failure`` # reads, and "not configured" must not page or surface as a # failure. skipped.append( "Jira integration not configured or not enabled — skipped", ) # NOT a return any more (PR4): Atlassian is one phase of this job, # not the whole job. Google / Zoom seats are evaluated from the # persisted usage signals below and must not be skipped because a # different vendor is unwired. plans = [] else: plans = await get_jira_plans(db) if not plans: skipped.append( "No active Jira license plans found — nothing to scan", ) # ── External reads up front (no open write txn) ─────────────── # These two raise on hard failure (missing org_admin_api_key, auth, # 5xx). Letting them propagate fails the whole run — correct: we must # NOT warn/reclaim against a partial directory or an incomplete admin # set. The framework turns the raise into status='failed' + a deduped # failure notification. directory = None admin_account_ids: set[str] = set() # ``driver is not None`` is implied by ``plans`` being non-empty (an # unconfigured Jira sets plans to []), but it is spelled out so the # correlation is enforced rather than assumed — and so the narrowing # is visible to the type checker instead of being a comment. if plans and driver is not None: directory = await fetch_atlassian_activity(db, driver=driver) admin_account_ids = await fetch_admin_account_ids( driver, cfg.admin_group_names, ) summary["atlassian_accounts"] = directory.total_accounts summary["admin_accounts"] = len(admin_account_ids) for plan in plans if directory is not None else []: binding = await get_plan_binding(db, plan) if binding is None: skipped.append(f"{plan.name}: no active connector binding; skipped") continue group_name = str(binding.get("group_name") or "").strip() family = product_family_for_group(group_name) if not group_name or family is None: skipped.append( f"{plan.name}: group_name {group_name!r} not a tracked Jira " "product (software/servicedesk/confluence); skipped", ) continue # Shared-billing Marketplace add-on plans (Tempo / ScriptRunner / # eazyBI) reuse the Jira Software conferring group # ``jira-software-users`` as a per-head billing proxy. A per-seat # "reclaim" on them would remove the holder from that group and # strip their *Jira Software* access — never valid (and a tier- # licensed app has no per-person seat to reclaim). Such bindings # carry ``seat_revocable: false``; skip the whole plan. Benign skip # → summary['skipped'], never errors (so a steady-state run doesn't # false-page). ``jira.seat_sync`` ignores this field and keeps # counting these plans by product_key, so the statistics口径 is # unaffected. if not _seat_revocable(binding): skipped.append( f"{plan.name}: binding seat_revocable=false (shared-billing " f"add-on / tier license reusing group {group_name!r}); " "idle_reclaim skips it wholesale — not per-seat revocable", ) continue summary["plans_scanned"] += 1 seats = list((await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, SoftwareSeatAllocation.status == "active", ) )).scalars().all()) if not seats: continue holder_ids = {s.user_id for s in seats} # ``lazyload("*")`` suppresses IdentityPerson's eager selectin # relationships (user_roles / org_memberships / external_identities) # — the idle scan only reads scalar columns (work_email, # email_aliases, employment_status, display_name, country), so the # three extra selectin round-trips per plan batch are pure waste # for a several-hundred-person scan. people = (await db.execute( select(IdentityPerson) .where(IdentityPerson.id.in_(holder_ids)) .options(lazyload("*")) )).scalars().all() people_by_id = {p.id: p for p in people} for seat in seats: summary["seats_scanned"] += 1 person = people_by_id.get(seat.user_id) acct = ( directory.lookup_by_emails( person.work_email, *(person.email_aliases or []), ) if person is not None and directory is not None else None ) # Per-seat SAVEPOINT isolation: the loop shares one session, so # without this a single seat whose autoflush fails poisons the # session and silently fails every subsequent seat. The # savepoint rolls back only the bad seat. # # ``best_effort`` rather than a raw ``begin_nested`` + bare # ``except``: the raw form swallows the errno family a savepoint # rollback CANNOT rescue (1205 lock-wait / 1213 deadlock / 1305 # orphaned savepoint abort the transaction server-side), and the # rollback itself then fails — turning one transient deadlock # into "SAVEPOINT does not exist" for every remaining seat. # ``best_effort`` re-raises exactly those so the executor's # fresh-session retry restarts the run cleanly. plan_name, seat_id = plan.name, seat.id async with best_effort( db, label=f"idle-reclaim seat {seat_id}", log=logger, ) as seat_step: await _process_seat( db, seat=seat, person=person, plan=plan, group_name=group_name, family=family, evidence=_evidence_from_atlassian( acct=acct, family=family, admin_account_ids=admin_account_ids, ), cfg=cfg, now=now, summary=summary, ) if not seat_step.ok: errors.append( f"{plan_name}/seat {seat_id}: {seat_step.reason()}" ) if len(errors) > _MAX_ERRORS: errors.append("...truncated") break # ── Phase 2: every other vendor, from the persisted usage signals ── await _run_signal_backed_phase( db, cfg=cfg, now=now, summary=summary, errors=errors, skipped=skipped, ) await write_sync_audit(db, job_key="software_seat.idle_reclaim", summary=summary) await db.commit() raise_if_total_failure( job_key="software_seat.idle_reclaim", summary=summary, progress_keys=( "warned", "reclaimed", "reset_active", "in_buffer", "kept_skipped", "undetermined", "admin_skipped", "leaver_skipped", ), ) return summary async def _signal_backed_plans(db: AsyncSession) -> list[tuple[Any, str, str]]: """``(plan, provider, product_key)`` for every non-Atlassian seat plan whose provider has a usage collector. Atlassian is excluded here because phase 1 already handled it from the live directory — running it twice would double-count and could warn a holder from one phase and reclaim them from the other. """ from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, ) from app.domains.provisioning.models import SoftwareConnector from app.integrations.models import Integration from app.integrations.usage_collectors import ( PRODUCT_GOOGLE_WORKSPACE, PRODUCT_ZOOM, ) default_product = { "google_workspace": PRODUCT_GOOGLE_WORKSPACE, "zoom": PRODUCT_ZOOM, } rows = (await db.execute( select(SoftwareLicensePlan, Integration.provider) .join( AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwareLicensePlan.software_id, ) .join( SoftwareConnector, SoftwareConnector.software_id == AssetSoftwareCatalog.id, ) .join(Integration, Integration.id == SoftwareConnector.integration_id) .where( SoftwareConnector.is_active.is_(True), Integration.enabled.is_(True), Integration.provider.in_(tuple(default_product)), ) )).all() out: list[tuple[Any, str, str]] = [] for plan, provider in rows: product = default_product.get(provider) if product is None: continue out.append((plan, provider, product)) return out async def _run_signal_backed_phase( db: AsyncSession, *, cfg: _Config, now: datetime, summary: dict[str, Any], errors: list[str], skipped: list[str], ) -> None: """Idle evaluation for vendors read from ``asset_seat_usage_signals``. Same ladder as phase 1 — literally the same ``_process_seat`` — with the evidence built from the persisted signal instead of a live directory. The freshness gate inside ``signals_for_people`` means a collector that stopped running makes this phase report everything as undetermined and reclaim nothing, rather than acting on stale rows. """ from app.domains.asset.models import SoftwareSeatAllocation from app.identity.person_models import IdentityPerson from app.integrations.usage_signal_service import signals_for_people plans = await _signal_backed_plans(db) if not plans: skipped.append( "No active Google/Zoom seat plans with a usage collector — " "signal-backed phase had nothing to scan", ) return for plan, provider, product_key in plans: summary["plans_scanned"] += 1 seats = list((await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, SoftwareSeatAllocation.status == "active", ) )).scalars().all()) if not seats: continue holder_ids = [s.user_id for s in seats] people = (await db.execute( select(IdentityPerson) .where(IdentityPerson.id.in_(holder_ids)) .options(lazyload("*")) )).scalars().all() people_by_id = {p.id: p for p in people} signals = await signals_for_people( db, holder_ids, provider=provider, now=now, ) for seat in seats: summary["seats_scanned"] += 1 person = people_by_id.get(seat.user_id) evidence = _evidence_from_signal( signal=signals.get((seat.user_id, provider, product_key)), product_key=product_key, seat=seat, ) # ``best_effort``, same reason as phase 1 above. plan_name, seat_id = plan.name, seat.id async with best_effort( db, label=f"idle-reclaim seat {seat_id}", log=logger, ) as seat_step: await _process_seat( db, seat=seat, person=person, plan=plan, group_name=provider, family=product_key, evidence=evidence, cfg=cfg, now=now, summary=summary, ) if not seat_step.ok: errors.append( f"{plan_name}/seat {seat_id}: {seat_step.reason()}" ) if len(errors) > _MAX_ERRORS: errors.append("...truncated") return @dataclass(slots=True) class _SeatEvidence: """What one vendor says about one holder, in one shape (PR4). The idle ladder below used to read Atlassian's directory object directly, which is why it could only ever reclaim Atlassian seats. It now consumes this instead, so adding a vendor means adding a builder rather than forking the state machine — the warn → buffer → self-service-keep → reclaim behaviour stays byte-identical for everyone. ``state`` is authoritative, ``last_active`` is only meaningful when it is ``active``: * ``active`` — ``last_active`` holds a real date. * ``never`` — the vendor affirmatively says this account has never signed in. Still reclaimable, but the idle clock runs from ``granted_on`` (the seat grant), because a seat handed out yesterday and not yet used is new, not idle. * ``unknown`` / ``unsupported`` — never reclaim. Counted as ``undetermined`` so the run reports how much it could not see. ``is_admin`` requires a POSITIVE determination from the vendor. Any provider that cannot say is treated as an admin and skipped: being wrong that way costs a missed reclaim, being wrong the other way strips an account owner's access. """ state: str last_active: date | None = None granted_on: date | None = None #: False when the holder has a Helix seat but no row at all in the #: vendor roster. Checked at the SAME point in the ladder the old #: inline ``acct is None`` test occupied, so it keeps landing in #: ``undetermined`` rather than being re-classified by a later gate. account_found: bool = True vendor_account_active: bool = True is_admin: bool = True collected_at: datetime | None = None source: str = "" def _evidence_from_atlassian( *, acct: Any | None, family: str, admin_account_ids: set[str], ) -> _SeatEvidence: """Live Atlassian directory → evidence. Semantics unchanged from the pre-PR4 inline reads, deliberately: this path was correct and is the one carrying production history.""" from app.integrations.usage_signals import SIGNAL_ACTIVE, SIGNAL_UNKNOWN if acct is None: return _SeatEvidence( state=SIGNAL_UNKNOWN, account_found=False, source="atlassian_live", ) last = acct.last_active_for(family) return _SeatEvidence( # A blank Atlassian ``last_active`` stays ``unknown`` rather than # becoming ``never``: Atlassian does not distinguish "never # signed in" from "we have no telemetry for this account", and # admins / API-only / mobile-only users all land in the blank. state=SIGNAL_ACTIVE if last is not None else SIGNAL_UNKNOWN, last_active=last, vendor_account_active=acct.is_active, is_admin=( acct.account_id in admin_account_ids or acct.is_org_or_site_admin ), source="atlassian_live", ) def _evidence_from_signal( *, signal: dict[str, Any] | None, product_key: str, seat: Any, ) -> _SeatEvidence: """Persisted usage signal → evidence. ``signal`` has already been through the freshness gate, so a row nobody refreshed arrives here as ``unknown`` and can only ever be counted as undetermined. """ from app.integrations.usage_signals import ( SIGNAL_ACTIVE, SIGNAL_NEVER, SIGNAL_UNKNOWN, ) granted = _as_utc(getattr(seat, "created_at", None)) granted_on = granted.date() if granted else None if signal is None: # Holds a Helix seat but the vendor roster has no row for them — # an anomaly for a human, never an auto-reclaim. return _SeatEvidence( state=SIGNAL_UNKNOWN, granted_on=granted_on, account_found=False, source="signal_store", ) state = signal.get("state") or SIGNAL_UNKNOWN ev = signal.get("evidence") or {} last_dt = _as_utc(signal.get("last_active_at")) return _SeatEvidence( state=state if state in (SIGNAL_ACTIVE, SIGNAL_NEVER) else SIGNAL_UNKNOWN, last_active=last_dt.date() if last_dt else None, granted_on=granted_on, vendor_account_active=bool(ev.get("vendor_account_active", True)), # Absent key ⇒ admin ⇒ skipped. Fail closed. is_admin=bool(ev.get("vendor_admin", True)), collected_at=signal.get("collected_at"), source="signal_store", ) async def _process_seat( db: AsyncSession, *, seat: Any, person: Any, plan: Any, group_name: str, family: str, evidence: _SeatEvidence, cfg: _Config, now: datetime, summary: dict[str, Any], ) -> None: """Evaluate + act on one active seat. Mutates ``summary`` counters.""" if person is None: # Data anomaly (FK should prevent it), not a system failure — # record as a skip + undetermined, never an error that could # false-page on an otherwise-healthy run. summary["undetermined"] += 1 summary["skipped"].append( f"seat {seat.id}: holder person {seat.user_id} not found", ) return # Leavers go through offboarding, not this job. if person.employment_status != "active": summary["leaver_skipped"] += 1 return if not evidence.account_found: # Holds a Helix seat but no matching vendor account in the roster # pull — an anomaly worth a human's eyes, never an auto-reclaim. summary["undetermined"] += 1 return if not evidence.vendor_account_active: # Account already deactivated at the vendor — not occupying a # license; nothing to reclaim. Don't touch. summary["undetermined"] += 1 return tz = resolve_timezone(person) today_local = _local_date(now, tz) # Hard guardrail: never auto-reclaim a vendor administrator. A # provider that cannot tell us reports ``is_admin=True``, so an # undeterminable admin status skips the seat rather than risking it. if evidence.is_admin: if seat.idle_warned_at is not None or seat.idle_manual_review_at is not None: if not cfg.dry_run: seat.idle_warned_at = None seat.idle_manual_review_at = None summary["admin_skipped"] += 1 return # Self-service keep opt-out. kept_until = _as_utc(seat.idle_kept_until) if kept_until is not None and kept_until > now: summary["kept_skipped"] += 1 return from app.integrations.usage_signals import SIGNAL_ACTIVE, SIGNAL_NEVER if evidence.state == SIGNAL_ACTIVE and evidence.last_active is not None: last_active = evidence.last_active idle_days = (today_local - last_active).days elif evidence.state == SIGNAL_NEVER and evidence.granted_on is not None: # The vendor says this account has never signed in. That IS idle # — but measured from the day the seat was handed out, not from a # login that never happened. Without this, every seat granted # this morning would read as infinitely idle. last_active = None idle_days = (today_local - evidence.granted_on).days else: # ``unknown`` / ``unsupported`` / a ``never`` with no grant date. # NOT idle — telemetry gaps, admin/API/mobile-only accounts and # stale snapshots all land here. Never reclaim; count it so the # run reports how much it could not see. summary["undetermined"] += 1 return if idle_days <= cfg.idle_threshold_days: # In use again — clear any stale warn / manual-review state so a # future idle period starts a fresh warning + full buffer. if seat.idle_warned_at is not None or seat.idle_manual_review_at is not None: if not cfg.dry_run: seat.idle_warned_at = None seat.idle_manual_review_at = None summary["reset_active"] += 1 return # ── idle_days > threshold, determinate, active, not admin/kept ── if seat.idle_warned_at is None: summary["warned"] += 1 if cfg.dry_run: return seat.idle_warned_at = now await _audit_seat( db, action="software_seat.idle_warned", seat=seat, plan=plan, detail={ "idle_days": idle_days, "last_active": ( last_active.isoformat() if last_active else None ), "signal_state": evidence.state, "signal_source": evidence.source, "product_family": family, "buffer_days": cfg.buffer_days, }, ) await _notify_warned( db, person=person, seat=seat, plan=plan, idle_days=idle_days, cfg=cfg, today_local=today_local, ) return warned_at = _as_utc(seat.idle_warned_at) warned_local = warned_at.astimezone(tz).date() if warned_at else today_local buffer_elapsed_days = (today_local - warned_local).days if buffer_elapsed_days < cfg.buffer_days: summary["in_buffer"] += 1 return # ── Buffer elapsed, still idle, not kept → reclaim via pipeline ── # One path for every holder shape: the coordinator's # ``enqueue_revoke_for_seat`` routes Jira to a # ``target_type='product_access'`` task → ``revoke_product_access`` # strips EVERY group conferring the product (applicationrole ∪ # admin-declared), so a holder who also has the product via # ``jira-dev-engineers`` / ``RD Leader`` is genuinely reclaimed — # the old single-group ``revoke_seat`` + manual-review punt left # those seats occupied forever. Same code path as self-cancel and # admin-revoke; retries / dead-letter / sibling add-on cascade / # failure DMs come from the provisioning pipeline. if cfg.dry_run: summary["reclaimed"] += 1 return from app.core.coordinators.provisioning_coordinator import ( enqueue_revoke_for_seat, ) from app.domains.provisioning.models import SOURCE_IDLE_RECLAIM # Optimistic flip — coordinator contract: the caller flips the source # seat; ``enqueue_revoke_for_seat`` cascade-flips siblings in the SAME # tx. ``revert_seat_on_failure=True`` makes the dispatcher's failure # hook flip everything back (and DM admins) if the vendor revoke # ultimately fails — a failed reclaim must surface as an active seat # again, never a silently-lying ``revoked`` row. # # Flip + enqueue live in ONE savepoint: any failure inside (designed # refusal or unexpected DB error) rolls the flip back atomically, so # the end-of-run commit can never persist a ``revoked`` seat with no # vendor task behind it — and one bad seat can't poison the batch # (per-process SAVEPOINT isolation, same rule as the other cron # loops sharing a session). seat_id = seat.id # rollback expires modified instances; capture first try: async with db.begin_nested(): seat.status = "revoked" seat.revoked_at = now task_id = await enqueue_revoke_for_seat( db, seat=seat, trace_id=f"software_seat.idle_reclaim:seat:{seat_id}", source=SOURCE_IDLE_RECLAIM, # Idle reclaim is its own policy decision; the software's # offboarding_policy (a leaver-flow setting) must not # gate it. enforce_offboarding_policy=False, revert_seat_on_failure=True, ) if task_id is None: # Plan / software / connector resolution failed inside # the coordinator — shouldn't happen (the plan loop # already required an active binding). Raise inside the # savepoint so the optimistic flip rolls back too. raise _EnqueueReturnedNoTask except AppException as e: # e.g. SEAT_NOT_INDEPENDENTLY_REVOCABLE on a blank product_key — # defensive; ``seat_revocable=false`` plans are skipped wholesale # before the per-seat loop. summary["errors"].append( f"{plan.name}/seat {seat_id}: revoke enqueue refused: {e.message}", ) return except _EnqueueReturnedNoTask: summary["errors"].append( f"{plan.name}/seat {seat_id}: revoke enqueue returned no task " "(software/connector resolution failed)", ) return # Clear any pre-2026-06-11 manual-review stamp — that state no longer # exists; the seat is now genuinely being reclaimed. seat.idle_manual_review_at = None seat.reason = ( (seat.reason or "") + f"\n[idle-reclaimed by software_seat.idle_reclaim: {idle_days}d " f"unused on prod site; product_access revoke task {task_id}]" ).strip() summary["reclaimed"] += 1 await _audit_seat( db, action="software_seat.idle_reclaimed", seat=seat, plan=plan, detail={ "idle_days": idle_days, "last_active": ( last_active.isoformat() if last_active else None ), # ``never`` reclaims carry no last-login by definition; the # state + source say where the number came from instead. "signal_state": evidence.state, "signal_source": evidence.source, "signal_collected_at": ( evidence.collected_at.isoformat() if evidence.collected_at else None ), "product_family": family, "binding_group": group_name, "provisioning_task_id": str(task_id), }, ) await _notify_reclaimed( db, person=person, seat=seat, plan=plan, idle_days=idle_days, ) # ── Audit + notification helpers ────────────────────────────────── async def _audit_seat( db: AsyncSession, *, action: str, seat: Any, plan: Any, detail: dict[str, Any], ) -> None: from app.audit.service import AuditService payload = { "plan_id": str(plan.id), "target_user_id": str(seat.user_id), **detail, } async with best_effort( db, label="idle_reclaim_handlers best-effort step", log=logger, ) as be_657_step: await AuditService(db).log( trace_id=f"software_seat.idle_reclaim:seat:{seat.id}", actor_type="system", actor_user_id=None, action=action, status="success", resource_type="software_seat_allocation", resource_id=str(seat.id), detail=payload, ) if not be_657_step.ok: logger.error("idle_reclaim audit %s failed for seat %s", action, seat.id) def _software_name(plan: Any) -> str: sw = getattr(plan, "software", None) return (sw.name if sw is not None else None) or "—" async def _notify_warned( db: AsyncSession, *, person: Any, seat: Any, plan: Any, idle_days: int, cfg: _Config, today_local: date, ) -> None: from app.notifications.service import ( CardAction, LocalizedText, NotificationEnvelope, NotificationService, ) sw = _software_name(plan) reclaim_after = cfg.buffer_days body = ( f"你已 {idle_days} 天未使用 [b]{sw}[/b]({plan.name or '—'})。\n" f"为节省 license,该坐席将在 [r]{reclaim_after} 天后[/r]自动回收。\n\n" f"若仍需使用,请点击下方[保留坐席],或直接登录一次 {sw} 即可" f"(登录后系统会自动判定为在用,不再回收)。" ) try: await NotificationService(db).dispatch( NotificationEnvelope( recipient=person, title=f"坐席闲置提醒:{sw}", body_text=body, channels=["feishu", "email"], card_subtitle=LocalizedText( zh=f"{idle_days} 天未使用 · {reclaim_after} 天后回收", en=f"Unused for {idle_days} days · reclaimed in {reclaim_after}", ja=f"{idle_days} 日間未使用 · {reclaim_after} 日後に回収されます", ), card_actions=[ CardAction( label=LocalizedText( zh="保留坐席", en="Keep my seat", ja="ライセンスを継続する", ), url=_SELF_SERVICE_URL, style="primary", ), ], url=_SELF_SERVICE_URL, related_resource_type="software_seat_allocation", related_resource_id=seat.id, ), trace_id=f"software_seat.idle_reclaim:warn:{seat.id}", ) except Exception: # noqa: BLE001 — best-effort logger.warning("idle_reclaim warn notify failed for seat %s", seat.id) async def _notify_reclaimed( db: AsyncSession, *, person: Any, seat: Any, plan: Any, idle_days: int, ) -> None: from app.notifications.service import ( NotificationEnvelope, NotificationService, ) sw = _software_name(plan) body = ( f"你的 [b]{sw}[/b]({plan.name or '—'})坐席因连续 {idle_days} 天" f"未使用已被回收(仅移除该产品访问,Atlassian 账号未停用)。\n\n" f"如需再次使用,随时可在 Helix 自助申请,通常几分钟内开通。" ) try: await NotificationService(db).dispatch( NotificationEnvelope( recipient=person, title=f"坐席已回收:{sw}", body_text=body, channels=["feishu", "email"], url=_SELF_SERVICE_URL, related_resource_type="software_seat_allocation", related_resource_id=seat.id, ), trace_id=f"software_seat.idle_reclaim:reclaimed:{seat.id}", ) except Exception: # noqa: BLE001 logger.warning("idle_reclaim reclaimed notify failed for seat %s", seat.id) __all__ = ["run_software_seat_idle_reclaim"]