"""Cloud-side orphan-assignment cleanup. Background — AWS Identity Center keeps a stale ``DescribeAccountAssignment`` row indefinitely after the SCIM-synced principal is deleted from the Identity Store. The console renders these as ``"User not found (UUID)"`` attached to a permission set on a sub-account; ``list_scope_grants`` silently drops them (the email lookup returns None). The result: a months-old leaver still shows ``DBaaS-Prod-Admin-Role`` on ``tidbcloud-marketplace`` long after Helix has fully offboarded them. This module is the cleanup machinery that surfaces and undoes them: * :func:`scan_orphan_assignments` — read-only enumeration. Returns one record per orphan principal × permission-set × scope, with a best-effort recovered email pulled from past Helix grant rows that referenced the same ``principal_id``. * :func:`cleanup_orphan_assignments` — delete-on-cloud + insert history row + audit. Idempotent per-orphan: a second run after the assignment is already gone is a no-op success. Designed for one-shot bulk cleanup AND continuous defence-in-depth: :func:`scan_orphan_assignments` is also called at the end of ``sync_grants_for_integration`` so admins see new orphans accumulating in the historical-grants tab without explicitly running a scan. The continuous path NEVER auto-deletes; the admin must confirm. """ from __future__ import annotations import logging import re import uuid from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.database import async_session_factory from app.core.exceptions import ( AppException, NotFoundError, ProdCloudLockedError, ) from app.core.txn import best_effort from app.domains.cmdb.models import ( GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, GRANT_STATUS_REVOKED, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudScope, ) from app.domains.cmdb.prod_write_lock import ensure_prod_scope_writable from app.identity.person_models import IdentityPerson from app.integrations.driver_base import ( CloudDriver, CloudScopeRef, OrphanAssignment, ) from app.integrations.models import Integration logger = logging.getLogger(__name__) @dataclass(slots=True) class OrphanRecord: """One scanned orphan, augmented with Helix-side recovery data.""" integration_id: str integration_name: str | None provider: str | None scope_id: str scope_external_id: str scope_display_name: str | None principal_id: str role_key: str role_display_name: str | None external_ref: str | None # Best-effort email recovered from past grant rows that referenced # the same ``principal_id`` in their ``external_ref``. ``None`` # when no Helix history exists for this principal. recovered_email: str | None # When the same principal_id maps to a current IdentityPerson via # the recovered email, surface their display_name so the admin sees # "张三 · zhang@pingcap.com" instead of just an opaque UUID. Falls # back to ``None`` when the leaver's row is hard-deleted. recovered_display_name: str | None @dataclass(slots=True) class OrphanCleanupResult: """Per-orphan outcome for :func:`cleanup_orphan_assignments`.""" principal_id: str role_key: str scope_id: str ok: bool already_absent: bool = False grant_id: str | None = None error_message: str | None = None # ── Recovery: principal_id → past Helix email ────────────────────── _AWS_IDENTITYSTORE_USER_ID_RE = re.compile( # AWS Identity Store user IDs are 8-4-4-4-12 hex (UUID-shaped) but # may carry a region-encoded prefix in some accounts (observed # form: ``95670a3c83-c108d80c-cbbe-4c76-a5c7-d95928e4b566``). # Allow alphanumeric + hyphen only, length 32-80 — this both # validates the AWS shape and prevents SQL LIKE-pattern injection # via ``%`` / ``_`` / ``\`` characters smuggled in from a # compromised AWS API response. (UUIDs and the prefixed form fit; # malicious ``;DROP TABLE`` or ``%`` would be rejected before # touching the LIKE pattern.) r"\A[A-Za-z0-9\-]{32,80}\Z", ) async def _recover_emails_for_principals( session: AsyncSession, integration_id: str, principal_ids: set[str], ) -> dict[str, tuple[str | None, str | None]]: """Look up past grant rows whose ``external_ref`` ends with the same ``principal_id`` to recover (email, display_name). The AWS driver writes ``external_ref`` in the canonical form ``account_id::permission_set_arn::principal_id``. Any past helix-issued or discovered grant for the same principal therefore carries that suffix, and joins back to the IdentityPerson row that Helix attributed at the time. When the person has since been hard-deleted, the FK will be NULL but the row is still there. For principals never SCIM-synced into Helix, returns ``(None, None)`` — the cleanup endpoint stores the row with email NULL in that case (the principal_id itself is the only attribution we can offer). Inputs are validated against the AWS Identity Store user-id shape before being used in the SQL LIKE pattern. This serves two purposes: 1. SQL LIKE-pattern hygiene — a principal id containing ``%`` or ``_`` would either match too many rows (false-recover an unrelated person's email onto this orphan's history) or, in a cross-pattern collision, return an unrelated grant's row. 2. Shape gate — IAM-mode rows use the ``account_id::user_name::policy_arn`` ref shape; if a stray IAM principal somehow reached this codepath, the regex rejects it and we return ``(None, None)`` rather than match on the policy_arn suffix. """ if not principal_ids: return {} out: dict[str, tuple[str | None, str | None]] = {} for pid in principal_ids: if not _AWS_IDENTITYSTORE_USER_ID_RE.match(pid): # Either malformed (unexpected AWS payload) or a non-SSO # ref shape leaking in. Skip recovery — the cleanup row # will store ``external_principal_email=NULL`` and the # principal id alone is the attribution. logger.info( "orphan-cleanup: skip email recovery for non-conforming " "principal_id %r", pid, ) out[pid] = (None, None) continue # Resolve via the latest grant row that pointed at this # principal — we want the most recent attribution in case the # email changed (e.g. legal name change). ORDER BY granted_at # DESC + LIMIT 1 keeps the work bounded. # # Round-42 audit: scope by ``integration_id`` so a multi-AWS-Org # tenant cannot cross-attribute principals. AWS Identity Store # principal_ids are unique WITHIN an Identity Store but NOT # across Orgs — without the integration filter a coincidental # UUID collision in another tenant would stamp the wrong # email/display_name on this orphan-cleanup history row. rows = (await session.execute( select( CmdbCloudAccessGrant.person_id, CmdbCloudAccessGrant.external_principal_email, ) .where( CmdbCloudAccessGrant.external_ref.like(f"%::{pid}"), CmdbCloudAccessGrant.scope_id.in_( select(CmdbCloudScope.id).where( CmdbCloudScope.integration_id == integration_id, ) ), ) .order_by(CmdbCloudAccessGrant.granted_at.desc()) .limit(1) )).first() email: str | None = None display_name: str | None = None if rows: person_id, prior_email = rows if person_id: person = await session.get( IdentityPerson, uuid.UUID(str(person_id)), ) if person is not None: email = person.work_email display_name = person.display_name if not email: email = prior_email out[pid] = (email, display_name) return out # ── Scan ──────────────────────────────────────────────────────────── async def scan_orphan_assignments( integration_id: str, *, scope_id: str | None = None, ) -> list[OrphanRecord]: """Enumerate orphan assignments on a cloud integration. Read-only. ``scope_id`` (optional): when set, scan ONLY that one scope. Used by the per-scope detail page's "扫描云端孤儿权限" button — the user is asking about ONE sub-account, not the whole AWS Org. A naive integration-wide scan against e.g. an AWS Org with 50+ member accounts × 20 permission sets each + IdentityStore lookups can issue tens of thousands of API calls and easily exceed any reasonable HTTP timeout (observed: 90s+ on a real prod-shaped integration). Always pass ``scope_id`` for interactive admin actions. ``scope_id`` omitted: scan every active scope on the integration. Reserved for batch / scheduled callers that have an explicit "scan everything" use-case and accept the long wallclock cost. Loads the cloud driver via the same path as ``sync_grants_for_integration`` so credential decryption + config layout stays consistent. Drivers without orphan support return ``[]`` (handled by the base raising NotImplementedError). """ from app.domains.cmdb.lifecycle_service import _load_cloud_driver async with async_session_factory() as session: integration = await session.get( Integration, uuid.UUID(integration_id), ) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) if not integration.enabled: return [] scope_q = select(CmdbCloudScope).where( CmdbCloudScope.integration_id == str(integration.id), CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, ) if scope_id: # Filter to the one scope the admin clicked. Reject if the # given scope_id doesn't belong to this integration — defends # against a stale URL where the integration_id and scope_id # got out of sync (otherwise the WHERE returns 0 rows and # we silently respond with empty list, hiding the bug). try: scope_uuid = uuid.UUID(scope_id) except ValueError as exc: raise NotFoundError("CmdbCloudScope", scope_id) from exc scope_q = scope_q.where(CmdbCloudScope.id == scope_uuid) scopes = (await session.execute( scope_q.order_by(CmdbCloudScope.first_seen_at) )).scalars().all() if not scopes: if scope_id: # Specific scope requested but not found / not active — # surface so the UI can show a meaningful error rather # than "0 orphans" (which the admin would interpret as # "all clean"). raise NotFoundError("CmdbCloudScope", scope_id) return [] driver: CloudDriver = await _load_cloud_driver(session, integration) all_orphans: list[tuple[CmdbCloudScope, OrphanAssignment]] = [] try: for scope in scopes: ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) try: found = await driver.list_scope_orphan_assignments(ref) except NotImplementedError: # Whole driver opted out — bail the scan rather # than waste calls on remaining scopes. return [] except Exception as exc: # noqa: BLE001 logger.warning( "orphan-scan %s scope=%s failed: %s", integration.name, scope.external_id, exc, ) continue for o in found: all_orphans.append((scope, o)) finally: try: await driver.close() except Exception: # noqa: BLE001 pass principal_ids = {o.principal_id for _scope, o in all_orphans} recovered = await _recover_emails_for_principals( session, str(integration.id), principal_ids, ) out: list[OrphanRecord] = [] for scope, o in all_orphans: email, display_name = recovered.get( o.principal_id, (None, None), ) out.append(OrphanRecord( integration_id=str(integration.id), integration_name=integration.name, provider=integration.provider, scope_id=str(scope.id), scope_external_id=scope.external_id, scope_display_name=scope.display_name, principal_id=o.principal_id, role_key=o.role_key, role_display_name=o.role_display_name, external_ref=o.external_ref, recovered_email=email or o.recovered_email, recovered_display_name=display_name, )) return out # ── Cleanup ───────────────────────────────────────────────────────── async def cleanup_orphan_assignments( integration_id: str, targets: list[dict[str, Any]], *, actor_user_id: uuid.UUID | None, trace_id: str, ) -> list[OrphanCleanupResult]: """Delete each requested orphan from the cloud + record a history row. ``targets`` is the list the admin confirmed in the modal — each entry is ``{"scope_id": str, "role_key": str, "principal_id": str, "external_ref": str | None, "recovered_email": str | None}``. Per-target flow: 1. Resolve scope, build CloudScopeRef + OrphanAssignment. 2. Call ``driver.delete_orphan_assignment`` (idempotent). 3. On success, INSERT a CmdbCloudAccessGrant row with ``status='revoked'``, ``revoke_reason='orphan_principal_swept'``, ``person_id=NULL``, ``external_principal_id=principal_id``, ``external_principal_email=recovered_email``. 4. Write ``cloud.grant.orphan_principal_swept`` audit row. Returns one result per target, in input order, so the admin UI can correlate failures back to the row in the confirm table. """ from app.domains.cmdb.lifecycle_service import _load_cloud_driver if not targets: return [] out: list[OrphanCleanupResult] = [] async with async_session_factory() as session: integration = await session.get( Integration, uuid.UUID(integration_id), ) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) if not integration.enabled: raise AppException( code="INTEGRATION_DISABLED", message="cannot clean up orphans on a disabled integration", status_code=400, ) scope_cache: dict[str, CmdbCloudScope] = {} async def _scope(sid: str) -> CmdbCloudScope: cached = scope_cache.get(sid) if cached is not None: return cached scope = await session.get(CmdbCloudScope, uuid.UUID(sid)) if ( scope is None or scope.deleted_at is not None or str(scope.integration_id) != str(integration.id) ): raise NotFoundError("CmdbCloudScope", sid) scope_cache[sid] = scope return scope driver: CloudDriver = await _load_cloud_driver(session, integration) audit = AuditService(session) # Round-22 TOCTOU hardening: between the admin opening the # orphan-scan modal and clicking "确认清理", SCIM may re-sync # the user (HR fixed an offboarding glitch, a tombstone-then- # recreate flow ran, email change re-bound IdentityStore). The # principal that LOOKED orphan at scan time is now a valid # accessor. Re-scan each affected scope ONCE here, cache the # live orphan triples, and only delete principals that are # STILL absent in the current snapshot. Cheap (1 API call per # scope, capped by len(distinct scope_ids) which is ≤ targets), # and prevents the "I deleted access for a real employee who # just rejoined" foot-gun. target_scope_ids: set[str] = { str(t.get("scope_id") or "").strip() for t in targets if t.get("scope_id") } live_orphan_sets: dict[str, set[tuple[str, str]]] = {} rescan_errors: dict[str, str] = {} for sid in target_scope_ids: try: scope_for_rescan = await _scope(sid) except NotFoundError as exc: rescan_errors[sid] = str(exc) continue ref_for_rescan = CloudScopeRef( scope_type=scope_for_rescan.scope_type, external_id=scope_for_rescan.external_id, display_name=scope_for_rescan.display_name, ) try: fresh = await driver.list_scope_orphan_assignments( ref_for_rescan, ) except NotImplementedError: # Driver doesn't support orphan discovery; the original # ``scan_orphan_assignments`` path would have raised this # too. Treat as empty so the target loop emits a clean # "driver does not implement" error per target instead # of crashing the whole batch. live_orphan_sets[sid] = set() continue except Exception as exc: # noqa: BLE001 # A transient cloud-API failure during re-scan. We # MUST NOT proceed with delete on a stale list — that # is exactly the bug class the re-scan is here to # prevent. Surface the rescan failure per-target so # the admin sees "couldn't verify the principal is # still orphan" instead of a silent partial delete. rescan_errors[sid] = ( f"orphan re-scan failed during TOCTOU check: {exc}" ) logger.exception( "orphan_cleanup: TOCTOU re-scan failed for scope=%s " "trace_id=%s", sid, trace_id, ) continue live_orphan_sets[sid] = { (o.role_key, o.principal_id) for o in fresh } try: for target in targets: scope_id = str(target.get("scope_id") or "").strip() role_key = str(target.get("role_key") or "").strip() principal_id = str( target.get("principal_id") or "", ).strip() if not (scope_id and role_key and principal_id): out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message="invalid target payload", )) continue try: scope = await _scope(scope_id) except NotFoundError as exc: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=str(exc), )) continue # Prod-write kill-switch: orphan cleanup issues a REAL # provider-side IAM revoke (delete_orphan_assignment), so a # prod target while the global lock is engaged must be # refused BEFORE any cloud write — same invariant every # other human-initiated grant-lifecycle write enforces. # Role-independent (even global_admin is blocked). Reported # per-target so one locked prod scope does not abort the # rest of the batch; non-prod / unlocked targets are no-ops. try: await ensure_prod_scope_writable(session, scope) except ProdCloudLockedError as exc: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=str(exc), )) continue # Round-22 TOCTOU: refuse the delete if the principal is # NO LONGER orphan in the fresh re-scan (admin's modal # was stale; SCIM re-onboarded the user; etc.). rescan_err = rescan_errors.get(scope_id) if rescan_err is not None: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=rescan_err, )) continue live_set = live_orphan_sets.get(scope_id) if live_set is None: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=( "orphan re-scan did not return a set for " "this scope; aborting delete to avoid acting " "on a stale modal" ), )) continue if (role_key, principal_id) not in live_set: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=( "principal is no longer orphan in the live " "cloud scan — refusing to delete. Re-open " "the orphan scan and reconfirm." ), )) continue ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) external_ref = ( str(target.get("external_ref")) if target.get("external_ref") else f"{scope.external_id}::{role_key}::{principal_id}" ) orphan = OrphanAssignment( principal_id=principal_id, role_key=role_key, scope=ref, role_display_name=( target.get("role_display_name") or None ), external_ref=external_ref, ) try: result = await driver.delete_orphan_assignment( ref, orphan, ) except NotImplementedError: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=( f"driver {integration.provider} does not " "implement orphan cleanup" ), )) continue except Exception as exc: # noqa: BLE001 out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=str(exc), )) continue if not result.ok: out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=result.error_message or "driver returned ok=False", )) continue # Per-target durability: commit each successful target # IMMEDIATELY after the history row + audit are flushed. # Cloud-side ``DeleteAccountAssignment`` already # succeeded by the time we reach this branch — if we # batched all the DB writes into one trailing # ``session.commit()`` and the 50th target's audit # write threw (DB blip, FK race), every prior target's # history + audit row would roll back even though the # cloud side is already deleted, leaving cloud-side # state ahead of Helix-side bookkeeping. Per-target # commit limits the inconsistency window to just the # single failing target. The grant insert + audit log # for one target are wrapped in a try-block that, on # failure, surfaces a CRITICAL log line attributing # the orphan-row, principal_id, and integration so # SREs can hand-recover the missing history. grant_id: str | None = None # Capture non-PK ORM columns into plain locals BEFORE the # commit below. ``_safe_rollback_cleanup_session`` in the # except branch expires every persistent instance, so # reading ``scope.external_id`` / ``integration.name`` / # ``integration.provider`` afterwards triggers a sync # lazy-refresh → ``MissingGreenlet`` raised FROM INSIDE the # error handler, which would abort cleanup of the remaining # N-1 targets AND skip the compliance-incident write (the # whole point of the recovery block). PK reads (.id) are # safe — the identity map retains them past expiry. scope_external_id = scope.external_id integration_name = integration.name integration_provider = integration.provider async with best_effort( session, label="cloud: orphan-cleanup grant + history row", log=logger, ) as be_636_step: if not result.already_absent: now = datetime.now(UTC).replace(tzinfo=None) grant = CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=None, role_key=role_key, role_display_name=( target.get("role_display_name") or None ), status=GRANT_STATUS_REVOKED, granted_at=now, revoked_at=now, revoked_by_person_id=( str(actor_user_id) if actor_user_id else None ), revoke_reason=( GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL ), external_ref=external_ref, external_principal_id=principal_id, external_principal_email=( target.get("recovered_email") or None ), source="discovered", detail={ "swept_from": "orphan_cleanup", "trace_id": trace_id, }, ) session.add(grant) await session.flush() grant_id = str(grant.id) await audit.log( trace_id=trace_id, actor_type=( "user" if actor_user_id else "system" ), actor_user_id=actor_user_id, action="cloud.grant.orphan_principal_swept", status="success", resource_type="cmdb_cloud_access_grant", resource_id=grant_id or external_ref, detail={ "integration_id": str(integration.id), "provider": integration.provider, "scope_id": str(scope.id), "scope_external_id": scope.external_id, "role_key": role_key, "principal_id": principal_id, "recovered_email": ( target.get("recovered_email") or None ), "already_absent": bool(result.already_absent), }, ) await session.commit() if not be_636_step.ok: # Cloud-side delete ALREADY happened. We failed # to write Helix-side bookkeeping. Log loud so # SREs can manually reconcile, then keep going — # don't let one DB blip block cleanup of the # remaining N-1 cloud-side stale rows. await _safe_rollback_cleanup_session(session) logger.critical( "orphan-cleanup: cloud-side delete SUCCEEDED " "but Helix-side history/audit FAILED — manual " "reconciliation required. integration=%s " "scope_external_id=%s role_key=%s " "principal_id=%s recovered_email=%s " "trace_id=%s err=%s", integration.id, scope_external_id, role_key, principal_id, target.get("recovered_email"), trace_id, be_636_step.reason(), ) # Pre-launch hardening 2026-05-16 follow-up (CR4 P1): # CRITICAL log alone relies on someone tailing — SRE # only notices if they happen to be watching. Write a # compliance_incident so the operator UI surfaces a # red queue entry, and push a failure_notifier card # so on-call gets paged via the dedup-aware channel. # Best-effort — if either side-effect fails, log # again but never re-raise (cleanup of the remaining # cloud-side rows takes priority over notification). try: import uuid as _uuid from app.compliance.models import SEVERITY_CRITICAL from app.compliance.service import ( ComplianceIncidentService, ) # Round-1 + Round-3 review follow-up # (Review-E P1-2 + LS3 P1-A): # ``create_or_upsert_incident`` dedups on # ``(incident_type, resource_type, resource_id)``. # Pre-fix every orphan-cleanup failure across # every scope / principal collapsed into ONE # shared row. Round-1 anchored to ``scope.id``, # which split per-scope but still collapsed # multiple stuck principals on the SAME scope # into one row — operator UI only saw the most # recent principal. Round-3 fix: synthesise a # uuid5 over ``(scope, role, principal)`` so # each distinct stuck principal becomes its own # incident row. The synthetic resource_id won't # click-through to a real cmdb_cloud_scope row, # but ``detail.scope_id`` carries the real id # for operator drill-down, and ``resource_label`` # is the readable triple. dedup_id = _uuid.uuid5( _uuid.NAMESPACE_OID, f"orphan_cleanup:{scope.id}:{role_key}:" f"{principal_id}", ) resource_label = ( f"scope={scope_external_id} " f"role={role_key} principal={principal_id}" )[:255] async with async_session_factory() as inc_db: inc_svc = ComplianceIncidentService(inc_db) await inc_svc.create_or_upsert_incident( incident_type=( "cloud_orphan_cleanup_helix_write_failed" ), severity=SEVERITY_CRITICAL, person_id=None, resource_type=( "cmdb_cloud_orphan_cleanup_target" ), resource_id=dedup_id, resource_label=resource_label, title=( "Cloud-side delete ok, Helix write " f"failed — {integration_name} " f"scope={scope_external_id} " f"role={role_key} principal=" f"{principal_id}" ), detail={ "integration_id": str(integration.id), "provider": integration_provider, "scope_id": str(scope.id), "scope_external_id": ( scope_external_id ), "role_key": role_key, "principal_id": principal_id, "recovered_email": ( target.get("recovered_email") ), "trace_id": trace_id, "error": be_636_step.reason(max_len=400), }, ) await inc_db.commit() except Exception: # noqa: BLE001 logger.error( "orphan-cleanup: compliance_incident " "write also failed — operator UI will not " "surface this drift. trace_id=%s", trace_id, ) # NB: we deliberately do NOT call # ``maybe_notify_failure`` directly here — that helper # is keyed on ``ScheduledJob.id`` and orphan_cleanup # runs nested inside a parent cron tick whose job_id # we don't carry through. The compliance_incident # above is the operator-UI surface; on-call paging # happens via the parent cron's failure rollup when # the orphan-cleanup summary surfaces a non-zero # ``cloud_succeeded_helix_failed`` counter and the # handler raises. Surfacing the counter happens via # the caller (sync.py / lifecycle_service) that reads # ``OrphanCleanupResult.error_message`` — see the # parent handler for the raise-on-errors plumbing. out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=False, error_message=( "cloud delete succeeded but Helix history " "write failed — see logs for trace " f"{trace_id}" ), )) continue out.append(OrphanCleanupResult( principal_id=principal_id, role_key=role_key, scope_id=scope_id, ok=True, already_absent=bool(result.already_absent), grant_id=grant_id, )) finally: try: await driver.close() except Exception: # noqa: BLE001 pass return out async def _safe_rollback_cleanup_session(session: AsyncSession) -> None: """Best-effort rollback after a Helix-side write failure inside the per-target commit block. We swallow any rollback error because the outer loop must keep going regardless — leaving the session dirty would cascade-fail every remaining target. """ try: await session.rollback() except Exception: # noqa: BLE001 logger.warning( "orphan-cleanup: rollback after failure also failed; " "continuing with potentially-dirty session", exc_info=True, )