"""API-key AK/SK credential rotation — dual-active mint + delayed retire.

Employees self-service "rotate my credential" on a cloud-auto-granted
API key: a *fresh* credential is minted and goes live immediately while
the *old* one keeps working for an employee-chosen grace period (立即 /
3 天 / 7 天 / 14 天), after which the old credential is torn down
automatically. This is the industry-standard dual-active rotation
pattern and every provider we auto-grant supports it natively:

* **AWS Bedrock** — the assignment is backed by a dedicated IAM user.
  IAM allows two access keys per user simultaneously, so rotation mints
  a *second* access key (``create_access_key``) and retires the old one
  (``delete_access_key``). The IAM user, policy, group, tags and
  cost-allocation rule are untouched — only the credential pair changes.

* **Aliyun Bailian** — the credential is a DashScope ``sk-`` API key
  under a workspace; a workspace holds many independent keys. Rotation
  mints a new ``sk-`` (``CreateApiKey``) and retires the old
  (``DeleteApiKey``). The companion RAM user is untouched.

* **Azure OpenAI / Foundry** — the credential is the account KEY1/KEY2.
  A Cognitive-Services account natively carries two keys that rotate
  independently. Rotation regenerates the *other* slot (so both the old
  slot and the fresh slot are valid = overlap), hands out the fresh
  slot, and at retire regenerates the *old* slot to invalidate it.

State machine (all provider-agnostic, stored on
``detail.<provider>.rotation``):

    (no rotation)
        │  rotate_key(grace_days)
        ▼
    status="minting"  (rotation_retire_at already stamped → sweep-visible)
        │  ├─ cloud mint fails → blob cleared + ``mint_failed`` audit row
        │  │    (durable marker: the vendor may hold an orphan credential)
        │  └─ worker crash mid-mint → stale tombstone; the hourly sweep
        │       clears it after MINTING_STALE_AFTER_HOURS with a
        │       ``mint_interrupted`` audit row (unblocks re-rotation)
        │  cloud mint ok
        ▼
    status="pending_retire"  (rotation_retire_at = now + grace_days)
        │                    │                          │
        │ retire cron        │ force_complete           │ cancel
        │ (retire_at <= now) │ (employee: "已切换")     │ (employee: revert)
        ▼                    ▼                          ▼
    old credential deleted, rotation cleared   new credential deleted,
                                               key_value reverted to old

    A FAILED cloud retire keeps ``pending_retire`` (``retire_attempts``
    counted in the blob) so the hourly sweep retries, audited as
    ``retire_failed``; after MAX_RETIRE_ATTEMPTS the state is cleared so
    the row doesn't strand, still audited as failed (manual cleanup).

The rotation blob carries the provider-specific *handle* of the old
credential (what to delete at retire) and of the new credential (what
to delete on cancel), plus the encrypted prior ``key_value`` so a
cancel can restore the employee's original credential bundle verbatim.

``key_value`` always holds the CURRENT credential the employee should
use — after a successful mint it is the NEW credential; reveal returns
it unchanged. The old credential is what the employee already has in
their running app; it keeps working until retire, so there is no
breakage window.

Security note: the rotation blob's ``prev_key_value`` is ciphertext
(AES-GCM, AAD = assignment.id) but is nonetheless stripped from every
API response in ``ApiKeyService._to_response`` — it never needs to
leave the DB.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.domains.apikey.models import ApiKeyAssignment
from app.domains.apikey.service import _encrypt_key_value
from app.integrations.models import Integration

logger = logging.getLogger(__name__)

# Providers whose auto-grant mints a rotatable cloud credential. Kept as
# a tuple so the order is deterministic (matters for detection when — in
# theory — a row carried more than one grant blob, which the platform
# never produces but we scan defensively).
ROTATABLE_PROVIDERS: tuple[str, ...] = ("bedrock", "bailian", "azure_openai")

ROTATION_STATUS_MINTING = "minting"
ROTATION_STATUS_PENDING_RETIRE = "pending_retire"
_ACTIVE_ROTATION_STATES = (
    ROTATION_STATUS_MINTING,
    ROTATION_STATUS_PENDING_RETIRE,
)

# Grace-period options the employee picks from (whole days). 0 = "立即"
# (hard cut — the old credential is retired the instant the new one is
# minted, no overlap). 14 days caps the dual-active exposure window.
ALLOWED_GRACE_DAYS: tuple[int, ...] = (0, 3, 7, 14)
DEFAULT_GRACE_DAYS = 7

# A failed cloud retire keeps the rotation in ``pending_retire`` so the
# hourly cron retries; after this many consecutive failures we give up
# (clearing the state so the row doesn't strand forever) — but callers
# must audit the give-up as ``status="failed"``: the old credential may
# still be live and needs manual cleanup.
MAX_RETIRE_ATTEMPTS = 6

# A ``minting`` tombstone older than this is a crashed mint (the cloud
# call window is seconds, not hours). The sweep clears it so the key
# stops being wedged in "轮换中" with no rotate/complete/cancel path.
MINTING_STALE_AFTER_HOURS = 2


# ── Detection / state helpers ────────────────────────────────────────

def detect_rotation_provider(assignment: ApiKeyAssignment) -> str | None:
    """Return the rotatable provider kind for a granted cloud assignment.

    ``None`` for non-cloud (OpenAI / Anthropic SaaS) keys, manually
    filled keys (no auto-grant), and any assignment whose grant blob
    isn't marked ``granted=True``.
    """
    detail = assignment.detail or {}
    for kind in ROTATABLE_PROVIDERS:
        blob = detail.get(kind) or {}
        if blob.get("granted"):
            return kind
    return None


def get_active_rotation(
    assignment: ApiKeyAssignment,
) -> tuple[str | None, dict[str, Any]]:
    """Return ``(kind, rotation_blob)`` for an in-flight rotation.

    ``(None, {})`` when the assignment has no rotation in ``minting`` or
    ``pending_retire`` state.
    """
    detail = assignment.detail or {}
    for kind in ROTATABLE_PROVIDERS:
        blob = (detail.get(kind) or {}).get("rotation")
        if isinstance(blob, dict) and blob.get("status") in _ACTIVE_ROTATION_STATES:
            return kind, blob
    return None, {}


def _set_provider_blob(
    assignment: ApiKeyAssignment, kind: str, updates: dict[str, Any],
) -> dict[str, Any]:
    """Merge ``updates`` into ``detail.<kind>`` and re-assign ``detail``
    (JSON columns need a fresh object for SQLAlchemy to flush the change).
    Returns the merged provider sub-blob.
    """
    detail = dict(assignment.detail or {})
    blob = dict(detail.get(kind) or {})
    blob.update(updates)
    detail[kind] = blob
    assignment.detail = detail
    return blob


def _clear_rotation_blob(assignment: ApiKeyAssignment, kind: str) -> None:
    detail = dict(assignment.detail or {})
    blob = dict(detail.get(kind) or {})
    blob.pop("rotation", None)
    detail[kind] = blob
    assignment.detail = detail
    assignment.rotation_retire_at = None


# ── Integration resolution ───────────────────────────────────────────

async def _resolve_integration(
    db: AsyncSession, assignment: ApiKeyAssignment, kind: str,
) -> Integration:
    """Resolve the cloud integration the credential lives in.

    Prefers the per-grant ``detail.<kind>.integration_id`` and falls back
    to the denormalised ``assignment.iam_integration_id`` column. Raises
    a structured 502 when neither resolves — the same posture the revoke
    path uses (a rotation can't proceed without knowing which tenant to
    mint against).
    """
    blob = (assignment.detail or {}).get(kind) or {}
    integ_id = blob.get("integration_id") or (
        str(assignment.iam_integration_id)
        if assignment.iam_integration_id else None
    )
    if not integ_id:
        raise AppException(
            code="ROTATION_NO_INTEGRATION",
            message=(
                f"{kind} rotation: assignment has no integration_id on "
                "detail nor iam_integration_id column; cannot resolve the "
                "cloud tenant to rotate against."
            ),
            status_code=502,
        )
    integration = await db.get(Integration, uuid.UUID(str(integ_id)))
    if integration is None:
        raise AppException(
            code="ROTATION_NO_INTEGRATION",
            message=(
                f"{kind} rotation: integration {integ_id} not found."
            ),
            status_code=502,
        )
    return integration


def _build_aws_provider(integration: Integration):
    """Construct a real boto3-backed AWSProvider from an integration.

    Mirrors ``bedrock_autogrant.maybe_auto_grant_bedrock`` credential
    handling exactly (config_public merged under decrypted secrets).
    """
    from app.domains.cmdb.credential_crypto import decrypt_fields
    from app.domains.cmdb.providers.aws import AWSProvider
    from app.domains.cmdb.providers.base import ProviderCredentials

    cred_fields = (
        decrypt_fields(
            integration.secrets_encrypted, integration.secrets_nonce,
            integration_id=str(integration.id),
        )
        if integration.secrets_encrypted else {}
    )
    merged = {
        k: str(v)
        for k, v in (integration.config_public or {}).items()
        if v is not None
    }
    merged.update(cred_fields)
    return AWSProvider(ProviderCredentials(
        provider_id=str(integration.id),
        provider_type="aws",
        account_alias=integration.name,
        fields=merged,
    ))


# ── Pre-flight check ─────────────────────────────────────────────────

async def precheck_rotation(
    db: AsyncSession, assignment: ApiKeyAssignment, kind: str,
) -> None:
    """Provider-specific feasibility check before a rotation starts.

    Raises ``AppException`` when the rotation cannot proceed. Currently
    only Bedrock has a hard constraint: AWS caps an IAM user at two
    access keys, so if the user already holds two (a prior rotation
    didn't retire, or an admin minted a console key) a mint would fail
    with ``LimitExceeded`` — we surface an actionable error first.
    """
    if kind != "bedrock":
        return
    blob = (assignment.detail or {}).get("bedrock") or {}
    user_name = blob.get("iam_user_name")
    if not user_name:
        raise AppException(
            code="ROTATION_NOT_SUPPORTED",
            message=(
                "This Bedrock assignment has no bound IAM user to rotate "
                "(manually filled or not auto-granted)."
            ),
            status_code=400,
        )
    integration = await _resolve_integration(db, assignment, kind)
    provider = _build_aws_provider(integration)
    try:
        keys = await provider.list_access_keys(user_name)
    except Exception as exc:  # noqa: BLE001
        from app.integrations.service import _scrub_message
        # A missing user is a real, actionable state; anything else is a
        # transient cloud problem the employee should retry. The boto
        # driver re-raises the raw ``ClientError`` (code=NoSuchEntity) for
        # a deleted user, so match on the code string rather than a type.
        if "NoSuchEntity" in str(exc):
            raise AppException(
                code="ROTATION_USER_ABSENT",
                message=(
                    "The bound IAM user no longer exists on AWS — the key "
                    "may have been revoked. Nothing to rotate."
                ),
                status_code=409,
            ) from exc
        raise AppException(
            code="ROTATION_PRECHECK_FAILED",
            message=(
                "Could not verify the IAM user's access-key count before "
                f"rotating: {_scrub_message(str(exc))[:200]}. Retry shortly."
            ),
            status_code=502,
        ) from exc
    finally:
        try:
            await provider.close()
        except Exception:  # noqa: BLE001
            pass
    if len(keys) >= 2:
        raise AppException(
            code="ROTATION_KEY_LIMIT",
            message=(
                "The IAM user already holds two access keys (AWS hard "
                "limit). A previous rotation may not have retired yet, or "
                "a console key exists. Wait for the pending rotation to "
                "retire, or ask an admin to clean up the extra key."
            ),
            status_code=409,
        )


# ── Per-provider mint ────────────────────────────────────────────────

async def _mint_bedrock(
    db: AsyncSession, assignment: ApiKeyAssignment, integration: Integration,
) -> tuple[str, dict[str, Any], dict[str, Any]]:
    """Mint a second IAM access key. Returns (new_bundle, prev_handle,
    new_handle)."""
    blob = (assignment.detail or {}).get("bedrock") or {}
    user_name = blob["iam_user_name"]
    region = blob.get("region") or "ap-southeast-1"
    user_arn = blob.get("user_arn") or assignment.iam_user_arn or ""
    provider = _build_aws_provider(integration)
    try:
        result = await provider.create_access_key(user_name)
        ak = result["access_key_id"]
        # Resolve the OLD key from a list taken AFTER mint (there should
        # now be exactly two keys: old + new) rather than a pre-mint list.
        # A pre-mint ListAccessKeys that returned stale/empty would leave
        # prev_handle=None and the old key would never retire (dual-active
        # never ends). Post-mint the new key is guaranteed present, so the
        # "not new" key is unambiguously the old one to retire.
        try:
            after = await provider.list_access_keys(user_name)
        except Exception:
            # create succeeded but we can't resolve the old key to
            # schedule its retire. Best-effort delete the just-minted key
            # so we don't leave an orphan (which would also trip the
            # 2-key ROTATION_KEY_LIMIT on the next attempt), then re-raise
            # so start_rotation surfaces ROTATION_CLOUD_FAILED and clears
            # the tombstone.
            try:
                await provider.delete_access_key(user_name, ak)
            except Exception:  # noqa: BLE001
                logger.exception(
                    "apikey.rotation: failed to clean up freshly-minted "
                    "Bedrock key %s for user %s after list_access_keys "
                    "failed — MANUAL CLEANUP of the extra IAM key required",
                    ak, user_name,
                )
            raise
    finally:
        try:
            await provider.close()
        except Exception:  # noqa: BLE001
            pass
    sk = result["secret_access_key"]
    prev_ids = [k["access_key_id"] for k in after if k["access_key_id"] != ak]
    if not prev_ids:
        # No prior key found alongside the new one — nothing to retire, so
        # dual-active degenerates to "just a fresh key". Rare (IAM
        # ListAccessKeys is strongly consistent); log so the operator can
        # tell why no retire will happen rather than discovering it later.
        logger.warning(
            "apikey.rotation: Bedrock rotation for IAM user %s found no "
            "prior access key beside the new one; retire will be a no-op",
            user_name,
        )
    prev_handle = {"access_key_id": prev_ids[0] if prev_ids else None}
    new_handle = {"access_key_id": ak}
    bundle = (
        f"ACCESS_KEY_ID={ak}\nSECRET_ACCESS_KEY={sk}\n"
        f"REGION={region}\nUSER_ARN={user_arn}"
    )
    return bundle, prev_handle, new_handle


async def _mint_bailian(
    db: AsyncSession, assignment: ApiKeyAssignment, integration: Integration,
) -> tuple[str, dict[str, Any], dict[str, Any]]:
    """Mint a fresh DashScope sk- key. Returns (new_bundle, prev_handle,
    new_handle)."""
    from app.domains.bailian.client import (
        BailianClient,
        dashscope_endpoint_for_provider,
        default_region_for_provider,
        resolve_credentials,
    )
    from app.domains.bailian.models import BailianApiKey, BailianWorkspace

    blob = (assignment.detail or {}).get("bailian") or {}
    workspace_id = str(blob.get("workspace_id") or "")
    ram_user_name = blob.get("ram_user_name") or blob.get("iam_user_name") or ""
    prev_api_key_id = blob.get("api_key_id")
    if not workspace_id:
        raise AppException(
            code="ROTATION_NOT_SUPPORTED",
            message="This Bailian assignment has no workspace_id to rotate.",
            status_code=400,
        )
    creds = resolve_credentials(integration)
    region = creds.get("bailian_region") or default_region_for_provider(
        integration.provider,
    )
    dashscope_endpoint = dashscope_endpoint_for_provider(integration.provider)
    description = (
        f"Helix · {assignment.request_number} · {ram_user_name} (rotation)"
    )
    client = BailianClient(creds)
    # ModelStudio 分区域:workspace 在哪个区域,Key 就得在哪个区域建。
    from app.domains.bailian.service import BailianService
    _ws_region = await BailianService(db).workspace_region(
        integration_id=str(integration.id),
        workspace_id=str(workspace_id) if workspace_id else None,
    )
    api_key_resp = await client.create_api_key(
        workspace_id=workspace_id, description=description,
        region=_ws_region,
    )
    new_api_key_id = api_key_resp["api_key_id"]
    bundle = (
        f"API_KEY={api_key_resp['api_key_value']}\n"
        f"WORKSPACE_ID={workspace_id}\n"
        f"DASHSCOPE_ENDPOINT={dashscope_endpoint}\n"
        f"REGION={region}\n"
        f"API_KEY_ID={new_api_key_id}\n"
        f"RAM_USER_NAME={ram_user_name}\n"
        f"NOTE=Rotated by Helix on {datetime.now(UTC).isoformat()}. Use as "
        f"`dashscope.api_key = \"<API_KEY>\"` with the DashScope Python "
        f"SDK, or export DASHSCOPE_API_KEY=<API_KEY>. The previous key "
        f"keeps working until it is retired. See /help#bailian-employee."
    )
    # Register the new sk- in bailian_api_keys (active) so the admin
    # Bailian page shows it. The OLD row stays active until retire.
    workspace = (await db.execute(
        select(BailianWorkspace).where(
            BailianWorkspace.integration_id == str(integration.id),
            BailianWorkspace.workspace_id == workspace_id,
            BailianWorkspace.deleted_at.is_(None),
        )
    )).scalar_one_or_none()
    if workspace is not None:
        db.add(BailianApiKey(
            id=uuid.uuid4(),
            integration_id=str(integration.id),
            workspace_pk=workspace.id,
            api_key_id=new_api_key_id,
            display_name=description,
            owner_person_id=(
                str(assignment.assigned_to) if assignment.assigned_to else None
            ),
            ram_user_name=ram_user_name,
            assignment_id=assignment.id,
            status="active",
            raw_data={
                "source": "auto-grant.rotation",
                "request_number": assignment.request_number,
                "created_by": api_key_resp.get("created_by"),
                "rotated_from_api_key_id": prev_api_key_id,
            },
        ))
    prev_handle = {
        "api_key_id": str(prev_api_key_id) if prev_api_key_id else None,
        "workspace_id": workspace_id,
    }
    new_handle = {"api_key_id": str(new_api_key_id), "workspace_id": workspace_id}
    # Update the canonical detail fields to point at the new key.
    _set_provider_blob(assignment, "bailian", {
        "api_key_id": str(new_api_key_id),
        "api_key_created_by": api_key_resp.get("created_by"),
        "api_key_description": description,
        "region": region,
        "dashscope_endpoint": dashscope_endpoint,
    })
    return bundle, prev_handle, new_handle


async def _mint_azure_openai(
    db: AsyncSession, assignment: ApiKeyAssignment, integration: Integration,
) -> tuple[str, dict[str, Any], dict[str, Any]]:
    """Regenerate the *other* key slot for overlap. Returns (new_bundle,
    prev_handle, new_handle)."""
    from app.domains.cmdb.lifecycle_service import _load_cloud_driver

    blob = (assignment.detail or {}).get("azure_openai") or {}
    subscription_id = str(blob.get("subscription_id") or "")
    resource_group = str(blob.get("resource_group") or "")
    account_name = str(
        blob.get("foundry_account_name") or blob.get("account_name") or "",
    )
    if not (subscription_id and resource_group and account_name):
        raise AppException(
            code="ROTATION_NOT_SUPPORTED",
            message=(
                "This Azure OpenAI assignment is missing subscription / "
                "resource-group / account and cannot be rotated."
            ),
            status_code=400,
        )
    # The slot currently handed to the employee. Autogrant always issues
    # key1, so default there; a prior rotation flips this to key2, and
    # the next rotation flips it back.
    current_slot = str(blob.get("active_key_slot") or "key1").lower()
    other_slot = "key2" if current_slot == "key1" else "key1"

    driver = await _load_cloud_driver(db, integration)
    try:
        keys = await driver.regenerate_foundry_account_key(
            subscription_id=subscription_id,
            resource_group=resource_group,
            account_name=account_name,
            key_name=other_slot,
        )
    finally:
        try:
            await driver.close()
        except Exception:  # noqa: BLE001
            pass
    new_key = keys.get(other_slot) or ""
    if not new_key:
        raise AppException(
            code="ROTATION_CLOUD_FAILED",
            message=(
                "Azure regenerateKey returned an empty key for slot "
                f"{other_slot}."
            ),
            status_code=502,
        )
    account_endpoint_url = str(blob.get("foundry_account_endpoint") or "")
    foundry_project_endpoint = str(blob.get("foundry_project_endpoint") or "")
    project_name = str(blob.get("foundry_project_name") or "")
    bundle = (
        f"# ── Microsoft Foundry SDK (azure-ai-projects) ──\n"
        f"# Rotated by Helix on {datetime.now(UTC).isoformat()} — the\n"
        f"# previous key keeps working until it is retired.\n"
        f"FOUNDRY_PROJECT_ENDPOINT={foundry_project_endpoint}\n"
        f"FOUNDRY_API_KEY={new_key}\n"
        f"FOUNDRY_PROJECT_NAME={project_name}\n"
        f"FOUNDRY_ACCOUNT_NAME={account_name}\n"
        f"FOUNDRY_ACCOUNT_ENDPOINT={account_endpoint_url}\n"
        f"\n"
        f"# ── openai python client (AzureOpenAI) ──\n"
        f"AZURE_OPENAI_API_KEY={new_key}\n"
        f"AZURE_OPENAI_ENDPOINT={account_endpoint_url}\n"
        f"AZURE_OPENAI_API_VERSION=2024-10-21\n"
        f"# AZURE_OPENAI_DEPLOYMENT_NAME=<ask-IT-or-see-How-to-use-panel>"
        f"  # model= / deployment, provisioned by IT\n"
        f"\n"
        f"AZURE_RESOURCE_GROUP={resource_group}\n"
        f"AZURE_SUBSCRIPTION_ID={subscription_id}\n"
        f"# Model deployments are managed by IT — the 'How to use' panel\n"
        f"# lists the live deployment name; contact IT if none / on\n"
        f"# DeploymentNotFound. See /help#foundry-employee."
    )
    # ``prev_handle`` = the slot to invalidate at retire (the OLD active
    # slot); ``new_handle`` = the slot to invalidate on cancel (the one
    # we just regenerated).
    prev_handle = {"slot": current_slot}
    new_handle = {"slot": other_slot}
    # Flip the canonical active slot to the newly-issued one.
    _set_provider_blob(assignment, "azure_openai", {
        "active_key_slot": other_slot,
    })
    return bundle, prev_handle, new_handle


_MINTERS = {
    "bedrock": _mint_bedrock,
    "bailian": _mint_bailian,
    "azure_openai": _mint_azure_openai,
}


async def start_rotation(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
    *,
    kind: str,
    grace_days: int,
    initiated_by: uuid.UUID | None,
) -> dict[str, Any]:
    """Mint a fresh credential and set up the pending-retire state.

    Leak-safety mirrors the autogrant flow: we stash the prior
    ``key_value`` and write a ``minting`` tombstone (committed) BEFORE
    the cloud call. On cloud success we durably commit the new
    ``key_value`` + rotation blob + ``rotation_retire_at`` together. On
    cloud failure the rotation blob is cleared and the exception
    re-raised.

    Crash-window honesty (matches autogrant's tradeoff): the two-commit
    shape bounds — but does not eliminate — orphans.
    * Crash DURING the cloud mint → the committed ``minting`` tombstone
      is the discoverable marker; ``key_value`` still points at the old
      (working) credential, so nothing broke for the user.
    * Crash AFTER cloud success but BEFORE the second commit → the new
      credential is live on the cloud but its handle was rolled back;
      the row is left in ``status='minting'`` (tombstone committed).
      ``key_value`` is still the OLD bundle (it's swapped only in the
      atomic final commit), so the user is never handed a rolled-back
      credential. For Bedrock this orphan is a 2nd IAM access key, which
      trips the ``ROTATION_KEY_LIMIT`` precheck on the next attempt —
      an admin clears the extra key (and the stuck ``minting`` blob) to
      recover, same manual-reconciliation posture as autogrant's
      ``in_flight`` tombstone. This window is ~1s and rare.

    Returns ``{"prev_handle", "new_handle", "retire_at"}``.
    """
    minter = _MINTERS.get(kind)
    if minter is None:
        raise AppException(
            code="ROTATION_NOT_SUPPORTED",
            message=f"Provider {kind!r} does not support rotation.",
            status_code=400,
        )
    integration = await _resolve_integration(db, assignment, kind)
    now = datetime.now(UTC)
    retire_at = now if grace_days <= 0 else now + _days(grace_days)

    # Leak-safety tombstone: stash the prior key_value (for cancel) and
    # mark minting. Committed before touching the cloud. Setting
    # ``rotation_retire_at`` here (not only on success) makes a crashed
    # mint visible to the hourly sweep, which repairs stale ``minting``
    # tombstones — otherwise the row would 409 every future rotate with
    # no recovery surface.
    prev_key_value = assignment.key_value
    _set_provider_blob(assignment, kind, {"rotation": {
        "status": ROTATION_STATUS_MINTING,
        "initiated_by": str(initiated_by) if initiated_by else None,
        "grace_days": grace_days,
        "minted_at": now.isoformat(),
        "prev_key_value": prev_key_value,
    }})
    assignment.rotation_retire_at = retire_at
    await db.commit()
    await db.refresh(assignment)

    try:
        bundle, prev_handle, new_handle = await minter(db, assignment, integration)
    except AppException as exc:
        _clear_rotation_blob(assignment, kind)
        await _audit_mint_failed(db, assignment, kind=kind, error=str(exc))
        await db.commit()
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "apikey.rotation: %s mint failed for assignment %s",
            kind, assignment.id,
        )
        _clear_rotation_blob(assignment, kind)
        await _audit_mint_failed(db, assignment, kind=kind, error=str(exc))
        await db.commit()
        from app.integrations.service import _scrub_message
        raise AppException(
            code="ROTATION_CLOUD_FAILED",
            message=(
                f"{kind} credential mint failed: "
                f"{_scrub_message(str(exc))[:200]}. No change was made; "
                "retry shortly."
            ),
            status_code=502,
        ) from exc

    # Cloud mint succeeded — promote to pending_retire and swap key_value.
    assignment.key_value = _encrypt_key_value(bundle, assignment_id=assignment.id)
    _set_provider_blob(assignment, kind, {"rotation": {
        "status": ROTATION_STATUS_PENDING_RETIRE,
        "initiated_by": str(initiated_by) if initiated_by else None,
        "grace_days": grace_days,
        "minted_at": now.isoformat(),
        "retire_at": retire_at.isoformat(),
        "prev_handle": prev_handle,
        "new_handle": new_handle,
        "prev_key_value": prev_key_value,
        "reminded_at": None,
    }})
    assignment.rotation_retire_at = retire_at
    # Durable commit: cloud credential + Helix state land together so a
    # rollback in an outer transaction can't orphan the new credential.
    await db.commit()
    await db.refresh(assignment)
    logger.info(
        "apikey.rotation: %s rotation started for assignment=%s "
        "grace_days=%s retire_at=%s",
        kind, assignment.id, grace_days, retire_at.isoformat(),
    )
    return {
        "prev_handle": prev_handle,
        "new_handle": new_handle,
        "retire_at": retire_at,
    }


async def _audit_mint_failed(
    db: AsyncSession, assignment: ApiKeyAssignment, *, kind: str, error: str,
) -> None:
    """Durable marker for a failed / ambiguous mint.

    The failure may have arrived AFTER the vendor created the credential
    (e.g. an httpx timeout on Bailian CreateApiKey, which is documented
    non-idempotent) — this audit row is the discoverable trail ops uses
    to reconcile a possible orphan credential on the vendor side, since
    the rotation tombstone is cleared to unblock re-rotation.
    """
    from app.audit.service import AuditService
    from app.integrations.service import _scrub_message

    await AuditService(db).log(
        trace_id=f"apikey.rotation_mint.{assignment.id}",
        actor_type="system",
        actor_user_id=None,
        action="apikey.rotation.mint_failed",
        resource_type="api_key_assignment",
        resource_id=str(assignment.id),
        status="failed",
        detail={
            "request_number": assignment.request_number,
            "kind": kind,
            "error": _scrub_message(error)[:400],
        },
    )


# ── Per-provider retire (delete the OLD credential) ──────────────────

async def _retire_bedrock(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    access_key_id = handle.get("access_key_id")
    blob = (assignment.detail or {}).get("bedrock") or {}
    user_name = blob.get("iam_user_name")
    if not access_key_id or not user_name:
        return {"kind": "bedrock", "skipped": "no-handle"}
    provider = _build_aws_provider(integration)
    try:
        result = await provider.delete_access_key(user_name, access_key_id)
    finally:
        try:
            await provider.close()
        except Exception:  # noqa: BLE001
            pass
    return {"kind": "bedrock", "access_key_id": access_key_id, "result": result}


async def _retire_bailian(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.bailian.client import BailianClient, resolve_credentials
    from app.domains.bailian.models import BailianApiKey

    api_key_id = handle.get("api_key_id")
    workspace_id = handle.get("workspace_id")
    if not api_key_id:
        return {"kind": "bailian", "skipped": "no-handle"}
    client = BailianClient(resolve_credentials(integration))
    # 打错区域会 HTTP 200 但删不掉(不报错),必须按 workspace 的区域走。
    from app.domains.bailian.service import BailianService
    _ws_region = await BailianService(db).workspace_region(
        integration_id=str(integration.id),
        workspace_id=str(workspace_id) if workspace_id else None,
    )
    result = await client.delete_api_key(
        api_key_id=str(api_key_id),
        workspace_id=str(workspace_id) if workspace_id else None,
        region=_ws_region,
    )
    # Soft-delete the old registry row(s).
    rows = (await db.execute(
        select(BailianApiKey).where(
            BailianApiKey.assignment_id == assignment.id,
            BailianApiKey.api_key_id == str(api_key_id),
            BailianApiKey.deleted_at.is_(None),
        )
    )).scalars().all()
    for row in rows:
        row.status = "deleted"
        row.deleted_at = datetime.now(UTC)
    return {"kind": "bailian", "api_key_id": api_key_id, "result": result}


async def _retire_azure_openai(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.cmdb.lifecycle_service import _load_cloud_driver

    slot = handle.get("slot")
    blob = (assignment.detail or {}).get("azure_openai") or {}
    subscription_id = str(blob.get("subscription_id") or "")
    resource_group = str(blob.get("resource_group") or "")
    account_name = str(
        blob.get("foundry_account_name") or blob.get("account_name") or "",
    )
    if not (slot and subscription_id and resource_group and account_name):
        return {"kind": "azure_openai", "skipped": "no-handle"}
    driver = await _load_cloud_driver(db, integration)
    try:
        await driver.regenerate_foundry_account_key(
            subscription_id=subscription_id,
            resource_group=resource_group,
            account_name=account_name,
            key_name=slot,
        )
    finally:
        try:
            await driver.close()
        except Exception:  # noqa: BLE001
            pass
    return {"kind": "azure_openai", "slot": slot, "regenerated": True}


_RETIRERS = {
    "bedrock": _retire_bedrock,
    "bailian": _retire_bailian,
    "azure_openai": _retire_azure_openai,
}


async def retire_rotation(
    db: AsyncSession, assignment: ApiKeyAssignment, *, kind: str,
) -> dict[str, Any]:
    """Delete the OLD credential and clear the rotation state.

    Idempotent: if the cloud identity is already gone (revoked /
    offboarding teardown) the provider retire returns ``already_absent``
    or is a no-op. On success the rotation blob + ``rotation_retire_at``
    column are cleared so the row leaves the "轮换中" state.

    On a cloud-teardown FAILURE the rotation state is KEPT (row stays
    ``pending_retire``, attempt count recorded in the blob) so the hourly
    sweep re-attempts — the old credential is still live and must not be
    reported as retired. The return carries ``retire_ok=False`` (plus
    ``attempts``); callers must inspect ``retire_ok``, audit the failure
    as a non-success, and must not send the owner a "旧凭证已停用"
    notification. After ``MAX_RETIRE_ATTEMPTS`` consecutive failures the
    state is cleared so the row doesn't strand forever, with
    ``gave_up=True`` in the summary — the credential may still be live
    and needs manual cleanup.

    Does NOT commit — the caller (cron / force-complete) owns the
    transaction so the retire + audit land atomically.
    """
    _kind, rotation = get_active_rotation(assignment)
    if not rotation:
        return {"kind": kind, "skipped": "no-active-rotation", "retire_ok": True}
    prev_handle = rotation.get("prev_handle") or {}
    retirer = _RETIRERS.get(kind)
    summary: dict[str, Any] = {"kind": kind, "retire_ok": True}
    if retirer is not None:
        try:
            integration = await _resolve_integration(db, assignment, kind)
            summary["retire"] = await retirer(
                db, assignment, integration, prev_handle,
            )
        except Exception as exc:  # noqa: BLE001
            # The cloud teardown of the OLD credential failed — that
            # credential is still LIVE (billable, usable, reachable by a
            # leaver). We must NOT clear the rotation state here: clearing
            # it would strand a live credential with no retry and let the
            # caller/cron report "retired" while the key lives on. Instead
            # keep the row in ``pending_retire`` (attempt counter recorded)
            # so the hourly sweep re-attempts, and signal the failure to
            # callers via ``retire_ok=False`` so audit/alerting/
            # notifications reflect reality. After MAX_RETIRE_ATTEMPTS
            # consecutive failures the state is cleared (the row must not
            # strand forever) with ``gave_up=True`` — still not a success.
            attempts = int(rotation.get("retire_attempts") or 0) + 1
            logger.exception(
                "apikey.rotation: %s retire failed for assignment %s "
                "(attempt %s/%s)",
                kind, assignment.id, attempts, MAX_RETIRE_ATTEMPTS,
            )
            from app.integrations.service import _scrub_message
            summary["error"] = _scrub_message(str(exc))[:400]
            summary["retire_ok"] = False
            summary["attempts"] = attempts
            if attempts >= MAX_RETIRE_ATTEMPTS:
                summary["gave_up"] = True
                _clear_rotation_blob(assignment, kind)
            else:
                summary["gave_up"] = False
                _set_provider_blob(assignment, kind, {"rotation": {
                    **rotation,
                    "retire_attempts": attempts,
                    "last_retire_error": summary["error"],
                }})
            return summary
    _clear_rotation_blob(assignment, kind)
    return summary


# ── Per-provider cancel (delete the NEW credential, revert) ──────────

async def _cancel_bedrock(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    access_key_id = handle.get("access_key_id")
    blob = (assignment.detail or {}).get("bedrock") or {}
    user_name = blob.get("iam_user_name")
    if not access_key_id or not user_name:
        return {"kind": "bedrock", "skipped": "no-handle"}
    provider = _build_aws_provider(integration)
    try:
        result = await provider.delete_access_key(user_name, access_key_id)
    finally:
        try:
            await provider.close()
        except Exception:  # noqa: BLE001
            pass
    return {"kind": "bedrock", "access_key_id": access_key_id, "result": result}


async def _cancel_bailian(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.bailian.client import BailianClient, resolve_credentials
    from app.domains.bailian.models import BailianApiKey

    api_key_id = handle.get("api_key_id")
    workspace_id = handle.get("workspace_id")
    if not api_key_id:
        return {"kind": "bailian", "skipped": "no-handle"}
    client = BailianClient(resolve_credentials(integration))
    # 打错区域会 HTTP 200 但删不掉(不报错),必须按 workspace 的区域走。
    from app.domains.bailian.service import BailianService
    _ws_region = await BailianService(db).workspace_region(
        integration_id=str(integration.id),
        workspace_id=str(workspace_id) if workspace_id else None,
    )
    result = await client.delete_api_key(
        api_key_id=str(api_key_id),
        workspace_id=str(workspace_id) if workspace_id else None,
        region=_ws_region,
    )
    # Soft-delete the NEW registry row we inserted at mint.
    rows = (await db.execute(
        select(BailianApiKey).where(
            BailianApiKey.assignment_id == assignment.id,
            BailianApiKey.api_key_id == str(api_key_id),
            BailianApiKey.deleted_at.is_(None),
        )
    )).scalars().all()
    for row in rows:
        row.status = "deleted"
        row.deleted_at = datetime.now(UTC)
    return {"kind": "bailian", "api_key_id": api_key_id, "result": result}


async def _cancel_azure_openai(
    db: AsyncSession, assignment: ApiKeyAssignment,
    integration: Integration, handle: dict[str, Any],
) -> dict[str, Any]:
    # Cancel = invalidate the freshly-regenerated slot. The old slot was
    # never touched, so regenerating the new slot restores the invariant
    # "only the old key works".
    from app.domains.cmdb.lifecycle_service import _load_cloud_driver

    slot = handle.get("slot")
    blob = (assignment.detail or {}).get("azure_openai") or {}
    subscription_id = str(blob.get("subscription_id") or "")
    resource_group = str(blob.get("resource_group") or "")
    account_name = str(
        blob.get("foundry_account_name") or blob.get("account_name") or "",
    )
    if not (slot and subscription_id and resource_group and account_name):
        return {"kind": "azure_openai", "skipped": "no-handle"}
    driver = await _load_cloud_driver(db, integration)
    try:
        await driver.regenerate_foundry_account_key(
            subscription_id=subscription_id,
            resource_group=resource_group,
            account_name=account_name,
            key_name=slot,
        )
    finally:
        try:
            await driver.close()
        except Exception:  # noqa: BLE001
            pass
    return {"kind": "azure_openai", "slot": slot, "regenerated": True}


_CANCELLERS = {
    "bedrock": _cancel_bedrock,
    "bailian": _cancel_bailian,
    "azure_openai": _cancel_azure_openai,
}


async def cancel_rotation(
    db: AsyncSession, assignment: ApiKeyAssignment, *, kind: str,
) -> dict[str, Any]:
    """Invalidate the NEW credential and restore the prior ``key_value``.

    The employee decided not to switch. We delete the just-minted
    credential, revert ``key_value`` to the stashed prior bundle (so
    reveal hands back the original credential the employee is still
    using), restore the canonical provider handle to the old one, and
    clear the rotation state.

    Does NOT commit — the caller owns the transaction.
    """
    _kind, rotation = get_active_rotation(assignment)
    if not rotation:
        return {"kind": kind, "skipped": "no-active-rotation"}
    new_handle = rotation.get("new_handle") or {}
    prev_handle = rotation.get("prev_handle") or {}
    prev_key_value = rotation.get("prev_key_value")
    canceller = _CANCELLERS.get(kind)
    summary: dict[str, Any] = {"kind": kind}
    if canceller is not None:
        try:
            integration = await _resolve_integration(db, assignment, kind)
            summary["cancel"] = await canceller(
                db, assignment, integration, new_handle,
            )
        except Exception as exc:  # noqa: BLE001
            logger.exception(
                "apikey.rotation: %s cancel failed for assignment %s",
                kind, assignment.id,
            )
            from app.integrations.service import _scrub_message
            raise AppException(
                code="ROTATION_CANCEL_FAILED",
                message=(
                    f"Could not invalidate the new {kind} credential: "
                    f"{_scrub_message(str(exc))[:200]}. Rotation left "
                    "in place; retry cancel or let it retire."
                ),
                status_code=502,
            ) from exc
    # Restore the prior credential bundle + canonical provider handle.
    if prev_key_value is not None:
        assignment.key_value = prev_key_value
    _restore_canonical_handle(assignment, kind, prev_handle)
    _clear_rotation_blob(assignment, kind)
    return summary


def _restore_canonical_handle(
    assignment: ApiKeyAssignment, kind: str, prev_handle: dict[str, Any],
) -> None:
    """Point the canonical provider detail fields back at the OLD
    credential after a cancel."""
    if kind == "bailian":
        api_key_id = prev_handle.get("api_key_id")
        if api_key_id:
            _set_provider_blob(assignment, "bailian", {"api_key_id": str(api_key_id)})
    elif kind == "azure_openai":
        slot = prev_handle.get("slot")
        if slot:
            _set_provider_blob(assignment, "azure_openai", {"active_key_slot": slot})
    # bedrock: key_value restore already carries the old AK/SK; there is
    # no separate canonical access-key-id field on the detail blob.


# ── Notifications ────────────────────────────────────────────────────

async def _dispatch_to_assignee(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
    *,
    title: str,
    body_text: str,
    template_key: str,
    context: dict[str, Any],
    trace_id: str,
) -> None:
    """Send one in-app + Feishu card to the key's owner. Swallows
    dispatch errors — a notification failure must never fail (or
    reverse) a rotation state change."""
    from app.identity.person_models import IdentityPerson
    from app.notifications.service import (
        NotificationEnvelope,
        NotificationService,
    )

    assignee = await db.get(IdentityPerson, assignment.assigned_to)
    if assignee is None:
        return
    try:
        await NotificationService(db).dispatch(
            NotificationEnvelope(
                recipient=assignee,
                title=title,
                body_text=body_text,
                template_key=template_key,
                url="/me/api-keys",
                related_resource_type="api_key_assignment",
                related_resource_id=assignment.id,
                context=context,
            ),
            trace_id=trace_id,
        )
    except Exception:  # noqa: BLE001
        logger.exception(
            "apikey.rotation: notify (%s) failed for assignment %s",
            template_key, assignment.id,
        )


async def _provider_label(
    db: AsyncSession, assignment: ApiKeyAssignment,
) -> str:
    # Never raise: this feeds the notification helpers, whose whole
    # contract is "a notification failure must not affect (or reverse)
    # committed rotation state". A DB hiccup here would otherwise
    # propagate out of a notify_* call — in the sweep it would double-
    # count an already-retired row as an error, and in rotate/complete/
    # cancel it would 500 the caller after the rotation already committed.
    from app.domains.apikey.models import ApiKeyProvider
    try:
        provider = await db.get(ApiKeyProvider, assignment.provider_id)
        return provider.name if provider else "AI API Key"
    except Exception:  # noqa: BLE001
        logger.warning(
            "apikey.rotation: provider label lookup failed for assignment "
            "%s; using generic label", assignment.id,
        )
        return "AI API Key"


async def notify_rotation_started(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
    *,
    retire_at: datetime,
    grace_days: int,
    trace_id: str,
) -> None:
    label = await _provider_label(db, assignment)
    when = retire_at.date().isoformat()
    if grace_days <= 0:
        tail_zh = "旧凭证已立即停用,请确认应用已切换到新凭证。"
        tail_en = "The previous credential was retired immediately."
    else:
        tail_zh = (
            f"新旧凭证在宽限期内同时有效,旧凭证将于 {when}({grace_days} 天后)"
            "自动停用。请在此之前把应用里的凭证换成新的。"
        )
        tail_en = (
            f"Both credentials work during the grace window; the old one "
            f"retires on {when} ({grace_days} days)."
        )
    await _dispatch_to_assignee(
        db, assignment,
        title=f"API Key 已轮换:{label}",
        body_text=(
            f"你的 API Key 申请单 {assignment.request_number}({label})"
            f"已生成新凭证。\n{tail_zh}\n到「我的 API Keys」查看并复制新凭证:"
            "/me/api-keys\n---\n"
            f"Your {label} API key ({assignment.request_number}) has a new "
            f"credential. {tail_en} View it at /me/api-keys."
        ),
        template_key="apikey.rotation.started",
        context={
            "request_number": assignment.request_number,
            "provider_name": label,
            "retire_at": retire_at.isoformat(),
            # Date-only rendering for the template body; ``retire_at``
            # stays as the full timestamp for anything reading the raw
            # context (bell payload, outbox replay).
            "retire_date": when,
            "grace_days": grace_days,
        },
        trace_id=trace_id,
    )


async def notify_rotation_retired(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
    *,
    trace_id: str,
    forced: bool = False,
) -> None:
    label = await _provider_label(db, assignment)
    how_zh = "(你手动确认已切换)" if forced else "(宽限期结束)"
    await _dispatch_to_assignee(
        db, assignment,
        title=f"API Key 旧凭证已停用:{label}",
        body_text=(
            f"你的 API Key 申请单 {assignment.request_number}({label})"
            f"的旧凭证已停用{how_zh}。如你的应用仍在使用旧凭证,请立即"
            "改用新凭证:/me/api-keys\n---\n"
            f"The previous credential for your {label} API key "
            f"({assignment.request_number}) has been retired. Make sure "
            "your app uses the new one: /me/api-keys."
        ),
        template_key="apikey.rotation.retired",
        context={
            "request_number": assignment.request_number,
            "provider_name": label,
            "forced": forced,
        },
        trace_id=trace_id,
    )


async def notify_rotation_cancelled(
    db: AsyncSession, assignment: ApiKeyAssignment, *, trace_id: str,
) -> None:
    label = await _provider_label(db, assignment)
    await _dispatch_to_assignee(
        db, assignment,
        title=f"API Key 轮换已取消:{label}",
        body_text=(
            f"你的 API Key 申请单 {assignment.request_number}({label})"
            "的轮换已取消,新凭证已作废,继续使用原有凭证即可。\n---\n"
            f"The rotation for your {label} API key "
            f"({assignment.request_number}) was cancelled; the new "
            "credential was discarded. Keep using your original credential."
        ),
        template_key="apikey.rotation.cancelled",
        context={
            "request_number": assignment.request_number,
            "provider_name": label,
        },
        trace_id=trace_id,
    )


async def notify_rotation_reminder(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
    *,
    retire_at: datetime,
    trace_id: str,
) -> None:
    label = await _provider_label(db, assignment)
    await _dispatch_to_assignee(
        db, assignment,
        title=f"API Key 旧凭证即将停用:{label}",
        body_text=(
            f"你的 API Key 申请单 {assignment.request_number}({label})"
            f"的旧凭证将于 {retire_at.date().isoformat()} 停用(不足 24 小时)。"
            "如尚未切换,请尽快把应用凭证换成新的:/me/api-keys\n---\n"
            f"The previous credential for your {label} API key "
            f"({assignment.request_number}) retires on "
            f"{retire_at.date().isoformat()} (< 24h). Switch to the new "
            "credential now if you haven't: /me/api-keys."
        ),
        template_key="apikey.rotation.reminder",
        context={
            "request_number": assignment.request_number,
            "provider_name": label,
            "retire_at": retire_at.isoformat(),
            "retire_date": retire_at.date().isoformat(),
        },
        trace_id=trace_id,
    )


# ── Retire cron sweep ────────────────────────────────────────────────

async def sweep_rotation_retire(
    session: AsyncSession, *, now: datetime | None = None,
) -> dict[str, Any]:
    """Cron entrypoint (``apikey.rotation_retire``).

    Scans every assignment carrying a ``rotation_retire_at`` timestamp
    and:

    * ``retire_at <= now`` → retire the old credential, clear rotation,
      audit ``apikey.rotation.retired``, notify the owner. A failed
      cloud retire keeps the row for retry (see ``retire_rotation``),
      audits ``apikey.rotation.retire_failed`` and never notifies.
    * ``retire_at - now <= 24h`` and not yet reminded → send the
      "old key retires soon" card and stamp ``reminded_at`` (the dedup —
      no separate table, the field itself gates re-firing).
    * ``status='minting'`` older than ``MINTING_STALE_AFTER_HOURS`` →
      crashed mid-mint tombstone; cleared with a ``mint_interrupted``
      audit row so the key stops being wedged in "轮换中".

    Commits per assignment so a mid-sweep cloud failure doesn't lose the
    retires already applied. Per-assignment errors accumulate into
    ``summary['errors']`` and don't abort the sweep.

    Rollback-safety: we snapshot the candidate **ids** up front, then
    re-fetch each row fresh (``FOR UPDATE``) inside its own try. A
    per-row ``session.rollback()`` on error expires the whole identity
    map, so we must NOT keep iterating pre-loaded ORM instances after a
    rollback (accessing an expired attr triggers async lazy IO →
    ``MissingGreenlet`` and would abort the sweep — the exact transient
    DB failure the "keep sweeping" design exists to tolerate). The
    per-row lock also serialises the sweep against a concurrent
    complete/cancel so it can't retire a credential that was just
    cancelled/restored.
    """
    from app.audit.service import AuditService
    from app.core.exceptions import format_exception_for_log

    now = now or datetime.now(UTC)
    remind_before = now + _days(1)
    summary: dict[str, Any] = {
        "now": now.isoformat(),
        "scanned": 0,
        "retired": 0,
        "reminded": 0,
        # ``retire_attempted`` / ``retire_failed`` count ONLY the due-row
        # retire path (not not-due / reminder / skipped rows). The cron
        # handler's systemic-failure raise gates on these so a
        # cloud-wide retire outage is detected even when most scanned
        # rows are still in their grace window — ``errors == scanned``
        # would be diluted by not-due rows and essentially never fire.
        "retire_attempted": 0,
        "retire_failed": 0,
        "errors": [],
        "details": [],
    }
    # Snapshot ids only — the row objects are re-fetched fresh per
    # iteration so a rollback in one iteration can't poison the next.
    ids = (await session.execute(
        select(ApiKeyAssignment.id).where(
            ApiKeyAssignment.rotation_retire_at.is_not(None),
            # Revoked rows had their whole cloud identity torn down —
            # nothing to retire. Any other status (active / expired) can
            # still carry a live old credential worth deleting.
            ApiKeyAssignment.status != "revoked",
        )
    )).scalars().all()

    for aid in ids:
        summary["scanned"] += 1
        # Human-readable label captured up front so the error path never
        # has to touch a possibly-expired instance.
        label = str(aid)
        # Whether this iteration entered the actual retire (cloud-delete)
        # path — used to attribute an exception to a retire failure vs a
        # not-due / reminder / fetch error for the systemic-failure gate.
        attempting_retire = False
        try:
            # Detach every instance from the identity map first. A prior
            # iteration's ``session.rollback()`` (error path) EXPIRES the
            # whole identity map; a later sync attribute access on any such
            # instance (this row, its Integration, …) would trigger an
            # implicit lazy-load SELECT outside an await → MissingGreenlet
            # and abort the sweep. Expunging guarantees the fetch below
            # builds brand-new, fully-populated instances every iteration.
            session.expunge_all()
            # Re-fetch fresh + lock the row so complete/cancel can't
            # interleave (racing a concurrent cancel here could delete the
            # OLD credential right after cancel restored it — leaving the
            # user with zero working credentials).
            assignment = (await session.execute(
                select(ApiKeyAssignment)
                .where(ApiKeyAssignment.id == aid)
                .with_for_update()
            )).scalar_one_or_none()
            if assignment is None:
                await session.rollback()
                continue
            label = assignment.request_number
            # Re-check under the lock: status may have flipped to revoked,
            # or the rotation may have been completed/cancelled between the
            # id snapshot and now.
            if assignment.status == "revoked":
                await session.rollback()
                continue
            kind, rotation = get_active_rotation(assignment)
            if kind is None:
                # Column set but no active rotation blob (resolved by a
                # concurrent complete/cancel, or stale) — clear the
                # column so it stops being scanned.
                assignment.rotation_retire_at = None
                await session.commit()
                continue

            if rotation.get("status") == ROTATION_STATUS_MINTING:
                # A row still in ``minting`` is either an in-flight mint
                # (start_rotation released the row lock at its tombstone
                # commit while the cloud call runs — leave it alone; its
                # handles aren't written yet) or a crashed mid-mint
                # tombstone. The mint window is seconds, so anything
                # older than the stale threshold is wreckage from a
                # worker death between the tombstone commit and the
                # outcome commit. Clear it (unblocks rotate/cancel) and
                # surface loudly — the vendor may hold a credential
                # Helix never recorded.
                minted_at_raw = rotation.get("minted_at")
                try:
                    minted_at = (
                        datetime.fromisoformat(minted_at_raw)
                        if minted_at_raw else None
                    )
                except (TypeError, ValueError):
                    minted_at = None
                if minted_at is not None and minted_at.tzinfo is None:
                    minted_at = minted_at.replace(tzinfo=UTC)
                stale_cutoff = now - _hours(MINTING_STALE_AFTER_HOURS)
                if minted_at is None or minted_at <= stale_cutoff:
                    _clear_rotation_blob(assignment, kind)
                    await AuditService(session).log(
                        trace_id=f"apikey.rotation_retire.{aid}",
                        actor_type="system",
                        actor_user_id=None,
                        action="apikey.rotation.mint_interrupted",
                        resource_type="api_key_assignment",
                        resource_id=str(aid),
                        status="failed",
                        detail={
                            "request_number": label,
                            "kind": kind,
                            "minted_at": minted_at_raw,
                        },
                    )
                    await session.commit()
                    summary["details"].append({
                        "request_number": label,
                        "action": "mint_interrupted",
                    })
                    summary["errors"].append({
                        "request_number": label,
                        "error": (
                            f"stale minting tombstone cleared ({kind}); "
                            "the vendor may hold an unrecorded credential "
                            "— reconcile manually"
                        ),
                    })
                else:
                    # Fresh mid-mint row — its worker may still land.
                    await session.rollback()
                continue

            # The column is the authoritative scheduling field (set
            # atomically with the blob at start_rotation); fall back to
            # the blob's iso string only if the column is somehow unset.
            retire_at = assignment.rotation_retire_at
            if retire_at is None:
                retire_at_raw = rotation.get("retire_at")
                try:
                    retire_at = (
                        datetime.fromisoformat(retire_at_raw)
                        if retire_at_raw else None
                    )
                except (TypeError, ValueError):
                    retire_at = None
            if retire_at is not None and retire_at.tzinfo is None:
                retire_at = retire_at.replace(tzinfo=UTC)

            # Only act on a fully-minted rotation (``minting`` was
            # repaired or skipped above; anything else is defensive).
            if rotation.get("status") != ROTATION_STATUS_PENDING_RETIRE:
                await session.rollback()
                continue

            if retire_at is not None and retire_at <= now:
                attempting_retire = True
                summary["retire_attempted"] += 1
                retire_summary = await retire_rotation(
                    session, assignment, kind=kind,
                )
                retire_ok = bool(retire_summary.get("retire_ok", True))
                await AuditService(session).log(
                    trace_id=f"apikey.rotation_retire.{aid}",
                    actor_type="system",
                    actor_user_id=None,
                    # A failed cloud teardown left the OLD credential live
                    # (row kept in pending_retire for retry, unless the
                    # attempt cap cleared it) — never log it as a retired
                    # success; land a distinct, queryable failed action.
                    action=(
                        "apikey.rotation.retired"
                        if retire_ok else "apikey.rotation.retire_failed"
                    ),
                    resource_type="api_key_assignment",
                    resource_id=str(aid),
                    status="success" if retire_ok else "failed",
                    detail={
                        "request_number": label,
                        "kind": kind,
                        "retire": retire_summary,
                    },
                )
                await session.commit()
                # Retire + audit are durably committed. Clear the
                # in-flight flag so a failure of the *follow-up*
                # notification commit below can't be miscounted as a
                # retire failure (which would page a healthy retire).
                attempting_retire = False
                if retire_ok:
                    summary["retired"] += 1
                    summary["details"].append({
                        "request_number": label,
                        "action": "retired",
                    })
                    await notify_rotation_retired(
                        session, assignment,
                        trace_id=f"apikey.rotation_retire.{aid}",
                    )
                    await session.commit()
                else:
                    # Old credential may still be live; the row stays
                    # pending_retire (until the attempt cap) and the next
                    # sweep retries. Count it so the handler's raise can
                    # fire, and don't tell the owner it was retired.
                    summary["retire_failed"] += 1
                    summary["errors"].append({
                        "request_number": label,
                        "error": retire_summary.get("error"),
                        "attempts": retire_summary.get("attempts"),
                        "gave_up": bool(retire_summary.get("gave_up")),
                    })
                    summary["details"].append({
                        "request_number": label,
                        "action": "retire_failed",
                    })
            elif (
                retire_at is not None
                and retire_at <= remind_before
                and not rotation.get("reminded_at")
            ):
                # Stamp the dedup marker (``reminded_at``) and stage the
                # notification, then commit ONCE so they land atomically.
                # Committing ``reminded_at`` before the notification (the
                # old order) meant a crash/commit-failure between the two
                # permanently suppressed the "retires in <24h" warning —
                # the marker is the only dedup gate, there is no separate
                # delivery table.
                _set_provider_blob(assignment, kind, {"rotation": {
                    **rotation,
                    "reminded_at": now.isoformat(),
                }})
                await notify_rotation_reminder(
                    session, assignment, retire_at=retire_at,
                    trace_id=f"apikey.rotation_retire.{aid}",
                )
                await session.commit()
                summary["reminded"] += 1
                summary["details"].append({
                    "request_number": label,
                    "action": "reminded",
                })
            else:
                # Not due, not in the reminder window (or already
                # reminded) — release the row lock without a write.
                await session.rollback()
        except Exception as exc:  # noqa: BLE001 — keep sweeping
            await session.rollback()
            if attempting_retire:
                summary["retire_failed"] += 1
            logger.exception(
                "apikey.rotation_retire: failed for assignment %s", aid,
            )
            summary["errors"].append({
                "request_number": label,
                "error": format_exception_for_log(exc, max_len=300),
            })

    if len(summary["details"]) > 200:
        summary["details_truncated"] = True
        summary["details"] = summary["details"][:200]
    return summary


# ── small helpers ────────────────────────────────────────────────────

def _days(n: int):
    from datetime import timedelta
    return timedelta(days=n)


def _hours(n: int):
    from datetime import timedelta
    return timedelta(hours=n)


__all__ = [
    "ALLOWED_GRACE_DAYS",
    "DEFAULT_GRACE_DAYS",
    "ROTATABLE_PROVIDERS",
    "ROTATION_STATUS_PENDING_RETIRE",
    "cancel_rotation",
    "detect_rotation_provider",
    "get_active_rotation",
    "notify_rotation_cancelled",
    "notify_rotation_retired",
    "notify_rotation_started",
    "precheck_rotation",
    "retire_rotation",
    "start_rotation",
    "sweep_rotation_retire",
]