"""Provisioning service. Public surface: - create_task() — enqueue a grant/revoke (called by onboarding, offboarding, request flows, etc.) - execute_task() — actually run one task (dispatch to connector or mark for manual handling) - execute_pending_batch() — dispatch all pending tasks (used by scheduler or admin "run all") - complete_manual_task() — admin marks a manual_notify task done - fail_task() — admin marks a task failed (human couldn't do it) - retry_task() — admin retries a failed task - list_tasks() — paginated query with filters - Connector CRUD """ from __future__ import annotations import re from datetime import UTC, datetime, timedelta from datetime import time as _time from typing import Any from uuid import UUID from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import flag_modified from app.audit.service import AuditService from app.core.exceptions import AppException, ConflictError, format_exception_for_log from app.core.provision_match import policy_match_conditions from app.core.working_hours import resolve_timezone from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, ) from app.integrations.driver_base import ( IntegrationDriver, ProvisioningContext, ProvisioningResult, SoftwareProvisioningDriver, ) from app.integrations.models import Integration from app.integrations.registry import get_driver_class, list_drivers from app.integrations.schemas import IntegrationDriverMetadata, IntegrationFieldInfo from app.integrations.service import build_driver_by_id, get_platform_setting def _binding_field_visible( field: IntegrationFieldInfo, values: dict[str, Any], ) -> bool: """Mirror of the service-side visible_when resolver, for bindings.""" if not field.visible_when: return True for k, expected in field.visible_when.items(): if str(values.get(k, "")) != str(expected): return False return True def _integration_registry() -> dict[tuple[str, str], type[IntegrationDriver]]: """Access the registry as a mapping (kind, provider) → driver cls.""" result: dict[tuple[str, str], type[IntegrationDriver]] = {} for meta in list_drivers(): cls = get_driver_class(meta.kind, meta.provider) if cls is not None: result[(meta.kind, meta.provider)] = cls return result from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.domains.provisioning.models import ( ACTION_GRANT, ACTION_REVOKE, EXEC_AUTO_API, EXEC_MANUAL_NOTIFY, EXEC_SSO_AUTO, SOURCE_ASSET_DISPOSAL, SOURCE_AUDIT, SOURCE_BUNDLE_APPLY, SOURCE_IDLE_RECLAIM, SOURCE_MANUAL, SOURCE_OFFBOARDING, SOURCE_ONBOARDING, SOURCE_REQUEST, SOURCE_ROLE_CHANGE, STATUS_CANCELLED, STATUS_COMPLETED, STATUS_FAILED, STATUS_IN_PROGRESS, STATUS_NOTIFIED, STATUS_PENDING, TARGET_HARDWARE, TARGET_SSO_GRANT, TERMINAL_TASK_STATUSES, ProvisioningTask, ProvisionPolicy, ) from app.domains.provisioning.models import SoftwareConnector as SoftwareConnectorRow from app.domains.provisioning.schemas import ( BillingAggregate, BillingLineItem, BulkExecuteRequest, BulkExecuteResponse, OnboardingPreview, OnboardingPreviewItem, ProvisioningTaskCreate, ProvisioningTaskResponse, ProvisionPolicyCreate, ProvisionPolicyResponse, ProvisionPolicyUpdate, ReconcileApplyRequest, ReconcileDiffItem, RoleChangePreview, SoftwareConnectorCreate, SoftwareConnectorResponse, SoftwareConnectorUpdate, ) from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) def _onboarding_not_before(person: IdentityPerson) -> datetime | None: """UTC moment at which a new-hire provisioning task becomes eligible for dispatch. Semantics: onboarding accounts are created at the START of the hire's first working day in their own wall-clock zone — i.e. ``hire_date 00:00`` local TZ. A hire in Asia/Shanghai with ``hire_date=2026-07-10`` gets ``not_before = 2026-07-09 16:00 UTC`` (= 2026-07-10 00:00 +08:00); a hire in America/Los_Angeles with the same nominal date gets the UTC instant mapping to 2026-07-10 00:00 -07:00. Mirrors the offboarding ``_leaver_revoke_not_before`` local-TZ pattern. Returns None (no gate — run ASAP once dispatched) when ``hire_date`` is unknown, so a hire with a missing start date is never stranded behind a gate that never opens. When ``hire_date`` is today or in the past the computed instant is already behind ``now``, so the dispatcher runs the task immediately — the gate only ever defers a hire whose start day has not yet begun locally. The returned datetime is timezone-aware UTC — directly comparable to ``datetime.now(UTC)`` and to ``provisioning_tasks.not_before``. """ if person.hire_date is None: return None try: tz = resolve_timezone(person) except Exception: # noqa: BLE001 — TZ resolution must never break onboarding return datetime.combine(person.hire_date, _time(0, 0), tzinfo=UTC) local_midnight = datetime.combine(person.hire_date, _time(0, 0), tzinfo=tz) return local_midnight.astimezone(UTC) VALID_ACTIONS = {ACTION_GRANT, ACTION_REVOKE} VALID_SOURCES = { SOURCE_ONBOARDING, SOURCE_OFFBOARDING, SOURCE_ROLE_CHANGE, SOURCE_REQUEST, SOURCE_AUDIT, SOURCE_BUNDLE_APPLY, # #1807: SOURCE_MANUAL / SOURCE_IDLE_RECLAIM are documented model # constants used by coordinator direct-build paths (e.g. # ``enqueue_revoke_for_seat``). They bypassed ``create_task`` so the # gap was latent, but any future caller routing through the single # ``create_task`` entry point would have hit a spurious INVALID_SOURCE # 400. Keep ``create_task`` validation in sync with what the models # (and direct-build callers) actually emit. SOURCE_MANUAL, SOURCE_IDLE_RECLAIM, # An asset left company ownership while still enrolled in an MDM, so # the device has to come out of the tenant. Kept distinct from # SOURCE_OFFBOARDING: a scrapped conference-room iPad has no leaver, # and filing these under offboarding would drop device retirements # into leaver dashboards and SLA clocks measuring something else. SOURCE_ASSET_DISPOSAL, } VALID_EXECUTION_MODES = {EXEC_AUTO_API, EXEC_MANUAL_NOTIFY, EXEC_SSO_AUTO} TERMINAL_STATUSES = TERMINAL_TASK_STATUSES # Account-creation config surfaced by ``onboarding_account_overview`` — the # switches that decide whether a new hire gets a *new* Google / Feishu account # (distinct from the seat/permission provision-policies). _PROVISIONING_GOOGLE_SETTING_KEY = "google.account_provisioning" _GOOGLE_WORKSPACE_PROVIDER = "google_workspace" _FEISHU_SOFTWARE_PROVIDER = "feishu_group" _FEISHU_OP_USER_CREATE = "user_create" # Structured-error-tag → operator-facing remediation hint. Drivers # prefix ``error_message`` with ``[TAG] ...`` for failure modes that # have a known, specific admin remediation path (e.g. Zoom 3412 means # "free up Basic seats on Zoom side"). The DM dispatcher detects the # prefix and appends the matching hint after the generic # "seat reverted to active" line so the admin knows exactly what to # do without parsing vendor error bodies. # # Naming convention: tags are vendor-prefixed (``ZOOM_*``, ``SLACK_*``, # ``GITHUB_*``, ``GOOGLE_*``, ``JIRA_*``, ``FEISHU_*``) to keep this # global namespace collision-free. Cross-driver bug-classes (e.g. # rate-limited) get vendor prefixes anyway — never reuse a tag name # across drivers, otherwise the hint dict can only hold one # remediation and the other driver's failures get the wrong advice. # # Adding a new tag: add a row here AND prefix the driver # ``error_message`` with ``[TAG] ...``. Unrecognised tags fall through # silently (the bare error_message still appears in the card). _PROVISIONING_REMEDIATION_HINTS: dict[str, str] = { "ZOOM_BASIC_CAP_EXCEEDED": ( "处理建议:打开 Zoom Admin → User Management → 筛 User Type = " "Basic → 清理长期未登录 / 已离职但仍占位的 Basic 用户(直接 " "Disassociate 即可,managed-domain 的 Basic 用户没有 1107 限制)" ",然后回到 Helix 重试本回收任务。" ), "ZOOM_LICENSED_CAP_EXCEEDED": ( "处理建议:Zoom Licensed 坐席已用满,在 Zoom Admin 回收闲置 " "Licensed 用户或扩容订阅后重试。" ), "ZOOM_USER_IN_OTHER_ACCOUNT": ( "处理建议:该 Zoom 账号目前属于另一 Zoom 租户;让对方租户管" "理员先 disassociate,或在 Zoom 工单提 transfer 后重试。" ), "ZOOM_MANAGED_DOMAIN_REFUSED": ( "处理建议:managed-domain 租户的 disassociate 路径已被 Zoom " "永久拒绝,Licensed→Basic 降级也失败 — 检查 Zoom 后台用户状" "态后联系 Zoom 支持。" ), # ── Feishu (added 2026-05-25) ─────────────────────────────── "FEISHU_HARD_DELETE_REFUSED": ( "处理建议:飞书拒绝硬删除该账号(常见原因:云盘/聊天文件归属" "未转交、上级部门冻结级联、租户合规策略要求人工确认)。系统" "已自动回退到冻结(`is_frozen=true`),坐席已释放、登录已" "禁用,合规上等价于离职。若必须最终删除,请在飞书后台处理" "文件归属后,从 Helix 任务面板手动触发一次 retry。" ), # ── Google Workspace (added 2026-05-25) ───────────────────── "GOOGLE_SUCCESSOR_NOT_IN_DIRECTORY": ( "处理建议:接收人(successor)在 Google Workspace 目录中不存在 " "— 先在 Google Admin 创建该账号(或确认拼写 / 域名),然后" "回到 Helix 重试本回收任务。" ), "GOOGLE_TRANSFER_PERMISSION_DENIED": ( "处理建议:Google Data Transfer 接口拒绝访问(常见原因:接收" "人所在域未授权 Data Transfer / Service Account 缺 " "admin.datatransfer scope / 域间委派被回收)。检查 Google " "Admin → Security → API controls → Domain-wide delegation 后" "重试;若 transfer_id 已生成,Google 端可能在服务端继续完成。" ), "GOOGLE_TRANSFER_TIMEOUT": ( "处理建议:Data Transfer 在窗口内未到达终态。再次重试本任务" "会继续轮询同一个 transfer_id(Google 端无重复触发);如果" ">10 分钟仍未完成,去 Google Admin → Data → Data Migration " "查看 transfer 状态或联系 Google Workspace 支持。" ), "GOOGLE_SCOPE_MISSING": ( "处理建议:Service Account 缺少对应的 admin.directory.* " "scope(组成员 / 用户管理)。在 Google Admin → Security → " "API controls → Domain-wide delegation 给该 Service Account " "补齐相应 scope 后重试。" ), # ── Jira / Atlassian (added 2026-05-25) ───────────────────── "JIRA_ORG_ADMIN_PERMISSION_DENIED": ( "处理建议:Atlassian org remove-access API 返回 403 拒绝撤销" "访问。坐席的 group removal 已成功(license 已释放),但账号的" "非 group 残留访问(Trello / 默认访问 / 直接授权)未清除。" "常见原因:org_unscoped_api_key 缺失或填的是带 scope 的 key" "(撤访问端点不在 granular scope 体系里,scoped key 会被拒)。" "在 admin.atlassian.com → Settings → API keys 新建一把**不选" "任何 scope** 的 org API key,填入集成的 org_unscoped_api_key。" ), "JIRA_ORG_ADMIN_AUTH_FAILED": ( "处理建议:Atlassian org remove-access API 返回 401," "org_unscoped_api_key 失效(已轮换 / 已撤销 / 已过期)。坐席的" " group removal 已成功(license 已释放),但账号的非 group 残留" "访问未清除。在 admin.atlassian.com → Settings → API keys 重新" "签发一把**不选 scope** 的 org key 并更新集成的 " "org_unscoped_api_key。" ), # ── Slack (added 2026-05-25) ──────────────────────────────── "SLACK_RATE_LIMITED": ( "处理建议:Slack SCIM 在重试预算内仍持续限流 / 5xx — 通常" "是 Slack 端 rate-limit 或临时维护。等若干分钟后重试本任务" "即可,如持续命中可查看 status.slack.com 是否有 incident。" "无需 admin 配置改动。" ), "SLACK_SCIM_UNAVAILABLE": ( "处理建议:Slack SCIM 在当前 plan 不可用(plus_teams_only " "等)。把该 Slack 集成 binding 的 execution_mode 切回 " "manual_notify 让 IT 手动操作。" ), } # Transient-error classification (2026-06-10, execution-plane PR3). # Prod audit: every ``status='failed'`` provisioning task had # ``attempt_count=1`` — the dispatcher only ever picks PENDING rows, # so a Zoom 429 ("Too many concurrent requests") or a Google 502 # died on its FIRST attempt and sat failed until a human clicked # retry, despite the task model already carrying ``attempt_count`` / # ``not_before`` / a 10-attempt dead-letter cap. Failures matching # these fragments now go back to PENDING with an exponential # ``not_before`` backoff instead of FAILED; the existing max-attempts # gate dead-letters runaway retries. Fragments are matched # case-insensitively against the canonicalised error text. Kept # deliberately conservative — an unknown error class must stay a # loud FAILED, not silently spin. _TRANSIENT_ERROR_FRAGMENTS: tuple[str, ...] = ( "too many concurrent requests", "too many requests", "rate limit", "rate_limited", "timed out", "timeout", "bad gateway", "service unavailable", "connection reset", "connection refused", "temporarily unavailable", ) # HTTP status codes that mark a failure transient — matched on WORD # BOUNDARIES (post-review fix 2026-06-10). The original bare-substring # match ("429" in text) would misclassify PERMANENT errors whose text # merely contains those digits inside an id or count ("group 4290 not # found", "15034 rows updated"), silently deferring the loud-failure # side effects through ~16 h of pointless backoff. \b excludes # digit-adjacent matches; an exact standalone "429" in a vendor body # almost always IS the status code. _TRANSIENT_STATUS_CODE_RE = re.compile(r"\b(429|502|503|504)\b") # attempt_count → minutes until the next dispatcher pickup. Indexed by # attempts already made; past the end the last value repeats until the # 10-attempt dead-letter cap closes the loop (total window ≈ 16 h). _TRANSIENT_BACKOFF_MINUTES: tuple[int, ...] = (5, 15, 45, 90, 120) def _is_transient_connector_error(err_text: str | None) -> bool: if not err_text: return False lowered = err_text.lower() if any(frag in lowered for frag in _TRANSIENT_ERROR_FRAGMENTS): return True return bool(_TRANSIENT_STATUS_CODE_RE.search(lowered)) def _transient_backoff_delay(attempt_count: int) -> timedelta: idx = min(max(attempt_count - 1, 0), len(_TRANSIENT_BACKOFF_MINUTES) - 1) return timedelta(minutes=_TRANSIENT_BACKOFF_MINUTES[idx]) # Pre-compiled prefix matcher for the tag in ``error_message``. # Tag content is ``[A-Z][A-Z0-9_]+`` followed by a single space, so # a stray ``[admin]`` style square-bracket comment in a raw vendor # body or a digit-prefixed ``[1107]`` Zoom-code-as-tag can't # false-match. _PROVISIONING_TAG_RE = re.compile(r"^\[([A-Z][A-Z0-9_]+)\]\s") def _extract_remediation_hint(error_message: str | None) -> str | None: """Return the operator-facing remediation hint for a tagged error_message, or ``None`` if no recognised tag is present. Pure function; safe to unit-test in isolation without DB or notification machinery. """ if not error_message: return None m = _PROVISIONING_TAG_RE.match(error_message) if not m: return None return _PROVISIONING_REMEDIATION_HINTS.get(m.group(1)) # Defense-in-depth credential scrub for revoke-failure DMs (added # 2026-05-15). Connector ``error_message`` strings occasionally embed # truncated raw vendor response bodies — e.g. Slack SCIM 401 with the # offending Bearer token echoed back. Recipients are already privileged # (software owner / it_admin), so this is information-sharing inside # the trust boundary, but the regex collapses common token shapes to # ```` before the body lands in Feishu + email anyway. # Patterns deliberately conservative — only hit shapes that are NEVER # legitimate content in an error message. _DM_CREDENTIAL_SCRUB = re.compile( r"(?i)(" r"Bearer\s+[A-Za-z0-9._\-]+" # OAuth bearer tokens r"|xox[abprso]-[A-Za-z0-9-]+" # Slack tokens r"|sk-[A-Za-z0-9]{16,}" # OpenAI-style API keys r"|ghp_[A-Za-z0-9]{16,}" # GitHub personal access tokens r"|gho_[A-Za-z0-9]{16,}" # GitHub OAuth tokens r"|AKIA[0-9A-Z]{16}" # AWS access key ids r"|aws_secret_access_key=\S+" # AWS secrets in conn strs r"|password=\S+" # generic password=... r")" ) # Keys in ``ProvisioningTask.context_data`` that are stamped by # trusted endpoints (e.g. /seat-allocations/{id}/change-plan sets # ``allow_demote=True``) and MUST NOT be overridable by an admin # editing a connector binding's ``binding_config`` JSON. The # coordinator merges binding_config over context_data; without the # allow-list, an admin who PATCHes ``binding_config={"allow_demote": # true}`` once would bypass the silent-demote refusal forever, # defeating the audit-gated tier-change guarantee. See DECISIONS # 2026-05-05 sub-decision #3 + round 2 security P0. _TRUSTED_RUNTIME_ONLY_CONTEXT_KEYS: frozenset[str] = frozenset({ "allow_demote", "tier_change_from_plan_id", "tier_change_from_seat_id", "tier_change_reason", }) class ProvisioningService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ───────────────────────── Task lifecycle ───────────────────────── async def create_task(self, body: ProvisioningTaskCreate) -> ProvisioningTask: """Create a provisioning task row. Does NOT execute it. Callers (onboarding, offboarding, approval-completion) use this to enqueue work, then invoke execute_task() or execute_pending_batch(). """ if body.action not in VALID_ACTIONS: raise AppException( code="INVALID_ACTION", message=f"action must be one of {sorted(VALID_ACTIONS)}", status_code=400, ) if body.source not in VALID_SOURCES: raise AppException( code="INVALID_SOURCE", message=f"source must be one of {sorted(VALID_SOURCES)}", status_code=400, ) execution_mode = body.execution_mode or await self._resolve_execution_mode( software_id=body.software_id, target_type=body.target_type, target_id=body.target_id, ) if execution_mode not in VALID_EXECUTION_MODES: raise AppException( code="INVALID_EXECUTION_MODE", message=f"execution_mode must be one of {sorted(VALID_EXECUTION_MODES)}", status_code=400, ) assignee_id = body.assignee_id if ( assignee_id is None and body.software_id and body.source == SOURCE_OFFBOARDING and body.action == ACTION_REVOKE and body.target_type in ("seat", "permission") ): # Offboarding software cleanup belongs to the software owner, # regardless of whether the vendor operation is automatic or # manual. Keep this decision at the single task-creation # chokepoint so all offboarding callers get the same assignee; # an explicitly supplied assignee still wins for a deliberate # transfer/escalation. owner = await self._get_software_owner(body.software_id) assignee_id = owner elif execution_mode == EXEC_MANUAL_NOTIFY and assignee_id is None and body.software_id: # Non-offboarding manual tasks retain the historical default. owner = await self._get_software_owner(body.software_id) assignee_id = owner # ``approval_request_id`` is FK-bound to ``approval_requests.id``, but # several FK-less columns of the same name across the codebase hold an # EmployeeRequest id instead (SoftwareSeatAllocation, SoftwareAddonCharge). # Feeding one of those in raises IntegrityError 1452 and — because most # callers swallow enqueue failures — deactivates their transaction, which # is how the 2026-08-06 Zoom seat incident wedged the approval callback. # ``ProvisioningTaskCreate`` has no validator for this field, so validate # here: this is the second door into the same FK, and the schema accepts # any UUID from any caller. approval_request_id = body.approval_request_id if approval_request_id is not None: from app.approval.models import ApprovalRequest # noqa: PLC0415 resolved = await self.db.scalar( select(ApprovalRequest.id).where( ApprovalRequest.id == approval_request_id, ), ) if resolved is None: logger.info( "create_task: approval_request_id %s does not resolve to an " "approval_requests row — dropping the link rather than " "violating the FK", approval_request_id, ) approval_request_id = resolved task = ProvisioningTask( user_id=body.user_id, software_id=body.software_id, action=body.action, target_type=body.target_type, target_id=body.target_id, source=body.source, execution_mode=execution_mode, status=STATUS_PENDING, assignee_id=assignee_id, approval_request_id=approval_request_id, context_data=body.context_data, # Defer-execution gate. NULL = run ASAP (default for grants, # approval-completion, manual admin actions). Offboarding # auto_revoke writes a future timestamp so the dispatcher # doesn't pick up the task until the leaver's last working # day has ended in their local TZ. See migration 221. not_before=body.not_before, ) self.db.add(task) await self.db.flush() return task async def _has_inflight_task( self, user_id: UUID, target_type: str, target_id: UUID, action: str, ) -> bool: """True if a non-terminal task already exists for this exact (user, target, action). Used to keep task-creating callers idempotent against double-submits — the same guard inlined in execute_onboarding_for_person, promoted to a shared helper so apply_role_change (and future callers) can reuse it.""" dup = ( await self.db.execute( select(ProvisioningTask.id).where( and_( ProvisioningTask.user_id == user_id, ProvisioningTask.target_type == target_type, ProvisioningTask.target_id == target_id, ProvisioningTask.action == action, ProvisioningTask.status.in_([ STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED, ]), ) ) ) ).first() return dup is not None async def execute_pending_batch( self, *, limit: int = 50, trace_id: str = "provisioning.dispatch", ) -> dict[str, int]: """Picks up to ``limit`` pending tasks and runs them. Intended for the ``provisioning.dispatch`` scheduled handler — no actor_user_id because the scheduler runs as "system". Returns a compact summary dict that the scheduler persists in ``scheduled_job_runs.last_result_summary``. **Per-(user_id, software_id) serialization**: vendor APIs (Zoom, Google Workspace, Slack) impose per-user rate limits that fire as "concurrent request" 429s when multiple operations on the SAME user hit within a short window. A user with N seat allocations on the same software (e.g. multiple Zoom plan tiers held over time) would generate N revoke tasks, all of which the dispatcher previously dispatched back-to-back in one tick — only the first survived; the rest cascaded into 429s and dead-lettered. This filter keeps the leading task per (user_id, software_id) and defers the rest to the next tick; ordered by ``created_at`` so the oldest fires first. Bug class caught 2026-05-08 (kevin.lu@pingcap.com — 3 Zoom revokes in one tick: 1×1107 managed-domain + 2×429 cascade). ``not_before`` gate (migration 221): tasks may carry a future timestamp that defers execution until the leaver's last working day ends / the new hire's hire_date arrives. The SELECT filters ``not_before IS NULL OR not_before <= NOW()`` so future-dated tasks stay invisible to the dispatcher until they're due. Production incident 2026-05-28 (Thalie Zhuang): without this gate, an HRIS sync that captured a leaver_date 30 days out triggered Gmail / Atlassian revoke within 30s of the sync. The ``execute_task`` defensive check below re-evaluates the gate at dispatch time so a manually-pulled task (admin clicking "retry" in the UI on a future-dated row) can't bypass it either. """ now_utc = datetime.now(UTC) # #1770: reap stale IN_PROGRESS rows first. With the durable claim # (below), a worker that crashed DURING a vendor call leaves the row # committed IN_PROGRESS — the dispatcher's ``status == PENDING`` filter # would then skip it FOREVER. Any IN_PROGRESS row whose ``started_at`` # is older than the stuck threshold is a crashed/abandoned attempt: # dead-letter it (FAILED) with a loud message so an operator verifies # the vendor side before retrying, rather than silently stranding it # OR blindly replaying (which would risk the double-grant this whole # change prevents). Threshold is generous (30 min) so a legitimately # slow vendor call in-flight is never reaped mid-execution. reaped = await self._reap_stale_in_progress(now_utc, trace_id=trace_id) rows = ( await self.db.execute( select(ProvisioningTask) .where( ProvisioningTask.status == STATUS_PENDING, or_( ProvisioningTask.not_before.is_(None), ProvisioningTask.not_before <= now_utc, ), ) .order_by(ProvisioningTask.created_at.asc()) .limit(limit), ) ).scalars().all() seen_user_software: set[tuple[Any, Any]] = set() runnable: list[ProvisioningTask] = [] deferred = 0 for task in rows: # Only auto_api tasks make outbound vendor calls — sso_auto # and manual_notify settle locally and don't hit per-user # rate limits, so they don't need serialization. if task.execution_mode != EXEC_AUTO_API: runnable.append(task) continue key = (task.user_id, task.software_id) if key in seen_user_software: deferred += 1 continue seen_user_software.add(key) runnable.append(task) completed = failed = notified = skipped = transient_retried = 0 errors: list[str] = [] for task in runnable: try: await self.execute_task( task, trace_id=trace_id, actor_user_id=None, durable_claim=True, ) except Exception as exc: # noqa: BLE001 exc_text = format_exception_for_log(exc) # Capture the id while the row is still loaded — after the # rollback below the instance is expired and a bare # ``task.id`` read would lazy-load (illegal in async). task_id = task.id errors.append(f"task {task_id}: {exc_text}") failed += 1 # Pre-launch P0 fix (Round 1, LG-004): per-task rollback # on failure so the next task's commit doesn't drag # contaminated session state. Mirrors ``bulk_execute`` # at line 1236. try: await self.db.rollback() except Exception: # noqa: BLE001 pass # #1769: an exception that *escapes* ``execute_task`` (e.g. # ``_post_success_side_effects`` → ``ApiKeyService.revoke_key`` # re-raising on a real vendor teardown failure) also rolls # back the ``attempt_count += 1`` / STATUS_IN_PROGRESS flip # that ``execute_task`` flushed — so the row returns to # PENDING with a FROZEN attempt_count. The 10-attempt dead- # letter cap then never trips: every 30s tick re-runs the # same real teardown forever and the failure DM is rolled # back too. Persist the failed attempt durably in a fresh # transaction so the counter climbs and the task eventually # dead-letters (surfacing an actionable FAILED row + DM). try: await self._persist_escaped_failure( task_id, exc_text, trace_id=trace_id, ) except Exception: # noqa: BLE001 — must never break the tick logger.warning( "provisioning dispatch: could not persist escaped-" "failure attempt for task %s", task_id, exc_info=True, ) try: await self.db.rollback() except Exception: # noqa: BLE001 pass continue # Capture the in-memory outcome BEFORE committing. Two reasons: # (1) ``commit()`` expires all attributes (expire_on_commit), # so reading ``task.status`` afterwards triggers a lazy refresh # — illegal mid-loop in async SQLAlchemy. (2) Counting only # after a *successful* commit stops the summary from # over-reporting: if the per-task commit below fails, the # status flip is rolled back and the task stays PENDING (it's # re-picked next tick), so it must NOT be counted as # completed/failed/notified here. Post-review fix 2026-06-14: # the old code incremented these counters before the commit, # so a commit failure produced a summary claiming N completed # while the DB had fewer COMPLETED rows. outcome = task.status outcome_not_before = task.not_before # Pre-launch P0 fix (Round 1, LG-004): commit per-task so a # subsequent worker crash (OOM, deploy roll, network drop) # doesn't roll back the prior tasks' completed vendor calls. # Without this commit, ``run_provisioning_dispatch`` runs # the whole batch under a single outer commit — if the # worker dies between connector returns and the outer # commit, vendor side-effects already executed (Slack # DELETE, Google PATCH, Zoom DELETE) but Helix has no # ProvisioningTask=COMPLETED record. The next dispatch tick # picks the same PENDING task back up and re-invokes the # connector — for grants that mints a SECOND seat (real # cloud cost), for revokes that 404s. This pairs with the # 2026-05-06 hardening on ``bulk_execute`` — the cron path # was missed in that round. try: await self.db.commit() except Exception as exc: # noqa: BLE001 errors.append( f"task {task.id} post-commit: " f"{format_exception_for_log(exc)}" ) try: await self.db.rollback() except Exception: # noqa: BLE001 pass continue if outcome == STATUS_COMPLETED: completed += 1 elif outcome == STATUS_FAILED: failed += 1 # Connector-REPORTED failures (``result.ok is False``) flip the # task to FAILED without raising, so they never hit the ``except`` # above and never land in ``errors``. Record them here too: the # total-failure escalation (``raise_if_total_failure`` in # ``run_provisioning_dispatch``) early-returns on an empty # ``errors`` list, so a full vendor outage where every task # returned ok=False would otherwise report ``status='ok'`` and # the failure-notifier would never fire. errors.append( f"task {task.id}: connector reported failure " f"(status={outcome}; see provisioning_task row for detail)" ) elif outcome == STATUS_NOTIFIED: notified += 1 elif outcome == STATUS_PENDING and outcome_not_before is not None: # Transient connector error re-queued with backoff — # post-review fix 2026-06-10: previously lumped into # ``skipped``, hiding retry churn from the job summary. transient_retried += 1 else: skipped += 1 return { "picked": len(rows), "completed": completed, "failed": failed, "notified": notified, "skipped": skipped, "transient_retried": transient_retried, "deferred_per_user_serial": deferred, "stale_in_progress_reaped": reaped, # #1770 "errors": errors[:5], # cap — full list stays in task rows } async def _reap_stale_in_progress( self, now_utc: datetime, *, trace_id: str, stuck_minutes: int = 30, ) -> int: """#1770: dead-letter IN_PROGRESS tasks abandoned by a crashed worker. With the durable claim, a worker that died mid-vendor-call leaves the row committed IN_PROGRESS; the dispatcher's ``status == PENDING`` filter then skips it forever. Any IN_PROGRESS row whose ``started_at`` is older than ``stuck_minutes`` is such an abandonment (a live tick never spans that long — execute_task is awaited sequentially). Mark it FAILED with a loud "verify vendor side" message so an operator triages it, rather than stranding it OR blindly replaying (double-grant risk). Runs in its own committed transaction at batch start. """ cutoff = now_utc - timedelta(minutes=stuck_minutes) stale = ( await self.db.execute( select(ProvisioningTask) .where( ProvisioningTask.status == STATUS_IN_PROGRESS, ProvisioningTask.started_at.isnot(None), ProvisioningTask.started_at < cutoff, ) .with_for_update() ) ).scalars().all() if not stale: return 0 for row in stale: row.status = STATUS_FAILED row.error_message = ( f"stuck IN_PROGRESS > {stuck_minutes}min — the worker likely " "crashed mid vendor call. VERIFY the vendor side (the " "grant/revoke may or may not have applied) before admin-" "retrying; do NOT blind-retry." )[:2000] await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="provisioning.stale_in_progress_reaped", status="failure", resource_type="provisioning_task", resource_id=str(row.id), detail={ "attempt_count": row.attempt_count, "started_at": ( row.started_at.isoformat() if row.started_at else None ), "stuck_minutes": stuck_minutes, }, ) await self.db.commit() return len(stale) async def _persist_escaped_failure( self, task_id: UUID, error_text: str, *, trace_id: str, ) -> None: """Durably record a failed attempt for a task whose exception escaped ``execute_task`` (see #1769), reconciled with the #1770 durable claim. Runs in a fresh transaction on the (already rolled-back, clean) session. The #1770 durable claim already COMMITTED the ``attempt_count += 1`` + IN_PROGRESS flip BEFORE the vendor call, so this compensator must NOT re-increment (that would double-count and dead-letter early). It re-reads the row under a lock, stamps the error, and: * dead-letters (FAILED) when the (already-persisted) attempt count has hit the cap; * otherwise sends the row back to PENDING so the dispatcher retries (an escaped CODE exception with the process still alive — the worker-CRASH case never reaches here, so the row stays IN_PROGRESS and the reaper fails it instead). A concurrent admin cancel/retry that already advanced the row wins (terminal short-circuit). """ row = ( await self.db.execute( select(ProvisioningTask) .where(ProvisioningTask.id == task_id) .with_for_update() ) ).scalar_one_or_none() if row is None or row.status in TERMINAL_STATUSES: await self.db.commit() return # #1770: attempt_count was already committed by the durable claim in # ``execute_task`` — do NOT bump it again here. Just record the error # and transition out of the durable IN_PROGRESS state. row.error_message = f"unhandled dispatch error: {error_text}"[:2000] # Mirror ``execute_task``'s dead-letter cap (max_attempts = 10). max_attempts = 10 if row.attempt_count >= max_attempts: row.status = STATUS_FAILED try: await self._post_failure_side_effects(row) except Exception: # noqa: BLE001 — DB state still written logger.warning( "escaped-failure dead-letter side-effects for task %s " "raised", row.id, exc_info=True, ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="provisioning.dead_lettered", status="failure", resource_type="provisioning_task", resource_id=str(row.id), detail={ "attempt_count": row.attempt_count, "max_attempts": max_attempts, "reason": "unhandled dispatch exception", }, ) else: # #1770: the durable claim left the row IN_PROGRESS; send it back # to PENDING so the dispatcher re-picks it (the crash case never # reaches this compensator — that row stays IN_PROGRESS for the # reaper). started_at is cleared so the reaper's staleness clock # doesn't count this alive-process retry against it. row.status = STATUS_PENDING row.started_at = None await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="provisioning.auto_api_failed", status="failure", resource_type="provisioning_task", resource_id=str(row.id), detail={ "error": error_text[:500], "attempt_count": row.attempt_count, "unhandled": True, }, ) await self.db.commit() async def execute_task( self, task: ProvisioningTask, *, trace_id: str, actor_user_id: UUID | None = None, durable_claim: bool = False, ) -> ProvisioningTask: """Dispatch a single task. For sso_auto: mark completed immediately (external lifecycle handles it). For manual_notify: mark notified and record the assignee (no API call). For auto_api: look up the software's connector and invoke grant_*/revoke_*. ``durable_claim`` (#1770): COMMIT the IN_PROGRESS + attempt_count claim BEFORE the vendor call. Set ONLY by the batch dispatcher (``execute_pending_batch``). Without it, IN_PROGRESS was merely flushed and the per-task commit happened AFTER the connector returned — so a worker that died DURING ``driver.grant_seat`` (OOM / deploy roll) rolled the row silently back to PENDING, and the next tick re-invoked the connector → a SECOND real seat/invite (double-grant). With the durable claim, a crash leaves the row IN_PROGRESS (the dispatcher's ``status == PENDING`` filter skips it — no blind replay), and the stale-IN_PROGRESS reaper (batch start) fails it for admin review. NOT used by the admin/offboarding callers — they run inside a single atomic transaction the caller commits, so a crash there rolls back cleanly with no partial vendor state to replay. """ # Re-acquire the task under a row lock so concurrent execute # callers (dispatcher cron + admin force-execute via MCP / HTTP # + offboarding service) can't both observe a non-terminal # status and double-fire the connector — which would mean two # real Google / Feishu / Slack / Jira account writes for one # task. The lock holds until ``tool_runtime`` (or the parent # caller's session) commits or rolls back. After the lock we # re-check status and bail out if another path already # advanced the task. # ``populate_existing=True`` is load-bearing: the dispatcher # batch-loads PENDING tasks into this session before calling # execute_task, so the row is already in the identity map. A bare # ``with_for_update()`` would take the DB lock but hand back the # STALE cached instance (expire_on_commit=False, so a sibling # txn's commit never refreshes it), letting the terminal/in- # progress guards below read an out-of-date status and double-fire # the connector — the exact race the lock exists to prevent. task = ( await self.db.execute( select(ProvisioningTask) .where(ProvisioningTask.id == task.id) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() or task if task.status in TERMINAL_STATUSES: return task if task.status == STATUS_NOTIFIED and task.execution_mode == EXEC_MANUAL_NOTIFY: # Already notified, waiting for human return task if task.status == STATUS_IN_PROGRESS: # Another caller is mid-flight inside its own transaction # (rare under our SELECT FOR UPDATE, but possible if it # advanced status before committing). Bail rather than # double-fire. return task # ``not_before`` defer-execution gate (migration 221). The # dispatcher's SELECT already filters this, but a manual admin # retry / MCP tool call / direct router invocation can still # land here for a future-dated task. Refuse to dispatch — leave # status PENDING so the next dispatcher tick re-evaluates the # gate naturally. Audit the skip so the operator UI can show # "won't run until " instead of a silent no-op. # # MySQL DATETIME round-trip strips tzinfo (asyncmy returns # naive on fresh load even when the column is ``DateTime( # timezone=True)``). Treat naive ``not_before`` as UTC so the # comparison can't raise TypeError on a freshly-loaded task — # which would crash the dispatcher tick mid-batch. if task.not_before is not None: nb = task.not_before if nb.tzinfo is None: nb = nb.replace(tzinfo=UTC) if nb > datetime.now(UTC): await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="provisioning.deferred_not_before", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "not_before": nb.isoformat(), "action": task.action, "target_type": task.target_type, "software_id": ( str(task.software_id) if task.software_id else None ), "user_id": str(task.user_id), }, ) return task # Pre-launch hardening 2026-05-04: dead-letter on max attempts # so a connector that always raises (bad config, expired # credentials, target tenant deleted) doesn't get retried # forever on every dispatch tick. The cron's failure_count # would otherwise climb without bound while the operator # gets no actionable signal beyond log noise. max_attempts = 10 if task.attempt_count >= max_attempts: task.status = STATUS_FAILED # Preserve the last vendor error — it's the only diagnostic an # operator triaging the dead-letter has; clobbering it with a # generic message defeats the loud terminal state's purpose. last_err = (task.error_message or "").strip() task.error_message = ( f"max retries exceeded ({task.attempt_count} attempts) — " "dead-lettered. Use admin retry to reset and try again " "after fixing the underlying issue." + (f" Last error: {last_err[:500]}" if last_err else "") ) await self._post_failure_side_effects(task) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.dead_lettered", status="failure", resource_type="provisioning_task", resource_id=str(task.id), detail={ "attempt_count": task.attempt_count, "max_attempts": max_attempts, }, ) return task task.status = STATUS_IN_PROGRESS task.started_at = datetime.now(UTC) task.attempt_count += 1 await self.db.flush() if durable_claim: # #1770: commit the claim so a crash during the vendor call below # leaves the row durably IN_PROGRESS (not rolled back to PENDING # → replayed → double-grant). The row lock acquired at the top is # released by this commit, but the IN_PROGRESS status itself now # guards re-entry (the guard above bails on IN_PROGRESS), and the # dispatcher only picks PENDING — so nothing double-fires. await self.db.commit() now = datetime.now(UTC) if task.execution_mode == EXEC_SSO_AUTO: task.status = STATUS_COMPLETED task.completed_at = now await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.sso_auto_resolved", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "action": task.action, "target_type": task.target_type, "software_id": str(task.software_id) if task.software_id else None, "user_id": str(task.user_id), }, ) return task if task.execution_mode == EXEC_MANUAL_NOTIFY: # Offboarding software-seat / software-permission revokes do # NOT wait for a human to confirm the vendor-side action. # Product decision 2026-06-13: once the reminder is # dispatched, Helix releases the local allocation and # resolves the offboarding blocker immediately, so a leaver # is never parked in the red "需要你处理" queue waiting on a # manual click. The DM still goes out — but as a reminder to # finish vendor-side cleanup, not as a gate that blocks the # offboarding. Scoped to source=offboarding + revoke + # seat/permission; ad-hoc admin revokes keep the explicit # confirm step (``complete_manual_task``) so we never mark a # still-active vendor seat reclaimed mid-employment. auto_release = ( task.source == SOURCE_OFFBOARDING and task.action == ACTION_REVOKE and task.target_type in ("seat", "permission") ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action=( "provisioning.offboarding_auto_released" if auto_release else "provisioning.manual_notified" ), status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "assignee_id": str(task.assignee_id) if task.assignee_id else None, "action": task.action, "target_type": task.target_type, "software_id": str(task.software_id) if task.software_id else None, "user_id": str(task.user_id), "auto_release": auto_release, }, ) # Outbound ping so the human knows there's work to do. Best- # effort — any send failure goes into the notification outbox # and retries itself; we don't want a Feishu blip to keep the # task in IN_PROGRESS. ``reminder_only`` renders the honest # auto-release variant (no "click to confirm" button — the # ledger is already closed). try: await self._notify_manual_assignees( task, trace_id=trace_id, reminder_only=auto_release, ) except Exception as e: # noqa: BLE001 logger.warning( "manual_notify ping for task %s failed: %s", task.id, e, ) if auto_release: # Reclaim locally + resolve the offboarding blocker now. # ``external_ref`` records that the vendor side is a # best-effort reminder, not a confirmed teardown, so the # seat ledger / audit trail stays honest about what Helix # actually did. task.status = STATUS_COMPLETED task.completed_at = now task.external_ref = ( task.external_ref or "auto-released on reminder (vendor cleanup pending IT)" ) await self._post_success_side_effects(task) else: task.status = STATUS_NOTIFIED return task # auto_api path try: result = await self._invoke_connector(task) except Exception as e: # noqa: BLE001 — any connector error → failed # Round-22 (2026-05-13) follow-up to Round-21 prod incident: # connector raises wrapped in ``AppException`` (e.g. Slack # 401 → ``AppException(code="SLACK_REVOKE_DENIED", message=...)``) # used to land here as a useless ``"connector raised # exception: "`` because ``str(AppException)`` is empty — # admins reading the failed provisioning_task row had no # idea WHY. Canonicalised via ``format_exception_for_log`` # which captures ``.code`` + ``.message`` for AppException- # derived raises and falls back to ``str``/``repr`` for the # standard-exception paths. err_text = format_exception_for_log(e) # #1791: a connector that *raises* (expired credentials → 401 # AppException, network error, "No active connector binding" # ValueError) must update binding health telemetry too — the # ``result.ok is False`` branch below records it, but a raise # previously left ``software_connectors`` untouched, so a # connector that fails on every call still showed a stale # ``last_success_at`` / no ``last_error_*`` on the admin # connector-health UI while tasks piled up FAILED. Record on # both the transient-retry and terminal-fail sub-branches for # parity with the ``ok=False`` path. await self._record_connector_failure( task.software_id, now, f"connector raised exception: {err_text}", ) if await self._maybe_schedule_transient_retry( task, err_text=f"connector raised exception: {err_text}", trace_id=trace_id, actor_user_id=actor_user_id, ): return task task.status = STATUS_FAILED task.error_message = f"connector raised exception: {err_text}" await self._post_failure_side_effects(task) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.auto_api_failed", status="failure", resource_type="provisioning_task", resource_id=str(task.id), detail={"error": err_text}, ) return task if result.ok: task.status = STATUS_COMPLETED task.external_ref = result.external_ref task.completed_at = now task.error_message = None # Preserve the connector's ``raw_response`` into # ``context_data['vendor_response']`` so downstream callers # (cancel_process restoration for software_account blockers, # forensic audits, MCP tools) can read the structured # outcome — most importantly the list of groups removed by # ``revoke_all_access`` so a cancelled offboarding can # re-grant exactly what was stripped. Before this hook the # raw_response was discarded after audit and the Atlassian # restoration was un-recoverable on cancel (Thalie Zhuang # incident 2026-05-28). Only set when non-empty to avoid # bloating context_data with empty dicts on drivers that # don't return a body. raw_resp = getattr(result, "raw_response", None) if raw_resp: ctx = dict(task.context_data or {}) ctx["vendor_response"] = raw_resp task.context_data = ctx flag_modified(task, "context_data") await self._record_connector_success(task.software_id, now) await self._post_success_side_effects(task) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.auto_api_success", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "external_ref": result.external_ref, "software_id": str(task.software_id) if task.software_id else None, "user_id": str(task.user_id), "action": task.action, }, ) elif result.capability_unsupported: # The vendor plan/tier exposes no API for this operation # (Slack Pro = no SCIM, GitHub org without Enterprise-Cloud # SCIM). Retrying never helps, and marking it FAILED both # surfaces a misleading "error" to admins and — because the # owning offboarding blocker stays RESO_PENDING — pins the # leaver's process at ``awaiting_provisioning`` with no # signal to anyone. Auto-downgrade to manual_notify: flip the # task to NOTIFIED and ping the assignee so a human does it by # hand and marks it complete (which then resolves the blocker # the normal way). This is the runtime safety net behind the # create-time fix in ``_resolve_execution_mode`` (which now # honours ``binding_config.execution_mode='manual_notify'``): # it catches tasks that were already created as auto_api # before the binding was corrected. Idempotent: once # downgraded, a re-dispatch short-circuits at the # ``NOTIFIED + manual_notify`` early-return near the top of # this method and never re-enters this branch (no re-notify, # no double side-effect). task.execution_mode = EXEC_MANUAL_NOTIFY task.status = STATUS_NOTIFIED task.error_message = result.error_message await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.auto_api_capability_downgraded", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "downgraded_to": EXEC_MANUAL_NOTIFY, "reason": result.error_message, "software_id": ( str(task.software_id) if task.software_id else None ), "user_id": str(task.user_id), "action": task.action, }, ) try: await self._notify_manual_assignees(task, trace_id=trace_id) except Exception as e: # noqa: BLE001 logger.warning( "capability-downgrade manual_notify ping for task %s " "failed: %s", task.id, e, ) else: err_text = result.error_message or "connector reported failure" if await self._maybe_schedule_transient_retry( task, err_text=err_text, trace_id=trace_id, actor_user_id=actor_user_id, ): # Connector-health stats still record the blip so the # software page's failure telemetry stays honest. await self._record_connector_failure( task.software_id, now, result.error_message, ) return task task.status = STATUS_FAILED task.error_message = err_text await self._record_connector_failure( task.software_id, now, result.error_message, ) await self._post_failure_side_effects(task) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=actor_user_id, action="provisioning.auto_api_failed", status="failure", resource_type="provisioning_task", resource_id=str(task.id), detail={"error": result.error_message}, ) return task async def complete_manual_task( self, task_id: UUID, *, note: str | None, external_ref: str | None, trace_id: str, actor_user_id: UUID, actor_is_admin: bool = False, via_token: bool = False, ) -> ProvisioningTask: """Mark a manual_notify task done. ``via_token`` (2026-06-09): the caller arrived through a signed one-click link in the manual-notify notification (no Helix login). The signature already proved the recipient's identity and authorised THIS task, so we skip the session-bound assignee re-check and the explicit proof requirement — the click itself is the attestation that the recipient removed the access in the external system. ``external_ref`` is defaulted to a marker and the audit row records ``via='token'`` so the trail still shows the completion was a one-click confirmation, not an in-console action. Idempotent on an already-terminal task (returns it). """ task = await self._get_task(task_id, lock=True) # Idempotent for the token path: a recipient who clicks the # email/Feishu button twice (or after the dispatcher already # auto-resolved it) must see "already done", not a 400 — the # one-click page can't ask them to "refresh and retry". if via_token and task.status in TERMINAL_STATUSES: return task # Re-validate assignee under the row lock — closes the race # window where an admin reassigns the task between the # router's pre-check and this lock acquire. Pre-launch # hardening 2026-05-05 (P1 from CR3): the router docstring # promised this re-validation but the service didn't perform # it. Admins bypass via the explicit ``actor_is_admin`` flag; # the signed-token path bypasses because the token IS the auth. if ( not actor_is_admin and not via_token and task.assignee_id != actor_user_id ): raise AppException( code="TASK_NOT_ASSIGNED_TO_YOU", message=( "Task was reassigned to another user before you " "could complete it. Refresh and check the new " "assignee." ), status_code=403, ) if task.status in TERMINAL_STATUSES: raise AppException( code="TASK_NOT_ACTIONABLE", message=f"Task is already {task.status}", status_code=400, ) if task.execution_mode != EXEC_MANUAL_NOTIFY: raise AppException( code="TASK_NOT_MANUAL", message="Only manual_notify tasks can be completed by humans", status_code=400, ) # Non-admins completing a REVOKE task must supply an external_ref # (ticket number, timestamp, screenshot URL, etc.) as evidence the # revoke was actually performed in the external system. Admins may # complete without proof when overriding operationally; the # one-click token path records the click itself as the proof. if ( not actor_is_admin and not via_token and task.action == ACTION_REVOKE and not (external_ref or "").strip() ): raise AppException( code="REVOKE_PROOF_REQUIRED", message=( "Provide an external_ref (e.g. ticket number or " "system reference) to confirm the access was revoked." ), status_code=400, ) # Evidence-required targets (SSO removal, migration 603): the click # itself is NOT accepted as proof — a revoked SAML grant that the # IdP still honours is exactly the failure this lifecycle exists to # catch, so the one-click page carries a mandatory reference field # and this guard refuses a bare token submission. Admins are held to # the same bar here: the point is the trail, not the permission. requires_evidence = bool( (task.context_data or {}).get("requires_evidence") ) or task.target_type == TARGET_SSO_GRANT if ( requires_evidence and task.action == ACTION_REVOKE and not (external_ref or "").strip() ): raise AppException( code="REVOKE_PROOF_REQUIRED", message=( "This removal needs evidence: a ticket id, the vendor " "console URL, or a short note of what was disabled." ), status_code=400, ) if via_token and not (external_ref or "").strip(): external_ref = "one-click confirmation (notification link)" now = datetime.now(UTC) task.status = STATUS_COMPLETED task.completed_at = now task.external_ref = external_ref ctx = dict(task.context_data or {}) if note: ctx["completion_note"] = note # Who closed it — read back by target-specific hooks (the SSO removal # ledger records ``removal_confirmed_by``). The audit row below has # the same actor, but hooks must not have to query audit_logs. ctx["completed_by_person_id"] = str(actor_user_id) ctx["completed_via"] = "token" if via_token else "console" task.context_data = ctx await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.manual_completed", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "note": note, "external_ref": external_ref, "via": "token" if via_token else "console", }, ) # Pre-launch P0 fix (Round 45, CR-301): propagate the manual # completion to the linked SoftwareSeatAllocation / # UserPermission / ApiKeyAssignment / OffboardingBlocker rows. # Without this, manual-mode (Slack / Zoom / Atlassian Free) # admins clicking "Done" leave the seat row in ``active`` # forever — billing keeps charging and the offboarding blocker # never moves to READY. Mirror of execute_task's success path # at line 488. Idempotent. await self._post_success_side_effects(task) return task async def fail_task( self, task_id: UUID, *, reason: str, trace_id: str, actor_user_id: UUID, ) -> ProvisioningTask: task = await self._get_task(task_id, lock=True) if task.status in TERMINAL_STATUSES: raise AppException( code="TASK_NOT_ACTIONABLE", message=f"Task is already {task.status}", status_code=400, ) task.status = STATUS_FAILED task.error_message = reason await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.marked_failed", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={"reason": reason}, ) # Pre-launch P1 fix (Round 45 follow-up, CR-309): admin-marked # failure must also run the failure side-effects (revert the # optimistic seat revoke + DM the software owner / IT admins). # Without this, an admin who knows the vendor revoke actually # failed and marks the task FAILED leaves Helix in "seat # revoked" while the vendor still bills. await self._post_failure_side_effects(task) return task async def cancel_task( self, task_id: UUID, *, reason: str, trace_id: str, actor_user_id: UUID, ) -> ProvisioningTask: # Pre-launch P1 fix: row-lock the read so a concurrent # ``execute_task`` (which holds FOR UPDATE while making the # vendor call) cannot complete with status=COMPLETED + # external_ref AFTER our in-memory copy was loaded — without # the lock our pending UPDATE would overwrite COMPLETED with # CANCELLED, dropping the external_ref and leaving Helix + # the vendor in shadow state. Mirrors ``fail_task``. task = await self._get_task(task_id, lock=True) # Allow cancel on FAILED tasks — admins may want to abandon a # dead-lettered task that cannot be retried (bad connector config, # person already fully deprovisioned externally, etc.). We no longer # need to refuse FAILED here; the retry path remains available while # the task is in FAILED state, but once cancelled it is permanently # terminal. Refuse only genuinely terminal statuses (COMPLETED, # CANCELLED) where the action has either succeeded or already been # intentionally abandoned. if task.status in TERMINAL_STATUSES: raise AppException( code="TASK_NOT_ACTIONABLE", message=f"Task is already {task.status}", status_code=400, ) task.status = STATUS_CANCELLED task.error_message = reason await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.cancelled", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={"reason": reason}, ) # Pre-launch P1 fix (Round 45 follow-up, CR-309): cancelled # revoke task must also run the failure side-effects so a # pre-flipped seat reverts to active. Cancel of a GRANT task # is intentionally NOT side-effected — granting was never # attempted so the original "pending" seat state is correct # already (or will be cleaned by the standard pending-task # sweep). # # #1792: pass ``suppress_failure_dm`` — the seat still reverts to # active for opt-in (``revert_seat_on_failure``) tasks, but the # owner does NOT get a spurious "软件坐席回收失败 / 已自动回滚为 # active, 需要管理员介入" DM for a deliberate admin cancel (and for # the non-opt-in families the "已回滚为 active" wording was itself # false). if task.action == ACTION_REVOKE: await self._post_failure_side_effects(task, suppress_failure_dm=True) return task async def retry_task( self, task_id: UUID, *, reset_assignee: bool, trace_id: str, actor_user_id: UUID, actor_is_admin: bool = False, ) -> ProvisioningTask: # Pre-launch P1 fix: row-lock so two concurrent admin retries # can't both observe FAILED and both reset attempt_count to 0. # The inner FOR UPDATE in ``execute_task`` already serializes # the vendor call so we don't double-grant, but two # ``provisioning.retried`` audit rows + two attempt-count # resets cause operator confusion. task = await self._get_task(task_id, lock=True) # Re-validate assignee under the row lock — mirrors # ``complete_manual_task``. The router's pre-check is unlocked, so # without this an assignee who was reassigned between the router read # and this lock acquire could still re-dispatch the vendor # grant/revoke on a task they no longer own. The docstring on the # ``/retry`` route promised this guard; this makes it real. Admins # bypass via the explicit ``actor_is_admin`` flag. if not actor_is_admin and task.assignee_id != actor_user_id: raise AppException( code="TASK_NOT_ASSIGNED_TO_YOU", message=( "Task was reassigned to another user before you " "could retry it. Refresh and check the new assignee." ), status_code=403, ) # Only FAILED tasks are retryable. CANCELLED is a terminal "abandoned" # state — an admin explicitly gave up on it; retrying would reopen a # consciously-closed action. If the admin decides they actually want # to re-attempt, they should create a new task instead. if task.status != STATUS_FAILED: raise AppException( code="TASK_NOT_RETRYABLE", message=( f"Only failed tasks can be retried (current: {task.status}). " f"Cancelled tasks are permanently abandoned — create a new task " f"if you need to re-attempt." if task.status == STATUS_CANCELLED else f"Only failed tasks can be retried (current: {task.status})" ), status_code=400, ) task.status = STATUS_PENDING task.error_message = None # Pre-launch hardening 2026-05-04: admin retry resets the # attempt counter so the dead-letter cap doesn't immediately # re-trigger if the operator already burned 9 attempts before # fixing the underlying connector. prev_attempts = task.attempt_count task.attempt_count = 0 # Clear the transient-retry backoff stamp — otherwise the # immediate ``execute_task`` below sees ``not_before > now`` and # silently defers (logs ``deferred_not_before`` and returns the # task untouched), so the operator's explicit "retry now" does # nothing until the (up to 120-min) backoff elapses. task.not_before = None # Clear the per-task revoke-failure DM stamp so the next failure # (if any) re-fires the signal — admin explicitly retried, so a # fresh outcome deserves a fresh push. Pairs with # ``_post_failure_side_effects``'s dedup stamp at the failure # write-back. Safe no-op when the task never had the stamp. if (task.context_data or {}).get("failure_dm_sent_at") is not None: ctx = dict(task.context_data or {}) ctx.pop("failure_dm_sent_at", None) task.context_data = ctx flag_modified(task, "context_data") if reset_assignee and task.execution_mode == EXEC_MANUAL_NOTIFY and task.software_id: owner = await self._get_software_owner(task.software_id) task.assignee_id = owner await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.retried", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "reset_assignee": reset_assignee, "previous_attempts": prev_attempts, }, ) return await self.execute_task(task, trace_id=trace_id, actor_user_id=actor_user_id) # ───────────────────────── Queries ───────────────────────── async def list_tasks( self, *, page: int = 1, page_size: int = 20, status: str | None = None, statuses: list[str] | None = None, source: str | None = None, user_id: UUID | None = None, software_id: UUID | None = None, assignee_id: UUID | None = None, action: str | None = None, execution_modes: list[str] | None = None, keyword: str | None = None, order_by: str | None = None, ) -> tuple[list[ProvisioningTaskResponse], int]: from sqlalchemy import or_ from app.identity.person_models import IdentityPerson filters = [] # ``statuses`` (multi-value) takes precedence over ``status`` # (single-value) — sending both is undefined-but-tolerated: # the multi-value wins. Matches the router-level doc. if statuses: filters.append(ProvisioningTask.status.in_(statuses)) elif status: filters.append(ProvisioningTask.status == status) if source: filters.append(ProvisioningTask.source == source) if user_id: filters.append(ProvisioningTask.user_id == user_id) if software_id: filters.append(ProvisioningTask.software_id == software_id) if assignee_id: filters.append(ProvisioningTask.assignee_id == assignee_id) if action: filters.append(ProvisioningTask.action == action) if execution_modes: filters.append(ProvisioningTask.execution_mode.in_(execution_modes)) keyword_clean = (keyword or "").strip() needs_person_join = bool(keyword_clean) needs_software_join = bool(keyword_clean) if keyword_clean: # Escape LIKE metacharacters (``%``, ``_``, ``\``) before # wrapping with our own ``%`` wildcards. Without this: # - ``keyword='%'`` would expand to ``%%%`` and match # every row — defeats the filter. # - ``keyword='a_c'`` would silently match ``abc``, # ``adc``, ``a1c`` etc., letting a caller enumerate # names via result-count side-channel. # Order matters: backslash MUST be escaped first or the # later replacements would re-escape it. escaped = ( keyword_clean .replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) pattern = f"%{escaped}%" filters.append( or_( IdentityPerson.display_name.ilike(pattern, escape="\\"), IdentityPerson.work_email.ilike(pattern, escape="\\"), AssetSoftwareCatalog.name.ilike(pattern, escape="\\"), ) ) condition = and_(*filters) if filters else None count_q = select(func.count()).select_from(ProvisioningTask) q = select(ProvisioningTask) if needs_person_join: count_q = count_q.join( IdentityPerson, ProvisioningTask.user_id == IdentityPerson.id, ) q = q.join( IdentityPerson, ProvisioningTask.user_id == IdentityPerson.id, ) if needs_software_join: # Outer join — a provisioning task can have software_id=NULL # (e.g. cloud-account grants) and we still want those rows # to be considered when the keyword only matches the user. count_q = count_q.outerjoin( AssetSoftwareCatalog, ProvisioningTask.software_id == AssetSoftwareCatalog.id, ) q = q.outerjoin( AssetSoftwareCatalog, ProvisioningTask.software_id == AssetSoftwareCatalog.id, ) if condition is not None: count_q = count_q.where(condition) q = q.where(condition) total = (await self.db.execute(count_q)).scalar() or 0 # ``order_by`` is an opt-in switch for callers that need a # different ordering primary key than ``created_at`` — most # notably the cancellations-overview "Completed" segment, where # ordering by completion timestamp matches the "show me what # just finished" reading order. A task created months ago that # only completed today should bubble to the top, not sink. We # tie-break on ``id`` so pagination is deterministic across # rows that share the primary timestamp (MySQL/TiDB's row # ordering is otherwise undefined). Unknown values fall back # to ``created_at`` rather than 400-ing — keeps old clients # working if we add more values later. # Note: every code path that sets ``status=completed`` also # sets ``completed_at`` (verified in service.py — see the # three call sites), so the column is non-NULL for the # Completed segment in practice. MySQL/TiDB sort NULLs last # on DESC natively anyway, so we don't need ``NULLS LAST`` # (which isn't valid MySQL syntax). if order_by == "completed_at": q = q.order_by( ProvisioningTask.completed_at.desc(), ProvisioningTask.id.desc(), ) else: q = q.order_by( ProvisioningTask.created_at.desc(), ProvisioningTask.id.desc(), ) q = q.offset((page - 1) * page_size).limit(page_size) rows = list((await self.db.execute(q)).scalars().all()) # Enrich in a single pass to avoid N+1 user_ids = {r.user_id for r in rows} | {r.assignee_id for r in rows if r.assignee_id} software_ids = {r.software_id for r in rows if r.software_id} # Account- AND product_access-type revoke tasks have # ``software_id=NULL`` by design and carry the integration id # in ``target_id``: # - ``account`` — offboarding ``_apply_account_policy`` # full-access revoke ("remove from all # groups", Jira / Atlassian) # - ``product_access`` — per-product Atlassian revoke # (Jira Software / Confluence / JSM) # where the human-readable product # name lives in # ``context_data.product_label`` # Bulk-load both so ``_to_response`` can fill ``target_label`` # with a meaningful name instead of letting the Software # Cancellations table render "—". Both target types resolve # ``target_id`` against the integrations catalog as a fallback; # ``product_access`` prefers its richer context label first. integration_ids = { r.target_id for r in rows if r.target_type in ("account", "product_access") and r.target_id is not None } users = await self._load_users(user_ids) softwares = await self._load_softwares(software_ids) integrations = await self._load_integrations(integration_ids) responses = [self._to_response(r, users, softwares, integrations) for r in rows] return responses, total async def get_task_response(self, task_id: UUID) -> ProvisioningTaskResponse: task = await self._get_task(task_id) users = await self._load_users({task.user_id, *( {task.assignee_id} if task.assignee_id else set())}) softwares = await self._load_softwares({task.software_id} if task.software_id else set()) integrations = await self._load_integrations( {task.target_id} if ( task.target_type in ("account", "product_access") and task.target_id is not None ) else set() ) return self._to_response(task, users, softwares, integrations) async def analytics_summary(self, *, window_days: int = 30) -> dict[str, Any]: """Single-trip aggregate for the provisioning-queue analytics. Replaces the page's former 4× narrow ``list_tasks`` count probes. Status / action / execution_mode / source / target_type rollups are lifetime-current (the queue's monitoring surface); the daily created-vs-completed trend and ``completed_today`` are windowed. Every aggregate is a single GROUP BY — O(1) round-trips in fleet size. """ from app.core.timeseries import day_axis db = self.db now = datetime.now(UTC) axis = day_axis(window_days=window_days, now=now) today = axis.days[-1] async def _group(column) -> dict[str, int]: rows = (await db.execute( select(column, func.count()).group_by(column) )).all() return {(k if k is not None else "unknown"): int(n) for k, n in rows} by_status = await _group(ProvisioningTask.status) by_action = await _group(ProvisioningTask.action) by_execution_mode = await _group(ProvisioningTask.execution_mode) by_source = await _group(ProvisioningTask.source) by_target_type = await _group(ProvisioningTask.target_type) total = sum(by_status.values()) completed = by_status.get(STATUS_COMPLETED, 0) failed = by_status.get(STATUS_FAILED, 0) pending = by_status.get(STATUS_PENDING, 0) notified = by_status.get(STATUS_NOTIFIED, 0) in_flight = pending + notified + by_status.get(STATUS_IN_PROGRESS, 0) # Success rate over terminal-attempted tasks (completed vs failed); # pending/cancelled excluded so an empty queue isn't "0% success". terminal_attempted = completed + failed success_rate = ( round(completed / terminal_attempted * 100, 1) if terminal_attempted else None ) # ── daily created-vs-completed trend ── created_rows = (await db.execute( select(func.date(ProvisioningTask.created_at), func.count()) .where(ProvisioningTask.created_at >= axis.window_start) .group_by(func.date(ProvisioningTask.created_at)) )).all() completed_rows = (await db.execute( select(func.date(ProvisioningTask.completed_at), func.count()) .where(ProvisioningTask.completed_at >= axis.window_start) .group_by(func.date(ProvisioningTask.completed_at)) )).all() created_map = {str(d): int(n) for d, n in created_rows} completed_map = {str(d): int(n) for d, n in completed_rows} daily = [ { "date": d.isoformat(), "created": created_map.get(d.isoformat(), 0), "completed": completed_map.get(d.isoformat(), 0), } for d in axis.days ] completed_today = completed_map.get(today.isoformat(), 0) return { "window_days": window_days, "kpi": { "total": total, "pending": pending, "notified": notified, "failed": failed, "completed": completed, "completed_today": completed_today, "in_flight": in_flight, "success_rate": success_rate, }, "by_status": by_status, "by_action": by_action, "by_execution_mode": by_execution_mode, "by_source": by_source, "by_target_type": by_target_type, "daily": daily, } # ───────────────────────── Connector CRUD ───────────────────────── async def list_connectors(self) -> list[SoftwareConnectorResponse]: rows = ( await self.db.execute( select(SoftwareConnectorRow).order_by(SoftwareConnectorRow.created_at.desc()) ) ).scalars().all() software_ids = {r.software_id for r in rows} softwares = await self._load_softwares(software_ids) integrations = await self._load_integrations( {r.integration_id for r in rows if r.integration_id} ) return [self._to_connector_response(r, softwares, integrations) for r in rows] async def onboarding_account_overview(self) -> dict[str, Any]: """Snapshot of the ACCOUNT-CREATION half of onboarding — the part that mints (or doesn't mint) real Google / Feishu accounts, which lives OUTSIDE the seat/permission provision-policies shown on the onboarding-strategy page. The strategy page's policy cards only ever add-to-group / unsuspend an already-existing account (or, for Feishu ``user_create`` bindings, create one). Whether a brand-new hire actually gets a *new* mailbox is governed by two switches that live on two other admin pages: * Google — ``google.account_provisioning.auto_create_on_hire`` (a platform setting; account minting runs in ``account_provisioner``, independent of any seat policy). * Feishu — the Feishu software connector's ``binding_config.operation`` mode: only ``user_create`` mints a new account; ``user_suspend`` / ``group_member`` merely unfreeze / add-to-group an existing one. Surfacing both here lets the page present ONE coherent picture instead of the admin having to cross-reference /admin/platform-settings and /admin/integrations. Read-only. """ google_cfg = await get_platform_setting( _PROVISIONING_GOOGLE_SETTING_KEY, default={}, session=self.db, ) if not isinstance(google_cfg, dict): google_cfg = {} fte = google_cfg.get("fte") if isinstance(google_cfg.get("fte"), dict) else {} contractor = ( google_cfg.get("contractor") if isinstance(google_cfg.get("contractor"), dict) else {} ) # Which provider integrations are actually wired up + enabled. enabled_providers = set( ( await self.db.execute( select(Integration.provider).where(Integration.enabled.is_(True)) ) ).scalars().all() ) # Active Feishu software connector's operation mode (first active one). feishu_row = ( await self.db.execute( select(SoftwareConnectorRow) .where(SoftwareConnectorRow.is_active.is_(True)) .order_by(SoftwareConnectorRow.created_at.desc()) ) ).scalars().all() feishu_operation: str | None = None for r in feishu_row: integ = (await self._load_integrations({r.integration_id})).get( r.integration_id ) if r.integration_id else None if integ is not None and integ.provider == _FEISHU_SOFTWARE_PROVIDER: feishu_operation = (r.binding_config or {}).get("operation") break google_configured = _GOOGLE_WORKSPACE_PROVIDER in enabled_providers # ── Feishu account source: the identity-sync feishu_contact target ── # New Feishu accounts are NOT created by the software connector above # (its user_suspend / group_member operations only touch existing # accounts). The real 建号 owner is the outbound identity-sync target # (HRIS → 飞书通讯录), whose runner creates missing in-scope people # (Authing-parity create-missing, gated by write_mode + filter + # skip-list). Surface its live state so the strategy page can point # the admin at the true switch instead of the connector. from app.domains.identity_sync.models import IdentitySyncTarget feishu_targets = ( await self.db.execute( select(IdentitySyncTarget) .where(IdentitySyncTarget.kind == "feishu_contact") .order_by( IdentitySyncTarget.enabled.desc(), IdentitySyncTarget.created_at.desc(), ) ) ).scalars().all() feishu_target = feishu_targets[0] if feishu_targets else None # Offboarding side — is the feishu_contact backstop reconciler live? # (It rides this target's enablement; on_offboard=keep = never freeze.) feishu_backstop_enabled = any( t.enabled and t.on_offboard != "keep" for t in feishu_targets ) feishu_account_source: dict[str, Any] | None = None if feishu_target is not None: fc = feishu_target.filter_config or {} # ``person_ids`` = the explicit allow-list used for staged # rollouts (锁人灰度). While it is set, people outside the list # are never created — the single most important truth to show. person_locked = bool(fc.get("person_ids")) # Attribute-scoped rollout (the OTHER staged-rollout shape): # a departments / countries / job-title / email-suffix filter # limits WHO auto-creation applies to. Prod today locks to one # department subtree — a page that shows "auto-create · on" # without this qualifier claims company-wide coverage it # doesn't have. (employee_no_pattern and employment_statuses # are data-quality gates, not rollout scoping — excluded.) scope_dept_count = len(fc.get("departments") or []) scope_country_count = len(fc.get("countries") or []) attr_scoped = bool( scope_dept_count or scope_country_count or fc.get("job_title_includes") or fc.get("email_suffixes") ) # ``mode`` matters: ``matches()`` evaluates the non-empty filter # dimensions with ``any()`` under ``mode='or'``. The default # ``employment_statuses`` dimension (active/on_leave) matches every # new hire, so an attribute filter OR'd with it does NOT narrow # auto-creation at all — it's effectively company-wide. Reporting # ``scope_limited=True`` there would tell the admin new hires are # NOT auto-created when in fact all are (the inverse of the truth). # Prod locks ``mode='and'`` (the UI hardcodes it), but the API # accepts 'or', so compute honestly. mode = str(fc.get("mode") or "and").strip().lower() if mode == "or" and fc.get("employment_statuses"): scope_limited = False else: scope_limited = attr_scoped skip_ids = (feishu_target.runner_options or {}).get( "skip_provision_person_ids" ) or [] feishu_account_source = { "target_name": feishu_target.name, "enabled": bool(feishu_target.enabled), # create_update / create_only → mints missing accounts; # update_only → never creates. "write_mode": feishu_target.write_mode, "creates_missing": feishu_target.write_mode != "update_only", "person_locked": person_locked, "person_locked_count": len(fc.get("person_ids") or []), "scope_limited": scope_limited, "scope_department_count": scope_dept_count, "scope_country_count": scope_country_count, "skip_count": len(skip_ids), "last_run_at": ( feishu_target.last_run_at.isoformat() if feishu_target.last_run_at else None ), "last_run_status": feishu_target.last_run_status, } from app.domains.asset.models import ( EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, ) return { "timing": "hire_date_local_midnight", "google": { "auto_create_on_hire": bool(google_cfg.get("auto_create_on_hire", False)), "integration_configured": google_configured, "fte_domain": (fte.get("domain") or None), "contractor_domain": (contractor.get("domain") or None), }, "feishu": { "integration_configured": _FEISHU_SOFTWARE_PROVIDER in enabled_providers, "operation": feishu_operation, "creates_accounts": feishu_operation == _FEISHU_OP_USER_CREATE, # The REAL account-creation owner (identity-sync target); # None when no feishu_contact target is configured at all. "account_source": feishu_account_source, }, # ── Offboarding account disposition (what happens on leave) ── "offboarding": { # leaver's Google account auto-suspends at leaver_date+1 00:00 # local (event path, retention-exempt) — only actually acts # when the Google integration is wired. "timing": "leaver_date_plus_one_local_midnight", "google": { "auto_suspend": google_configured, "session_force_logout": google_configured, "retention_exempt_classes": list( EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS ), }, "feishu": { # seat-revoke operation (freeze/delete/group) + whether the # common-path backstop reconciler is live. "seat_revoke_operation": feishu_operation, "backstop_enabled": feishu_backstop_enabled, }, }, } async def _load_integrations(self, ids: set[UUID]) -> dict[UUID, Integration]: """Bulk-load Integration rows by id, keyed by id.""" if not ids: return {} rows = ( await self.db.execute( select(Integration).where(Integration.id.in_(ids)) ) ).scalars().all() return {r.id: r for r in rows} async def _validate_binding( self, *, integration_id: UUID, binding_config: dict[str, Any], ) -> tuple[Integration, type[IntegrationDriver]]: """Load the integration, validate it's a software_provisioning one, validate ``binding_config`` against the driver's binding_schema. Returns the integration row + the driver class. """ integration = await self.db.get(Integration, integration_id) if integration is None: raise AppException( code="INTEGRATION_NOT_FOUND", message=f"Integration {integration_id} not found", status_code=400, ) if integration.kind != "software_provisioning": raise AppException( code="INTEGRATION_WRONG_KIND", message=( f"Integration {integration.slug} is kind={integration.kind!r}; " f"software bindings require kind='software_provisioning'" ), status_code=400, ) if not integration.enabled: raise AppException( code="INTEGRATION_DISABLED", message=f"Integration {integration.slug} is disabled", status_code=400, ) driver_cls = get_driver_class(integration.kind, integration.provider) if driver_cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=( f"No driver for ({integration.kind!r}, {integration.provider!r})" ), status_code=500, ) for field in driver_cls.binding_schema: if not _binding_field_visible(field, binding_config): continue if field.required and not binding_config.get(field.key): raise AppException( code="MISSING_BINDING_CONFIG", message=f"Missing required binding field: {field.label}", status_code=400, ) return integration, driver_cls async def create_connector( self, body: SoftwareConnectorCreate, *, trace_id: str, actor_user_id: UUID, ) -> SoftwareConnectorResponse: await self._validate_binding( integration_id=body.integration_id, binding_config=body.binding_config, ) existing = await self.db.execute( select(SoftwareConnectorRow).where( SoftwareConnectorRow.software_id == body.software_id, ) ) if existing.scalar_one_or_none(): raise AppException( code="CONNECTOR_ALREADY_EXISTS", message="This software already has a connector binding configured", status_code=400, ) row = SoftwareConnectorRow( software_id=body.software_id, integration_id=body.integration_id, binding_config=body.binding_config or None, is_active=body.is_active, ) self.db.add(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.connector_created", status="success", resource_type="software_connector", resource_id=str(row.id), detail={ "software_id": str(body.software_id), "integration_id": str(body.integration_id), }, ) softwares = await self._load_softwares({row.software_id}) integrations = await self._load_integrations({row.integration_id}) return self._to_connector_response(row, softwares, integrations) async def update_connector( self, connector_id: UUID, body: SoftwareConnectorUpdate, *, trace_id: str, actor_user_id: UUID, ) -> SoftwareConnectorResponse: row = await self._get_connector(connector_id) changes: dict[str, Any] = {} effective_integration_id = body.integration_id or row.integration_id effective_binding = ( body.binding_config if body.binding_config is not None else dict(row.binding_config or {}) ) await self._validate_binding( integration_id=effective_integration_id, binding_config=effective_binding, ) if body.integration_id is not None and body.integration_id != row.integration_id: changes["integration_id"] = (str(row.integration_id), str(body.integration_id)) row.integration_id = body.integration_id if body.binding_config is not None: row.binding_config = body.binding_config or None changes["binding_config_updated"] = True if body.is_active is not None: row.is_active = body.is_active changes["is_active"] = body.is_active await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.connector_updated", status="success", resource_type="software_connector", resource_id=str(row.id), detail=changes, ) softwares = await self._load_softwares({row.software_id}) integrations = await self._load_integrations({row.integration_id}) return self._to_connector_response(row, softwares, integrations) async def delete_connector( self, connector_id: UUID, *, trace_id: str, actor_user_id: UUID, ) -> None: row = await self._get_connector(connector_id) # Guard: refuse to delete a connector that has in-flight tasks. # Deleting it mid-execution would leave tasks with a dangling # software_id that can't resolve a driver, causing silent failures # on the next dispatch tick. in_flight_count = await self.db.scalar( select(func.count()).select_from(ProvisioningTask).where( and_( ProvisioningTask.software_id == row.software_id, ProvisioningTask.status.in_((STATUS_PENDING, STATUS_IN_PROGRESS)), ) ) ) if (in_flight_count or 0) > 0: raise ConflictError( f"Cannot delete connector: {in_flight_count} task(s) are " f"pending or in-progress for this software. Cancel or wait " f"for tasks to finish before removing the connector." ) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.connector_deleted", status="success", resource_type="software_connector", resource_id=str(connector_id), detail={"software_id": str(row.software_id)}, ) def list_available_connector_types(self) -> list[IntegrationDriverMetadata]: """Driver catalog, scoped to software_provisioning drivers. The admin UI uses this to render the "pick an integration type" step of the binding form. Reuses the central integration registry so there's exactly one source of truth for what providers exist. """ return [ driver_cls.metadata() for (kind, _provider), driver_cls in _integration_registry().items() if kind == "software_provisioning" ] # ───────────────────────── Policy CRUD ───────────────────────── async def list_policies( self, *, department_name: str | None = None, active_only: bool = False, ) -> list[ProvisionPolicyResponse]: q = select(ProvisionPolicy) if active_only: q = q.where(ProvisionPolicy.is_active.is_(True)) if department_name is not None: q = q.where(ProvisionPolicy.department_name == department_name) rows = list((await self.db.execute(q.order_by(ProvisionPolicy.created_at.desc()))).scalars().all()) return await self._enrich_policies(rows) async def create_policy( self, body: ProvisionPolicyCreate, *, trace_id: str, actor_user_id: UUID, ) -> ProvisionPolicyResponse: if body.target_type not in {"seat", "permission"}: raise AppException( code="INVALID_TARGET_TYPE", message="target_type must be 'seat' or 'permission'", status_code=400, ) # Validate target exists if body.target_type == "seat": plan = ( await self.db.execute( select(SoftwareLicensePlan).where(SoftwareLicensePlan.id == body.target_id) ) ).scalar_one_or_none() if not plan: raise AppException( code="TARGET_NOT_FOUND", message=f"License plan {body.target_id} not found", status_code=400, ) elif body.target_type == "permission": from app.domains.permission.models import SoftwarePermission perm = ( await self.db.execute( select(SoftwarePermission).where(SoftwarePermission.id == body.target_id) ) ).scalar_one_or_none() if not perm: raise AppException( code="TARGET_NOT_FOUND", message=f"Software permission {body.target_id} not found", status_code=400, ) # Enforce uniqueness on the full match-key + target tuple. existing = ( await self.db.execute( select(ProvisionPolicy).where( and_( ProvisionPolicy.department_name == body.department_name, ProvisionPolicy.match_job_title == body.match_job_title, ProvisionPolicy.match_location == body.match_location, ProvisionPolicy.target_type == body.target_type, ProvisionPolicy.target_id == body.target_id, ) ) ) ).scalar_one_or_none() if existing: raise AppException( code="POLICY_ALREADY_EXISTS", message="A policy with this match key + target already exists", status_code=400, ) row = ProvisionPolicy( department_name=body.department_name, match_job_title=body.match_job_title, match_location=body.match_location, target_type=body.target_type, target_id=body.target_id, is_active=body.is_active, notes=body.notes, created_by=actor_user_id, ) self.db.add(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.policy_created", status="success", resource_type="provision_policy", resource_id=str(row.id), detail={ "department_name": body.department_name, "match_job_title": body.match_job_title, "match_location": body.match_location, "target_type": body.target_type, "target_id": str(body.target_id), }, ) enriched = await self._enrich_policies([row]) return enriched[0] async def update_policy( self, policy_id: UUID, body: ProvisionPolicyUpdate, *, trace_id: str, actor_user_id: UUID, ) -> ProvisionPolicyResponse: row = await self._get_policy(policy_id) changes: dict[str, Any] = {} # Collect match-key changes and re-check uniqueness once against # the post-update tuple. match_updates: dict[str, Any] = {} if body.department_name is not None: match_updates["department_name"] = body.department_name if body.match_job_title is not None: match_updates["match_job_title"] = body.match_job_title if body.match_location is not None: match_updates["match_location"] = body.match_location if match_updates: effective = { "department_name": row.department_name, "match_job_title": row.match_job_title, "match_location": row.match_location, **match_updates, } conflict = ( await self.db.execute( select(ProvisionPolicy).where( and_( ProvisionPolicy.id != row.id, ProvisionPolicy.department_name == effective["department_name"], ProvisionPolicy.match_job_title == effective["match_job_title"], ProvisionPolicy.match_location == effective["match_location"], ProvisionPolicy.target_type == row.target_type, ProvisionPolicy.target_id == row.target_id, ) ) ) ).scalar_one_or_none() if conflict: raise AppException( code="POLICY_ALREADY_EXISTS", message="Another policy already covers this match key + target", status_code=400, ) for k, v in match_updates.items(): changes[k] = (getattr(row, k), v) setattr(row, k, v) if body.is_active is not None: changes["is_active"] = body.is_active row.is_active = body.is_active if body.notes is not None: row.notes = body.notes changes["notes_updated"] = True await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.policy_updated", status="success", resource_type="provision_policy", resource_id=str(row.id), detail=changes, ) enriched = await self._enrich_policies([row]) return enriched[0] async def delete_policy( self, policy_id: UUID, *, trace_id: str, actor_user_id: UUID, ) -> None: row = await self._get_policy(policy_id) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.policy_deleted", status="success", resource_type="provision_policy", resource_id=str(policy_id), detail={ "department_name": row.department_name, "target_type": row.target_type, "target_id": str(row.target_id), }, ) # ───────────────────────── Onboarding orchestrator ───────────────────────── async def preview_onboarding( self, person_id: UUID, ) -> OnboardingPreview: """Compute (but do not execute) the provisioning items a person would receive based on current policies. Marks items already provisioned so the admin can see duplicates before confirming. """ person = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id == person_id) ) ).scalar_one_or_none() if not person: raise AppException( code="PERSON_NOT_FOUND", message=f"Person {person_id} not found", status_code=404, ) department = person.primary_department policies = await self._match_policies_for_person(person) items = await self._build_preview_items(person_id, policies) return OnboardingPreview( person_id=person.id, person_name=person.display_name, person_email=person.work_email, person_employee_no=person.employee_no, person_employment_status=person.employment_status, department=department, items=items, ) async def execute_onboarding_for_person( self, person_id: UUID, *, trace_id: str, actor_user_id: UUID | None, source: str = SOURCE_ONBOARDING, ) -> list[ProvisioningTask]: """Create pending provisioning_tasks for every applicable policy that isn't already satisfied. Does NOT dispatch them (call execute_task / bulk_execute afterwards). Idempotent: safe to call twice for the same person — skips already-provisioned targets. """ person = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id == person_id) ) ).scalar_one_or_none() if not person: return [] # Leaver invariant: never mint GRANT tasks for someone who is no # longer an active employee (soft-deleted rows, left / # leaver_pending states). The admin execute endpoint and event # hooks all funnel through here. if person.deleted_at is not None or person.employment_status != "active": logger.warning( "execute_onboarding_for_person: refusing GRANT task creation " "for %s (status=%s, deleted=%s)", person_id, person.employment_status, bool(person.deleted_at), ) return [] department = person.primary_department policies = await self._match_policies_for_person(person) items = await self._build_preview_items(person_id, policies) # Defer every new-hire task until 00:00 on the hire's first day in # their own timezone — accounts come alive at local midnight of the # start date rather than the moment IT confirms the sync. NULL when # hire_date is unknown (run ASAP). Mirrors the offboarding gate. onboarding_gate = _onboarding_not_before(person) created_tasks: list[ProvisioningTask] = [] for item in items: if item.already_provisioned: continue # Skip if there's already an active pending/in-progress task for this target dup = ( await self.db.execute( select(ProvisioningTask).where( and_( ProvisioningTask.user_id == person_id, ProvisioningTask.target_type == item.target_type, ProvisioningTask.target_id == item.target_id, ProvisioningTask.action == ACTION_GRANT, ProvisioningTask.status.in_([ STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED, ]), ) ) ) ).scalars().first() # ``.first()`` (not ``scalar_one_or_none``): nothing serializes # concurrent task creation for a person and there is no unique # constraint on (user_id, target_type, target_id, action, status), # so two overlapping onboarding/apply flows can leave two # non-terminal GRANT rows. scalar_one_or_none would then raise # MultipleResultsFound and 500 this person on every subsequent # run — presence is all this idempotency guard needs (mirrors # ``_has_inflight_task``). if dup: continue task = await self.create_task(ProvisioningTaskCreate( user_id=person_id, software_id=item.software_id, action=ACTION_GRANT, target_type=item.target_type, target_id=item.target_id, source=source, context_data={ "policy_id": str(item.policy_id), "matched_department": department, }, not_before=onboarding_gate, )) created_tasks.append(task) if created_tasks: await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="provisioning.onboarding_tasks_created", status="success", resource_type="identity_person", resource_id=str(person_id), detail={ "source": source, "task_count": len(created_tasks), "tasks": [str(t.id) for t in created_tasks], }, ) else: # Zero tasks created — emit an explicit audit row so IT can tell # "provisioned nothing because no policy matched" apart from # "this hire was never processed". Without it, a hire who # matches no ProvisionPolicy (none of the department × title × # location rules, and no NULL-wildcard catch-all) is silently # under-provisioned with no trace — a real risk at 500-person # scale. Distinguish the three zero causes precisely for triage. if not policies: reason = "no_matching_policy" elif not items: # Policies matched but their targets didn't resolve to an # actionable item (e.g. the bound plan/permission row is # gone) — a misconfiguration, not "nothing to do". reason = "policy_without_resolvable_target" else: # Items existed but every one was skipped — either already # an active allocation OR an in-flight pending/in-progress # task already covers it. "satisfied" covers both so the # triage row doesn't claim "provisioned" when grants are # merely queued and not yet dispatched. reason = "all_targets_already_satisfied" await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="provisioning.onboarding_no_tasks", status="success", resource_type="identity_person", resource_id=str(person_id), detail={ "source": source, "reason": reason, "matched_policy_count": len(policies), "matched_department": department, }, ) return created_tasks async def reschedule_pending_onboarding_tasks_for_hire_date_change( self, person_id: UUID, *, trace_id: str, ) -> int: """Re-anchor unstarted onboarding grants after HRIS corrects a date. The onboarding process state machine owns its own date, but its pending auto-grant tasks persist a concrete ``not_before`` timestamp. A date correction must update that timestamp too; otherwise a task created for an earlier date can run before the corrected arrival, or a task for a later date can remain unnecessarily deferred. Only untouched auto/onboarding ``pending`` grants move — in-progress, completed, and manual-notification work retains its operator-visible history. """ person = await self.db.get(IdentityPerson, person_id) if person is None: return 0 not_before = _onboarding_not_before(person) tasks = ( await self.db.execute( select(ProvisioningTask) .where( ProvisioningTask.user_id == person_id, ProvisioningTask.source == SOURCE_ONBOARDING, ProvisioningTask.action == ACTION_GRANT, ProvisioningTask.status == STATUS_PENDING, ) .with_for_update() ) ).scalars().all() changed = 0 for task in tasks: if task.not_before != not_before: task.not_before = not_before changed += 1 if changed: await self.audit.log( trace_id=trace_id, actor_type="system", action="provisioning.onboarding_tasks_rescheduled", status="success", resource_type="identity_person", resource_id=str(person_id), detail={ "task_count": changed, "hire_date": person.hire_date.isoformat() if person.hire_date else None, "not_before": not_before.isoformat() if not_before else None, }, ) return changed # ───────────────────────── Bulk task execution ───────────────────────── async def bulk_execute( self, body: BulkExecuteRequest, *, trace_id: str, actor_user_id: UUID, ) -> BulkExecuteResponse: """Dispatch a batch of pending tasks. Narrowed by explicit task_ids, or by (source, user_id) filters. Only tasks in status=pending are picked up; anything else is skipped. """ conditions = [ProvisioningTask.status == STATUS_PENDING] if body.task_ids: conditions.append(ProvisioningTask.id.in_(body.task_ids)) else: if body.source: conditions.append(ProvisioningTask.source == body.source) if body.user_id: conditions.append(ProvisioningTask.user_id == body.user_id) tasks = list( (await self.db.execute(select(ProvisioningTask).where(and_(*conditions)))).scalars().all() ) # Pre-launch hardening 2026-05-06: COMMIT after each task. Without # this, ``execute_task`` (which calls real grant/revoke driver # APIs against Slack/Google/Zoom/Jira/GitHub) makes external # state changes that survive a request-level rollback while the # local ``status=COMPLETED`` + ``external_ref`` rollback with # the surrounding session — leaving real cloud-side grants # without a Helix record (shadow grants). Per-task commit makes # each connector outcome durable independently of subsequent # iterations. completed = failed = notified = skipped = errored = transient_retried = 0 for task in tasks: # Pre-launch P1 fix: previously a single ``execute_task`` # exception (plan deleted mid-run, FOR UPDATE deadlock, # network drop in ``_invoke_connector``) tore down the # whole batch loop, dropping the per-batch audit row and # leaving the operator with no completion signal — even # though the per-task commits already made earlier outcomes # durable. Now: catch + count + continue, so the batch # always completes and the summary truthfully reports # ``errored`` count alongside completed/failed/notified. try: await self.execute_task(task, trace_id=trace_id, actor_user_id=actor_user_id) except Exception as exc: # noqa: BLE001 errored += 1 logger.exception( "bulk_execute: task %s raised %s — leaving in current " "status %s, continuing batch", task.id, type(exc).__name__, task.status, ) # Roll back any mid-task ORM state attached to this # session iteration so the next task's commit isn't # contaminated. try: await self.db.rollback() except Exception: # noqa: BLE001 pass continue # Capture outcome before committing — ``commit()`` expires # attributes, and we must only count a task once its status # flip is durably persisted (see the mirror fix in # ``execute_pending_batch``). Post-review fix 2026-06-14. outcome = task.status outcome_not_before = task.not_before # Commit per-task so a downstream failure (network drop, # request timeout) doesn't roll back the prior tasks' # state mutations + audit rows. A commit failure on one task # must not tear down the whole batch (and lose the per-batch # audit row below) — roll back, count it as ``errored``, and # carry on, mirroring the connector-exception branch above. try: await self.db.commit() except Exception as exc: # noqa: BLE001 errored += 1 logger.exception( "bulk_execute: task %s post-commit failed %s — " "status flip rolled back, continuing batch", task.id, type(exc).__name__, ) try: await self.db.rollback() except Exception: # noqa: BLE001 pass continue if outcome == STATUS_COMPLETED: completed += 1 elif outcome == STATUS_FAILED: failed += 1 elif outcome == STATUS_NOTIFIED: notified += 1 elif outcome == STATUS_PENDING and outcome_not_before is not None: # Transient connector error re-queued with backoff — the task # is armed to re-hit the vendor shortly, NOT terminal. Mirror # ``execute_pending_batch``: counting it as ``skipped`` told # the operator "nothing to do" when the task is still live. transient_retried += 1 else: skipped += 1 await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.bulk_executed", status="success", resource_type="provisioning_task_batch", resource_id=None, detail={ "source_filter": body.source, "user_id_filter": str(body.user_id) if body.user_id else None, "task_ids": [str(i) for i in (body.task_ids or [])] or None, "executed": len(tasks), "completed": completed, "failed": failed, "notified": notified, "skipped": skipped, "errored": errored, "transient_retried": transient_retried, }, ) return BulkExecuteResponse( executed=len(tasks), completed=completed, failed=failed, notified=notified, skipped=skipped, errored=errored, transient_retried=transient_retried, ) # ───────────────────────── Policy helpers ───────────────────────── async def _get_policy(self, policy_id: UUID) -> ProvisionPolicy: row = ( await self.db.execute( select(ProvisionPolicy).where(ProvisionPolicy.id == policy_id) ) ).scalar_one_or_none() if not row: raise AppException( code="POLICY_NOT_FOUND", message="Provision policy not found", status_code=404, ) return row async def _match_policies_for_department( self, department: str | None, ) -> list[ProvisionPolicy]: """Backward-compatible wrapper: match by department only, with job_title / location unrestricted. Kept for the preview endpoint that hasn't been updated to pass a full person. New code should call :meth:`_match_policies_for_person` instead — it applies all three match keys (department + job_title + location). """ return await self._match_policies_for_attrs( department=department, job_title=None, location=None, strict_null_on_missing=True, ) async def _match_policies_for_person( self, person: IdentityPerson, ) -> list[ProvisionPolicy]: """Return every active policy whose match keys are satisfied by this person. For each of (department_name, match_job_title, match_location), the rule column may be NULL (wildcard) or equal to the person's corresponding attribute. Multiple rules can match a single person — callers dedup the resulting target set by (target_type, target_id). """ return await self._match_policies_for_attrs( department=person.primary_department, job_title=person.job_title, location=person.work_location_code, strict_null_on_missing=False, ) async def _match_policies_for_attrs( self, *, department: str | None, job_title: str | None, location: str | None, strict_null_on_missing: bool, ) -> list[ProvisionPolicy]: # Canonical match semantics live in app/core/provision_match so # the onboarding readiness scanner answers "what do policies # promise this person" with the exact same WHERE clause. conditions = policy_match_conditions( ProvisionPolicy, department=department, job_title=job_title, location=location, strict_null_on_missing=strict_null_on_missing, ) rows = list( (await self.db.execute(select(ProvisionPolicy).where(and_(*conditions)))).scalars().all() ) # Dedup by (target_type, target_id) — union across multiple matches. seen: set[tuple[str, UUID]] = set() deduped: list[ProvisionPolicy] = [] for r in rows: key = (r.target_type, r.target_id) if key in seen: continue seen.add(key) deduped.append(r) return deduped async def _build_preview_items( self, person_id: UUID, policies: list[ProvisionPolicy], ) -> list[OnboardingPreviewItem]: from app.domains.asset.models import SoftwareSeatAllocation plan_ids = [p.target_id for p in policies if p.target_type == "seat"] plans_by_id: dict[UUID, SoftwareLicensePlan] = {} if plan_ids: rows = ( await self.db.execute( select(SoftwareLicensePlan).where(SoftwareLicensePlan.id.in_(plan_ids)) ) ).scalars().all() plans_by_id = {p.id: p for p in rows} software_ids = {p.software_id for p in plans_by_id.values()} softwares = await self._load_softwares(software_ids) # Check already-active seat allocations active_plan_ids: set[UUID] = set() if plan_ids: active = ( await self.db.execute( select(SoftwareSeatAllocation).where( and_( SoftwareSeatAllocation.user_id == person_id, SoftwareSeatAllocation.plan_id.in_(plan_ids), SoftwareSeatAllocation.status == "active", ) ) ) ).scalars().all() active_plan_ids = {a.plan_id for a in active} # Permissions: load permission catalog + already-granted user_permissions async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_2869_step: from app.domains.permission.models import ( PERM_ACTIVE, PERM_PENDING, SoftwarePermission, UserPermission, ) perm_ids = [p.target_id for p in policies if p.target_type == "permission"] perms_by_id: dict[UUID, SoftwarePermission] = {} if perm_ids: perm_rows = ( await self.db.execute( select(SoftwarePermission).where(SoftwarePermission.id.in_(perm_ids)) ) ).scalars().all() perms_by_id = {p.id: p for p in perm_rows} perm_software_ids = {p.software_id for p in perms_by_id.values()} if perm_software_ids: extra_sw = ( await self.db.execute( select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id.in_(perm_software_ids)) ) ).scalars().all() for sw in extra_sw: softwares[sw.id] = sw active_perm_ids: set[UUID] = set() if perm_ids: actives = ( await self.db.execute( select(UserPermission).where( and_( UserPermission.user_id == person_id, UserPermission.permission_id.in_(perm_ids), UserPermission.status.in_([PERM_ACTIVE, PERM_PENDING]), ) ) ) ).scalars().all() active_perm_ids = {a.permission_id for a in actives} if not be_2869_step.ok: logger.warning( "get_onboarding_preview: failed to load permission data " "(person_id=%s) — preview will show permissions as not-yet-provisioned", person_id, ) perms_by_id = {} active_perm_ids = set() items: list[OnboardingPreviewItem] = [] for pol in policies: if pol.target_type == "seat": plan = plans_by_id.get(pol.target_id) if not plan: continue software = softwares.get(plan.software_id) target_label = f"{software.name if software else '?'} — {plan.name}" items.append(OnboardingPreviewItem( policy_id=pol.id, target_type="seat", target_id=pol.target_id, target_label=target_label, software_id=plan.software_id, software_name=software.name if software else None, already_provisioned=pol.target_id in active_plan_ids, )) elif pol.target_type == "permission": perm = perms_by_id.get(pol.target_id) if not perm: continue software = softwares.get(perm.software_id) target_label = f"{software.name if software else '?'} — {perm.display_name}" items.append(OnboardingPreviewItem( policy_id=pol.id, target_type="permission", target_id=pol.target_id, target_label=target_label, software_id=perm.software_id, software_name=software.name if software else None, already_provisioned=pol.target_id in active_perm_ids, )) return items # ───────────────────────── Role change diff ───────────────────────── async def preview_role_change( self, person_id: UUID, ) -> RoleChangePreview: """Compute the diff between the policies that should apply to this person (based on their current department) and what they currently have. Returns items to grant (missing) and items to revoke (no longer covered by policy).""" from app.domains.asset.models import SoftwareSeatAllocation person = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id == person_id) ) ).scalar_one_or_none() if not person: raise AppException( code="PERSON_NOT_FOUND", message=f"Person {person_id} not found", status_code=404, ) department = person.primary_department policies = await self._match_policies_for_person(person) # Expected: plan_ids / permission_ids from policies expected_plan_ids = {p.target_id for p in policies if p.target_type == "seat"} expected_perm_ids = {p.target_id for p in policies if p.target_type == "permission"} # Current seats current_seats = list(( await self.db.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan, AssetSoftwareCatalog) .join(SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id) .join(AssetSoftwareCatalog, SoftwareLicensePlan.software_id == AssetSoftwareCatalog.id) .where( and_( SoftwareSeatAllocation.user_id == person_id, SoftwareSeatAllocation.status == "active", ) ) ) ).all()) current_seat_plan_ids = {row[0].plan_id for row in current_seats} # Current permissions async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_2999_step: from app.domains.permission.models import ( PERM_ACTIVE, PERM_PENDING, SoftwarePermission, UserPermission, ) up_rows = list(( await self.db.execute( select(UserPermission, SoftwarePermission, AssetSoftwareCatalog) .join(SoftwarePermission, UserPermission.permission_id == SoftwarePermission.id) .join(AssetSoftwareCatalog, SoftwarePermission.software_id == AssetSoftwareCatalog.id) .where( and_( UserPermission.user_id == person_id, UserPermission.status.in_([PERM_ACTIVE, PERM_PENDING]), ) ) ) ).all()) current_perm_ids = {row[0].permission_id for row in up_rows} if not be_2999_step.ok: logger.warning( "get_role_change_diff: failed to load permission data " "(person_id=%s) — diff will omit permission deltas", person_id, ) up_rows = [] current_perm_ids = set() # Diff to_add_plan_ids = expected_plan_ids - current_seat_plan_ids to_remove_plan_ids = current_seat_plan_ids - expected_plan_ids to_add_perm_ids = expected_perm_ids - current_perm_ids to_remove_perm_ids = current_perm_ids - expected_perm_ids items_to_add: list[ReconcileDiffItem] = [] items_to_remove: list[ReconcileDiffItem] = [] # Load plan details for all ids all_plan_ids = to_add_plan_ids | to_remove_plan_ids plans_by_id = {} if all_plan_ids: plans = ( await self.db.execute( select(SoftwareLicensePlan, AssetSoftwareCatalog) .join(AssetSoftwareCatalog, SoftwareLicensePlan.software_id == AssetSoftwareCatalog.id) .where(SoftwareLicensePlan.id.in_(all_plan_ids)) ) ).all() plans_by_id = {p.id: (p, sw) for p, sw in plans} for pid in to_add_plan_ids: bag = plans_by_id.get(pid) if not bag: continue plan, sw = bag items_to_add.append(ReconcileDiffItem( action="grant", target_type="seat", target_id=pid, target_label=f"{sw.name} — {plan.name}", software_id=sw.id, software_name=sw.name, reason="policy_match", )) for pid in to_remove_plan_ids: bag = plans_by_id.get(pid) if not bag: continue plan, sw = bag items_to_remove.append(ReconcileDiffItem( action="revoke", target_type="seat", target_id=pid, target_label=f"{sw.name} — {plan.name}", software_id=sw.id, software_name=sw.name, reason="policy_unmatch", )) # Permissions. The ``permission`` domain is optional — guard # the import only. DB errors during the SELECT must propagate # so partial results don't masquerade as a clean preview. try: from app.domains.permission.models import SoftwarePermission except ImportError: logger.info( "permission domain not installed; role-change preview " "will omit permission-level diff", ) else: all_perm_ids = to_add_perm_ids | to_remove_perm_ids if all_perm_ids: perms = ( await self.db.execute( select(SoftwarePermission, AssetSoftwareCatalog) .join(AssetSoftwareCatalog, SoftwarePermission.software_id == AssetSoftwareCatalog.id) .where(SoftwarePermission.id.in_(all_perm_ids)) ) ).all() perm_by_id = {p.id: (p, sw) for p, sw in perms} for pid in to_add_perm_ids: bag = perm_by_id.get(pid) if not bag: continue p, sw = bag items_to_add.append(ReconcileDiffItem( action="grant", target_type="permission", target_id=pid, target_label=f"{sw.name} — {p.display_name}", software_id=sw.id, software_name=sw.name, reason="policy_match", )) for pid in to_remove_perm_ids: bag = perm_by_id.get(pid) if not bag: continue p, sw = bag items_to_remove.append(ReconcileDiffItem( action="revoke", target_type="permission", target_id=pid, target_label=f"{sw.name} — {p.display_name}", software_id=sw.id, software_name=sw.name, reason="policy_unmatch", )) return RoleChangePreview( person_id=person.id, person_name=person.display_name, person_email=person.work_email, person_employee_no=person.employee_no, person_employment_status=person.employment_status, current_department=department, policies_matched=len(policies), items_to_add=items_to_add, items_to_remove=items_to_remove, ) async def apply_role_change( self, person_id: UUID, body: ReconcileApplyRequest, *, trace_id: str, actor_user_id: UUID, ) -> int: """Create provisioning tasks for a chosen subset of the diff. Returns the number of tasks created.""" preview = await self.preview_role_change(person_id) add_set = set(body.add_target_ids) remove_set = set(body.remove_target_ids) # Leaver invariant: GRANT tasks may only be minted for an ACTIVE, # non-deleted employee. When a person enters offboarding # (leaver_pending / left) their seats are revoked, so the policy diff # RE-SURFACES those seats as "to add" — clicking Apply here would mint # GRANT tasks that re-provision paid seats/permissions to a departing # employee and fight the offboarding flow. execute_task never re-gates # on employment status, so this is the gate. REVOKE tasks are still # allowed (you DO want to revoke a leaver's access). Mirrors the # active-employee guard in execute_onboarding_for_person / # bundle apply_to_person — this task-creating path was missing it. person = await self.db.get(IdentityPerson, person_id) grants_allowed = ( person is not None and person.deleted_at is None and person.employment_status == "active" ) count = 0 for item in preview.items_to_add: if item.target_id not in add_set: continue if not grants_allowed: # Skip grant minting for a non-active / deleted person; any # selected revokes below still proceed. continue # Dedup against an already-queued task for the same target + # action. The preview diff is computed against ACTIVE seat / # permission allocations only — a still-pending provisioning # task is not yet an allocation, so it is invisible to the # diff. Without this guard a second "Apply" click (or an MCP # retry) before the dispatcher tick stacks N duplicate GRANT # tasks → N duplicate vendor grant_seat calls → duplicate paid # seats. Mirrors the dedup in execute_onboarding_for_person / # bundle apply_to_person (this was the one task-creating path # missing it). if await self._has_inflight_task( person_id, item.target_type, item.target_id, ACTION_GRANT, ): continue await self.create_task(ProvisioningTaskCreate( user_id=person_id, software_id=item.software_id, action=ACTION_GRANT, target_type=item.target_type, target_id=item.target_id, source=SOURCE_ROLE_CHANGE, context_data={"role_change": True}, )) count += 1 for item in preview.items_to_remove: if item.target_id not in remove_set: continue if await self._has_inflight_task( person_id, item.target_type, item.target_id, ACTION_REVOKE, ): continue await self.create_task(ProvisioningTaskCreate( user_id=person_id, software_id=item.software_id, action=ACTION_REVOKE, target_type=item.target_type, target_id=item.target_id, source=SOURCE_ROLE_CHANGE, context_data={"role_change": True}, )) count += 1 if count: await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.role_change_applied", status="success", resource_type="identity_person", resource_id=str(person_id), detail={ "tasks_created": count, "added_count": len(add_set), "removed_count": len(remove_set), }, ) return count # ───────────────────────── Billing ───────────────────────── async def compute_billing( self, *, user_id: UUID | None = None, user_ids: list[UUID] | None = None, ) -> BillingAggregate: """Compute current monthly cost from every active seat allocation. Normalisation: - monthly → price - yearly → price / 12 - one_time → 0 (one-time charges aren't recurring) ``user_id`` / ``user_ids`` narrow the scope for the employee self-service billing view (``/my`` and ``/my?scope=subordinates``). Admin callers pass neither and get the whole org. Currency: aggregation refuses to silently mix. Each line carries its plan's currency; ``by_currency`` exposes the per-currency sum. ``total_monthly_cost`` is populated only when every active seat is in a single currency; multi-currency tenants get ``total_monthly_cost=0.0`` plus ``currency_warning=True`` on the aggregate (the legacy field stays a float so the schema contract — ``float`` — doesn't break older clients). """ from sqlalchemy.orm import selectinload from app.domains.asset.models import SoftwareSeatAllocation stmt = ( select(SoftwareSeatAllocation, SoftwareLicensePlan, AssetSoftwareCatalog, IdentityPerson) .join(SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id) .join(AssetSoftwareCatalog, SoftwareLicensePlan.software_id == AssetSoftwareCatalog.id) .join(IdentityPerson, SoftwareSeatAllocation.user_id == IdentityPerson.id) .options(selectinload(IdentityPerson.org_memberships)) .where(SoftwareSeatAllocation.status == "active") ) # Narrow to the caller's scope. ``user_ids`` takes precedence # over ``user_id`` so callers can always pass a list (eg. the # BFS-expanded subordinate set) without having to special-case # single-user lookups. if user_ids is not None: stmt = stmt.where(SoftwareSeatAllocation.user_id.in_(user_ids)) elif user_id is not None: stmt = stmt.where(SoftwareSeatAllocation.user_id == user_id) rows = list((await self.db.execute(stmt)).all()) def monthly(price: float, cycle: str) -> float: if cycle == "monthly": return float(price) if cycle == "yearly": return float(price) / 12 return 0.0 lines: list[BillingLineItem] = [] by_currency: dict[str, float] = {} by_sw: dict[UUID, dict] = {} by_dep: dict[str, dict] = {} for seat, plan, sw, person in rows: cost = monthly(plan.price, plan.billing_cycle) currency = (plan.currency or "CNY").upper() by_currency[currency] = by_currency.get(currency, 0.0) + cost leaf_dep = person.primary_department or "未分配" lines.append(BillingLineItem( seat_id=seat.id, user_id=person.id, user_name=person.display_name, department=leaf_dep, software_id=sw.id, software_name=sw.name, plan_id=plan.id, plan_name=plan.name, price=float(plan.price), billing_cycle=plan.billing_cycle, currency=currency, monthly_cost=cost, )) sw_bucket = by_sw.setdefault(sw.id, { "software_id": str(sw.id), "software_name": sw.name, "monthly_cost": 0.0, "seat_count": 0, "currency": currency, "currency_warning": False, "by_currency": {}, }) sw_bucket["monthly_cost"] += cost sw_bucket["seat_count"] += 1 sw_bucket["by_currency"][currency] = ( sw_bucket["by_currency"].get(currency, 0.0) + cost ) # If a single software somehow spans multiple currencies # (e.g. an admin reassigned a plan's currency mid-month), # surface the warning via a structured field instead of # poisoning the ``currency`` field with a magic-string. # The UI can render `${monthly_cost} ${currency}` safely # for the happy path; mixed-currency tenants render via # by_currency. (Round 2 logic-scan P1.) if sw_bucket["currency"] != currency: sw_bucket["currency_warning"] = True # Bump every ancestor department so "Business Operations" # subsumes everyone under it (including "IT Services & # Security"). Falls back to the leaf name when the org path # is missing (mostly pre-sync fixtures). chain = person.department_chain or [leaf_dep] for idx, dep in enumerate(chain): is_leaf = idx == len(chain) - 1 bucket = by_dep.setdefault(dep, { "department": dep, "parent_department": chain[idx - 1] if idx > 0 else None, "depth": idx, "monthly_cost": 0.0, "seat_count": 0, "direct_monthly_cost": 0.0, "direct_seat_count": 0, }) bucket["monthly_cost"] += cost bucket["seat_count"] += 1 if is_leaf: bucket["direct_monthly_cost"] += cost bucket["direct_seat_count"] += 1 # Single-currency: surface the legacy flat total. Multi-currency: # zero out the legacy field and set ``currency_warning=True`` — # callers MUST read ``by_currency`` instead. We keep # ``total_monthly_cost`` as a float (not Optional) because the # schema contract is ``float`` and a ``None`` would force every # existing dashboard to handle nulls; the warning flag is the # explicit signal that mixing happened. if len(by_currency) <= 1: single_total = round(next(iter(by_currency.values()), 0.0), 2) currency_warning = False else: single_total = 0.0 currency_warning = True return BillingAggregate( total_monthly_cost=single_total, currency_warning=currency_warning, by_currency={k: round(v, 2) for k, v in by_currency.items()}, total_currency_count=len(by_currency), seat_count=len(lines), by_software=sorted(by_sw.values(), key=lambda x: x["monthly_cost"], reverse=True), by_department=sorted(by_dep.values(), key=lambda x: (x["depth"], -x["monthly_cost"])), lines=lines, ) async def _enrich_policies( self, rows: list[ProvisionPolicy], ) -> list[ProvisionPolicyResponse]: plan_ids = {r.target_id for r in rows if r.target_type == "seat"} plans_by_id: dict[UUID, SoftwareLicensePlan] = {} if plan_ids: plans = ( await self.db.execute( select(SoftwareLicensePlan).where(SoftwareLicensePlan.id.in_(plan_ids)) ) ).scalars().all() plans_by_id = {p.id: p for p in plans} software_ids = {p.software_id for p in plans_by_id.values()} # Permission lookup. The ``permission`` domain is optional in # earlier deployments — guard the import so callers running on # older codebases still get useful (if perm-less) responses. # DB errors must propagate so we don't silently lose data. perm_ids = {r.target_id for r in rows if r.target_type == "permission"} perms_by_id: dict[UUID, Any] = {} if perm_ids: try: from app.domains.permission.models import SoftwarePermission except ImportError: logger.info( "permission domain not installed; provisioning preview " "will skip permission-target rows", ) else: perms = ( await self.db.execute( select(SoftwarePermission).where(SoftwarePermission.id.in_(perm_ids)) ) ).scalars().all() perms_by_id = {p.id: p for p in perms} for p in perms: software_ids.add(p.software_id) softwares = await self._load_softwares(software_ids) creator_ids = {r.created_by for r in rows if r.created_by} creators = await self._load_users(creator_ids) responses = [] for r in rows: label = None sw_id = None sw_name = None if r.target_type == "seat": plan = plans_by_id.get(r.target_id) if plan: sw_id = plan.software_id sw = softwares.get(plan.software_id) sw_name = sw.name if sw else None label = f"{sw_name or '?'} — {plan.name}" elif r.target_type == "permission": perm = perms_by_id.get(r.target_id) if perm: sw_id = perm.software_id sw = softwares.get(perm.software_id) sw_name = sw.name if sw else None label = f"{sw_name or '?'} — {perm.display_name}" creator = creators.get(r.created_by) if r.created_by else None responses.append(ProvisionPolicyResponse( id=r.id, department_name=r.department_name, match_job_title=r.match_job_title, match_location=r.match_location, target_type=r.target_type, target_id=r.target_id, target_label=label, software_id=sw_id, software_name=sw_name, is_active=r.is_active, notes=r.notes, created_by=r.created_by, created_by_name=creator.display_name if creator else None, created_at=r.created_at, updated_at=r.updated_at, )) return responses # ───────────────────────── Internal helpers ───────────────────────── async def _get_task( self, task_id: UUID, *, lock: bool = False, ) -> ProvisioningTask: # ``lock=True`` adds SELECT ... FOR UPDATE so callers that mutate # status (complete_manual_task, fail_task, the auto-execute path # in execute_task) can read-then-write without a TOCTOU window # that would let two concurrent clients both observe a non-terminal # status and both flip the row, double-firing audit + downstream # side effects. stmt = select(ProvisioningTask).where(ProvisioningTask.id == task_id) if lock: # populate_existing forces a refresh from the locked row — the # task may already be cached in this session from an earlier # unlocked read, in which case a bare with_for_update would # lock yet return the stale instance (identity-map trap). stmt = stmt.with_for_update().execution_options(populate_existing=True) row = (await self.db.execute(stmt)).scalar_one_or_none() if not row: raise AppException( code="TASK_NOT_FOUND", message="Provisioning task not found", status_code=404, ) return row async def _get_connector(self, connector_id: UUID) -> SoftwareConnectorRow: row = ( await self.db.execute( select(SoftwareConnectorRow).where(SoftwareConnectorRow.id == connector_id) ) ).scalar_one_or_none() if not row: raise AppException( code="CONNECTOR_NOT_FOUND", message="Software connector not found", status_code=404, ) return row async def _get_software_owner(self, software_id: UUID) -> UUID | None: row = ( await self.db.execute( select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id == software_id) ) ).scalar_one_or_none() if not row: return None return row.owner_person_id async def _resolve_execution_mode( self, *, software_id: UUID | None, target_type: str | None = None, target_id: UUID | None = None, ) -> str: """Infer execution_mode from the task target. Catalog-software tasks (seat / permission) resolve from the software's binding: 1. software has a binding whose integration provider != 'manual' → auto_api 2. software has sso_provider set → sso_auto 3. binding exists but provider == 'manual' → manual_notify 4. no binding, no sso_provider → manual_notify Account / api_key tasks carry ``software_id=None`` BY DESIGN (they aren't tied to a single catalog row) and so cannot be resolved via the binding table. Before this branch existed the ``software_id is None`` short-circuit below sent them all to ``manual_notify`` — which silently downgraded Jira leaver full-access revoke (#537) from "driver runs ``revoke_all_access`` automatically" to "DM an admin to do it by hand", so the leaver kept Jira access until someone actioned the notification. Resolve them by target instead: * ``account`` — build the driver from the integration id in ``target_id``; auto_api iff it supports full-access revoke, else manual_notify (a provider we can't auto-revoke still needs a human). * ``api_key`` — auto_api: ``_invoke_connector`` short-circuits OK and the real cloud-credential teardown runs in ``_post_success_side_effects`` via ``ApiKeyService.revoke_key``. """ if target_type == "account": if target_id is None: return EXEC_MANUAL_NOTIFY try: driver = await build_driver_by_id(target_id, session=self.db) except Exception: # noqa: BLE001 — bad config → human handles it return EXEC_MANUAL_NOTIFY if ( isinstance(driver, SoftwareProvisioningDriver) and driver.supports_full_access_revoke ): return EXEC_AUTO_API return EXEC_MANUAL_NOTIFY # Per-product access revoke — auto_api iff the driver class # opts in via ``supports_product_access_revoke``. Same shape as # ``account`` above (software_id=None, target_id=integration_id), # different driver method. Without this branch the # ``software_id is None`` short-circuit would silently downgrade # every self-cancel-via-product_access task to manual_notify — # repeating the #544 bug for the new code path. if target_type == "product_access": if target_id is None: return EXEC_MANUAL_NOTIFY try: driver = await build_driver_by_id(target_id, session=self.db) except Exception: # noqa: BLE001 — bad config → human handles it return EXEC_MANUAL_NOTIFY if ( isinstance(driver, SoftwareProvisioningDriver) and driver.supports_product_access_revoke ): return EXEC_AUTO_API return EXEC_MANUAL_NOTIFY if target_type == "api_key": return EXEC_AUTO_API if software_id is None: return EXEC_MANUAL_NOTIFY # Identity-managed catalog rows (Feishu / Google account) — the # lifecycle is owned by an ``identity_sync`` target, not by a # seat-layer driver. Route any task that does sneak through here # to manual_notify so the admin sees it and can sanity-check # whether identity_sync is actually doing its job — calling the # seat driver would either no-op (tenant_users binding rejects # grant/revoke) or duplicate the freeze that identity_sync's # ``deactivate`` already ran (vendor returns "already frozen" # 230002, audit trail forks). Look up the policy from the # catalog row up-front before the binding probe so it wins over # the auto_api branch below. from app.domains.offboarding.models import POLICY_IDENTITY_SYNC_MANAGED software_for_policy = ( await self.db.execute( select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id == software_id) ) ).scalar_one_or_none() if ( software_for_policy is not None and software_for_policy.offboarding_policy == POLICY_IDENTITY_SYNC_MANAGED ): return EXEC_MANUAL_NOTIFY binding = ( await self.db.execute( select(SoftwareConnectorRow).where( and_( SoftwareConnectorRow.software_id == software_id, SoftwareConnectorRow.is_active.is_(True), ) ) ) ).scalar_one_or_none() if binding is not None: integ = await self.db.get(Integration, binding.integration_id) if integ is not None and integ.provider != "manual": # Honour the binding's explicit ``execution_mode`` when the # admin has pinned it to manual_notify. Some providers expose # a SCIM/API surface only on higher plan tiers (Slack Pro has # no SCIM → ``binding_config.execution_mode='manual_notify'``, # see slack_software.binding_schema). Without this, a seat # revoke is forced to auto_api, the driver hard-fails the # plan-gate (``plus_teams_only``), the task lands FAILED, the # offboarding blocker stays RESO_PENDING and the process wedges # at ``awaiting_provisioning`` forever (prod incident 2026-06). # account/product_access paths already gate on the driver # ``supports_*`` capability flags above; this closes the same # gap for the seat path. Only ``manual_notify`` is honoured # here — an absent/auto_api value keeps the auto_api default. cfg = binding.binding_config or {} if cfg.get("execution_mode") == EXEC_MANUAL_NOTIFY: return EXEC_MANUAL_NOTIFY # The binding's integration must actually be a # seat/permission provisioning driver. A catalog row can # be bound to a non-provisioning integration (e.g. Jamf, # kind='mdm', where seat==device) — resolving that to # auto_api sends the task to ``_invoke_connector``, which # then raises ``Binding references non-provisioning # integration`` and the task lands FAILED, wedging the # offboarding blocker (prod: kind='mdm' provider='jamf'). # Gate on the driver type, same as the account / # product_access branches above. A build failure or a # non-provisioning driver degrades to manual_notify so a # human handles it instead of a crash-loop. try: seat_driver = await build_driver_by_id( binding.integration_id, session=self.db, ) except Exception: # noqa: BLE001 — bad config → human return EXEC_MANUAL_NOTIFY if not isinstance(seat_driver, SoftwareProvisioningDriver): return EXEC_MANUAL_NOTIFY return EXEC_AUTO_API if software_for_policy and software_for_policy.sso_provider: return EXEC_SSO_AUTO return EXEC_MANUAL_NOTIFY async def _invoke_connector(self, task: ProvisioningTask) -> ProvisioningResult: """Dispatch one auto_api task to the software's bound driver. Resolution path: 1. Find the active ``software_connectors`` binding for this task's software. 2. Build the integration driver via :func:`build_driver_by_id` (handles decrypt + driver instantiation). 3. Construct ``ProvisioningContext`` with user info + the binding's ``binding_config`` merged into ``extra`` so the driver can read its per-software params. 4. Call the appropriate ``grant_*``/``revoke_*`` based on (action, target_type). (The legacy ``cloud_account`` target type was removed in migration 053; cloud access is now managed by the ``cmdb_cloud_access_grants`` lifecycle, not ProvisioningTask.) """ # PR3-1 short-circuit (2026-05-19): ``target_type='api_key'`` # tasks come from ``OffboardingService.auto_revoke_blocker`` and # carry ``software_id=None`` (api_keys aren't bound to a # Software row in the license catalog). The pre-PR3 shape # raised ``ValueError`` here and the task landed FAILED — admin # saw a stuck blocker and the api_key never got revoked. The # cloud teardown ITSELF lives in ``_post_success_side_effects`` # (which routes into ApiKeyService.revoke_key with # actor_role=SYSTEM_OFFBOARDING for real cloud-side IAM user / # RAM user / AOAI account deletion), so the connector-level # path returns OK and lets the post-success hook do the work. # ``external_ref`` mirrors the assignment id for the audit # row's resource_id link. if task.target_type == "api_key": return ProvisioningResult( ok=True, external_ref=( str(task.target_id) if task.target_id else None ), ) # Full-access revoke (Jira offboarding). The leaver action strips # the WHOLE provider footprint ONCE — every group, not just one # tracked seat/group — so these tasks carry ``software_id=None`` # and stamp the provider integration id in ``target_id``. We build # the driver directly from that integration and call # ``revoke_all_access`` (idempotent; the account itself is kept). if task.target_type == "account": if task.target_id is None: raise ValueError( "full-access-revoke task has no target_id (integration_id)" ) driver = await build_driver_by_id(task.target_id, session=self.db) if ( not isinstance(driver, SoftwareProvisioningDriver) or not driver.supports_full_access_revoke ): raise ValueError( f"Integration {task.target_id} does not support full-" f"access revoke (provider={getattr(driver, 'provider', '?')!r})" ) acct_user = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id == task.user_id ) ) ).scalar_one_or_none() if not acct_user: raise ValueError(f"User {task.user_id} not found") ctx = ProvisioningContext( user_email=acct_user.work_email, user_display_name=acct_user.display_name, extra=dict(task.context_data or {}), ) return await driver.revoke_all_access(ctx) # Per-product access revoke (Jira self-cancel / admin-revoke / # idle-reclaim). Distinct from ``account`` (whole provider, leaver # scorched-earth) and from ``seat`` (single bound group): this # strips one product family's access for a still-employed user by # removing them from every group on this site that confers that # product (Atlassian applicationrole API). The dispatcher carries # ``product_key`` and ``sites`` in ``context_data`` set by the # coordinator from the source connector binding. Task layout # mirrors ``account``: ``software_id=None`` (one task covers a # product family that may span multiple Helix software rows — # Jira Software plus its Marketplace add-ons share the same # vendor-side group set), ``target_id=integration_id`` (the # provider instance the API call lands on). if task.target_type == "product_access": if task.target_id is None: raise ValueError( "product-access-revoke task has no target_id " "(integration_id)" ) driver = await build_driver_by_id(task.target_id, session=self.db) if ( not isinstance(driver, SoftwareProvisioningDriver) or not driver.supports_product_access_revoke ): raise ValueError( f"Integration {task.target_id} does not support per-" f"product access revoke " f"(provider={getattr(driver, 'provider', '?')!r}). " "Coordinator routed a task here but driver class " "lacks supports_product_access_revoke=True." ) pa_user = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id == task.user_id ) ) ).scalar_one_or_none() if not pa_user: raise ValueError(f"User {task.user_id} not found") ctx_extra = dict(task.context_data or {}) # Load admin-declared conferring groups for products # ``/rest/api/3/applicationrole`` doesn't cover (Confluence # — see driver docstring). Setting shape: # { "": ["group-1", "group-2", …], … } # Looked up per task so an admin edit takes effect on the # next task without a worker restart. Round 3 audit P1.4: # narrow except + explicit shape validation. Pre-fix, a # malformed setting silently produced an "unconfigured" # observable identical to setting-absent — admin re-pasted # the same broken JSON forever with no breadcrumb. # Distinguish "absent" from "present but shape-wrong" via a # diagnostic marker the driver propagates into its refusal # message. product_key = str( ctx_extra.get("product_key") or "" ).strip().lower() if product_key: from sqlalchemy.exc import SQLAlchemyError from app.integrations.models import PlatformSetting row = None try: row = await self.db.get( PlatformSetting, "software_seat.product_access_revoke." "admin_declared_conferring_groups", ) # Round 3 Round 2 (P1-1): broaden the except. Pool- # checkout timeouts (``asyncio.TimeoutError``), raw # asyncmy ``OperationalError`` that doesn't inherit # from ``SQLAlchemyError``, and ``OSError`` (broken # pipe to MySQL) all need the same "convert to refusal # marker" treatment so the driver's refusal message # can attribute the failure. We deliberately DO NOT # broaden to ``Exception`` — that would swallow # programmer bugs (NameError / TypeError on a refactor # mistake) and reintroduce the silent-degrade pattern # Round 1's narrow except set out to fix. except (TimeoutError, SQLAlchemyError, OSError) as exc: logger.warning( "product_access dispatch: infra failure loading " "admin_declared_conferring_groups setting: %s", exc, ) ctx_extra["admin_declared_groups_load_error"] = ( f"{type(exc).__name__}: {exc}" ) if row is not None: raw = row.value if not isinstance(raw, dict): ctx_extra["admin_declared_groups_load_error"] = ( "platform_setting value is not a JSON object; " f"got {type(raw).__name__} — admin must re-PUT " "as {product_key: [group_name, ...]}" ) else: groups = raw.get(product_key) if groups is None: # Absent entry for this product_key — no # error, just nothing for this product # (other product_keys may be configured). pass elif not isinstance(groups, list): ctx_extra["admin_declared_groups_load_error"] = ( f"platform_setting entry for product_key=" f"{product_key!r} is not a list; got " f"{type(groups).__name__}" ) else: cleaned = [ str(g).strip() for g in groups if isinstance(g, str) and str(g).strip() ] if not cleaned and not groups: # Round 3 Round 2 (P2-3): explicit # empty list means "admin intentionally # cleared this entry". Surface as a # load_error so the driver refusal # tells the admin the difference # between "entry missing → add it" vs # "entry empty → either fill it or # delete the key entirely". Without # this distinction the empty-list case # rendered an identical refusal to # "key absent" → admin can't tell why # their `{"confluence": []}` PUT does # nothing. ctx_extra["admin_declared_groups_load_error"] = ( f"platform_setting entry for product_key=" f"{product_key!r} is an EMPTY list — " "to remove configuration delete the key " "entirely; to enable revoke add at least " "one conferring group name" ) elif not cleaned: # Non-empty list with all entries # invalid (admin typo'd all of them). ctx_extra["admin_declared_groups_load_error"] = ( f"platform_setting entry for product_key=" f"{product_key!r} has {len(groups)} " "entries but ALL are empty / non-string" ) # Only attach when we actually have # something to use — empty-list attach was # dead-weight (Round 3 Round 3 P2). The # driver checks truthiness; an empty list # would behave identically to "no key" but # cluttered ctx.extra. if cleaned: ctx_extra["admin_declared_conferring_groups"] = ( cleaned ) # Protected org-wide identity groups — the driver refuses to # strip these to reclaim a seat (they're mirrored from the # IdP via SCIM, so removal bounces forever and rips a core # directory group off an active employee). Flat list, # site-agnostic; merged with the driver's built-in defaults # (``employee_in_china`` / ``All users from G Suite``). Same # infra-error tolerance as the admin-declared load above — a # DB blip just falls back to the driver defaults rather than # failing the revoke. from sqlalchemy.exc import SQLAlchemyError as _SQLAlchemyError from app.integrations.models import ( PlatformSetting as _PlatformSetting, ) try: prot_row = await self.db.get( _PlatformSetting, "software_seat.product_access_revoke." "protected_identity_groups", ) except (TimeoutError, _SQLAlchemyError, OSError) as exc: prot_row = None logger.warning( "product_access dispatch: infra failure loading " "protected_identity_groups setting (falling back to " "driver defaults): %s", exc, ) if prot_row is not None and isinstance(prot_row.value, list): prot_clean = [ str(g).strip() for g in prot_row.value if isinstance(g, str) and str(g).strip() ] if prot_clean: ctx_extra["protected_identity_groups"] = prot_clean ctx = ProvisioningContext( user_email=pa_user.work_email, user_display_name=pa_user.display_name, extra=ctx_extra, ) return await driver.revoke_product_access(ctx) if task.software_id is None: raise ValueError("auto_api task has no software_id") binding = ( await self.db.execute( select(SoftwareConnectorRow).where( and_( SoftwareConnectorRow.software_id == task.software_id, SoftwareConnectorRow.is_active.is_(True), ) ) ) ).scalar_one_or_none() if binding is None: raise ValueError( f"No active connector binding configured for software {task.software_id}" ) driver = await build_driver_by_id( binding.integration_id, session=self.db, ) if not isinstance(driver, SoftwareProvisioningDriver): # A catalog row bound to a non-provisioning integration # (e.g. Jamf, kind='mdm', provider='jamf', where seat==device # and there's no seat grant/revoke API). ``_resolve_execution_mode`` # now gates on the driver type at create-time, but a task # already created as auto_api before that fix — or before the # binding was corrected — still lands here. Raising made it # FAILED permanently, wedging the owning offboarding blocker # at RESO_PENDING. Return capability_unsupported instead so # ``execute_task`` downgrades it to manual_notify and pings a # human, exactly like the plan-gated seat path. return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( "Binding references non-provisioning integration " f"(kind={driver.kind!r}, provider={driver.provider!r}); " "no automatic seat grant/revoke API — routed to manual." ), ) user = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id == task.user_id) ) ).scalar_one_or_none() if not user: raise ValueError(f"User {task.user_id} not found") # Leaver invariant (execute-time TOCTOU gate, symmetric to the # successor re-validation below): a GRANT enqueued while the # person was active can fire hours/days later (transient-retry # backoff alone can hold a task ~16h). If the grantee has since # left or been soft-deleted, refusing here is the difference # between a clear 409 and silently minting a vendor seat for a # departed employee that offboarding already revoked. if task.action == ACTION_GRANT and ( user.deleted_at is not None or user.employment_status != "active" ): raise AppException( code="GRANTEE_NO_LONGER_ACTIVE", message=( f"Grant target {user.work_email} is now " f"{user.employment_status!r}" f"{' (soft-deleted)' if user.deleted_at else ''}. " "Refusing to provision a vendor account/seat for an " "inactive person." ), status_code=409, ) plan_name = None plan_row: SoftwareLicensePlan | None = None if task.target_type == "seat" and task.target_id: # ``target_id`` history: older callers set it to the plan_id, # newer callers (``enqueue_grant_for_seat`` / # ``enqueue_revoke_for_seat``, PR-B 2026-04-21) set it to # the seat allocation id and stash ``plan_id`` in # context_data. Handle both conventions so the plan is # always resolved regardless of who queued the task — # needed by ``_maybe_evict_for_zoom`` which can't enforce # capacity without the plan row. from app.domains.asset.models import SoftwareSeatAllocation plan_row = ( await self.db.execute( select(SoftwareLicensePlan).where(SoftwareLicensePlan.id == task.target_id) ) ).scalar_one_or_none() if plan_row is None: seat = await self.db.get(SoftwareSeatAllocation, task.target_id) candidate_plan_id = None if seat is not None: candidate_plan_id = seat.plan_id else: # Last resort: context_data stamped at enqueue time. cd_plan = (task.context_data or {}).get("plan_id") if cd_plan: try: candidate_plan_id = UUID(str(cd_plan)) except (ValueError, TypeError): candidate_plan_id = None if candidate_plan_id: plan_row = ( await self.db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.id == candidate_plan_id, ) ) ).scalar_one_or_none() plan_name = plan_row.name if plan_row else None # Merge binding_config into extra so the driver sees per-software # params (group_email, operation, ...). task.context_data may add # runtime metadata; binding_config takes precedence for its keys # EXCEPT for the trusted-runtime allow-list below — those keys # are stamped by privileged endpoints (e.g. /change-plan sets # ``allow_demote=True`` on the grant task) and MUST NOT be # overridable by an admin who edits the connector binding. If # binding_config could override, an IT admin who PATCHes # ``binding_config={"allow_demote": true}`` once would bypass # the silent-demote refusal forever, defeating the audit-gated # tier-change guarantee. (Pre-launch round 2 security P0.) trusted_runtime_only_keys = _TRUSTED_RUNTIME_ONLY_CONTEXT_KEYS extra: dict[str, Any] = dict(task.context_data or {}) binding_cfg = dict(binding.binding_config or {}) for k in list(binding_cfg.keys()): if k in trusted_runtime_only_keys: # Don't let binding_config shadow trusted runtime keys. binding_cfg.pop(k, None) extra.update(binding_cfg) permission_key = None if task.target_type == "permission": permission_key = (task.context_data or {}).get("permission_key") # Successor (for revoke flows that must hand data to someone # else before the seat can be dropped — Zoom meetings, Google # Drive). Stamped at enqueue time; we resolve the person's # display name here so drivers don't have to hit the DB. # # P1-3 TOCTOU fix (2026-05-08): re-validate the successor is # still an active employee at execute time. The successor was # validated at enqueue time (offboarding scan or admin revoke), # but ``execute_task`` can run hours/days later — long enough # for the picked successor to themselves resign, change roles, # or lose their seat on the source software. Without this gate # the connector tries to hand Drive/Calendar/Zoom ownership to # a leaver_pending or deleted person, which the vendor either # silently accepts (creating a new orphan) or rejects with a # generic 4xx that's hard to attribute. Failing here surfaces # a clear error so the operator knows to re-pick. successor_email: str | None = None successor_display_name: str | None = None raw_successor_email = (task.context_data or {}).get("successor_email") raw_successor_id = (task.context_data or {}).get("successor_user_id") if isinstance(raw_successor_email, str) and raw_successor_email: successor_email = raw_successor_email successor_uuid: UUID | None = None if raw_successor_id: try: successor_uuid = UUID(str(raw_successor_id)) except (ValueError, TypeError): successor_uuid = None if successor_uuid is not None: successor = await self.db.get(IdentityPerson, successor_uuid) if successor is not None: if ( successor.deleted_at is not None or successor.employment_status != "active" ): raise AppException( code="SUCCESSOR_NO_LONGER_ACTIVE", message=( f"Successor {successor.work_email} stamped " f"on this task at enqueue time is now " f"{successor.employment_status!r}. Refusing " f"to hand vendor-side data to an inactive " f"person — re-pick a successor and retry " f"the revoke from the offboarding panel." ), status_code=409, ) successor_email = successor_email or successor.work_email successor_display_name = successor.display_name else: raise AppException( code="SUCCESSOR_NOT_FOUND", message=( "Successor person stamped on this task no longer " "exists. Re-pick a successor in the offboarding " "panel and retry." ), status_code=409, ) ctx = ProvisioningContext( user_email=user.work_email, user_display_name=user.display_name, plan_name=plan_name, permission_key=permission_key, successor_email=successor_email, successor_display_name=successor_display_name, extra=extra, ) if task.action == ACTION_GRANT and task.target_type == "seat": # Plans with ``over_quota_policy='evict_lru'`` get a # capacity probe + LRU eviction *before* the grant is # attempted, so Zoom never returns a quota error here. # Other plans (reject policy) are guarded earlier at the # request router by ``check_plan_quota``, so this is a # no-op for them. await self._maybe_evict_for_zoom(task, ctx, plan_row=plan_row) return await driver.grant_seat(ctx) if task.action == ACTION_REVOKE and task.target_type == "seat": result = await driver.revoke_seat(ctx) # Managed-domain Basic-pool retry. ``revoke_seat`` for an # SSO-bound tenant falls through to ``PATCH user.type=1`` # (Basic) after Zoom refuses ``DELETE ?action=disassociate`` # with 1107. If the Basic pool is at capacity that PATCH # lands 3412 (``ZOOM_BASIC_CAP_EXCEEDED``) and the leaver's # revoke stays stuck on a non-actionable error. Apply the # same LRU pattern as the Licensed pool: delete the # longest-idle Basic user (always a managed-domain-permitted # ``action=delete``, NOT a 1107-tripping disassociate) and # retry the leaver's revoke once. End-to-end this is how # offboarding actually completes on managed-domain tenants # when the Basic pool is full. result = await self._maybe_retry_zoom_on_basic_cap( task, ctx, driver, result, ) return result if task.action == ACTION_GRANT and task.target_type == "permission": return await driver.grant_permission(ctx) if task.action == ACTION_REVOKE and task.target_type == "permission": return await driver.revoke_permission(ctx) raise ValueError( f"Unsupported combination: action={task.action}, target_type={task.target_type}" ) async def _maybe_evict_for_zoom( self, task: ProvisioningTask, ctx: ProvisioningContext, *, plan_row: SoftwareLicensePlan | None, ) -> None: """If this is a Zoom seat grant on an over-quota ``evict_lru`` plan, run the LRU eviction before the grant proceeds. Narrowly scoped so non-Zoom grants pay a single ``is None`` check. The coordinator handles all Zoom-specific logic (driver resolution, victim selection, notification) — this method just decides whether to invoke it. Stamps the eviction outcome into ``task.context_data`` so audit can surface *which* teammate was bumped for *which* new request, without chasing provider-side records. """ if plan_row is None or (plan_row.over_quota_policy or "reject") != "evict_lru": return from app.core.coordinators.zoom_seat_coordinator import ( ensure_seat_capacity, load_zoom_driver, ) zoom_driver = await load_zoom_driver(self.db) if zoom_driver is None: # Plan says evict_lru but no Zoom integration is wired up — # fail loudly so admin notices rather than silently landing # a grant that Zoom will reject seconds later. raise RuntimeError( f"Plan {plan_row.name!r} has over_quota_policy='evict_lru' " "but no enabled Zoom integration is configured" ) outcome = await ensure_seat_capacity( self.db, plan=plan_row, new_requester_email=ctx.user_email, driver=zoom_driver, trace_id=f"provisioning_task:{task.id}", ) if outcome.evicted_user_email: ctx_data = dict(task.context_data or {}) ctx_data["zoom_eviction"] = { "victim_email": outcome.evicted_user_email, "victim_last_login": outcome.evicted_zoom_last_login, "victim_user_id": ( str(outcome.evicted_user_id) if outcome.evicted_user_id else None ), "victim_seat_allocation_id": ( str(outcome.evicted_seat_allocation_id) if outcome.evicted_seat_allocation_id else None ), "revoke_external_ref": outcome.revoke_external_ref, "evicted_at": datetime.now(UTC).isoformat(), } task.context_data = ctx_data flag_modified(task, "context_data") async def _maybe_retry_zoom_on_basic_cap( self, task: ProvisioningTask, ctx: ProvisioningContext, driver: SoftwareProvisioningDriver, result: ProvisioningResult, ) -> ProvisioningResult: """If a Zoom revoke failed with ``ZOOM_BASIC_CAP_EXCEEDED``, try to free a Basic slot via LRU eviction and retry once. Narrow scope: only applies to Zoom drivers — every other provider pays a single attr-check. We detect the tag via ``raw_response['error_tag']`` first (driver-stamped, locale- independent) and fall back to a prefix scan on ``error_message`` so a future driver tweak that drops the dict can't silently kill the retry. Failure semantics: * If the Basic pool eviction succeeds but the retry revoke still fails, the second ``ProvisioningResult`` is returned as-is — the admin sees the actual underlying error rather than the original 3412. * If eviction itself can't find a victim (everyone is whitelisted, list_users failed, …), the ORIGINAL 3412 result is returned untouched so the operator gets the same hint as before. Stamps ``task.context_data['zoom_basic_eviction']`` on success so audit / replay can correlate which Basic user was evicted for which leaver. The local seat-allocation flip happens inside the coordinator's ``_mark_local_alloc_revoked_if_exists`` helper. """ if getattr(driver, "provider", None) != "zoom": return result if result.ok: return result raw = result.raw_response or {} tag = raw.get("error_tag") if isinstance(raw, dict) else None msg_prefix_match = bool( (result.error_message or "").startswith( "[ZOOM_BASIC_CAP_EXCEEDED] " ) ) if tag != "ZOOM_BASIC_CAP_EXCEEDED" and not msg_prefix_match: return result # Local import: the coordinator depends on integration / driver # internals that the provisioning service intentionally doesn't # eager-load at module scope (matches the existing # ``_maybe_evict_for_zoom`` pattern). from app.core.coordinators.zoom_seat_coordinator import ( make_room_in_basic_pool, ) logger.info( "zoom_revoke: task %s hit Basic-cap on %s — running " "make_room_in_basic_pool then retrying revoke", task.id, ctx.user_email, ) try: evicted = await make_room_in_basic_pool( self.db, driver=driver, # type: ignore[arg-type] skip_emails={(ctx.user_email or "").lower()}, trace_id=f"provisioning_task:{task.id}", ) except Exception as exc: # noqa: BLE001 logger.warning( "zoom_revoke: make_room_in_basic_pool raised for task " "%s: %s — returning original 3412 result", task.id, exc, ) return result if evicted is None: logger.warning( "zoom_revoke: no eligible Basic victim for task %s — " "returning original 3412 result for admin remediation", task.id, ) return result ctx_data = dict(task.context_data or {}) ctx_data["zoom_basic_eviction"] = { "victim_email": str(evicted.get("email") or ""), "victim_last_login": evicted.get("last_login_time") or None, "evicted_at": datetime.now(UTC).isoformat(), "reason": "managed_domain_basic_pool_full", } task.context_data = ctx_data # Force-mark the JSON column dirty so SQLAlchemy emits the # UPDATE on commit. Plain JSON columns without MutableDict # don't always detect attribute reassignment of an unchanged # dict identity (Python's ``dict() ==`` returns True if # contents match, defeating the default change detection). # Sibling writes in this service (``fail_task`` line ~810) use # the same belt-and-braces pattern. flag_modified(task, "context_data") retry_result = await driver.revoke_seat(ctx) if retry_result.ok: logger.info( "zoom_revoke: task %s succeeded after Basic-pool " "eviction of %s", task.id, evicted.get("email"), ) else: logger.warning( "zoom_revoke: task %s still failing after Basic-pool " "eviction of %s: %s", task.id, evicted.get("email"), retry_result.error_message, ) return retry_result # Legacy ``_invoke_cloud_account_revoke`` removed in migration 053 — # per-user cloud access is now managed by ``cmdb_cloud_access_grants`` # (see app/domains/cmdb/lifecycle_service.py), not ProvisioningTask. async def _post_success_side_effects(self, task: ProvisioningTask) -> None: """After a task completes, propagate status to related domain rows. - Software seat allocations: grant → set status=active, revoke → status=revoked - User permissions: same pattern via permission_id match - API keys: revoke → status=revoked (grant path handled by existing apikey flow) - Offboarding blockers: resolved via reconcile (admin-triggered), or when this task is linked. Seat resolution is dual-keyed: ``enqueue_*_for_seat`` writes ``target_id = seat.id`` (the allocation UUID), but historical callers wrote ``target_id = plan.id``. We try the allocation UUID first and fall back to plan_id so both conventions land in the same revoked-seat ledger row. Without dual resolution the post-success hook silently no-ops on every modern revoke and the local seat row stays ``active`` forever — which is what the pre-launch audit caught (2026-05-05). """ now = datetime.now(UTC) if task.target_type == "seat" and task.target_id: from app.domains.asset.models import SoftwareSeatAllocation # Primary: target_id == allocation.id (current convention # since enqueue_grant_for_seat / enqueue_revoke_for_seat). seat = ( await self.db.execute( select(SoftwareSeatAllocation).where( and_( SoftwareSeatAllocation.id == task.target_id, SoftwareSeatAllocation.user_id == task.user_id, ) ) ) ).scalars().first() # Fallback: target_id == plan.id (legacy direct # ProvisioningTaskCreate callers that pre-date the # coordinator). Match the latest pending/active seat for # the user on that plan. if seat is None: seat = ( await self.db.execute( select(SoftwareSeatAllocation).where( and_( SoftwareSeatAllocation.user_id == task.user_id, SoftwareSeatAllocation.plan_id == task.target_id, SoftwareSeatAllocation.status.in_( ["active", "pending"], ), ) ).order_by(SoftwareSeatAllocation.created_at.desc()) ) ).scalars().first() if seat: if task.action == ACTION_REVOKE: if seat.status != "revoked": seat.status = "revoked" seat.revoked_at = now elif task.action == ACTION_GRANT: # Grant success: flip pending → active so the seat # billing roster matches what the vendor actually # provisioned. Idempotent: already-active rows are # left alone. if seat.status == "pending": seat.status = "active" seat.allocated_at = seat.allocated_at or now elif task.target_type == TARGET_HARDWARE and task.target_id: # An MDM cleanup todo. Stamp the ledger so the device page # shows a confirmation instead of an open request — but do # NOT mark the device gone: only the vendor's own inventory # can establish that, and ``jamf.device_sync`` will. A # confirmation the sync then contradicts is exactly the case # the reminder loop is there to catch. from app.core.coordinators.mdm_device_retirement_coordinator import ( confirm_cleanup_by_admin, ) await confirm_cleanup_by_admin(self.db, task_id=task.id, now=now) elif task.target_type == TARGET_SSO_GRANT and task.target_id: # SSO removal (migration 603): a human confirmed, with evidence, # that the IdP-side access is gone. Same posture as the MDM # branch above — this is a CONFIRMATION, not proof; the daily # ``sso.removal_reconcile`` reads Google's SAML login audit and # will flip it to ``contradicted`` if a login shows up later. from app.core.coordinators.access_certification_coordinator import ( sync_cert_items_for_sso_grant, ) from app.domains.sso.removal import SsoRemovalService grant = await SsoRemovalService(self.db).on_task_completed( task, trace_id=f"provisioning_task:{task.id}", ) if grant is not None: await sync_cert_items_for_sso_grant( self.db, grant_id=str(grant.id), removal_status=grant.removal_status, now=now, ) elif task.target_type == "permission" and task.target_id: from app.domains.permission.models import ( PERM_ACTIVE, PERM_PENDING, PERM_REVOKED, UserPermission, ) up = ( await self.db.execute( select(UserPermission).where( and_( UserPermission.user_id == task.user_id, UserPermission.permission_id == task.target_id, UserPermission.status.in_([PERM_ACTIVE, PERM_PENDING]), ) ).order_by(UserPermission.created_at.desc()) ) ).scalars().first() if up: if task.action == ACTION_GRANT: up.status = PERM_ACTIVE up.granted_at = now up.external_ref = task.external_ref elif task.action == ACTION_REVOKE: # Migration 606: an expiry-driven revoke lands on # ``expired`` so the ledger tells "validity ran out" # apart from "someone revoked it". The sweep stamps the # terminal status into the task context; anything else # is a plain revoke. from app.domains.permission.models import ( # noqa: PLC0415 PERM_EXPIRED, ) terminal = (task.context_data or {}).get("terminal_status") up.status = PERM_EXPIRED if terminal == PERM_EXPIRED else PERM_REVOKED up.revoked_at = now elif task.target_type == "api_key" and task.target_id: from app.domains.apikey.models import ApiKeyAssignment ak = ( await self.db.execute( select(ApiKeyAssignment).where( and_( ApiKeyAssignment.id == task.target_id, ApiKeyAssignment.assigned_to == task.user_id, ) ) ) ).scalar_one_or_none() if ak and task.action == ACTION_REVOKE: # PR3-1 (2026-05-19): the previous shape just flipped # ``ak.status = "revoked"`` here, which silently # bypassed every cloud-side teardown that # :meth:`ApiKeyService.revoke_key` performs — IAM user # deletion (AWS Bedrock), DashScope sk- key invalidation # + RAM user teardown (Aliyun Bailian), Cognitive # Services account DELETE (Azure OpenAI). Result: # offboarding admin clicks "auto-revoke" on an API-Key # blocker → Helix UI shows the row as revoked, BUT the # cloud credential keeps working until someone notices # and manually hits ``/apikeys/{id}/revoke``. The # offboarded employee (or anyone they shared the key # with) keeps invoking the provider on the company # bill, and the credential never appears in any # leak-scanner's "Helix says revoked" snapshot. # # Dispatch into the real service path so cloud # teardown actually fires, the new revoke-metadata # columns get populated, and the assignee + manager # are notified. ``actor_role=SYSTEM_OFFBOARDING`` # captures the surface that triggered the revoke; # ``current_user`` is the admin who dispatched the # provisioning task (``task.created_by``) so the audit # trail preserves the actor identity. from app.domains.apikey.service import ( ACTOR_ROLE_SYSTEM_OFFBOARDING, ApiKeyService, ) # Round-27 pre-launch fix (2026-05-25): the previous code # read ``task.created_by`` but ``ProvisioningTask`` has NO # such column (verified ``provisioning/models.py:65-113``; # ``created_by`` belongs to the unrelated # ``ProvisionPolicy`` model at line 145). Every api_key # revoke success path raised ``AttributeError`` before # the cloud-side IAM/RAM/AOAI teardown ran — the # offboarding blocker never resolved. The actor is # system-attributed via ``ACTOR_ROLE_SYSTEM_OFFBOARDING`` # so no Person lookup is needed. actor = None ctx_notes = (task.context_data or {}).get("notes") try: await ApiKeyService(self.db).revoke_key( assignment_id=ak.id, reason=ctx_notes or "offboarding auto-revoke", current_user=actor, trace_id=f"provisioning_task:{task.id}", ip_address=None, user_agent=None, reason_code="offboarding", actor_role=ACTOR_ROLE_SYSTEM_OFFBOARDING, ) except Exception: # noqa: BLE001 # The cloud teardown may fail (network blip / # already-deleted IAM user). We log and re-raise so # the provisioning task itself surfaces the # failure — admin retry then finds the row in its # previous status because revoke_key flushes the # status flip AFTER cloud teardown succeeds. This # preserves the "retry-safe" contract: a failed # auto-revoke leaves the row recoverable, not # half-flipped. logger.exception( "provisioning api_key revoke failed for " "task %s (assignment %s)", task.id, ak.id, ) raise # Offboarding blockers ctx = task.context_data or {} blocker_id = ctx.get("offboarding_blocker_id") if blocker_id: from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_READY, RESO_PENDING, RESO_REVOKED_AUTO, OffboardingBlocker, OffboardingProcess, ) blocker = ( await self.db.execute( select(OffboardingBlocker).where(OffboardingBlocker.id == blocker_id) ) ).scalar_one_or_none() if blocker and blocker.resolution == RESO_PENDING: blocker.resolution = RESO_REVOKED_AUTO blocker.resolution_ref = task.external_ref # Recompute process status. Use FOR UPDATE so two concurrent # tasks that both resolve the last blocker at the same time # don't both observe pending==0 and both write PROC_READY, # producing duplicate ready_at stamps and duplicate # downstream signals. proc = ( await self.db.execute( select(OffboardingProcess) .where(OffboardingProcess.id == blocker.process_id) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() if proc: pending = ( await self.db.execute( select(func.count()).select_from(OffboardingBlocker).where( and_( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, ) ) ) ).scalar() or 0 if pending == 0 and proc.status == PROC_BLOCKED: proc.status = PROC_READY proc.ready_at = now async def _maybe_schedule_transient_retry( self, task: ProvisioningTask, *, err_text: str, trace_id: str, actor_user_id: UUID | None, ) -> bool: """When a connector failure looks transient (429 / 5xx / timeout — see ``_TRANSIENT_ERROR_FRAGMENTS``), put the task back into PENDING with an exponential ``not_before`` backoff instead of terminal FAILED, and return True. The dispatcher's pending-selection already honours ``not_before``; ``attempt_count`` was incremented at dispatch start, so the existing 10-attempt dead-letter gate at the top of ``execute_task`` bounds the loop (≈16 h total window). Failure side-effects (admin DM, optimistic-state revert) are deliberately NOT fired here — the loud path is reserved for the terminal outcome. """ if not _is_transient_connector_error(err_text): return False delay = _transient_backoff_delay(task.attempt_count) next_at = datetime.now(UTC) + delay task.status = STATUS_PENDING task.not_before = next_at task.error_message = ( f"transient — auto-retry " f"{task.attempt_count}/10 scheduled at " f"{next_at.isoformat(timespec='seconds')}: {err_text}" )[:2000] await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="provisioning.auto_api_transient_retry", status="failure", resource_type="provisioning_task", resource_id=str(task.id), detail={ "error": err_text[:500], "attempt_count": task.attempt_count, "next_attempt_at": next_at.isoformat(), }, ) return True async def _post_failure_side_effects( self, task: ProvisioningTask, *, suppress_failure_dm: bool = False, ) -> None: """Revert optimistic local state when an auto_api task fails. The admin seat-revoke endpoint flips ``alloc.status='revoked'`` immediately after enqueueing the revoke task so the UI reflects the action without waiting for the dispatcher tick. If the dispatcher then fails to call the vendor — managed- domain refusal (Zoom 1107), credentials expired, vendor 5xx beyond max retries — the seat would otherwise remain falsely- revoked while the vendor still holds the paid license. Helix UI shows "gone", reality shows "still here". Revert here so the admin sees the seat back in ``active`` plus the failed task's ``error_message``, and can decide to retry or escalate. **Opt-in via ``context_data["revert_seat_on_failure"]``** — not all seat pre-flippers want this revert behaviour: - **Admin revoke endpoint**: opts in. No sibling state; the revert is the right recovery. - **Leaver-blocked guard's flip-to-left path**: pre-flips but does NOT opt in. The Helix-side revoke is the billing- correctness invariant (the person is gone; we must stop billing internally); vendor cleanup retries via ``post_leaver_verify`` independently. A revert here would re-attribute the seat to a "left" person and break the billing roster. - **Change-plan flow**: pre-flips old seat AND creates a new active seat (different plan). Does NOT opt in. A unilateral revert of just the old seat would land Helix in a "double- billing" state (old=active + new=active for the same person on the same software) while vendor reality is one or the other. Change-plan failure recovery is a separate concern that needs both-task transactional coordination. Idempotent: re-entry on an already-active seat is a no-op. Bug class caught 2026-05-08 (kevin.lu@pingcap.com — 3 Zoom seats showed revoked in Helix while Zoom kept all 3 licensed seats because every disassociate hit Zoom's managed-domain refusal 1107). Mirror of ``_post_success_side_effects`` for the failure branch — same dual-key (allocation.id primary, plan_id legacy fallback) lookup so seats enqueued via either coordinator convention land back in active. """ if task.target_type == TARGET_SSO_GRANT and task.target_id: # SSO removal (migration 603): a failed / cancelled / dead- # lettered task means the IdP-side removal is STILL outstanding. # Detach it from the grant so the daily reconcile opens a fresh # one — a cancelled task must never read as "nothing to do". from app.domains.sso.removal import SsoRemovalService await SsoRemovalService(self.db).on_task_not_completed( task, outcome=task.status, trace_id=f"provisioning_task:{task.id}", ) return if ( task.action != ACTION_REVOKE or task.target_type not in ("seat", "product_access", "account") or not task.target_id ): return # ``account`` = Jira/Atlassian "remove account access" (often a # leaver full-revoke carrying an offboarding_blocker_id). Its # terminal failure MUST surface a DM — otherwise a leaver keeps # live vendor access and the offboarding blocker stays pending # with only an unwatched FAILED row as evidence. The seat-revert # below stays gated by ``revert_seat_on_failure`` (which account # tasks never set), so this only enables the alert. # DM the software owner / IT admins for ALL revoke seat failures # (not just the ``revert_seat_on_failure`` opt-in flow). The # billing-correctness invariants and the visible "task failed" # signal apply equally to leaver-blocked / change-plan failures — # those just don't want the seat *state* reverted. The DM is # orthogonal to the revert: it's the surface signal that the # platform needs human attention. # # **Per-task dedup** (added in same PR after round-1 review): # a single failed task can re-enter ``_post_failure_side_effects`` # via admin force-execute on an already-FAILED task (status check # at line 330 only short-circuits on COMPLETED / CANCELLED, not # FAILED — so attempt_count keeps climbing if an operator # repeatedly clicks Execute without fixing the underlying cause). # Without dedup that would re-DM the owner on every click. We # stash ``failure_dm_sent_at`` in ``context_data`` and skip the # DM if it's already set; ``retry_task`` clears the flag so an # admin retry that re-fails DOES produce a fresh signal. # #1792: an admin *cancelling* a still-pending revoke task is not a # vendor failure — but cancel_task routes through here to get the # seat-revert (for opt-in tasks). Firing the "[Helix] 软件坐席回收 # 失败 / 坐席已自动回滚为 active, 需要管理员介入" DM on a deliberate # cancel is a false alarm; for the non-``revert_seat_on_failure`` # families (leaver-blocked / change-plan / account revoke) the # "已回滚为 active" claim is itself untrue. ``suppress_failure_dm`` # keeps the seat-revert behaviour while skipping the misleading DM. already_dm = (task.context_data or {}).get("failure_dm_sent_at") if not already_dm and not suppress_failure_dm: try: await self._dispatch_revoke_failure_dms(task) except Exception as exc: # noqa: BLE001 — best-effort # exc_info=True so prod log scrape captures the stack — # without it a transient DB blip in _list_it_admins # silently drops the alert with only a one-line warning. logger.warning( "post_failure_side_effects: revoke-failure DM for task " "%s raised: %s — DB state still written", task.id, exc, exc_info=True, ) # Stamp regardless of dispatch outcome — a recipient-resolution # failure shouldn't make the next force-execute re-fire the # whole fan-out. The audit row from the underlying # ``provisioning.*_failed`` action is the source of truth. ctx = dict(task.context_data or {}) ctx["failure_dm_sent_at"] = datetime.now(UTC).isoformat() task.context_data = ctx flag_modified(task, "context_data") if not (task.context_data or {}).get("revert_seat_on_failure"): return from app.domains.asset.models import SoftwareSeatAllocation # ``target_type='product_access'`` tasks cascade-revoke MULTIPLE # allocations at enqueue time (the source seat plus every # sibling that rides the same product license — Jira Software # base plus Tempo / eazyBI / ScriptRunner). On failure all of # them must revert symmetrically; reverting only the source # seat would leave the admin with "I cancelled Jira, base is # back to active but the 3 add-on rows stayed revoked" — a # split-state that the next ``jira.seat_sync`` would then # ping-pong on. Read the cascade scope from ``context_data`` # (written by the coordinator's enqueue helper). ctx_data = task.context_data or {} if task.target_type == "product_access": ids_to_revert: list[UUID] = [] src = ctx_data.get("source_seat_allocation_id") if isinstance(src, str): try: ids_to_revert.append(UUID(src)) except (ValueError, TypeError): logger.warning( "post_failure_side_effects: product_access task %s " "has malformed source_seat_allocation_id %r — " "skipping that id", task.id, src, ) for sib in ctx_data.get("sibling_allocation_ids") or []: if not isinstance(sib, str): continue try: ids_to_revert.append(UUID(sib)) except (ValueError, TypeError): logger.warning( "post_failure_side_effects: product_access task %s " "has malformed sibling allocation id %r — " "skipping that id", task.id, sib, ) if not ids_to_revert: return rows = (await self.db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.id.in_(ids_to_revert), SoftwareSeatAllocation.user_id == task.user_id, ) )).scalars().all() reverted = 0 for seat in rows: if seat.status != "revoked": continue seat.status = "active" seat.revoked_at = None seat.reason = ( (seat.reason or "") + f"\n[product_access revoke task {task.id} failed → " f"seat reverted to active so admin can retry. Error: " f"{(task.error_message or '')[:200]}]" ).strip() reverted += 1 if reverted: logger.warning( "post_failure_side_effects: reverted %d seat(s) " "revoked→active after product_access task %s failed (%s)", reverted, task.id, (task.error_message or "")[:120], ) return seat = ( await self.db.execute( select(SoftwareSeatAllocation).where( and_( SoftwareSeatAllocation.id == task.target_id, SoftwareSeatAllocation.user_id == task.user_id, ) ) ) ).scalars().first() if seat is None: seat = ( await self.db.execute( select(SoftwareSeatAllocation).where( and_( SoftwareSeatAllocation.user_id == task.user_id, SoftwareSeatAllocation.plan_id == task.target_id, ) ).order_by(SoftwareSeatAllocation.revoked_at.desc()) ) ).scalars().first() if seat is None or seat.status != "revoked": return seat.status = "active" seat.revoked_at = None # Append the failure context to the seat's reason so the # admin reading the row sees WHY it's back in active without # cross-referencing the provisioning_tasks table. Cap the # error snippet so a verbose vendor body can't blow past the # column's TEXT capacity over many retries. seat.reason = ( (seat.reason or "") + f"\n[revoke task {task.id} failed → seat reverted to " f"active so admin can retry. Error: " f"{(task.error_message or '')[:200]}]" ).strip() logger.warning( "post_failure_side_effects: reverted seat %s status " "revoked→active after revoke task %s failed (%s)", seat.id, task.id, (task.error_message or "")[:120], ) async def _dispatch_revoke_failure_dms(self, task: ProvisioningTask) -> None: """Notify the software owner (or IT admins) when a revoke task terminally fails — surfaces "this seat is back to active, please retry or escalate" without the admin having to poll the queue. Fan-out policy mirrors ``_notify_manual_assignees``: 1. If ``task.assignee_id`` is set (software owner default), DM them directly. 2. Otherwise fan out to every active IT admin / global admin. Channels: Feishu + email + notification_outbox (so the bell card surfaces it for any admin reading NotificationCenter). Per-recipient ``try / except`` so one bad address can't sink the whole fan-out. Idempotent for caller bookkeeping: ``_post_failure_side_effects`` is the single call site, invoked once per failure transition. Admin-initiated retries that re-fail produce a fresh push (the admin asked for the retry, so a fresh result deserves a fresh signal). Cross-task dedup window: if a SIBLING task (same user_id, software_id, action, target_type, target_id) already had its failure DM sent within the last 60 seconds, this dispatch is skipped. This is a hardening tier ON TOP of the upstream dedup fix (``find_in_flight_seat_revoke_task`` at enqueue time) — if a future enqueue path slips past the dedup gate anyway, the admin only ever sees one card per (user, software, seat) per minute, not the duplicate pair that triggered this PR. """ from app.integrations.service import get_platform_setting from app.notifications.service import ( NotificationEnvelope, NotificationService, ) # Cross-task DM dedup — see docstring. Stamp on context_data # is parsed ISO-8601; older sibling tasks without the stamp # are ignored (the stamp was added 2026-05-06 and is the only # signal a sibling already paged). # # ``task.id is not None`` guard: the sibling query filters # ``ProvisioningTask.id != task.id`` which in SQL evaluates # to NULL (not TRUE) when ``task.id`` is None, silently # disabling the dedup and letting every sibling slip through. # ``_dispatch_revoke_failure_dms`` is only called from # ``_post_failure_side_effects`` after ``db.flush()`` has # populated the PK, but the explicit guard pins the invariant # locally so a future caller can't trip it. dm_dedup_window_seconds = 60 if ( task.id is not None and task.user_id and task.target_id and task.target_type ): sibling_rows = ( await self.db.execute( select(ProvisioningTask).where( and_( ProvisioningTask.id != task.id, ProvisioningTask.user_id == task.user_id, ProvisioningTask.action == task.action, ProvisioningTask.target_type == task.target_type, ProvisioningTask.target_id == task.target_id, ( ProvisioningTask.software_id == task.software_id if task.software_id is not None else ProvisioningTask.software_id.is_(None) ), ), ), ) ).scalars().all() now_utc = datetime.now(UTC) for sib in sibling_rows: stamp_raw = (sib.context_data or {}).get( "failure_dm_sent_at", ) if not isinstance(stamp_raw, str): continue try: stamped = datetime.fromisoformat(stamp_raw) except ValueError: continue if stamped.tzinfo is None: stamped = stamped.replace(tzinfo=UTC) delta = (now_utc - stamped).total_seconds() if 0 <= delta < dm_dedup_window_seconds: logger.info( "revoke_failure DM for task %s suppressed: " "sibling task %s already DM'd %.1fs ago " "(window=%ds)", task.id, sib.id, delta, dm_dedup_window_seconds, ) return # Subject (target user whose seat the revoke targeted). subject = await self.db.get(IdentityPerson, task.user_id) actor_name = ( subject.display_name or subject.work_email or str(task.user_id) ) if subject else str(task.user_id) actor_email = subject.work_email if subject else "" software_name = "(未绑定软件)" if task.software_id: sw = await self.db.get(AssetSoftwareCatalog, task.software_id) if sw: software_name = sw.name portal_url = str( await get_platform_setting( "app.frontend_base_url", default="", session=self.db, ) or "", ) recipients: list[IdentityPerson] = [] seen: set[UUID] = set() if task.assignee_id: assignee = await self.db.get(IdentityPerson, task.assignee_id) if ( assignee is not None and assignee.deleted_at is None and assignee.employment_status == "active" ): recipients.append(assignee) seen.add(assignee.id) if not recipients: recipients = await self._list_it_admins(exclude_ids=seen) if not recipients: logger.info( "revoke_failure DM for task %s: no eligible recipients " "— skipping", task.id, ) return # Scrub credential-shaped substrings (defense-in-depth — see # ``_DM_CREDENTIAL_SCRUB`` at module scope). Recipients are # privileged, but vendor 4xx bodies sometimes echo the offending # token; we'd rather not splash that into a Feishu card. error_raw = (task.error_message or "").strip().split("\n", 1)[0] error_brief = _DM_CREDENTIAL_SCRUB.sub("", error_raw)[:240] # Operator-facing remediation hint for tagged errors (drivers # prefix ``error_message`` with ``[TAG] …``). Read off the # FULL ``task.error_message`` (not ``error_brief``) so a # 240-char truncation can't lose the tag prefix. remediation_hint = _extract_remediation_hint(task.error_message) title = f"[Helix] 软件坐席回收失败: {actor_name} · {software_name}" body_lines = [ f"{actor_name}（{actor_email}）的 {software_name} 坐席回收任务执行失败。", "", f"错误：{error_brief or '(无错误信息)'}", "", "坐席已自动回滚为 active，需要管理员介入。", ] if remediation_hint: body_lines.extend(["", remediation_hint]) body_text = "\n".join(body_lines) cancellations_url = ( f"{portal_url}/admin/software/cancellations" if portal_url else None ) svc = NotificationService(self.db) for recipient in recipients: try: await svc.dispatch( NotificationEnvelope( recipient=recipient, title=title, body_text=body_text, channels=["feishu", "email"], template_key="provisioning.revoke_failure", context={ "actor": actor_name, "actor_name": actor_name, "actor_email": actor_email, "software_name": software_name, "task_id": str(task.id), "error_brief": error_brief, "remediation_hint": remediation_hint or "", "portal_url": cancellations_url or "", }, url=cancellations_url, related_resource_type="provisioning_task", related_resource_id=task.id, ), trace_id="provisioning.revoke_failure", actor_user_id=None, ) except Exception as exc: # noqa: BLE001 — best-effort logger.warning( "revoke_failure DM to %s (task %s) failed: %s", recipient.work_email, task.id, exc, exc_info=True, ) async def _record_connector_success(self, software_id: UUID | None, at: datetime) -> None: if software_id is None: return connector_row = ( await self.db.execute( select(SoftwareConnectorRow).where(SoftwareConnectorRow.software_id == software_id) ) ).scalar_one_or_none() if connector_row: connector_row.last_success_at = at async def _record_connector_failure( self, software_id: UUID | None, at: datetime, message: str | None, ) -> None: if software_id is None: return connector_row = ( await self.db.execute( select(SoftwareConnectorRow).where(SoftwareConnectorRow.software_id == software_id) ) ).scalar_one_or_none() if connector_row: connector_row.last_error_at = at connector_row.last_error_message = (message or "")[:2000] async def _notify_manual_assignees( self, task: ProvisioningTask, *, trace_id: str, reminder_only: bool = False, ) -> None: """Push a Feishu DM (+ email fallback) to each person who is expected to carry out the manual action for a notified task. ``reminder_only`` switches to the offboarding auto-release variant: Helix has ALREADY reclaimed the local seat/permission and resolved the blocker, so the message is a reminder to finish the vendor-side cleanup — NOT a gate, and with NO "click to confirm" button (the ledger is already closed; a confirm button would be a misleading no-op). Recipient resolution: 1. If ``task.assignee_id`` is set (software owner default), DM them directly. 2. Otherwise, fan out to every active person holding the ``it_admin`` or ``global_admin`` role. This is the case when the bound software has no owner — e.g. Slack on the Pro plan where the IT team handles invites as a group. The template is platform-setting ``email_template.provisioning. manual_notify`` (falls back to a built-in Chinese default in NotificationService). Variables: {actor} — the subject user's display name or email {actor_email} — subject user's work email {action} — "分配" / "回收" {software_name} — catalog name, or "(未绑定软件)" {plan_name} — from context_data.plan_name, or "" {task_id} — UUID of the provisioning task {portal_url} — base Helix URL """ from app.integrations.service import get_platform_setting from app.notifications.service import ( NotificationEnvelope, NotificationService, ) # Subject (target user whose seat is being granted/revoked). subject = await self.db.get(IdentityPerson, task.user_id) actor_name = ( subject.display_name or subject.work_email or str(task.user_id) ) if subject else str(task.user_id) actor_email = subject.work_email if subject else "" software_name = "(未绑定软件)" if task.software_id: sw = await self.db.get(AssetSoftwareCatalog, task.software_id) if sw: software_name = sw.name ctx_data = task.context_data or {} plan_name = str(ctx_data.get("plan_name") or "") action_label = "分配" if task.action == ACTION_GRANT else "回收" # SSO removal tasks (migration 603) carry no software row; the # system name lives in context. Label them as what they are so the # owner is not asked to "回收 (未绑定软件) 坐席". is_sso_removal = task.target_type == TARGET_SSO_GRANT sso_system_name = str(ctx_data.get("system_name") or "").strip() sso_provider = str(ctx_data.get("sso_provider") or "").strip() if is_sso_removal and sso_system_name: software_name = f"SSO · {sso_system_name}" portal_url = str( await get_platform_setting( "app.frontend_base_url", default="", session=self.db, ) or "", ) recipients: list[IdentityPerson] = [] seen: set[UUID] = set() if task.assignee_id: assignee = await self.db.get(IdentityPerson, task.assignee_id) if ( assignee is not None and assignee.deleted_at is None and assignee.employment_status == "active" ): recipients.append(assignee) seen.add(assignee.id) if not recipients: recipients = await self._list_it_admins(exclude_ids=seen) if not recipients: logger.info( "manual_notify task %s: no eligible recipients " "(no assignee + no it_admin) — skipping DM", task.id, ) return # Body uses the template's ``{cta_line}`` placeholder; we do # NOT inline portal_url into body_text because the action # button (rendered from ``url=``) is the only CTA. Stops dev / # staging from leaking ``localhost:3000`` into the message. if reminder_only: title = ( f"[Helix] 离职回收提醒: 请在外部系统{action_label} " f"{software_name} 账号 (Helix 已自动回收坐席): {actor_name}" ) elif is_sso_removal: title = ( f"[Helix] 请在 {sso_provider or 'IdP'} 停用 SSO 访问: " f"{sso_system_name or software_name} · {actor_name}" ) else: title = ( f"[Helix] 请手工{action_label} {software_name} 坐席: {actor_name}" ) body_text = "(see template)" svc = NotificationService(self.db) # One-click "我已处理,标记完成" — a per-recipient signed link # (2026-06-09) so the owner can close the task straight from the # notification without logging into Helix. The token authorises # exactly THIS task for THIS recipient and expires; the public # ``/provisioning/quick-complete`` route renders a confirm page # (GET, no side effect → safe against email prefetch) whose # button POSTs the completion. The link rides ONLY on the # Feishu card buttons — the in-app bell deep-links to the # console instead (a logged-in web user should act in-app, not # on the token page). from app.core.security import create_action_token # noqa: PLC0415 from app.domains.provisioning.models import ( # noqa: PLC0415 ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE, ) from app.notifications.service import ( # noqa: PLC0415 CardAction, LocalizedText, ) deep_link_url = ( f"{portal_url}/admin/provisioning" if portal_url else None ) # Locale fragment triple for the action verb: each language's # template interpolates its own ``action_label_``. Mapped off # ``task.action`` rather than threaded through the call chain (only # two values). Frozen into the outbox row, so the in-app bell still # renders the right language after the reader switches preference. is_revoke = task.action != ACTION_GRANT action_label_en = "revoke" if is_revoke else "provision" action_label_ja = "回収" if is_revoke else "付与" plan_suffix = f" - {plan_name}" if plan_name else "" context = { "actor": actor_name, "actor_name": actor_name, "actor_email": actor_email, "action": action_label, "action_label_zh": action_label, "action_label_en": action_label_en, "action_label_ja": action_label_ja, "software_name": software_name, "plan_name": plan_name, "plan_suffix": plan_suffix, "task_id": str(task.id), "portal_url": ( f"{portal_url}/admin/provisioning" if portal_url else "" ), } confirm_label = LocalizedText( zh=f"✅ 我已{action_label},标记完成", en=( "✅ I've revoked it — mark done" if is_revoke else "✅ I've provisioned it — mark done" ), ja=( "✅ 回収しました。完了にする" if is_revoke else "✅ 付与しました。完了にする" ), ) open_task_button = LocalizedText( zh="打开任务详情", en="Open task details", ja="タスクの詳細を開く", ) template_key = ( "provisioning.offboarding_auto_release_reminder" if reminder_only else "provisioning.manual_notify" ) if is_sso_removal and not reminder_only: # Evidence is mandatory on completion (the quick-complete page # shows a required reference field), so the button says so — # a "mark done" label would promise a one-click close that the # server will refuse. template_key = "provisioning.sso_removal" confirm_label = LocalizedText( zh="✅ 已在目标系统停用,填写凭证", en="✅ Disabled it — record evidence", ja="✅ 無効化済み・根拠を記録", ) context.update({ "system_name": sso_system_name or software_name, "sso_provider": sso_provider or "IdP", "revoke_reason": str(ctx_data.get("revoke_reason") or ""), }) for recipient in recipients: quick_url = None card_actions: list[CardAction] = [] if reminder_only: # Helix already reclaimed the local seat — no confirm # button. Just a deep link to the task for context. if deep_link_url: card_actions = [ CardAction( label=open_task_button, url=deep_link_url, style="default", ), ] elif portal_url: token = create_action_token( purpose=ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE, subject_id=str(recipient.id), resource_id=str(task.id), ) quick_url = ( f"{portal_url}/api/v1/provisioning/quick-complete" f"?token={token}" ) card_actions = [ CardAction(label=confirm_label, url=quick_url, style="primary"), CardAction( label=open_task_button, url=deep_link_url, style="default", ), ] try: await svc.dispatch( NotificationEnvelope( recipient=recipient, title=title, body_text=body_text, channels=["feishu", "email"], template_key=template_key, context={**context, "confirm_url": quick_url or ""}, # ``url`` is what the IN-APP bell stores — a # logged-in web user clicking the notification # must land on the normal console page, NOT the # no-login confirm page (2026-06-10 complaint: # web bell opened the bare token page in a new # tab). The one-click confirm link lives only on # the Feishu card buttons (card_actions) where # login friction is real. Relative path → SPA # navigates in-app; Feishu/email resolve it # absolute via app.frontend_base_url. # ``/admin/provisioning`` (bare) 301s to the # lifecycle overview — land directly on the task # queue pre-filtered to the rows needing a human. url="/admin/provisioning/tasks?status=notified", card_actions=card_actions, related_resource_type="provisioning_task", related_resource_id=task.id, ), trace_id=trace_id, actor_user_id=None, ) except Exception as e: # noqa: BLE001 — best-effort logger.warning( "manual_notify DM to %s (task %s) failed: %s", recipient.work_email, task.id, e, ) async def _list_it_admins( self, *, exclude_ids: set[UUID], ) -> list[IdentityPerson]: """Active persons holding ``it_admin`` or ``global_admin``. Mirrors the resolver used by offboarding escalation so the two notification paths target the same group. Skips ids in ``exclude_ids`` (e.g. the assignee we already DM'd). """ from sqlalchemy import or_ from app.auth.models import AuthRole, AuthUserRole rows = ( await self.db.execute( select(IdentityPerson) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( and_( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", or_( AuthRole.name == "it_admin", AuthRole.name == "global_admin", ), ) ) .distinct() ) ).scalars().all() return [r for r in rows if r.id not in exclude_ids] async def _load_users(self, ids: set[UUID]) -> dict[UUID, IdentityPerson]: if not ids: return {} rows = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(ids)) ) ).scalars().all() return {r.id: r for r in rows} async def _load_softwares(self, ids: set[UUID]) -> dict[UUID, AssetSoftwareCatalog]: if not ids: return {} rows = ( await self.db.execute( select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id.in_(ids)) ) ).scalars().all() return {r.id: r for r in rows} def _to_response( self, task: ProvisioningTask, users: dict[UUID, IdentityPerson], softwares: dict[UUID, AssetSoftwareCatalog], integrations: dict[UUID, Integration] | None = None, ) -> ProvisioningTaskResponse: user = users.get(task.user_id) assignee = users.get(task.assignee_id) if task.assignee_id else None software = softwares.get(task.software_id) if task.software_id else None # ``target_label`` is the display string the UI shows in the # "Software" column of the cancellations / queue overviews. # Seat + permission tasks already have ``software_name`` — # mirror that. Two other shapes need fallbacks because they # legitimately carry NULL software: # - ``target_type='account'`` — Jira / Atlassian "remove # from all groups" (offboarding ``_apply_account_policy``). # ``target_id`` is the integration id; label = integration # name (e.g. "Jira (Atlassian)"). # - ``target_type='product_access'`` — per-product Atlassian # revoke (Jira Software / Confluence / JSM). ``target_id`` # is the integration id, BUT ``context_data.product_label`` # is more precise ("Jira Software" vs "Jira (Atlassian)"). # Prefer the context label; fall back to integration name # if the context didn't stamp one. # If every fallback misses (integration deleted, context # stripped) we leave the field NULL — the FE renders "—" # rather than fabricating a misleading label. target_label: str | None = None if software is not None: target_label = software.name elif task.target_type == TARGET_SSO_GRANT: # SSO removal: the system name was stamped into context at open # time so the queue never has to join sso_grants for a label. ctx = task.context_data if isinstance(task.context_data, dict) else {} label = ctx.get("system_name") if isinstance(label, str) and label.strip(): target_label = f"SSO · {label.strip()}" elif task.target_type == "product_access": # Defensive: ``context_data`` is typed dict | None on the # model, but a stray UPDATE / migration that stores a # JSON string would crash this code path with AttributeError # on ``.get()``. Coerce to {} when not-a-dict so the # endpoint can't 500 on corrupt rows; the fallback to # integration.name below still gives the row a label. ctx = task.context_data if isinstance(task.context_data, dict) else {} label = ctx.get("product_label") or ctx.get("plan_name") if isinstance(label, str) and label.strip(): target_label = label.strip() elif task.target_id is not None: integ = (integrations or {}).get(task.target_id) if integ is not None: target_label = integ.name elif task.target_type == "account" and task.target_id is not None: integ = (integrations or {}).get(task.target_id) if integ is not None: target_label = integ.name return ProvisioningTaskResponse( id=task.id, user_id=task.user_id, user_name=user.display_name if user else None, user_email=user.work_email if user else None, user_employee_no=user.employee_no if user else None, user_employment_status=user.employment_status if user else None, software_id=task.software_id, software_name=software.name if software else None, target_label=target_label, action=task.action, target_type=task.target_type, target_id=task.target_id, source=task.source, execution_mode=task.execution_mode, status=task.status, assignee_id=task.assignee_id, assignee_name=assignee.display_name if assignee else None, assignee_employee_no=assignee.employee_no if assignee else None, assignee_employment_status=assignee.employment_status if assignee else None, external_ref=task.external_ref, error_message=task.error_message, approval_request_id=task.approval_request_id, context_data=task.context_data, attempt_count=task.attempt_count, started_at=task.started_at, completed_at=task.completed_at, created_at=task.created_at, updated_at=task.updated_at, ) def _to_connector_response( self, row: SoftwareConnectorRow, softwares: dict[UUID, AssetSoftwareCatalog], integrations: dict[UUID, Integration] | None = None, ) -> SoftwareConnectorResponse: software = softwares.get(row.software_id) integ = (integrations or {}).get(row.integration_id) return SoftwareConnectorResponse( id=row.id, software_id=row.software_id, software_name=software.name if software else None, integration_id=row.integration_id, integration_name=integ.name if integ else None, integration_provider=integ.provider if integ else None, binding_config=dict(row.binding_config or {}), is_active=row.is_active, last_success_at=row.last_success_at, last_error_at=row.last_error_at, last_error_message=row.last_error_message, created_at=row.created_at, updated_at=row.updated_at, ) # ─── Connection test ─── async def test_connector( self, connector_id: UUID, *, trace_id: str, actor_user_id: UUID, ) -> tuple[bool, str | None, dict[str, Any] | None]: """Test the underlying integration's connectivity for this binding. Delegates to the integration's own ``test_connection`` (credentials live on the integration, not the binding). Records last_success_at/last_error_at on the binding row so admins see per-binding health in the provisioning UI even though the actual probe is shared across all bindings pointing at the same integration. Returns ``(ok, error_message, raw_detail)``. """ row = await self._get_connector(connector_id) try: driver = await build_driver_by_id(row.integration_id, session=self.db) except AppException: raise except Exception as e: # noqa: BLE001 return ( False, f"Failed to build integration driver: {format_exception_for_log(e)}", None, ) try: result = await driver.test_connection() except Exception as e: # noqa: BLE001 err_msg = format_exception_for_log(e) result = None else: err_msg = None now = datetime.now(UTC) integ_provider = driver.provider if result and result.ok: row.last_success_at = now row.last_error_at = None row.last_error_message = None await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.connector_test_success", status="success", resource_type="software_connector", resource_id=str(row.id), detail={"integration_provider": integ_provider, "message": result.message}, ) return (True, None, result.detail) msg = err_msg or (result.message if result else "Unknown error") row.last_error_at = now row.last_error_message = (msg or "")[:2000] await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="provisioning.connector_test_failed", status="failure", resource_type="software_connector", resource_id=str(row.id), detail={"integration_provider": integ_provider, "error": msg}, ) return (False, msg, None)