from __future__ import annotations

import asyncio
import calendar
import json
import logging
import re
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import TYPE_CHECKING, Any
from uuid import UUID, uuid4
from zoneinfo import ZoneInfo

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.approval.feishu_client import FeishuApprovalClient
from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent
from app.approval.submission_intents import (
    compact_terminal_intent_payload,
    defer_submission_intent_retry,
    reconciliation_due_predicate,
    reconciliation_order,
    submit_with_durable_intent,
)
from app.audit.service import AuditService, log_denied_action
from app.auth.models import AuthUser
from app.core.exceptions import (
    AppException,
    ForbiddenError,
    NotFoundError,
    ValidationError,
)
from app.core.rbac import can_manage_assets
from app.core.txn import best_effort
from app.domains.apikey.expiry import (
    effective_assignment_status,
    effective_status_condition,
    expiry_today,
)
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyAssignmentMember,
    ApiKeyProvider,
)
from app.domains.apikey.models import (
    ApiKeyReleaseRequest as ApiKeyReleaseRequestModel,
)
from app.domains.apikey.schemas import (
    ApiKeyApplyApprovalResponse,
    ApiKeyAssignmentResponse,
    ApiKeyExtendRequest,
    ApiKeyFillRequest,
    ApiKeyIamRebindRequest,
    ApiKeyMemberResponse,
    ApiKeyProviderCreate,
    ApiKeyProviderResponse,
    ApiKeyProviderUpdate,
    ApiKeyReleaseRequest,
    ApiKeyReleaseRequestResponse,
    ApiKeyRequestSubmit,
    ApiKeyTransferRequestSubmit,
    AssignmentUpdate,
)
from app.domains.asset.models import EmployeeRequest
from app.domains.cmdb.credential_crypto import decrypt_fields, encrypt_fields
from app.identity.person_models import IdentityPerson
from app.integrations.compat import get_feishu_config
from app.integrations.feishu_widget_values import date_interval_wire_value

if TYPE_CHECKING:
    # Annotation-only. ``app.integrations.models`` is imported lazily at
    # every call site to keep this module's import graph acyclic; without
    # this block the ``"Integration"`` annotation on
    # ``_build_aws_driver_for_tag_update`` resolved to nothing.
    from app.integrations.models import Integration

from app.core.safe_logging import get_safe_logger

logger = get_safe_logger(__name__)


def _esc(s: str) -> str:
    """Escape LIKE metacharacters to prevent wildcard injection."""
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _encrypt_key_value(plaintext: str, *, assignment_id: UUID | None = None) -> str:
    """Encrypt a key value string for DB storage.

    Pre-launch P2 fix: bind AES-GCM AAD to the owning row's id so a
    ciphertext swap from one assignment row onto another (e.g. an
    attacker with DB write moves Bob's ``key_value`` blob into Alice's
    row, then Alice reveals the key) is rejected at decrypt. We pass
    ``assignment.id`` through the ``integration_id`` parameter of
    ``encrypt_fields``; the parameter name is historical
    (originally bound to ``integrations.id``) but functionally the
    parameter is just an opaque AAD context, so the same primitive
    works for ``apikey_assignments.id`` without API churn.
    """
    if assignment_id is None:
        # Pre-launch P2 guard: encrypting without AAD allows ciphertext
        # swapping between rows (attacker copies Bob's key_value blob
        # onto Alice's row; decrypt succeeds because no row identity is
        # bound). Every caller MUST flush first to obtain the DB-assigned
        # UUID and pass it here. Log as error so the call site is visible
        # in prod logs and the caller can be fixed; we still encrypt
        # (without AAD) so the call doesn't hard-fail legacy callers, but
        # the log line makes the gap discoverable.
        logger.error(
            "_encrypt_key_value called without assignment_id — "
            "ciphertext will have no AAD binding; flush the row to "
            "obtain its UUID before encrypting",
        )
    aad_ctx = str(assignment_id) if assignment_id else None
    ct, nonce = encrypt_fields({"v": plaintext}, integration_id=aad_ctx)
    return json.dumps({"enc": ct, "nonce": nonce})


def _decrypt_key_value(stored: str, *, assignment_id: UUID | None = None) -> str:
    """Decrypt a stored key value; falls back to plaintext for legacy unencrypted values.

    The ``assignment_id`` parameter MUST match the value passed at
    encrypt time. Legacy rows encrypted before the AAD binding land in
    the no-AAD fallback inside ``decrypt_fields`` and round-trip
    correctly until they're re-saved (which will re-encrypt with the
    proper AAD).

    Round-31 audit S4 (2026-05-18): split "legacy plaintext" from
    "encrypted-blob-with-bad-AAD" so a real tamper / key-rotation
    mismatch surfaces as a hard error instead of returning the raw
    ciphertext envelope (which the masking code at ``_to_response``
    would then slice 4-chars-head-+-4-chars-tail of, leaking the
    ``{"enc":"..."`` literal into list responses). The legacy plaintext
    path stays — JSON-parse failure means the value was never
    encrypted and is the raw key — but the encrypted-blob path now
    raises ``KEY_DECRYPT_FAILED`` so the row surfaces as "unreadable"
    in the UI instead of masquerading as a usable masked key.
    """
    try:
        data = json.loads(stored)
    except Exception:
        # Not JSON → legacy plaintext row that was written before the
        # encryption migration. Return as-is; masking code slices the
        # plaintext, which is the same posture we've always had.
        return stored
    if not isinstance(data, dict):
        # JSON-but-not-a-dict (a literal number or list) — the legacy
        # plaintext path didn't produce these shapes, and the
        # encrypted envelope is always a dict. Treat as plaintext fall
        # back and log; same posture as raw-string legacy values.
        logger.warning(
            "_decrypt_key_value: stored value parsed as JSON but is "
            "not a dict (assignment_id=%s); returning raw — predates "
            "the encryption migration",
            assignment_id,
        )
        return stored
    has_enc = "enc" in data
    has_nonce = "nonce" in data
    if has_enc != has_nonce:
        # Round-3 audit LOW: partial envelope (one of enc/nonce
        # present, the other missing) cannot have been written by
        # ``_encrypt_key_value`` — fail loud so a future writer
        # introducing partial shapes can't slip past the AAD check.
        logger.error(
            "_decrypt_key_value: stored value has partial encrypted "
            "envelope (enc=%s, nonce=%s) for assignment_id=%s — "
            "refusing to expose raw ciphertext",
            has_enc, has_nonce, assignment_id,
        )
        raise AppException(
            code="KEY_DECRYPT_FAILED",
            message=(
                "Stored key has malformed encrypted envelope. "
                "Contact IT."
            ),
            status_code=500,
        )
    if not has_enc and not has_nonce:
        # Dict but neither enc nor nonce — pre-encryption legacy row
        # that happened to be JSON. Same posture as the non-dict
        # legacy fallback.
        logger.warning(
            "_decrypt_key_value: stored value parsed as JSON dict but "
            "missing both enc and nonce fields (assignment_id=%s); "
            "returning raw — predates the encryption migration",
            assignment_id,
        )
        return stored
    aad_ctx = str(assignment_id) if assignment_id else None
    try:
        fields = decrypt_fields(
            data["enc"], data["nonce"], integration_id=aad_ctx,
        )
        return fields["v"]
    except Exception as exc:
        # Encrypted envelope present but AEAD rejected the ciphertext.
        # This is either (a) ciphertext-swap (an attacker copied Bob's
        # ``key_value`` onto Alice's row, AAD = Alice's id doesn't
        # match), (b) master-key rotation without re-encrypt sweep, or
        # (c) torch-corrupted blob. None of these should silently
        # surface plaintext-looking ciphertext. Raise so the caller's
        # error path renders a hard "unreadable — contact IT" state
        # and the audit trail captures the failure.
        logger.error(
            "_decrypt_key_value: AEAD decrypt failed for encrypted "
            "row (assignment_id=%s) — possible ciphertext swap or "
            "master-key rotation without re-encrypt: %s",
            assignment_id, exc,
        )
        raise AppException(
            code="KEY_DECRYPT_FAILED",
            message=(
                "Stored key ciphertext failed authenticated decrypt — "
                "row may have been tampered with or the master key "
                "was rotated without a re-encrypt sweep. Contact IT."
            ),
            status_code=500,
        ) from exc

# Allowed statuses for assignments
STATUS_PENDING = "pending_approval"
STATUS_APPROVED = "approved"
STATUS_REJECTED = "rejected"
STATUS_ACTIVE = "active"  # key filled by admin
STATUS_EXPIRED = "expired"
STATUS_REVOKED = "revoked"
# ``pending_release`` parks a row that the employee has asked their
# leader to release. Cloud teardown does NOT fire on entry to this
# state — only on the subsequent leader-approved transition into
# STATUS_REVOKED. Reveals are blocked in this state so a window between
# self-request and approval can't leak the key. If the leader rejects
# the request, we revert to the prior state (STATUS_ACTIVE; or
# STATUS_EXPIRED if the expiry cron caught the row in the meantime).
STATUS_PENDING_RELEASE = "pending_release"

# Bounds for the "待我审批" apply-approval listing — it fans out one live
# Feishu instance lookup per pending apply, so cap the scan + concurrency
# rather than letting a large pending backlog flood Feishu on page load.
_APPLY_APPROVAL_SCAN_CAP = 80
_APPLY_APPROVAL_FEISHU_CONCURRENCY = 5

# Revocation actor role tags — written to api_key_assignments.revoke_actor_role
# at the moment of the status flip into STATUS_REVOKED. See migration
# 183 for the full enumeration; values are kept as module constants so
# code paths can't typo a value into the column.
ACTOR_ROLE_ADMIN = "admin"
ACTOR_ROLE_SELF_WITH_APPROVAL = "self_with_approval"
ACTOR_ROLE_SYSTEM_OFFBOARDING = "system_offboarding"  # reserved for PR2

# Self-service release reason codes. ``other`` requires a free-text note
# (validated at the schema layer); the rest are optional-note. ``admin_revoke``
# is the catch-all on the admin path so the admin's free-text reason still
# lives in revoke_reason_note while the column gets a stable bucket.
RELEASE_REASON_CODES = {
    "no_longer_needed",
    "project_ended",
    "wrong_provider",
    "other",
}
ADMIN_REVOKE_REASON_CODES = {
    "admin_revoke",
    "leaked",
    "compliance",
    "other",
}

# Cap on ``detail.last_transfer.retag_attempts`` so the blob can't grow
# unbounded when an integration is permanently broken and an admin keeps
# clicking retry.
_RETAG_ATTEMPTS_CAP = 20


def _trim_retag_attempts(
    attempts: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Cap the attempt list to :data:`_RETAG_ATTEMPTS_CAP` WITHOUT ever
    dropping the first successful attempt per cloud bucket.

    This list is not only forensics. ``spend.py``'s
    ``_transfer_epoch_end_with_retag`` bounds the previous holder's
    Bedrock Owner-tag epoch — and the user-facing spend caveat's
    ``until`` — at the FIRST attempt reporting ``tagged``. A plain
    newest-N trim would move that datum to a later attempt, extending
    the previous holder's window and shifting real money between two
    people's segments, with nothing in either file pointing at the
    coupling. So the first success is pinned and the cap is spent on the
    newest attempts around it: the reader keeps working unchanged and
    the datum is unlosable rather than merely unlikely to be lost.
    (Needs >20 retries on one hand-over to trigger at all.)
    """
    if len(attempts) <= _RETAG_ATTEMPTS_CAP:
        return attempts
    cutoff = len(attempts) - _RETAG_ATTEMPTS_CAP
    # Per bucket, because a bucket that succeeded LATE has its own first
    # success — pinning only the globally-earliest success would still
    # let Bedrock's first ``tagged`` fall off the front.
    pinned: dict[str, int] = {}
    for idx, attempt in enumerate(attempts[:cutoff]):
        for bucket, summary in ((attempt.get("summary") or {}).items()):
            if (
                isinstance(summary, dict)
                and summary.get("status") == "tagged"
                and bucket not in pinned
            ):
                pinned[bucket] = idx
    keep_idx = sorted(set(pinned.values()))
    if not keep_idx:
        return attempts[cutoff:]
    # Chronological order is what the "first tagged" reader relies on, so
    # the pinned rows lead and the newest attempts fill the remainder.
    tail = attempts[len(attempts) - (_RETAG_ATTEMPTS_CAP - len(keep_idx)):]
    return [attempts[i] for i in keep_idx] + tail


def _opt_money(value: Any) -> float | None:
    """可选金额 → 2 位小数 float；``None`` 原样透传。

    ``None`` 与 ``0.0`` 在这里语义不同，不能 coalesce：``None`` = 这段
    窗口没有任何一行带该口径（前端整档不渲染），``0.0`` = 确实是零。
    把前者塞成 0 会在自助页上凭空造出一个「折前 $0.00」的断言。
    """
    return None if value is None else float(round(Decimal(str(value)), 2))


class ApiKeyService:
    def __init__(self, db: AsyncSession):
        self.db = db
        # Client is lazy-constructed from the primary chat_approval
        # integration's credentials. Synchronous __init__ can't do a DB
        # read, so we defer until the first method that actually needs
        # the Feishu API. Most endpoints (list providers, list
        # assignments) never touch Feishu at all.
        self._client: FeishuApprovalClient | None = None
        self._transfer_notification_service: Any | None = None

    async def _get_client(self) -> FeishuApprovalClient:
        if self._client is None:
            cfg = await get_feishu_config(session=self.db)
            self._client = FeishuApprovalClient(
                app_id=str(cfg.get("app_id") or ""),
                app_secret=str(cfg.get("app_secret") or ""),
            )
        return self._client

    @property
    def client(self) -> FeishuApprovalClient:
        """Back-compat accessor (used by tests that set service.client)."""
        if self._client is None:
            raise RuntimeError(
                "FeishuApprovalClient not yet initialized — "
                "call await _get_client() first",
            )
        return self._client

    @client.setter
    def client(self, value: FeishuApprovalClient) -> None:
        self._client = value

    # ─── Provider CRUD (admin) ───

    async def list_providers(self, *, include_inactive: bool = False) -> list[ApiKeyProviderResponse]:
        query = select(ApiKeyProvider).order_by(ApiKeyProvider.sort_order, ApiKeyProvider.name)
        if not include_inactive:
            query = query.where(ApiKeyProvider.is_active == True)  # noqa: E712
        result = await self.db.execute(query)
        return [
            ApiKeyProviderResponse.model_validate(p, from_attributes=True)
            for p in result.scalars().all()
        ]

    async def create_provider(self, body: ApiKeyProviderCreate) -> ApiKeyProviderResponse:
        provider = ApiKeyProvider(
            name=body.name,
            description=body.description,
            is_active=body.is_active,
            sort_order=body.sort_order,
        )
        self.db.add(provider)
        await self.db.flush()
        return ApiKeyProviderResponse.model_validate(provider, from_attributes=True)

    async def update_provider(self, provider_id: UUID, body: ApiKeyProviderUpdate) -> ApiKeyProviderResponse:
        provider = await self._get_provider(provider_id)
        for field, value in body.model_dump(exclude_unset=True).items():
            setattr(provider, field, value)
        await self.db.flush()
        return ApiKeyProviderResponse.model_validate(provider, from_attributes=True)

    async def delete_provider(self, provider_id: UUID) -> None:
        provider = await self._get_provider(provider_id)
        # Block on ANY referencing assignment, not just active ones —
        # ``api_key_assignments.provider_id`` is FK-with-no-ondelete, so
        # historical rows (``revoked`` / ``rejected`` / ``expired``)
        # would also block the DELETE at the DB layer, surfacing as an
        # opaque MySQL 1451 → FastAPI 500 with no actionable message.
        # Pre-check here so the admin gets the same friendly 409 + a
        # count regardless of which status mix is keeping the provider
        # alive. The doc-string remains accurate: "Deactivate instead."
        any_count = (await self.db.execute(
            select(func.count()).select_from(ApiKeyAssignment).where(
                ApiKeyAssignment.provider_id == provider_id,
            )
        )).scalar() or 0
        if any_count > 0:
            # Break the count down by status so the admin sees whether
            # this is a "wait for X pending requests" vs "historical
            # tombstones" situation. Two-bucket split keeps the message
            # short while still surfacing both classes.
            active_count = (await self.db.execute(
                select(func.count()).select_from(ApiKeyAssignment).where(
                    and_(
                        ApiKeyAssignment.provider_id == provider_id,
                        ApiKeyAssignment.status.in_([
                            STATUS_PENDING, STATUS_APPROVED, STATUS_ACTIVE,
                        ]),
                    )
                )
            )).scalar() or 0
            historical = any_count - active_count
            raise AppException(
                code="PROVIDER_HAS_ASSIGNMENTS",
                message=(
                    f"Cannot delete provider: {any_count} assignment(s) "
                    f"reference it ({active_count} active, {historical} "
                    "historical). Deactivate it instead — "
                    "``is_active=false`` blocks new requests and keeps "
                    "the audit trail intact."
                ),
                status_code=409,
            )
        await self.db.delete(provider)
        await self.db.flush()

    # ─── Assignment: employee request submission ───

    async def submit_api_key_request(
        self,
        *,
        body: ApiKeyRequestSubmit,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        # Validate provider exists and is active
        provider = await self._get_provider(body.provider_id)
        if not provider.is_active:
            raise AppException(
                code="PROVIDER_INACTIVE",
                message=f"Provider '{provider.name}' is not available for new requests",
                status_code=400,
            )

        from app.integrations.service import get_approval_flows_map

        flows = await get_approval_flows_map(session=self.db)
        approval_code = (flows.get("api_key_request") or "").strip()
        if not approval_code:
            raise AppException(
                code="API_KEY_APPROVAL_NOT_CONFIGURED",
                message=(
                    "API Key 申请审批流程尚未配置。请到 管理后台 > 连接器 > "
                    "飞书连接器的「审批流」里新增 flow_key=api_key_request 的映射。"
                ),
                status_code=503,
            )

        # Round-27 pre-launch (2026-05-25): dedup + distributed_lock
        # against rapid double-submission, mirroring
        # ``scope_request_service.submit:454-501``. Before this guard a
        # double-click within the 30/h rate-limit window created TWO
        # Feishu approval cards + TWO ApiKeyAssignment rows + TWO
        # EmployeeRequest rows; approving both ran autogrant twice and
        # provisioned two cloud credentials. Lock-key is per
        # (requester, provider, project_name) so concurrent submissions
        # to DIFFERENT projects under the same provider still proceed.
        from app.core.distributed_lock import (  # noqa: PLC0415
            LockBusy,
            distributed_lock,
        )
        clean_project = body.project_name.strip().lower()
        lock_key = (
            f"apikey_submit:{current_user.id}:{provider.id}:{clean_project}"
        )

        async def _check_pending_dup() -> ApiKeyAssignment | None:
            return (await self.db.execute(
                select(ApiKeyAssignment).where(
                    ApiKeyAssignment.assigned_to == current_user.id,
                    ApiKeyAssignment.provider_id == provider.id,
                    func.lower(
                        func.trim(ApiKeyAssignment.project_name),
                    ) == clean_project,
                    ApiKeyAssignment.status == STATUS_PENDING,
                ).with_for_update().execution_options(populate_existing=True)
            )).scalars().first()

        try:
            async with distributed_lock(lock_key, ttl=120):
                dup = await _check_pending_dup()
                if dup is not None:
                    feishu = dict((dup.detail or {}).get("feishu") or {})
                    if not feishu.get("instance_code"):
                        await self._resume_api_key_submission(
                            dup,
                            current_user=current_user,
                            provider=provider,
                        )
                        return await self._to_response(
                            dup, current_user=current_user, provider=provider,
                        )
                    raise AppException(
                        code="API_KEY_REQUEST_DUPLICATE",
                        message=(
                            "You already have a pending API-key request "
                            "for this provider + project_name. Wait for "
                            "the approver to act, or withdraw the "
                            "previous submission before re-submitting. "
                            "Helix refuses to double-submit so two cloud "
                            "credentials don't get auto-granted from one "
                            "intent."
                        ),
                        status_code=409,
                    )
                req = await self._submit_api_key_request_inner(
                    body=body,
                    provider=provider,
                    current_user=current_user,
                    approval_code=approval_code,
                    trace_id=trace_id,
                    ip_address=ip_address,
                    user_agent=user_agent,
                )
                # Commit inside the lock so the pending request is durable
                # before release — else a racing double-submit's dup check
                # (separate txn) misses it and two cloud credentials can be
                # auto-granted from one intent.
                await self.db.commit()
                return req
        except LockBusy as exc:
            raise AppException(
                code="API_KEY_REQUEST_IN_FLIGHT",
                message=(
                    "Another submission for this provider + project is "
                    "being processed. Please retry in a few seconds."
                ),
                status_code=409,
            ) from exc

    async def _submit_api_key_request_inner(
        self,
        *,
        body: ApiKeyRequestSubmit,
        provider: ApiKeyProvider,
        current_user: AuthUser,
        approval_code: str,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        # Build Feishu form
        definition = await (await self._get_client()).get_approval_definition(approval_code)
        widgets = self._parse_template_widgets(definition.get("form", "[]"))

        # Auto-fill known fields into answers
        answers = dict(body.answers)
        self._auto_fill_answers(answers, widgets, provider, body)

        form_payload = self._serialize_feishu_form(widgets, answers)
        feishu_user_id = await (await self._get_client()).resolve_user_id_by_email(current_user.work_email)

        request_number = self._generate_request_number()
        feishu_base = {
            "template_key": "api_key_request",
            "approval_code": approval_code,
            "approval_name": definition.get("approval_name"),
            "instance_code": None,
            "instance_status": None,
            "submitted_form": json.loads(form_payload),
        }
        employee_request = EmployeeRequest(
            request_number=request_number,
            request_type="api_key_request",
            status="pending",
            requester_id=current_user.id,
            reason=body.reason,
            detail={
                "submitted_answers": body.answers,
                "feishu": feishu_base,
                "api_key": {
                    "provider_id": str(provider.id),
                    "provider_name": provider.name,
                    "project_name": body.project_name,
                    "monthly_budget_usd": str(body.monthly_budget_usd),
                    "valid_from": body.valid_from.isoformat() if body.valid_from else None,
                    "valid_until": body.valid_until.isoformat() if body.valid_until else None,
                },
            },
        )
        self.db.add(employee_request)
        await self.db.flush()
        assignment = ApiKeyAssignment(
            provider_id=provider.id,
            assigned_to=current_user.id,
            status=STATUS_PENDING,
            request_number=request_number,
            project_name=body.project_name,
            reason=body.reason,
            monthly_budget_usd=body.monthly_budget_usd,
            expected_usage=body.expected_usage,
            valid_from=body.valid_from,
            valid_until=body.valid_until,
            employee_request_id=employee_request.id,
            detail={"submitted_answers": body.answers, "feishu": feishu_base},
        )
        self.db.add(assignment)
        await self.db.flush()
        approval_request_id = uuid4()
        try:
            intent = await submit_with_durable_intent(
                self.db,
                client=await self._get_client(),
                scope="api_key_request",
                applicant_id=current_user.id,
                intent_key=str(assignment.id),
                semantic_payload={
                    "assignment_id": str(assignment.id),
                    "provider_id": str(provider.id),
                    "project_name": body.project_name.strip().lower(),
                    "reason": body.reason,
                    "monthly_budget_usd": str(body.monthly_budget_usd),
                    "expected_usage": body.expected_usage,
                    "valid_from": body.valid_from.isoformat() if body.valid_from else None,
                    "valid_until": body.valid_until.isoformat() if body.valid_until else None,
                    "answers": body.answers,
                },
                provider_payload={
                    "approval_code": approval_code,
                    "identity_type": "user_id",
                    "identity_value": feishu_user_id,
                    "form": form_payload,
                },
                finalization_payload={
                    "assignment_id": str(assignment.id),
                    "employee_request_id": str(employee_request.id),
                    "approval_request_id": str(approval_request_id),
                    "provider_id": str(provider.id),
                    "provider_name": provider.name,
                    "approval_code": approval_code,
                    "approval_name": definition.get("approval_name"),
                    "request_number": request_number,
                    "ip_address": ip_address,
                    "user_agent": user_agent,
                },
                trace_id=trace_id,
            )
            await self._finalize_api_key_submission(
                intent.id,
                current_user=current_user,
                recovered=intent.attempt_count > 1,
            )
        except AppException as exc:
            if exc.code not in {
                "APPROVAL_SUBMISSION_RESULT_UNKNOWN",
                "APPROVAL_SUBMISSION_IN_PROGRESS",
            }:
                await self._mark_api_key_submission_rejected(
                    assignment.id,
                    employee_request.id,
                )
            raise
        return await self._to_response(
            assignment, current_user=current_user, provider=provider,
        )

    async def _resume_api_key_submission(
        self,
        assignment: ApiKeyAssignment,
        *,
        current_user: AuthUser,
        provider: ApiKeyProvider,
        client: Any | None = None,
    ) -> None:
        """Resume the exact durable create generation for an assignment."""
        intent = await self.db.scalar(
            select(ApprovalSubmissionIntent).where(
                ApprovalSubmissionIntent.scope == "api_key_request",
                ApprovalSubmissionIntent.applicant_id == assignment.assigned_to,
                ApprovalSubmissionIntent.idempotency_key == str(assignment.id),
            )
        )
        if intent is None:
            raise AppException(
                code="API_KEY_SUBMISSION_INTENT_MISSING",
                message="The pending API-key request has no durable submission intent.",
                status_code=409,
            )
        stored = dict(intent.payload or {})
        final = dict(stored.get("finalization") or {})
        try:
            intent = await submit_with_durable_intent(
                self.db,
                client=client or await self._get_client(),
                scope=intent.scope,
                applicant_id=intent.applicant_id,
                intent_key=intent.idempotency_key,
                semantic_payload=dict(stored.get("semantic") or {}),
                provider_payload=dict(stored.get("provider") or {}),
                finalization_payload=final,
                trace_id=str(stored.get("trace_id") or f"api-key:{assignment.id}"),
            )
            await self._finalize_api_key_submission(
                intent.id,
                current_user=current_user,
                recovered=True,
            )
        except AppException as exc:
            if exc.code not in {
                "APPROVAL_SUBMISSION_RESULT_UNKNOWN",
                "APPROVAL_SUBMISSION_IN_PROGRESS",
            }:
                await self._mark_api_key_submission_rejected(
                    assignment.id,
                    UUID(str(final["employee_request_id"])),
                )
            raise

    async def _mark_api_key_submission_rejected(
        self,
        assignment_id: UUID,
        employee_request_id: UUID,
    ) -> None:
        """Close local pending mirrors after a definite provider refusal."""
        await self.db.rollback()
        failed_assignment = await self.db.get(ApiKeyAssignment, assignment_id)
        failed_request = await self.db.get(EmployeeRequest, employee_request_id)
        if failed_assignment is not None:
            failed_assignment.status = STATUS_REJECTED
        if failed_request is not None:
            failed_request.status = "rejected"
        await self.db.commit()

    async def _finalize_api_key_submission(
        self,
        intent_id: UUID,
        *,
        current_user: AuthUser | None,
        recovered: bool,
    ) -> None:
        """Atomically materialise mirrors, audit evidence, and completion."""
        intent = await self.db.scalar(
            select(ApprovalSubmissionIntent)
            .where(ApprovalSubmissionIntent.id == intent_id)
            .with_for_update()
            .execution_options(populate_existing=True)
        )
        if intent is None or intent.scope != "api_key_request":
            raise RuntimeError("API-key submission intent disappeared")
        if intent.state == "completed":
            return
        final = dict((intent.payload or {}).get("finalization") or {})
        assignment = await self.db.get(
            ApiKeyAssignment, UUID(str(final["assignment_id"])),
        )
        employee_request = await self.db.get(
            EmployeeRequest, UUID(str(final["employee_request_id"])),
        )
        if assignment is None or employee_request is None:
            raise RuntimeError("API-key submission intent lost its domain rows")
        if intent.state != "provider_created" or not intent.external_id:
            raise RuntimeError(f"cannot finalize API-key intent in {intent.state!r}")

        response = dict((intent.payload or {}).get("provider_response") or {})
        for row in (assignment, employee_request):
            detail = dict(row.detail or {})
            feishu = dict(detail.get("feishu") or {})
            feishu["instance_code"] = intent.external_id
            feishu["instance_status"] = response.get("instance_status", "PENDING")
            detail["feishu"] = feishu
            row.detail = detail

        approval_id = UUID(str(final["approval_request_id"]))
        approval = await self.db.get(ApprovalRequest, approval_id)
        if approval is None:
            approval = ApprovalRequest(
                id=approval_id,
                external_id=intent.external_id,
                approval_type="api_key_request",
                applicant_id=intent.applicant_id,
                status="pending",
                source_payload={
                    "api_key_assignment_id": str(assignment.id),
                    "employee_request_id": str(employee_request.id),
                    "provider_id": final["provider_id"],
                    "provider_name": final["provider_name"],
                    "approval_code": final["approval_code"],
                    "approval_name": final.get("approval_name"),
                },
            )
            self.db.add(approval)
            await self.db.flush()
        assignment.approval_request_id = approval.id

        trace_id = str(
            (intent.payload or {}).get("trace_id") or f"api-key:{assignment.id}"
        )
        await AuditService(self.db).log(
            trace_id=(f"approval-intent-reconcile:{intent.id}" if recovered else trace_id),
            actor_type="system" if recovered else "user",
            actor_user_id=None if recovered else intent.applicant_id,
            action=(
                "approval.submission_intent.reconciled"
                if recovered
                else "apikey.request.create"
            ),
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "intent_id": str(intent.id),
                "request_number": final["request_number"],
                "provider_name": final["provider_name"],
                "approval_instance_code": intent.external_id,
                "provider_request_uuid": intent.provider_request_uuid,
                "recovered": recovered,
            },
            ip_address=None if recovered else final.get("ip_address"),
            user_agent=None if recovered else final.get("user_agent"),
        )
        intent.employee_request_id = employee_request.id
        intent.approval_request_id = approval.id
        intent.state = "completed"
        intent.completed_at = datetime.now(UTC)
        intent.last_error = None
        compact_terminal_intent_payload(
            intent,
            resource_type="api_key_assignment",
            resource_id=assignment.id,
        )
        await self.db.commit()
        await self.db.refresh(assignment)

    async def reconcile_submission_intents(
        self,
        *,
        min_age_minutes: int = 2,
        batch_limit: int = 100,
        client: Any | None = None,
    ) -> dict[str, int]:
        """Recover API-key creates without requiring an HTTP retry."""
        cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes)
        ids = list(
            (
                await self.db.scalars(
                    select(ApprovalSubmissionIntent.id)
                    .where(
                        ApprovalSubmissionIntent.scope == "api_key_request",
                        ApprovalSubmissionIntent.state.in_(
                            ("prepared", "submitting", "unknown", "provider_created")
                        ),
                        (ApprovalSubmissionIntent.state == "provider_created")
                        | (ApprovalSubmissionIntent.updated_at < cutoff),
                        reconciliation_due_predicate(datetime.now(UTC)),
                    )
                    .order_by(*reconciliation_order())
                    .limit(batch_limit)
                )
            ).all()
        )
        summary = {
            "checked": len(ids),
            "reconciled": 0,
            "still_unknown": 0,
            "failed": 0,
        }
        for intent_id in ids:
            # ``_finalize_api_key_submission`` COMMITS per intent and the loop
            # keeps working afterwards — a SAVEPOINT cannot survive that (see
            # ``app.core.txn``; wrapping this very loop broke 13 idempotency
            # tests on 2026-08-06). The session is restored the other way
            # instead: every handler below rolls back before any bookkeeping —
            # ``defer_submission_intent_retry`` opens with ``db.rollback()``.
            # savepoint-exempt: commits per intent; both handlers rollback first
            try:
                intent = await self.db.get(ApprovalSubmissionIntent, intent_id)
                if intent is None:
                    continue
                assignment = await self.db.get(
                    ApiKeyAssignment, UUID(intent.idempotency_key),
                )
                if assignment is None:
                    raise RuntimeError("API-key intent has no assignment")
                stored = dict(intent.payload or {})
                intent = await submit_with_durable_intent(
                    self.db,
                    client=client or await self._get_client(),
                    scope=intent.scope,
                    applicant_id=intent.applicant_id,
                    intent_key=intent.idempotency_key,
                    semantic_payload=dict(stored.get("semantic") or {}),
                    provider_payload=dict(stored.get("provider") or {}),
                    finalization_payload=dict(stored.get("finalization") or {}),
                    trace_id=str(
                        stored.get("trace_id") or f"api-key:{assignment.id}"
                    ),
                )
                await self._finalize_api_key_submission(
                    intent.id,
                    current_user=None,
                    recovered=True,
                )
                summary["reconciled"] += 1
            except AppException as exc:
                if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN":
                    await self.db.rollback()
                else:
                    await defer_submission_intent_retry(self.db, intent_id, exc)
                summary[
                    "still_unknown"
                    if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN"
                    else "failed"
                ] += 1
            except Exception as exc:
                await defer_submission_intent_retry(self.db, intent_id, exc)
                summary["failed"] += 1
                logger.exception("Failed to reconcile API-key intent %s", intent_id)
        return summary

    # ─── Assignment: Feishu callback handling ───

    async def handle_feishu_callback(self, instance_code: str, instance_detail: dict[str, Any]) -> None:
        """Called from the shared Feishu callback handler when approval_type is api_key_request.

        Round-46 (2026-05-20): row-locks both the ApprovalRequest and the
        ApiKeyAssignment before any state inspection. Feishu retries the
        same callback aggressively on any 5xx / timeout — two concurrent
        workers used to both observe ``approval.status='pending'`` AND
        ``assignment.status != STATUS_APPROVED`` and BOTH ran the
        autogrant block, producing duplicate AWS IAM users / RAM users /
        Cognitive Services accounts plus duplicate cost-allocation rule
        rows. ``SELECT ... FOR UPDATE`` on each serialises so the second
        callback observes the post-commit ``status='approved'`` and
        short-circuits at the "already-granted" branch.
        """
        result = await self.db.execute(
            select(ApprovalRequest)
            .where(ApprovalRequest.external_id == instance_code)
            .with_for_update()
        )
        approval = result.scalar_one_or_none()
        if not approval:
            return

        if approval.status in ("approved", "rejected"):
            return

        status = self._map_feishu_status(instance_detail.get("status"))
        approval.status = status
        source = dict(approval.source_payload or {})
        source["latest_instance_detail"] = instance_detail
        approval.source_payload = source
        approval.updated_at = datetime.now(UTC)

        assignment_id = source.get("api_key_assignment_id")
        if not assignment_id:
            return

        assignment = await self._get_assignment(UUID(assignment_id), lock=True)
        previous_status = assignment.status

        # Pre-launch hardening 2026-05-06: re-validate requester is
        # still active before stamping APPROVED + firing autogrant.
        # API key approvals can sit pending for days; an offboarded
        # employee getting a fresh Bedrock/Bailian credential
        # provisioned post-departure is precisely the soft-launch
        # this gate prevents.
        if status == "approved":
            from app.core.grant_eligibility import is_blocked_from_new_grant
            from app.identity.person_models import IdentityPerson
            requester = await self.db.get(
                IdentityPerson, assignment.assigned_to,
            )
            # Blocks left / leaver_pending / soft-deleted AND the
            # hris_record_state='purged' tombstone the prior inline check
            # missed — a purged-but-active requester's stale pending
            # approval would otherwise mint a live AWS IAM / Aliyun RAM /
            # Azure OpenAI credential post-purge (2026-07-12).
            if is_blocked_from_new_grant(requester):
                assignment.status = STATUS_REJECTED
                approval.status = "rejected"
                requester_status = (
                    requester.employment_status if requester else None
                )
                from app.audit.service import AuditService
                await AuditService(self.db).log(
                    trace_id=f"feishu_callback:{instance_code}",
                    actor_type="integration",
                    actor_user_id=None,
                    action="api_key.callback_refused_leaver",
                    status="failure",
                    resource_type="api_key_assignment",
                    resource_id=str(assignment.id),
                    detail={
                        "instance_code": instance_code,
                        "feishu_status": instance_detail.get("status"),
                        "requester_status": requester_status,
                        "reason": "requester_offboarded_during_pending",
                    },
                )
                detail = dict(assignment.detail or {})
                feishu = dict(detail.get("feishu") or {})
                feishu["instance_status"] = instance_detail.get("status")
                feishu["latest_instance_detail"] = instance_detail
                feishu["refused_reason"] = "requester_offboarded_during_pending"
                detail["feishu"] = feishu
                assignment.detail = detail
                await self.db.flush()
                return
            assignment.status = STATUS_APPROVED
        elif status == "rejected":
            assignment.status = STATUS_REJECTED
        else:
            assignment.status = STATUS_PENDING

        # Auto-grant on the approved transition. Each helper is a
        # no-op for providers it doesn't recognise (matches by
        # provider name).
        #
        # Round-20 audit (A2F17): the previous shape invoked all three
        # helpers unconditionally on every approved assignment, trusting
        # that "at most one matcher fires" because provider names are
        # disjoint. They are NOT disjoint under admin renames — e.g.
        # a provider renamed to "Aliyun Bailian (deprecated — use
        # Bedrock instead)" matches both the Bedrock substring matcher
        # AND the Bailian one, double-firing cloud-resource minting
        # (one orphan AWS IAM user + one orphan Aliyun RAM user). We
        # now check matchers explicitly and refuse to auto-grant if
        # more than one claims the provider — the assignment lands
        # ``approved`` without a key and IT manually fills it. Loud
        # rather than silent double-grant.
        if status == "approved" and previous_status != STATUS_APPROVED:
            from app.domains.apikey.azure_openai_autogrant import (
                _is_azure_openai_provider,
                maybe_auto_grant_azure_openai,
            )
            from app.domains.apikey.bailian_autogrant import (
                _is_bailian_provider,
                maybe_auto_grant_bailian,
            )
            from app.domains.apikey.bedrock_autogrant import (
                _is_bedrock_provider,
                maybe_auto_grant_bedrock,
            )
            # Load provider row once (none of the matchers do DB I/O,
            # they take the row directly).
            provider = await self._get_provider(assignment.provider_id)
            matchers: list[tuple[str, Any]] = []
            if _is_bedrock_provider(provider):
                matchers.append(("bedrock", maybe_auto_grant_bedrock))
            if _is_bailian_provider(provider):
                matchers.append(("bailian", maybe_auto_grant_bailian))
            if _is_azure_openai_provider(provider):
                matchers.append(
                    ("azure_openai", maybe_auto_grant_azure_openai),
                )
            if len(matchers) > 1:
                kinds = [m[0] for m in matchers]
                logger.error(
                    "apikey auto-grant: provider %r (id=%s) matched "
                    "multiple auto-grant kinds %s for assignment %s — "
                    "refusing to auto-grant. Admin must rename the "
                    "provider to avoid the overlap, then manually fill "
                    "the key.",
                    getattr(provider, "name", None),
                    assignment.provider_id, kinds, assignment.id,
                )
                detail = dict(assignment.detail or {})
                detail["autogrant_ambiguous"] = {
                    "kinds": kinds,
                    "provider_name": getattr(provider, "name", None),
                }
                assignment.detail = detail
            elif len(matchers) == 1:
                kind, fn = matchers[0]
                try:
                    await fn(self.db, assignment)
                except Exception:  # noqa: BLE001
                    logger.exception(
                        "%s auto-grant threw unexpectedly for "
                        "assignment %s",
                        kind, assignment.id,
                    )

        detail = dict(assignment.detail or {})
        feishu = dict(detail.get("feishu") or {})
        feishu["instance_status"] = instance_detail.get("status")
        feishu["latest_instance_detail"] = instance_detail
        detail["feishu"] = feishu
        assignment.detail = detail

        # Mirror the status + feishu sub-detail onto the unified
        # employee_requests row so "我的申请" reflects the callback result
        # without any extra query.
        employee_request_id = (
            source.get("employee_request_id") or (str(assignment.employee_request_id) if assignment.employee_request_id else None)
        )
        if employee_request_id:
            er_result = await self.db.execute(
                select(EmployeeRequest).where(EmployeeRequest.id == UUID(employee_request_id))
            )
            employee_request = er_result.scalar_one_or_none()
            if employee_request is not None:
                employee_request.status = status
                er_detail = dict(employee_request.detail or {})
                er_feishu = dict(er_detail.get("feishu") or {})
                er_feishu["instance_status"] = instance_detail.get("status")
                er_feishu["latest_instance_detail"] = instance_detail
                er_detail["feishu"] = er_feishu
                employee_request.detail = er_detail
                if status in ("approved", "rejected"):
                    employee_request.processed_at = datetime.now(UTC)

        await self.db.flush()

    # ─── Self-service release request (Helix-internal approval) ───
    #
    # An employee initiates "I want to release this API key" → we
    # park the :class:`ApiKeyAssignment` in ``pending_release`` and
    # insert a new :class:`ApiKeyReleaseRequestModel` row in
    # ``status='pending'``. The assignee's direct manager
    # (``IdentityPerson.manager_person_id``) receives a Feishu PUSH
    # notification linking to the Helix Web UI approval page. The
    # manager — OR any IT Admin (escape valve when the manager link
    # is broken / the manager is unavailable) — clicks Approve /
    # Reject *inside Helix*; no Feishu approval template, no
    # callback round-trip. On approve we call :meth:`revoke_key`
    # with ``actor_role=self_with_approval`` so the cloud-side
    # teardown runs through the same path as an admin revoke. On
    # reject we restore the assignment to its prior status
    # (captured in :class:`ApiKeyReleaseRequestModel.previous_status`)
    # and notify the requester.

    async def submit_release_request(
        self,
        assignment_id: UUID,
        body: ApiKeyReleaseRequest,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Employee asks their manager to revoke an API key.

        Why not just let the employee revoke directly: a key may be
        wired into a long-running production workflow that the
        employee themselves no longer thinks about (a colleague took
        over, a cron runs in some cluster). Cloud revoke kills the
        credential immediately and irrecoverably (see
        :meth:`_revoke_cloud_resources` for what AWS / Aliyun /
        Azure each delete). A leader gate gives a human a chance to
        ask "are you SURE this isn't still running in prod?" before
        the AK/SK pair vanishes.

        State transition: ``active`` / ``expired`` → ``pending_release``.
        Reveal is blocked while in ``pending_release`` (see
        :meth:`reveal_key`) so an employee can't grab the plaintext
        during the window between "I want to release" and "manager
        approved". ``pending_release`` does NOT call the cloud —
        teardown only fires on the approve path.

        If the manager rejects (or the employee cancels), the row
        reverts to ``ApiKeyReleaseRequestModel.previous_status``.
        """
        # Authz: only the assignee may submit. Admins use the direct
        # ``/revoke`` endpoint — they don't need a leader gate because
        # an admin pulling a key IS the compliance event.
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment owner may submit a release request.",
            )

        # State machine: release is only meaningful for live keys.
        # ``approved`` (admin hasn't filled the key yet) → release
        # would terminate a request that never granted a credential;
        # use the cancel-request endpoint for that. ``pending_release``
        # → idempotent re-call: if a pending request row already
        # exists, return it; otherwise (a stale flip from a prior
        # release path that never landed a request row) create one
        # to heal the orphan. ``pending_approval`` / ``rejected`` /
        # ``revoked`` are obvious no-ops.
        if assignment.status == STATUS_PENDING_RELEASE:
            existing = await self._find_open_release_request(assignment.id)
            if existing is not None:
                return await self._to_response(assignment)
            # Stale pending_release with no open request row — fall
            # through to create one. We treat the assignment's
            # ``detail.release.previous_status`` as the rollback
            # target if present, else default to STATUS_ACTIVE.
        if assignment.status not in (
            STATUS_ACTIVE, STATUS_EXPIRED, STATUS_PENDING_RELEASE,
        ):
            raise AppException(
                code="INVALID_STATUS_FOR_RELEASE",
                message=(
                    f"Cannot request release for assignment in status "
                    f"'{assignment.status}'. Release is only available "
                    "for keys in 'active' or 'expired' status."
                ),
                status_code=400,
            )

        # Cross-flow mutual exclusion (Round-45 hardening): a pending
        # AssetTransferRequest represents an active intent to hand the
        # key to a colleague who still wants it; a release would
        # invalidate that colleague's incoming credential. Without
        # this gate, a double-action lands a contradictory state:
        # manager approves release → row goes ``revoked`` → recipient
        # accepts the stale transfer → ``_apply_transfer``'s api_key
        # branch tries to flip ``assigned_to`` on a revoked row.
        from app.domains.offboarding.models import (
            XFER_PENDING,
            AssetTransferRequest,
        )

        pending_xfer = (
            await self.db.execute(
                select(AssetTransferRequest).where(
                    and_(
                        AssetTransferRequest.asset_type == "api_key",
                        AssetTransferRequest.asset_ref_id == assignment.id,
                        AssetTransferRequest.status == XFER_PENDING,
                    ),
                ),
            )
        ).scalar_one_or_none()
        if pending_xfer is not None:
            raise AppException(
                code="TRANSFER_PENDING",
                message=(
                    "Cannot request release: a transfer request for "
                    "this API Key is already pending. Cancel the "
                    "transfer (or wait for the recipient to respond) "
                    "before requesting release."
                ),
                status_code=409,
            )

        # Resolve previous_status. On a clean ``active`` / ``expired``
        # row this is the current status. On a stale pending_release
        # row (rare) we consult the legacy ``detail.release.previous_status``
        # stash; if missing, default to STATUS_ACTIVE — the worst case
        # is "key reverts to active even though it was expired", which
        # the expiry cron heals on the next tick.
        if assignment.status == STATUS_PENDING_RELEASE:
            prev_meta = dict((assignment.detail or {}).get("release") or {})
            previous_status = prev_meta.get(
                "previous_status",
            ) or STATUS_ACTIVE
        else:
            previous_status = assignment.status

        # Insert the release-request row + flip the assignment status
        # in a single flush — both writes are part of the outer HTTP
        # transaction so a downstream notification failure does NOT
        # roll the row back (we fire notifications best-effort below).
        release_request = ApiKeyReleaseRequestModel(
            assignment_id=assignment.id,
            requester_id=current_user.id,
            reason_code=body.reason_code,
            reason_note=body.reason_note,
            status="pending",
            previous_status=previous_status,
            requested_at=datetime.now(UTC),
        )
        self.db.add(release_request)
        await self.db.flush()

        # Flip the assignment + stash a small forensic blob on the
        # row's detail JSON. Keeping ``detail.release`` populated
        # parallel to the dedicated table lets the existing admin
        # UI (which reads the JSON) keep working without a schema
        # cut-over; the table is the source of truth for the new
        # approver-side flow.
        assignment.status = STATUS_PENDING_RELEASE
        detail = dict(assignment.detail or {})
        detail["release"] = {
            "release_request_id": str(release_request.id),
            "previous_status": previous_status,
            "reason_code": body.reason_code,
            "reason_note": body.reason_note,
            "requested_at": release_request.requested_at.isoformat(),
        }
        assignment.detail = detail
        await self.db.flush()

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.release.request",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "release_request_id": str(release_request.id),
                "reason_code": body.reason_code,
                "previous_status": previous_status,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        # Push the approval notification — best-effort, never blocks
        # the submit. If the notification provider is down we still
        # want the request row + assignment flip persisted so the
        # manager can find the pending item on the Helix Web UI.
        try:
            await self._notify_release_pending(
                release_request, assignment, trace_id=trace_id,
            )
        except Exception as notif_err:  # noqa: BLE001
            logger.warning(
                "release-pending notify failed for %s: %s",
                release_request.id, notif_err,
            )
        return await self._to_response(assignment)

    async def cancel_release_request(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Employee withdraws their own pending release request.

        Reverts the assignment to its pre-release status (from the
        request row's ``previous_status``) and marks the request row
        ``cancelled``. Idempotent on already-decided requests (the
        assignment is no longer in ``pending_release``) so a
        double-click on the cancel button is safe.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment owner may cancel a release request.",
            )
        if assignment.status != STATUS_PENDING_RELEASE:
            # No pending release to cancel. Return current state —
            # the client UI should refresh and hide the cancel button.
            return await self._to_response(assignment)

        request = await self._find_open_release_request(assignment.id)
        # No open request row but the assignment is in pending_release
        # (rare — stale state from a partial migration / aborted
        # write). Still revert the assignment so the user is not
        # locked into a useless status. Skip the audit row's
        # release_request_id field in that case.
        previous_status = STATUS_ACTIVE
        if request is not None:
            previous_status = request.previous_status or STATUS_ACTIVE
            request.status = "cancelled"
            request.decision_note = None
            request.responded_at = datetime.now(UTC)
        else:
            stash = dict((assignment.detail or {}).get("release") or {})
            previous_status = stash.get(
                "previous_status",
            ) or STATUS_ACTIVE

        assignment.status = previous_status
        detail = dict(assignment.detail or {})
        release_meta = dict(detail.get("release") or {})
        release_meta["cancelled_at"] = datetime.now(UTC).isoformat()
        detail["release"] = release_meta
        assignment.detail = detail

        await self.db.flush()

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.release.cancel",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "release_request_id": (
                    str(request.id) if request is not None else None
                ),
                "reverted_to": previous_status,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    async def approve_release_request(
        self,
        request_id: UUID,
        *,
        current_user: AuthUser,
        decision_note: str | None,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyReleaseRequestResponse:
        """Manager / IT Admin approves a pending release request.

        Authz: caller MUST be either (a) the requester's direct
        manager (``IdentityPerson.manager_person_id``) OR (b) hold
        ``GLOBAL_ADMIN_ROLE``. The IT Admin escape valve covers the
        case where the manager link is broken (e.g. HRIS sync drift,
        manager themselves on leave) — without it a stuck release
        request has no resolution path short of admin DB surgery.

        On approve we call :meth:`revoke_key` with
        ``actor_role=ACTOR_ROLE_SELF_WITH_APPROVAL`` so the
        cloud-side teardown runs through the same path as an admin
        revoke. The ``current_user`` argument to ``revoke_key`` is
        the *assignee* (not the manager) because the
        ``actor_role=self_with_approval`` guard inside ``revoke_key``
        expects the caller identity to match the assignee.

        Leaver-status check: if the assignee departed between
        submitting and the manager approving, we DO NOT execute the
        revoke — the offboarding pipeline owns that row. We treat
        the approve as auto-rejected ("assignee_offboarded_during_pending")
        so reports cleanly separate "manager said no" from "manager
        couldn't say yes because the person left".
        """
        from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role

        request = await self._lock_release_request(request_id)

        if request.status != "pending":
            # Already decided — idempotent: return the existing
            # state. A re-click on a stale UI doesn't fire revoke
            # twice; the second click is a no-op.
            return await self._hydrate_release_request(request)

        requester = await self.db.get(IdentityPerson, request.requester_id)
        if requester is None:
            # Pathological: requester hard-deleted while request was
            # pending. Return 410 so the UI can prune the stale card.
            raise AppException(
                code="REQUESTER_NOT_FOUND",
                message=(
                    "The original requester no longer exists. This "
                    "release request cannot be approved."
                ),
                status_code=410,
            )

        is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE)
        is_manager = (
            requester.manager_person_id is not None
            and requester.manager_person_id == current_user.id
        )
        if not (is_admin or is_manager):
            raise ForbiddenError(
                "Only the requester's direct manager or an IT Admin "
                "may approve this release.",
            )

        assignment = await self._get_assignment(
            request.assignment_id, lock=True,
        )

        # Round-46 (2026-05-20): if ownership changed between submit
        # and approve (e.g. via an offboarding-driven transfer that
        # completed between callback receipt and approver click), the
        # requester is no longer the assignee. ``revoke_key`` would
        # then 403 with FORBIDDEN inside the self_with_approval branch
        # at line ~2767, the entire HTTP transaction would roll back
        # (including the audit row at 1112 and the request.status
        # flip), and the row would stay parked in pending_release
        # forever. Detect this BEFORE we flip the request row so we
        # can rollback the release to its previous_status with a
        # dedicated error code instead of leaving it stranded.
        if assignment.assigned_to != request.requester_id:
            request.status = "rejected"
            request.approver_id = current_user.id
            request.decision_note = (
                "Auto-rejected: assignment was transferred to a new "
                "owner during the pending window; the new owner must "
                "submit their own release request."
            )
            request.responded_at = datetime.now(UTC)
            previous_status = request.previous_status or STATUS_ACTIVE
            if assignment.status == STATUS_PENDING_RELEASE:
                assignment.status = previous_status
            detail = dict(assignment.detail or {})
            release_meta = dict(detail.get("release") or {})
            release_meta["rejected_at"] = request.responded_at.isoformat()
            release_meta["rejected_reason"] = "transfer_intervened"
            detail["release"] = release_meta
            assignment.detail = detail
            await self.db.flush()
            await AuditService(self.db).log(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.release.refused_transfer_intervened",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="failure",
                detail={
                    "request_number": assignment.request_number,
                    "release_request_id": str(request.id),
                    "reverted_to": previous_status,
                    "original_requester_id": str(request.requester_id),
                    "current_assignee_id": str(assignment.assigned_to),
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            await self._sync_attestation_items(assignment.id, outcome="rejected")
            return await self._hydrate_release_request(request)

        # Leaver-status check — see docstring. Mirror the action /
        # status / detail shape from the old Feishu callback's
        # ``apikey.release.refused_leaver`` audit row so forensics
        # tooling can scan both flows with one predicate.
        leaver_status = (requester.employment_status or "").lower()
        if (
            requester.deleted_at is not None
            or leaver_status in ("left", "leaver_pending")
        ):
            request.status = "rejected"
            request.approver_id = current_user.id
            request.decision_note = (
                "Auto-rejected: requester departed during pending "
                "window — offboarding pipeline owns this row."
            )
            request.responded_at = datetime.now(UTC)

            previous_status = request.previous_status or STATUS_ACTIVE
            if assignment.status == STATUS_PENDING_RELEASE:
                assignment.status = previous_status
            detail = dict(assignment.detail or {})
            release_meta = dict(detail.get("release") or {})
            release_meta["rejected_at"] = request.responded_at.isoformat()
            release_meta["rejected_reason"] = (
                "assignee_offboarded_during_pending"
            )
            detail["release"] = release_meta
            assignment.detail = detail
            await self.db.flush()

            await AuditService(self.db).log(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.release.refused_leaver",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="failure",
                detail={
                    "request_number": assignment.request_number,
                    "release_request_id": str(request.id),
                    "reverted_to": previous_status,
                    "assignee_employment_status": leaver_status,
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            await self._sync_attestation_items(assignment.id, outcome="rejected")
            return await self._hydrate_release_request(request)

        # Happy path: flip the request row first (so a downstream
        # revoke_key failure doesn't leave the request in pending
        # forever), then call revoke_key with the assignee as
        # current_user.
        request.status = "approved"
        request.approver_id = current_user.id
        request.decision_note = decision_note
        request.responded_at = datetime.now(UTC)
        await self.db.flush()

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.release.approve",
            resource_type="api_key_release_request",
            resource_id=str(request.id),
            status="success",
            detail={
                "assignment_id": str(assignment.id),
                "request_number": assignment.request_number,
                "approver_role": (
                    "manager" if is_manager and not is_admin
                    else ("admin" if is_admin and not is_manager else "manager+admin")
                ),
                "reason_code": request.reason_code,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        # Drive the cloud-side teardown. The revoke_key call is
        # synchronous — if it raises, the outer HTTP transaction
        # rolls back the request.status='approved' flip we just
        # did. We deliberately propagate the exception so the
        # approver sees the failure (the alternative — swallow and
        # show "approved" — would silently leave the cloud-side
        # credential live).
        await self.revoke_key(
            assignment.id,
            reason=(
                request.reason_note
                or f"self_with_approval:{request.reason_code}"
            ),
            current_user=requester,
            trace_id=trace_id,
            ip_address=ip_address,
            user_agent=user_agent,
            reason_code=request.reason_code,
            actor_role=ACTOR_ROLE_SELF_WITH_APPROVAL,
        )
        await self._sync_attestation_items(assignment.id, outcome="revoked")

        # Notify the requester their release was approved + the key
        # is now revoked. Best-effort.
        try:
            await self._notify_release_decided(
                request, assignment, approved=True,
                trace_id=trace_id,
            )
        except Exception as notif_err:  # noqa: BLE001
            logger.warning(
                "release-approved notify failed for %s: %s",
                request.id, notif_err,
            )
        return await self._hydrate_release_request(request)

    async def reject_release_request(
        self,
        request_id: UUID,
        *,
        current_user: AuthUser,
        decision_note: str | None,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyReleaseRequestResponse:
        """Manager / IT Admin rejects a pending release request.

        Reverts the assignment to its pre-release status and
        notifies the requester. Same authz as :meth:`approve_release_request`
        (requester's direct manager OR IT Admin).
        """
        from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role

        request = await self._lock_release_request(request_id)
        if request.status != "pending":
            # Idempotent: return the already-decided state.
            return await self._hydrate_release_request(request)

        requester = await self.db.get(IdentityPerson, request.requester_id)
        if requester is None:
            raise AppException(
                code="REQUESTER_NOT_FOUND",
                message=(
                    "The original requester no longer exists. This "
                    "release request cannot be rejected."
                ),
                status_code=410,
            )

        is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE)
        is_manager = (
            requester.manager_person_id is not None
            and requester.manager_person_id == current_user.id
        )
        if not (is_admin or is_manager):
            raise ForbiddenError(
                "Only the requester's direct manager or an IT Admin "
                "may reject this release.",
            )

        assignment = await self._get_assignment(
            request.assignment_id, lock=True,
        )

        request.status = "rejected"
        request.approver_id = current_user.id
        request.decision_note = decision_note
        request.responded_at = datetime.now(UTC)

        previous_status = request.previous_status or STATUS_ACTIVE
        if assignment.status == STATUS_PENDING_RELEASE:
            assignment.status = previous_status
        detail = dict(assignment.detail or {})
        release_meta = dict(detail.get("release") or {})
        release_meta["rejected_at"] = request.responded_at.isoformat()
        if decision_note:
            release_meta["rejected_note"] = decision_note
        detail["release"] = release_meta
        assignment.detail = detail
        await self.db.flush()

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.release.reject",
            resource_type="api_key_release_request",
            resource_id=str(request.id),
            status="success",
            detail={
                "assignment_id": str(assignment.id),
                "request_number": assignment.request_number,
                "approver_role": (
                    "manager" if is_manager and not is_admin
                    else ("admin" if is_admin and not is_manager else "manager+admin")
                ),
                "reverted_to": previous_status,
                "decision_note": decision_note,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        await self._sync_attestation_items(assignment.id, outcome="rejected")

        try:
            await self._notify_release_decided(
                request, assignment, approved=False,
                trace_id=trace_id,
            )
        except Exception as notif_err:  # noqa: BLE001
            logger.warning(
                "release-rejected notify failed for %s: %s",
                request.id, notif_err,
            )
        return await self._hydrate_release_request(request)

    async def get_release_request(
        self,
        request_id: UUID,
        *,
        current_user: AuthUser,
    ) -> ApiKeyReleaseRequestResponse:
        """Fetch one release request. Visible to: requester,
        requester's manager, IT Admin, or management (read-only).
        """
        from app.core.rbac import (
            GLOBAL_ADMIN_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )

        request = await self.db.get(ApiKeyReleaseRequestModel, request_id)
        if request is None:
            raise NotFoundError(
                f"Release request {request_id} not found.",
            )
        if has_role(
            current_user.role_names, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE,
        ):
            return await self._hydrate_release_request(request)
        if request.requester_id == current_user.id:
            return await self._hydrate_release_request(request)
        requester = await self.db.get(IdentityPerson, request.requester_id)
        if (
            requester is not None
            and requester.manager_person_id == current_user.id
        ):
            return await self._hydrate_release_request(request)
        raise ForbiddenError(
            "You do not have permission to view this release request.",
        )

    async def list_release_requests_for_approver(
        self,
        *,
        current_user: AuthUser,
        status: str | None = "pending",
    ) -> list[ApiKeyReleaseRequestResponse]:
        """List release requests where ``current_user`` is the
        designated approver — i.e. the requester's direct manager,
        OR (when caller is IT Admin / management) every pending
        request.

        Management is the read-only leadership view; they see the
        full pending queue for governance even though approval
        authority itself stays with IT Admin and the direct manager.

        We compute the set at query time rather than materialising
        ``approver_id`` at submit because (a) the manager link can
        change between submit and the manager actually checking the
        UI (HRIS sync drift, re-org), and (b) the IT Admin escape
        valve must always see the full set without a separate
        index. The (requester_id, status) composite index handles
        the manager-view scan efficiently; the (approver_id, status)
        index covers a future "I personally decided this one"
        history query.
        """
        from app.core.rbac import (
            GLOBAL_ADMIN_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )

        is_admin = has_role(
            current_user.role_names, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE,
        )
        stmt = select(ApiKeyReleaseRequestModel)
        if status:
            stmt = stmt.where(ApiKeyReleaseRequestModel.status == status)
        stmt = stmt.order_by(
            ApiKeyReleaseRequestModel.requested_at.desc(),
        )

        if is_admin:
            rows = (await self.db.execute(stmt)).scalars().all()
        else:
            # Restrict to requests whose requester reports to
            # current_user. We join through IdentityPerson rather
            # than a flat WHERE on requester_id IN (...) because the
            # subordinate set can be large and the SQL planner
            # handles the join more cleanly.
            subq = (
                select(IdentityPerson.id)
                .where(IdentityPerson.manager_person_id == current_user.id)
            )
            stmt = stmt.where(
                ApiKeyReleaseRequestModel.requester_id.in_(subq),
            )
            rows = (await self.db.execute(stmt)).scalars().all()

        return [
            await self._hydrate_release_request(r) for r in rows
        ]

    async def list_release_requests_for_requester(
        self,
        *,
        current_user: AuthUser,
        status: str | None = None,
    ) -> list[ApiKeyReleaseRequestResponse]:
        """List release requests submitted by ``current_user``.

        Used by the requester-side "我的释放申请" view so they can
        see status / cancel still-pending ones.
        """
        stmt = select(ApiKeyReleaseRequestModel).where(
            ApiKeyReleaseRequestModel.requester_id == current_user.id,
        )
        if status:
            stmt = stmt.where(ApiKeyReleaseRequestModel.status == status)
        stmt = stmt.order_by(
            ApiKeyReleaseRequestModel.requested_at.desc(),
        )
        rows = (await self.db.execute(stmt)).scalars().all()
        return [
            await self._hydrate_release_request(r) for r in rows
        ]

    async def list_pending_apply_approvals_for_approver(
        self,
        *,
        current_user: AuthUser,
    ) -> list[ApiKeyApplyApprovalResponse]:
        """Pending AI API-Key *apply* (申请) requests the caller is the
        CURRENT Feishu approver of — i.e. "待我审批".

        The apply approval lives entirely in Feishu (approval_type=
        ``api_key_request``), so "who must act now" is the live Feishu
        pending-approver set, not a manager link or a stored approver_id.
        We resolve it the same way 催办 does — fetch each pending
        instance's ``task_list`` and keep only the rows where the caller's
        work-email is one of the currently-pending approvers. This keeps
        the list precise ("待我审批", not "everyone's queue") and mirrors
        the actual Feishu routing instead of guessing.

        Best-effort + bounded: a Feishu hiccup on one instance drops just
        that row (it stays in飞书审批 regardless); the scan is capped so a
        large pending backlog can't fan out unboundedly on page load.
        """
        if not current_user.work_email:
            return []
        caller_email = current_user.work_email.strip().lower()

        rows = (
            await self.db.execute(
                select(ApiKeyAssignment)
                .where(ApiKeyAssignment.status == STATUS_PENDING)
                .order_by(ApiKeyAssignment.created_at.desc())
                .limit(_APPLY_APPROVAL_SCAN_CAP)
            )
        ).scalars().all()
        if not rows:
            return []

        client = await self._get_client()
        sem = asyncio.Semaphore(_APPLY_APPROVAL_FEISHU_CONCURRENCY)

        async def _is_pending_for_caller(instance_code: str) -> bool:
            async with sem:
                try:
                    emails = await client.list_pending_approver_emails(
                        instance_code,
                    )
                except Exception:  # noqa: BLE001
                    return False
            return caller_email in emails

        # Resolve the live Feishu instance for each pending apply. The
        # instance_code is stamped on the assignment at submit time under
        # detail.feishu.instance_code (same field the requester drawer +
        # 催办 read).
        candidates: list[tuple[ApiKeyAssignment, str]] = []
        for row in rows:
            detail = row.detail if isinstance(row.detail, dict) else {}
            instance_code = (detail.get("feishu") or {}).get("instance_code")
            if isinstance(instance_code, str) and instance_code.strip():
                candidates.append((row, instance_code.strip()))
        if not candidates:
            return []

        flags = await asyncio.gather(
            *(_is_pending_for_caller(code) for _, code in candidates),
        )
        mine = [
            (row, code)
            for (row, code), pending in zip(candidates, flags, strict=True)
            if pending
        ]
        if not mine:
            return []

        # Hydrate display fields (requester + provider) for the survivors.
        out: list[ApiKeyApplyApprovalResponse] = []
        for row, instance_code in mine:
            requester = await self.db.get(IdentityPerson, row.assigned_to)
            provider = await self.db.get(ApiKeyProvider, row.provider_id)
            out.append(
                ApiKeyApplyApprovalResponse(
                    id=row.id,
                    request_number=row.request_number,
                    requester_id=row.assigned_to,
                    requester_display_name=(
                        requester.display_name if requester else None
                    ),
                    requester_email=requester.work_email if requester else None,
                    requester_employee_no=(
                        requester.employee_no if requester else None
                    ),
                    requester_employment_status=(
                        requester.employment_status if requester else None
                    ),
                    provider_name=provider.name if provider else None,
                    project_name=row.project_name,
                    reason=row.reason,
                    monthly_budget_usd=row.monthly_budget_usd,
                    expected_usage=row.expected_usage,
                    submitted_at=row.created_at,
                    feishu_instance_code=instance_code,
                    employee_request_id=row.employee_request_id,
                )
            )
        return out

    async def _find_open_release_request(
        self, assignment_id: UUID,
    ) -> ApiKeyReleaseRequestModel | None:
        """Return the pending release request for an assignment, or
        None. There is at most one — enforced by application logic
        (submit refuses to insert a second one) rather than a
        partial UNIQUE constraint (which TiDB does not support).
        """
        return (
            await self.db.execute(
                select(ApiKeyReleaseRequestModel).where(
                    and_(
                        ApiKeyReleaseRequestModel.assignment_id == assignment_id,
                        ApiKeyReleaseRequestModel.status == "pending",
                    ),
                ),
            )
        ).scalar_one_or_none()

    async def _lock_release_request(
        self, request_id: UUID,
    ) -> ApiKeyReleaseRequestModel:
        """Load + row-lock a release request for decision. Raises
        :class:`NotFoundError` if missing.
        """
        request = (
            await self.db.execute(
                select(ApiKeyReleaseRequestModel)
                .where(ApiKeyReleaseRequestModel.id == request_id)
                .with_for_update(),
            )
        ).scalar_one_or_none()
        if request is None:
            raise NotFoundError(
                f"Release request {request_id} not found.",
            )
        return request

    async def _hydrate_release_request(
        self, request: ApiKeyReleaseRequestModel,
    ) -> ApiKeyReleaseRequestResponse:
        """Build the response DTO, denormalising a few columns from
        the related assignment / requester / approver / provider so
        the approver UI can render without a second round-trip.
        """
        requester = await self.db.get(IdentityPerson, request.requester_id)
        approver = (
            await self.db.get(IdentityPerson, request.approver_id)
            if request.approver_id is not None else None
        )
        assignment = await self.db.get(ApiKeyAssignment, request.assignment_id)
        provider = None
        if assignment is not None:
            provider = await self.db.get(
                ApiKeyProvider, assignment.provider_id,
            )
        return ApiKeyReleaseRequestResponse(
            id=request.id,
            assignment_id=request.assignment_id,
            requester_id=request.requester_id,
            approver_id=request.approver_id,
            reason_code=request.reason_code,
            reason_note=request.reason_note,
            status=request.status,
            decision_note=request.decision_note,
            previous_status=request.previous_status,
            requested_at=request.requested_at,
            responded_at=request.responded_at,
            created_at=request.created_at,
            updated_at=request.updated_at,
            requester_display_name=(
                requester.display_name if requester else None
            ),
            requester_email=(
                requester.work_email if requester else None
            ),
            requester_employee_no=(
                requester.employee_no if requester else None
            ),
            requester_employment_status=(
                requester.employment_status if requester else None
            ),
            approver_display_name=(
                approver.display_name if approver else None
            ),
            provider_name=(provider.name if provider else None),
            project_name=(
                assignment.project_name if assignment else None
            ),
            request_number=(
                assignment.request_number if assignment else None
            ),
        )

    async def _notify_release_pending(
        self,
        request: ApiKeyReleaseRequestModel,
        assignment: ApiKeyAssignment,
        *,
        trace_id: str,
    ) -> None:
        """Push a Feishu / email card to the assignee's manager
        asking them to approve / reject the release inside Helix.

        Fallback recipient: if the manager link is broken (the
        assignee has ``manager_person_id IS NULL`` — a real
        condition for C-level / contractor accounts), notify every
        active IT Admin so the request is not stranded.
        """
        from app.auth.models import AuthRole, AuthUserRole
        from app.core.rbac import GLOBAL_ADMIN_ROLE
        from app.integrations.service import get_platform_setting
        from app.notifications.service import (
            NotificationEnvelope,
            NotificationService,
        )

        requester = await self.db.get(IdentityPerson, request.requester_id)
        if requester is None:
            return
        provider = await self.db.get(ApiKeyProvider, assignment.provider_id)
        provider_name = provider.name if provider else "(unknown provider)"
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="(未配置门户地址)",
            session=self.db,
        )
        approval_url = f"{portal_url}/my/todos?tab=apikey-release"

        recipients: list[IdentityPerson] = []
        if requester.manager_person_id is not None:
            manager = await self.db.get(
                IdentityPerson, requester.manager_person_id,
            )
            if (
                manager is not None
                and manager.deleted_at is None
                and (manager.employment_status or "active").lower() == "active"
            ):
                recipients.append(manager)
        if not recipients:
            admins = (
                await self.db.execute(
                    select(IdentityPerson)
                    .join(
                        AuthUserRole,
                        AuthUserRole.user_id == IdentityPerson.id,
                    )
                    .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
                    .where(
                        AuthRole.name == GLOBAL_ADMIN_ROLE,
                        IdentityPerson.deleted_at.is_(None),
                    )
                )
            ).scalars().all()
            recipients.extend(admins)

        if not recipients:
            logger.warning(
                "release-pending notify: no manager or IT admin found "
                "for assignment %s — request %s stranded",
                assignment.id, request.id,
            )
            return

        notifier = NotificationService(self.db)
        # Dedup recipients by id in case manager IS also an admin.
        seen: set[UUID] = set()
        for recipient in recipients:
            if recipient.id in seen:
                continue
            seen.add(recipient.id)
            await notifier.dispatch(
                NotificationEnvelope(
                    recipient=recipient,
                    title=(
                        f"API Key 释放申请待审批:{requester.display_name or requester.work_email}"
                    ),
                    body_text=(
                        f"{requester.display_name or requester.work_email} "
                        f"提交了 API Key 释放申请({provider_name},"
                        f"项目:{assignment.project_name})。\n"
                        f"理由:{request.reason_code}"
                        + (
                            f" — {request.reason_note}"
                            if request.reason_note else ""
                        )
                        + f"\n请到 Helix 审批:{approval_url}"
                    ),
                    channels=["feishu", "email"],
                    template_key="apikey.release.pending",
                    url=approval_url,
                    related_resource_type="api_key_release_request",
                    related_resource_id=request.id,
                    context={
                        "requester_name": (
                            requester.display_name or requester.work_email
                        ),
                        "provider_name": provider_name,
                        "project_name": assignment.project_name,
                        "request_number": assignment.request_number,
                        "reason_code": request.reason_code,
                        "reason_note": request.reason_note or "",
                        # Pre-joined reason line: the template is a static
                        # string with no conditional, so the " — note" tail
                        # has to be composed here or an empty note would
                        # leave a dangling dash in every language.
                        "reason_text": (
                            f"{request.reason_code} — {request.reason_note}"
                            if request.reason_note
                            else request.reason_code
                        ),
                        "approval_url": approval_url,
                    },
                ),
                trace_id=trace_id,
            )

    async def _notify_release_decided(
        self,
        request: ApiKeyReleaseRequestModel,
        assignment: ApiKeyAssignment,
        *,
        approved: bool,
        trace_id: str,
    ) -> None:
        """Notify the requester of the manager's decision. Approved
        → "your key is now revoked"; rejected → "your release was
        denied, the key is still live".
        """
        from app.integrations.service import get_platform_setting
        from app.notifications.service import (
            NotificationEnvelope,
            NotificationService,
        )

        requester = await self.db.get(IdentityPerson, request.requester_id)
        if requester is None:
            return
        provider = await self.db.get(ApiKeyProvider, assignment.provider_id)
        provider_name = provider.name if provider else "(unknown provider)"
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="(未配置门户地址)",
            session=self.db,
        )
        # Canonical personal API-key route is ``/me/api-keys`` (matches every
        # other apikey emitter + the mobile deep-link map ``/m/api-keys``).
        # ``/my/apikeys`` was a stale outlier that 404'd on desktop and fell
        # through to the mobile home screen.
        my_url = f"{portal_url}/me/api-keys"

        notifier = NotificationService(self.db)
        if approved:
            title = f"API Key 释放申请已通过:{provider_name}"
            body = (
                f"你提交的 API Key 释放申请({assignment.request_number}, "
                f"{provider_name})已被审批通过,该 Key 已立即作废。\n"
                "如需新 Key,请重新提交申请。"
            )
            template_key = "apikey.release.approved"
        else:
            title = f"API Key 释放申请已被驳回:{provider_name}"
            body_lines = [
                f"你提交的 API Key 释放申请({assignment.request_number}, "
                f"{provider_name})已被审批驳回,该 Key 仍可正常使用。",
            ]
            if request.decision_note:
                body_lines.append(f"审批意见:{request.decision_note}")
            body_lines.append(
                "如确实不再需要,请与领导沟通后重新提交。",
            )
            body = "\n".join(body_lines)
            template_key = "apikey.release.rejected"

        await notifier.dispatch(
            NotificationEnvelope(
                recipient=requester,
                title=title,
                body_text=body,
                channels=["feishu", "email"],
                template_key=template_key,
                url=my_url,
                related_resource_type="api_key_assignment",
                related_resource_id=assignment.id,
                context={
                    "request_number": assignment.request_number,
                    "provider_name": provider_name,
                    "decision_note": request.decision_note or "",
                    "my_url": my_url,
                },
            ),
            trace_id=trace_id,
        )

    # ─── Self-service ownership transfer (PR2) ───

    async def submit_transfer_request(
        self,
        assignment_id: UUID,
        body: ApiKeyTransferRequestSubmit,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Owner submits an ownership-transfer request to a colleague.

        Unlike :meth:`submit_release_request`, transfer does NOT tear
        down the cloud credential — the key keeps working through the
        handover so production code keeps running. The recipient
        receives a card in ``/my/transfers`` and either accepts (which
        runs the existing :meth:`OffboardingService._apply_transfer`'s
        api_key branch, now plumbed through :meth:`apply_transfer`
        below) or rejects (assignment stays with the current owner).

        We delegate to :meth:`OffboardingService.create_transfer`
        because that path already enforces the "recipient is active
        and not the initiator" + "no duplicate pending transfer" +
        "initiator actually holds the asset" invariants — those are
        exactly the same constraints we'd duplicate here. Passing
        ``offboarding_process_id=None`` keeps this transfer outside any
        leaver flow.

        Authz: assignment owner only. State machine: only ``active`` /
        ``expired`` assignments are transferable — a key in
        ``pending_release`` already has a leader-bound release in flight
        and would create a contradictory parallel flow; we 409 those.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment owner may submit a transfer.",
            )
        if assignment.status not in (STATUS_ACTIVE, STATUS_EXPIRED):
            raise AppException(
                code="INVALID_STATUS_FOR_TRANSFER",
                message=(
                    f"Cannot transfer an assignment in status "
                    f"'{assignment.status}'. Transfer is only "
                    "available for 'active' or 'expired' keys."
                ),
                status_code=400,
            )

        from app.domains.offboarding.service import OffboardingService

        ob = OffboardingService(self.db)
        self._transfer_notification_service = ob
        await ob.create_transfer(
            initiator_id=current_user.id,
            recipient_id=body.recipient_person_id,
            asset_type="api_key",
            asset_ref_id=assignment.id,
            reason=body.note,
            offboarding_process_id=None,
            trace_id=trace_id,
        )

        # The audit row written by OffboardingService.create_transfer
        # covers the transfer entity; the row below is the apikey-domain
        # mirror so a "show me everything that happened to AK-XXX"
        # query against ``apikey.*`` actions returns the transfer too.
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.transfer.request",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "recipient_person_id": str(body.recipient_person_id),
                "note": body.note,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    async def dispatch_transfer_notifications_after_commit(
        self, *, trace_id: str,
    ) -> None:
        """Drain the delegated transfer outbox only after the route commit."""
        if self._transfer_notification_service is None:
            return
        svc, self._transfer_notification_service = (
            self._transfer_notification_service, None
        )
        await svc.dispatch_transfer_notifications_after_commit(trace_id=trace_id)

    async def cancel_transfer_request(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Owner withdraws their own pending transfer request.

        Idempotent on assignments with no pending transfer — returns
        current state, no side effect. Delegates to
        :meth:`OffboardingService._cancel_pending_transfer_internal`-
        style logic via the public cancel endpoint, but since we want
        ownership of the audit row to live in the apikey domain we
        implement the state flip inline rather than calling the
        offboarding router.
        """
        from app.domains.offboarding.models import (
            XFER_CANCELLED,
            XFER_PENDING,
            AssetTransferRequest,
        )

        assignment = await self._get_assignment(assignment_id)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment owner may cancel a transfer.",
            )

        xfer = (
            await self.db.execute(
                select(AssetTransferRequest).where(
                    and_(
                        AssetTransferRequest.asset_type == "api_key",
                        AssetTransferRequest.asset_ref_id == assignment.id,
                        AssetTransferRequest.initiator_id == current_user.id,
                        AssetTransferRequest.status == XFER_PENDING,
                    ),
                ).with_for_update(),
            )
        ).scalar_one_or_none()
        if xfer is None:
            # No pending transfer to cancel. Idempotent return.
            return await self._to_response(assignment)

        xfer.status = XFER_CANCELLED
        xfer.responded_at = datetime.now(UTC)
        xfer.response_note = "cancelled by initiator"

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.transfer.cancel",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "transfer_request_id": str(xfer.id),
                "recipient_person_id": str(xfer.recipient_id),
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    async def _reconcile_bailian_owner_mirror(
        self, assignment, new_owner_id: UUID,
    ) -> int:
        """Re-point every纳管'd Bailian console key linked to this
        assignment at the current owner.

        A纳管'd Bailian console key carries its display owner on
        ``bailian_api_keys.owner_person_id`` (what the 云资产 → 百炼 →
        API Key 列表 renders) AND back-links the registry row via
        ``assignment_id``. Keep that denormalised mirror in lock-step with
        the registry's ``assigned_to`` so both surfaces agree and BSS spend
        attributes to the right person. Idempotent — a re-run observes the
        already-synced value and writes nothing. Returns rows updated.
        """
        try:
            from app.domains.bailian.models import BailianApiKey  # noqa: PLC0415
        except ImportError:  # pragma: no cover - module always present
            return 0
        linked_keys = (
            await self.db.execute(
                select(BailianApiKey).where(
                    BailianApiKey.assignment_id == assignment.id,
                    BailianApiKey.deleted_at.is_(None),
                )
            )
        ).scalars().all()
        synced = 0
        for bk in linked_keys:
            if bk.owner_person_id != str(new_owner_id):
                bk.owner_person_id = str(new_owner_id)
                synced += 1
        if synced:
            await self.db.flush()
        return synced

    async def apply_transfer(
        self,
        *,
        assignment_id: UUID,
        new_owner_id: UUID,
        transfer_request_id: UUID | None,
        offboarding_process_id: UUID | None,
    ) -> None:
        """Ownership-rewrite path called from
        :meth:`OffboardingService._apply_transfer` once the recipient
        has accepted.

        Three things happen, in this order so a downstream failure
        leaves a recoverable state:

        1. **Local row flip** — ``assigned_to``, ``transferred_from``.
           This is the authoritative source for Helix's UI and
           subsequent cost-attribution rule writes.

        2. **Cost-allocation rule swap** — for every provider bucket
           (``bedrock`` / ``azure_openai`` / ``bailian``) that has an
           ``allocation_rule_id`` stamped on
           ``detail.<bucket>.allocation_rule_id``, update the rule's
           ``assign_person_id`` to the new owner. Future cost-fact
           ingests (from the daily cost-ingest cron) pick up the new
           rule mapping and stamp ``attributed_person_id = new_owner``
           on the fact rows they create. Pre-transfer facts keep their
           already-stamped ``attributed_person_id = old_owner``
           (because the rule applies at ingest time, not at read
           time) — that's the "from transfer-effective date" temporal
           split the user asked for, achieved without adding
           time-window columns to the rule table.

        3. **Cloud-side tag rewrite** — for Bedrock assignments
           (AWS IAM user with ``Owner`` + ``helix-owner-email`` tags),
           push the new owner's email so AWS Cost Explorer attributes
           future spend correctly. Best-effort: a tag-update failure
           does NOT roll back the transfer (the local row + rule are
           already correct; an admin can retry the tag via the AWS
           console or via a future "rebind IAM" sweep). Aliyun Bailian
           and Azure OpenAI do not need tag rewrites — Bailian's
           autograny does not emit a cost-allocation rule (per-key BSS
           cost isn't available), and Azure OpenAI's rule keys on the
           per-employee resource-group name, which doesn't change on
           transfer.

        We do NOT touch ``status`` — an active key stays active through
        the handover, which is the whole point of preferring transfer
        over revoke. We do NOT recompute or backfill historical
        ``cmdb_cost_facts.attributed_person_id`` — that would rewrite
        history and break the temporal split.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        # Round-46 (2026-05-20) defence-in-depth: refuse to apply a
        # transfer onto a non-live row. The offboarding-layer still-
        # holds check (``_initiator_still_holds_asset``) already gates
        # this for the standard accept path, but a future caller that
        # bypasses it (force-transfer, admin tool, MCP) lands here
        # without the gate — a transfer onto a revoked / rejected /
        # pending_approval row would silently mis-attribute on AWS +
        # CmdbCostAllocationRule, re-arm a leaver-owned dead row, etc.
        if assignment.status not in (STATUS_ACTIVE, STATUS_EXPIRED, STATUS_PENDING_RELEASE):
            raise AppException(
                code="INVALID_STATUS_FOR_TRANSFER_APPLY",
                message=(
                    f"Cannot apply transfer onto assignment in status "
                    f"{assignment.status!r}. Only active / expired / "
                    "pending_release rows are transferable."
                ),
                status_code=409,
            )
        # Recipient must still be an active employee. A transfer
        # submitted weeks ago and accepted only now (or accepted by a
        # leaver via stale Feishu card) would otherwise flip ownership
        # to a departed person, re-arm the cost-allocation rule under
        # their email, and retag the AWS IAM user's Owner — the exact
        # leaver-grant pattern the runtime gate exists to prevent.
        new_owner = await self.db.get(IdentityPerson, new_owner_id)
        if new_owner is None or new_owner.deleted_at is not None:
            raise AppException(
                code="RECIPIENT_NOT_FOUND",
                message=(
                    "Cannot apply transfer: recipient identity is "
                    "deleted or missing."
                ),
                status_code=410,
            )
        if (new_owner.employment_status or "active").lower() in (
            "left", "leaver_pending",
        ):
            raise AppException(
                code="RECIPIENT_NOT_ACTIVE",
                message=(
                    "Cannot apply transfer: recipient employment "
                    f"status is {new_owner.employment_status!r}. "
                    "Cancel the transfer and route the key to a "
                    "current employee."
                ),
                status_code=409,
            )
        old_owner_id = assignment.assigned_to
        if old_owner_id == new_owner_id:
            # Defensive: ``create_transfer`` already rejects self-
            # transfer with 400, so this branch only fires under a
            # concurrent-accept race. Skip the owner-flip / roster / rule
            # swap (all already no-ops at this point) — but STILL reconcile
            # the denormalised Bailian owner-mirror: a historical pre-#1216
            # drift (``assigned_to`` already == target yet the Bailian
            # mirror stale) is only fixable on this path, and the adopt
            # reconciler does not re-sync owner on already-linked keys.
            logger.warning(
                "apply_transfer: assignment %s already owned by %s — "
                "skipping flip (concurrent accept?), reconciling mirror",
                assignment.id, new_owner_id,
            )
            await self._reconcile_bailian_owner_mirror(
                assignment, new_owner_id,
            )
            return

        # 1. Local row flip.
        assignment.transferred_from = old_owner_id
        assignment.assigned_to = new_owner_id
        await self.db.flush()

        # 1a. Shared-access roster handover (migration 200):
        #
        # * Auto-promote the old Owner to a member of the new key
        #   so the handover doesn't immediately strip read access
        #   on a colleague who may still need to consult spend /
        #   reveal during the transition. The user requested this
        #   behaviour explicitly when greenlighting the feature
        #   ("Owner 转交时旧 owner 自动降级为成员"). Skipped when
        #   the old owner is inactive — :meth:`_add_member_internal`
        #   handles that filter.
        # * Remove the new Owner from the roster if they happened to
        #   already be a member (avoid the "you're an Owner AND a
        #   member of yourself" UI confusion).
        await self._add_member_internal(
            assignment=assignment,
            person_id=old_owner_id,
            added_by_id=new_owner_id,
        )
        new_owner_member = (
            await self.db.execute(
                select(ApiKeyAssignmentMember).where(
                    ApiKeyAssignmentMember.assignment_id == assignment.id,
                    ApiKeyAssignmentMember.person_id == new_owner_id,
                )
            )
        ).scalar_one_or_none()
        if new_owner_member is not None:
            scrub_id = new_owner_member.id
            await self.db.delete(new_owner_member)
            await self.db.flush()
            # Forensic — the recipient was on the roster before the
            # transfer and is now the Owner. Without this row a
            # later audit query against ``apikey.member.*`` actions
            # for this assignment would render an unexplained roster
            # change.
            await AuditService(self.db).log(
                trace_id=(
                    str(transfer_request_id) if transfer_request_id
                    else (
                        str(offboarding_process_id)
                        if offboarding_process_id else "force-transfer"
                    )
                ),
                actor_type=(
                    "system" if offboarding_process_id is not None
                    else "user"
                ),
                actor_user_id=new_owner_id,
                action="apikey.member.remove",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="success",
                detail={
                    "request_number": assignment.request_number,
                    "member_person_id": str(new_owner_id),
                    "reason": "promoted_to_owner_via_transfer",
                    "membership_row_id": str(scrub_id),
                },
            )

        # 1b. Denormalised Bailian owner-mirror sync.
        #
        # Adoption seeds both ``bailian_api_keys.owner_person_id`` and the
        # registry from the same person, but every later owner change went
        # through ONE side only: this ``apply_transfer`` path (self-service
        # accept / admin transfer / T+0 offboarding force-transfer) rewrote
        # ``assigned_to`` while leaving the Bailian mirror pointing at the
        # departed / previous owner — so the 百炼 page showed a stale 负责人
        # and BSS spend mis-attributed. Re-point every linked Bailian key so
        # the registry stays the single source of truth and both surfaces
        # agree after any transfer.
        bailian_synced = await self._reconcile_bailian_owner_mirror(
            assignment, new_owner_id,
        )

        # 2. Cost-allocation rule swap. Per-bucket so one missing rule
        # (e.g. Bailian, which doesn't emit one) doesn't stop the
        # others. Idempotent under retry: the second call observes
        # ``assign_person_id == new_owner_id`` and writes a no-op.
        rule_swaps: list[dict[str, Any]] = []
        try:
            from app.domains.cmdb.models import CmdbCostAllocationRule
        except ImportError:  # pragma: no cover - module always present
            CmdbCostAllocationRule = None  # type: ignore[assignment]

        detail = dict(assignment.detail or {})
        if CmdbCostAllocationRule is not None:
            for bucket_key in ("bedrock", "azure_openai", "bailian"):
                sub = detail.get(bucket_key) or {}
                rule_id = sub.get("allocation_rule_id")
                if not rule_id:
                    continue
                rule = await self.db.get(CmdbCostAllocationRule, str(rule_id))
                if rule is None:
                    rule_swaps.append({
                        "bucket": bucket_key,
                        "rule_id": str(rule_id),
                        "status": "not_found",
                    })
                    continue
                if rule.assign_person_id == str(new_owner_id):
                    rule_swaps.append({
                        "bucket": bucket_key,
                        "rule_id": str(rule_id),
                        "status": "already_swapped",
                    })
                    continue
                rule.assign_person_id = str(new_owner_id)
                today_iso = datetime.now(UTC).date().isoformat()
                trace_token = (
                    f"xfer={transfer_request_id}"
                    if transfer_request_id is not None
                    else f"offboarding={offboarding_process_id}"
                )
                suffix = (
                    f"\n\n[transferred {today_iso}: "
                    f"{old_owner_id} → {new_owner_id} via "
                    f"{trace_token}]"
                )
                rule.description = (rule.description or "") + suffix
                rule_swaps.append({
                    "bucket": bucket_key,
                    "rule_id": str(rule_id),
                    "status": "swapped",
                })

        # 3. Cloud-side tag rewrite (AWS only — see docstring).
        cloud_tag_summary = await self._retag_cloud_resources_on_transfer(
            assignment, new_owner_id=new_owner_id,
        )

        # Stash the most-recent transfer's metadata on the row so the
        # admin "transfer history" tooltip + governance reports can
        # answer "when was this last transferred, to whom, via which
        # xfer id?" without a join.
        detail["last_transfer"] = {
            "transfer_request_id": (
                str(transfer_request_id) if transfer_request_id else None
            ),
            "from_person_id": str(old_owner_id),
            "to_person_id": str(new_owner_id),
            "completed_at": datetime.now(UTC).isoformat(),
            "offboarding_process_id": (
                str(offboarding_process_id)
                if offboarding_process_id else None
            ),
            "rule_swaps": rule_swaps,
            "cloud_tag_summary": cloud_tag_summary,
            "bailian_owner_synced": bailian_synced,
            "force_transfer": transfer_request_id is None,
        }
        assignment.detail = detail
        await self.db.flush()

        # Round-49: when called from the offboarding force-transfer
        # path there's no AssetTransferRequest row — derive the trace
        # from the offboarding process id (still a UUID; still
        # globally unique). Action label is differentiated so the
        # audit dashboard can distinguish voluntary accept from
        # force-transfer at a glance.
        audit_trace = (
            transfer_request_id if transfer_request_id else offboarding_process_id
        )

        # 4. Rotate on ownership change (migration 603, platform setting
        #    ``apikey.rotate_on_transfer``, default ON). The previous owner
        #    still knows the plaintext — a transferred-but-unrotated key is
        #    a shared credential, and on the leaver path it is a credential
        #    a departed person keeps. Never raises: a rotation that cannot
        #    start (non-rotatable provider, AWS two-key limit, cloud outage)
        #    is recorded on ``detail.transfer_rotation`` and the new owner
        #    + IT are told to rotate by hand.
        await self._rotate_after_transfer(
            assignment,
            new_owner_id=new_owner_id,
            trace_id=str(audit_trace) if audit_trace else "force-transfer",
            forced=transfer_request_id is None,
        )
        await AuditService(self.db).log(
            trace_id=str(audit_trace) if audit_trace else "force-transfer",
            actor_type=(
                "system"
                if offboarding_process_id is not None
                else "user"
            ),
            actor_user_id=new_owner_id,
            action=(
                "apikey.transfer.complete"
                if transfer_request_id is not None
                else "apikey.transfer.force_complete"
            ),
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "from_person_id": str(old_owner_id),
                "to_person_id": str(new_owner_id),
                "transfer_request_id": (
                    str(transfer_request_id) if transfer_request_id else None
                ),
                "offboarding_process_id": (
                    str(offboarding_process_id)
                    if offboarding_process_id else None
                ),
                "rule_swaps": rule_swaps,
                "cloud_tag_summary": cloud_tag_summary,
                "bailian_owner_synced": bailian_synced,
                "force_transfer": transfer_request_id is None,
            },
        )

    async def _rotate_after_transfer(
        self,
        assignment: ApiKeyAssignment,
        *,
        new_owner_id: UUID,
        trace_id: str,
        forced: bool,
    ) -> None:
        """Start a rotation right after ``apply_transfer`` rewrote ownership.

        Outcomes land on ``detail.transfer_rotation``:
          * ``started``          — new credential minted, old one retires
                                   after ``apikey.transfer_rotation_grace_days``
          * ``skipped_disabled`` — the platform setting is off
          * ``skipped_in_flight``— a rotation was already running
          * ``manual_required``  — provider cannot be rotated by Helix
                                   (manually filled key / SaaS provider):
                                   new owner + IT are told to rotate at
                                   the vendor and have an admin re-fill
          * ``failed``           — mint raised (AWS key limit, cloud 5xx):
                                   same manual card, error recorded
        """
        from app.domains.apikey import rotation as _rot  # noqa: PLC0415
        from app.integrations.service import get_platform_setting  # noqa: PLC0415

        enabled = await get_platform_setting(
            "apikey.rotate_on_transfer", default=True, session=self.db,
        )
        detail = dict(assignment.detail or {})
        outcome: dict[str, Any] = {
            "at": datetime.now(UTC).isoformat(),
            "forced": forced,
            "to_person_id": str(new_owner_id),
        }
        if not bool(enabled):
            outcome["status"] = "skipped_disabled"
            detail["transfer_rotation"] = outcome
            assignment.detail = detail
            await self.db.flush()
            return
        if assignment.status != STATUS_ACTIVE or not (assignment.key_value or ""):
            # expired / pending_release rows are transferable but there is
            # no live credential worth rotating.
            outcome["status"] = "skipped_not_live"
            outcome["assignment_status"] = assignment.status
            detail["transfer_rotation"] = outcome
            assignment.detail = detail
            await self.db.flush()
            return
        active_kind, _blob = _rot.get_active_rotation(assignment)
        if active_kind is not None:
            outcome["status"] = "skipped_in_flight"
            outcome["kind"] = active_kind
            detail["transfer_rotation"] = outcome
            assignment.detail = detail
            await self.db.flush()
            return
        try:
            grace_raw = await get_platform_setting(
                "apikey.transfer_rotation_grace_days", default=7, session=self.db,
            )
            grace_days = max(0, min(30, int(grace_raw)))
        except (TypeError, ValueError):
            grace_days = 7
        kind = _rot.detect_rotation_provider(assignment)
        if kind is None:
            outcome["status"] = "manual_required"
            outcome["reason"] = "provider_not_rotatable"
            detail["transfer_rotation"] = outcome
            assignment.detail = detail
            await self.db.flush()
            await self._notify_transfer_rotation_manual(
                assignment, new_owner_id=new_owner_id, reason="provider_not_rotatable",
                trace_id=trace_id,
            )
            return
        # ``start_rotation`` COMMITS twice — the leak-safety tombstone before
        # the cloud mint, then the durable credential swap — and keeps doing
        # DB work after each. A SAVEPOINT is released by that first commit
        # and every statement after it is silently dropped, so this block is
        # the shape ``app.core.txn`` explicitly refuses to wrap. The usual
        # fallback (``await db.rollback()`` at the top of the handler) is
        # wrong here too: ``precheck_rotation`` and the early part of
        # ``start_rotation`` can raise BEFORE that first commit, and a
        # rollback would then silently undo ``apply_transfer``'s ownership
        # rewrite — losing the transfer while reporting it succeeded. So the
        # session is left as-is; ``start_rotation`` commits on each of its
        # own error paths, which is what keeps it usable for the bookkeeping
        # below. Residual: a failure BETWEEN its two commits leaves a dead
        # session and this recovery raises — loud and retryable, which is the
        # right failure mode for a half-finished credential rotation.
        # savepoint-exempt: body commits mid-flight; a rollback would undo the transfer
        try:
            await _rot.precheck_rotation(self.db, assignment, kind)
            result = await _rot.start_rotation(
                self.db, assignment, kind=kind, grace_days=grace_days,
                initiated_by=None,
            )
        except Exception as exc:  # noqa: BLE001 — never block the transfer
            logger.warning(
                "apikey transfer rotation failed for %s (%s): %s",
                assignment.id, kind, exc,
            )
            # ``start_rotation`` may have replaced ``assignment.detail`` (it
            # writes the rotation blob before the cloud call and clears it on
            # failure) — re-read before stamping ours.
            detail = dict(assignment.detail or {})
            outcome["status"] = "failed"
            outcome["kind"] = kind
            code = getattr(exc, "code", None)
            outcome["error"] = (str(code) if code else f"{type(exc).__name__}")[:120]
            detail["transfer_rotation"] = outcome
            assignment.detail = detail
            await self.db.flush()
            await AuditService(self.db).log(
                trace_id=trace_id,
                actor_type="system",
                actor_user_id=None,
                action="apikey.transfer.rotation_failed",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="failure",
                detail={"kind": kind, "error": outcome["error"], "forced": forced},
            )
            await self._notify_transfer_rotation_manual(
                assignment, new_owner_id=new_owner_id,
                reason=outcome["error"], trace_id=trace_id,
            )
            return
        detail = dict(assignment.detail or {})
        outcome["status"] = "started"
        outcome["kind"] = kind
        outcome["grace_days"] = grace_days
        retire_at_val = result.get("retire_at")
        outcome["retire_at"] = (
            retire_at_val.isoformat() if isinstance(retire_at_val, datetime)
            else str(retire_at_val)
        )
        detail["transfer_rotation"] = outcome
        assignment.detail = detail
        await self.db.flush()
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="system",
            actor_user_id=None,
            action="apikey.transfer.rotation_started",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={"kind": kind, "grace_days": grace_days, "forced": forced},
        )
        try:
            await _rot.notify_rotation_started(
                self.db, assignment,
                retire_at=result["retire_at"], grace_days=grace_days,
                trace_id=trace_id,
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning("apikey transfer rotation notify failed: %s", exc)

    async def _notify_transfer_rotation_manual(
        self,
        assignment: ApiKeyAssignment,
        *,
        new_owner_id: UUID,
        reason: str,
        trace_id: str,
    ) -> None:
        """Tell the new owner (and IT) that this key must be rotated by hand."""
        from app.notifications.service import (  # noqa: PLC0415
            NotificationEnvelope,
            NotificationService,
        )

        recipients: list[IdentityPerson] = []
        owner = await self.db.get(IdentityPerson, new_owner_id)
        if owner is not None and owner.deleted_at is None and owner.work_email:
            recipients.append(owner)
        # Read-only: one SELECT plus per-id ``db.get``. Nothing is added to
        # the session and nothing flushes, so a failure here cannot deactivate
        # the transaction the caller keeps writing to afterwards.
        # savepoint-exempt: read-only role lookup, no write to poison the txn
        try:
            from app.auth.models import AuthRole, AuthUserRole  # noqa: PLC0415

            admin_ids = list((await self.db.execute(
                select(AuthUserRole.user_id)
                .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
                .where(AuthRole.name.in_(("global_admin", "it_admin")))
            )).scalars().all())
            for aid in admin_ids:
                if owner is not None and aid == owner.id:
                    continue
                a = await self.db.get(IdentityPerson, aid)
                if (
                    a is not None and a.deleted_at is None
                    and (a.employment_status or "").lower() == "active" and a.work_email
                ):
                    recipients.append(a)
        except Exception as exc:  # noqa: BLE001
            logger.warning("apikey transfer rotation: admin resolve failed: %s", exc)
        if not recipients:
            return
        provider = await self._get_provider(assignment.provider_id)
        label = provider.name if provider else "API Key"
        svc = NotificationService(self.db)
        for r in recipients:
            # SAVEPOINT + ``step.reason()``, replacing ``except Exception`` +
            # ``str(exc)``. ``dispatch`` self-isolates its own writes, so the
            # savepoint here is defence in depth; the fixed defect is the
            # error text (``str(AppException)`` is an empty sentinel).
            async with best_effort(
                self.db, label=f"api-key manual-rotation card to {r.id}",
                log=logger, log_level=logging.DEBUG,
            ) as step:
                await svc.dispatch(
                    NotificationEnvelope(
                        recipient=r,
                        title=(
                            f"API Key 转手后需手动轮换:{assignment.request_number}({label})"
                        ),
                        body_text=(
                            f"API Key {assignment.request_number}({label})已转交给新持有人,"
                            "但 Helix 无法自动轮换这把 Key。请在厂商控制台生成新凭证并由管理员"
                            f"在 Helix 更新;原持有人仍知道旧凭证。原因:{reason}"
                        ),
                        channels=["feishu", "email"],
                        template_key="apikey.transfer_rotation_manual",
                        context={
                            "request_number": assignment.request_number,
                            "provider_name": label,
                            "reason": reason,
                        },
                        url=f"/me/api-keys/{assignment.id}",
                        related_resource_type="api_key_assignment",
                        related_resource_id=assignment.id,
                    ),
                    trace_id=trace_id,
                    actor_user_id=None,
                )
            if not step.ok:
                logger.warning(
                    "apikey transfer rotation manual card failed: %s",
                    step.reason(),
                )

    async def _sync_attestation_items(
        self, assignment_id: UUID, *, outcome: str,
    ) -> None:
        """Mirror a release decision onto attestation items (migration 603)."""
        from app.core.coordinators.access_certification_coordinator import (  # noqa: PLC0415
            sync_cert_items_for_api_key,
        )

        # SAVEPOINT, not a bare ``except``: ``sync_cert_items_for_api_key``
        # flushes. Swallowing a failed flush deactivates the transaction the
        # release decision itself lives in, so "the decision must not fail"
        # was a lie — the caller's next write (its audit row) raised and the
        # whole approve/reject rolled back. See ``app.core.txn``.
        async with best_effort(
            self.db,
            label=f"attestation sync for api key assignment {assignment_id}",
            log=logger,
            log_level=logging.DEBUG,  # the call site logs it below
        ) as step:
            await sync_cert_items_for_api_key(
                self.db, assignment_id=assignment_id, outcome=outcome,
                now=datetime.now(UTC),
            )
        if not step.ok:
            logger.warning(
                "apikey release: attestation sync failed for %s: %s",
                assignment_id, step.reason(),
            )

    async def submit_release_for_attestation(
        self,
        *,
        assignment_id: UUID,
        holder_id: UUID,
        reason_note: str,
        trace_id: str,
    ) -> ApiKeyReleaseRequestModel:
        """Holder said "I don't need this key" on their self-attestation.

        Same state machine as :meth:`submit_release_request` (manager or IT
        approves before the credential is torn down) with two differences:
        the reason code is fixed to ``no_longer_needed`` and the call is
        idempotent on an already-pending release (returns it) — the
        attestation coordinator may retry.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != holder_id:
            raise ForbiddenError("Only the assignment owner may release it.")
        existing = await self._find_open_release_request(assignment.id)
        if existing is not None:
            return existing
        if assignment.status not in (STATUS_ACTIVE, STATUS_EXPIRED, STATUS_PENDING_RELEASE):
            raise AppException(
                code="INVALID_STATUS_FOR_RELEASE",
                message=(
                    f"Cannot request release for assignment in status "
                    f"'{assignment.status}'."
                ),
                status_code=400,
            )
        previous_status = (
            assignment.status if assignment.status != STATUS_PENDING_RELEASE
            else STATUS_ACTIVE
        )
        release_request = ApiKeyReleaseRequestModel(
            assignment_id=assignment.id,
            requester_id=holder_id,
            reason_code="no_longer_needed",
            reason_note=reason_note,
            status="pending",
            previous_status=previous_status,
            requested_at=datetime.now(UTC),
        )
        self.db.add(release_request)
        await self.db.flush()
        assignment.status = STATUS_PENDING_RELEASE
        detail = dict(assignment.detail or {})
        detail["release"] = {
            "release_request_id": str(release_request.id),
            "previous_status": previous_status,
            "reason_code": "no_longer_needed",
            "reason_note": reason_note,
            "requested_at": release_request.requested_at.isoformat(),
            "source": "access_certification",
        }
        assignment.detail = detail
        await self.db.flush()
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=holder_id,
            action="apikey.release.request",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "release_request_id": str(release_request.id),
                "reason_code": "no_longer_needed",
                "previous_status": previous_status,
                "source": "access_certification",
            },
        )
        try:
            await self._notify_release_pending(
                release_request, assignment, trace_id=trace_id,
            )
        except Exception as notif_err:  # noqa: BLE001
            logger.warning(
                "release-pending notify failed for %s: %s",
                release_request.id, notif_err,
            )
        return release_request

    async def retry_iam_retag(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Admin retry for a transfer whose cloud-side IAM tag rewrite
        failed (PR3-3).

        When ``apply_transfer`` runs, the cloud-tag push is best-effort
        — a failure lands in
        ``detail.last_transfer.cloud_tag_summary.<bucket>.status='failed'``
        but does NOT roll back the transfer. Without a retry path, the
        admin would have to either log in to the AWS console (out of
        Helix) or trigger a fresh autograny rebind (heavyweight) to
        restore tag consistency. This endpoint re-runs
        ``_retag_cloud_resources_on_transfer`` for the current owner
        and updates ``detail.last_transfer.cloud_tag_summary`` in place
        so the admin UI's "tag failed, click to retry" affordance
        clears itself.

        Authz: admin only. The retry IS a cloud-side mutation (calls
        ``iam.tag_user``) so it inherits the same gate as the
        ``/iam-binding`` rebind endpoint.

        Idempotent: re-running on an already-tagged row is a no-op AWS
        call (``iam.tag_user`` is an upsert; same key+value lands the
        same result). The summary is overwritten with the latest
        attempt's outcome.
        """
        if not can_manage_assets(current_user.role_names):
            raise AppException(
                code="FORBIDDEN",
                message=(
                    "Retrying an IAM tag rewrite requires asset / IT "
                    "admin role."
                ),
                status_code=403,
            )
        assignment = await self._get_assignment(assignment_id, lock=True)

        # Only meaningful when there's actually a tag binding to retry.
        if assignment.iam_integration_id is None or not assignment.iam_user_arn:
            raise AppException(
                code="NO_IAM_BINDING",
                message=(
                    "This assignment has no AWS IAM user binding — "
                    "nothing to re-tag."
                ),
                status_code=400,
            )
        # Self-review safeguard (PR3-3): re-tagging a revoked assignment
        # is meaningless — the cloud IAM user was deleted as part of
        # ``revoke_key``'s teardown chain (see
        # ``_revoke_cloud_resources``), and ``iam.tag_user`` would 404
        # with ``NoSuchEntity``. The audit row would then carry a
        # misleading "tag failed" instead of the truthful "cloud
        # resource already gone". Refuse up-front so the admin gets a
        # clear error.
        if assignment.status == STATUS_REVOKED:
            raise AppException(
                code="ASSIGNMENT_REVOKED",
                message=(
                    "Cannot retry IAM tag on a revoked assignment — "
                    "the cloud-side IAM user was already deleted by "
                    "the revoke teardown chain."
                ),
                status_code=400,
            )

        cloud_tag_summary = await self._retag_cloud_resources_on_transfer(
            assignment, new_owner_id=assignment.assigned_to,
        )

        # Stash the result on detail.last_transfer.cloud_tag_summary so
        # the admin UI clears the "failed" badge if this attempt
        # succeeded. We DON'T overwrite the rest of last_transfer
        # (transfer_request_id, from_person_id, etc. are still the
        # historically accurate values).
        detail = dict(assignment.detail or {})
        last_transfer = dict(detail.get("last_transfer") or {})
        last_transfer["cloud_tag_summary"] = cloud_tag_summary
        last_transfer.setdefault("retag_attempts", [])
        last_transfer["retag_attempts"].append({
            "at": datetime.now(UTC).isoformat(),
            "by_person_id": str(current_user.id),
            "summary": cloud_tag_summary,
        })
        # Cap the list, but NOT by recency alone — the first successful
        # attempt is load-bearing for spend attribution (see
        # :func:`_trim_retag_attempts`).
        last_transfer["retag_attempts"] = _trim_retag_attempts(
            last_transfer["retag_attempts"],
        )
        detail["last_transfer"] = last_transfer
        assignment.detail = detail
        await self.db.flush()

        any_failed = any(
            (bucket_summary or {}).get("status") == "failed"
            for bucket_summary in cloud_tag_summary.values()
        )
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.iam_retag.retry",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="failure" if any_failed else "success",
            detail={
                "request_number": assignment.request_number,
                "cloud_tag_summary": cloud_tag_summary,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        if any_failed:
            # Surface as 502 so the admin sees a clear error and the
            # UI can re-fetch + re-render the "failed" badge with the
            # fresh error reason. Audit row is already written above;
            # the failure detail also lives on
            # ``detail.last_transfer.retag_attempts[-1].summary`` for
            # the UI to render. We inline the first failure reason
            # into ``message`` so a basic toast surfaces actionable
            # text without the frontend needing to peek inside
            # ``detail``.
            first_reason: str | None = None
            for bucket_summary in cloud_tag_summary.values():
                if (bucket_summary or {}).get("status") == "failed":
                    first_reason = (bucket_summary or {}).get("reason")
                    break
            raise AppException(
                code="IAM_RETAG_FAILED",
                message=(
                    f"IAM tag retry failed: {first_reason or 'unknown'}. "
                    "See audit log + assignment detail for the full "
                    "per-bucket summary."
                ),
                status_code=502,
            )
        return await self._to_response(assignment)

    async def _build_aws_driver_for_tag_update(
        self, integration: Integration,
    ) -> Any:
        """Decrypt the AWS integration's secrets bundle and instantiate
        the cloud driver.

        Used only by the transfer flow's IAM tag rewrite. Mirrors the
        per-domain shim pattern in
        :func:`apikey.azure_openai_autogrant._build_driver` — apikey
        domain doesn't share a generic "give me a configured driver
        for this integration" helper with provisioning, so we keep the
        decrypt + merge + construct dance local. Pulled out as a
        method so the transfer-flow tests can monkeypatch it without
        coupling to ``_retag_cloud_resources_on_transfer``'s body.
        """
        from app.integrations.drivers.aws_cloud import AwsCloudDriver

        secret_fields: dict[str, Any] = {}
        if integration.secrets_encrypted:
            secret_fields = decrypt_fields(
                integration.secrets_encrypted,
                integration.secrets_nonce,
                integration_id=str(integration.id),
            )
        merged_config = {
            **{
                k: v for k, v in (integration.config_public or {}).items()
                if v is not None
            },
            **secret_fields,
        }
        return AwsCloudDriver(
            integration_id=str(integration.id), config=merged_config,
        )

    async def _retag_cloud_resources_on_transfer(
        self,
        assignment: ApiKeyAssignment,
        *,
        new_owner_id: UUID,
    ) -> dict[str, Any]:
        """Push the new owner's email into the cloud IAM user's
        ``Owner`` / ``helix-owner-email`` tags.

        AWS Bedrock only — see :meth:`apply_transfer` docstring for
        why Aliyun + Azure don't need this. Best-effort: a single
        failure logs + returns ``status='failed'`` in the per-bucket
        summary; the caller does NOT roll back the transfer (local
        row + rule are already authoritative for Helix attribution).

        Returns ``{"bedrock": {"status": "...", ...}}`` for the audit
        detail. Empty dict if the assignment has no AWS IAM binding.
        """
        if assignment.iam_integration_id is None or not assignment.iam_user_arn:
            return {}

        # Resolve the new owner's work_email — that's the tag value the
        # autograny path used (Bedrock autogrant tags
        # ``Owner=<email>`` + ``helix-owner-email=<email>``).
        new_owner = await self.db.get(IdentityPerson, new_owner_id)
        new_email = (new_owner.work_email or "").strip() if new_owner else ""
        if not new_email:
            return {
                "bedrock": {
                    "status": "skipped",
                    "reason": "new owner has no work_email",
                },
            }

        from app.integrations.models import Integration

        integ = await self.db.get(Integration, assignment.iam_integration_id)
        if integ is None or integ.provider != "aws":
            # Non-AWS bucket (or integration deleted) — nothing to do.
            return {}

        # Extract the IAM user name from the ARN. Canonical user ARN:
        # ``arn:aws:iam::<acct>:user/<optional-path>/<name>``. A naïve
        # ``split("/")[-1]`` on a role ARN (``...:role/<name>``) would
        # happily return the role's name, then ``iam.tag_user`` would
        # either ``NoSuchEntity`` (best case) or — if a user with the
        # same name happens to exist in the same account — silently
        # tag the wrong principal. Refuse non-user ARNs up-front so
        # admin-rebound rows whose ARN was hand-stamped surface as a
        # clear "skipped" rather than a misleading "tagged".
        import re

        arn = assignment.iam_user_arn or ""
        m = re.match(
            r"^arn:aws:iam::\d+:user/(?:[^/]+/)*([^/]+)$",
            arn,
        )
        if m is None:
            return {
                "bedrock": {
                    "status": "skipped",
                    "reason": (
                        "iam_user_arn does not match arn:aws:iam::<acct>:user/... "
                        "shape; refusing to tag a non-IAM-user principal"
                    ),
                },
            }
        user_name = m.group(1)

        try:
            driver = await self._build_aws_driver_for_tag_update(integ)
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "apply_transfer: failed to build AWS driver for "
                "integration %s: %s — IAM tag not updated",
                integ.id, exc,
            )
            return {
                "bedrock": {
                    "status": "failed",
                    "reason": f"driver-build-failed: {exc}",
                },
            }

        try:
            result = await driver.tag_iam_user(
                user_name=user_name,
                tags={
                    "Owner": new_email,
                    "helix-owner-email": new_email,
                },
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "apply_transfer: tag_iam_user on %s/%s failed: %s",
                integ.slug, user_name, exc,
            )
            return {
                "bedrock": {
                    "status": "failed",
                    "user_name": user_name,
                    "reason": str(exc)[:300],
                },
            }
        finally:
            try:
                close_fn = getattr(driver, "close", None)
                if close_fn is not None:
                    maybe = close_fn()
                    if maybe is not None:
                        await maybe
            except Exception:  # noqa: BLE001
                # close() failures don't matter — boto3 clients are
                # fire-and-forget for our purposes.
                pass

        if not result.ok:
            return {
                "bedrock": {
                    "status": "failed",
                    "user_name": user_name,
                    "reason": result.error_message or "unknown",
                },
            }
        return {
            "bedrock": {
                "status": "tagged",
                "user_name": user_name,
                "new_email": new_email,
            },
        }

    # ─── Assignment: admin fills key ───

    async def fill_key(
        self,
        assignment_id: UUID,
        body: ApiKeyFillRequest,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        # Defence-in-depth authz: HTTP route uses ``require_asset_admin``
        # but service-direct callers (worker, MCP, future tests) must not
        # be able to bypass admin-only enforcement. Mirrors the gate the
        # rest of the service applies (revoke / reveal / rebind).
        if not can_manage_assets(current_user.role_names):
            raise AppException(
                code="FORBIDDEN",
                message="Filling an API key requires asset / IT admin role.",
                status_code=403,
            )
        # Row-lock so two concurrent admin clicks serialise — the second
        # observes the already-filled state and either no-ops (same
        # key_value) or 409s (different value, requires overwrite=True).
        # Pre-fix the second click silently overwrote the first admin's
        # value with the second's, with both ``apikey.key.fill`` audit
        # rows landing — forensics couldn't disambiguate which key is
        # actually live on the cloud provider.
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.status not in (STATUS_APPROVED, STATUS_ACTIVE):
            raise AppException(
                code="INVALID_STATUS_FOR_KEY_FILL",
                message=f"Cannot fill key for assignment in status '{assignment.status}'",
                status_code=400,
            )
        # Overwrite guard: a row that's already ACTIVE with a stamped
        # key_value should only accept a re-fill on an explicit
        # ``overwrite=True``. Otherwise an admin double-click would
        # silently rotate the live credential.
        if assignment.status == STATUS_ACTIVE and assignment.key_value:
            if not getattr(body, "overwrite", False):
                raise AppException(
                    code="KEY_ALREADY_FILLED",
                    message=(
                        "This assignment already has a stamped key. Pass "
                        "overwrite=true to rotate it; otherwise revoke "
                        "and re-issue."
                    ),
                    status_code=409,
                )
        assignment.key_value = _encrypt_key_value(
            body.key_value, assignment_id=assignment.id,
        )
        assignment.status = STATUS_ACTIVE
        await self.db.flush()

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.key.fill",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={"request_number": assignment.request_number},
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    # ─── Assignment: admin re-mints a Bailian sk- key ───

    async def reissue_bailian_api_key(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> dict[str, Any]:
        """Admin migration helper: re-mint the DashScope sk- key on a
        Bailian assignment whose existing key_value is the now-broken
        legacy RAM AK/SK bundle.

        Wraps :func:`app.domains.apikey.bailian_autogrant.reissue_bailian_api_key`
        with audit logging. ``reissue_bailian_api_key`` does its own
        commit (it has to — Aliyun side-effects are already durable
        before the function returns) so we audit *after* the helper
        returns and write the audit row in a separate flush.
        """
        from app.domains.apikey.bailian_autogrant import (
            reissue_bailian_api_key as _reissue,
        )

        # Row-lock so two concurrent admin retries don't both mint a
        # fresh sk- key on the Aliyun side (the second key would orphan
        # — live on Aliyun, no Helix row pointing at it, no teardown
        # path). Same posture as fill_key.
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.status not in (STATUS_APPROVED, STATUS_ACTIVE):
            raise AppException(
                code="INVALID_STATUS_FOR_REISSUE",
                message=(
                    "Cannot reissue Bailian key for assignment in "
                    f"status '{assignment.status}' — must be 'approved' "
                    "or 'active'."
                ),
                status_code=400,
            )

        result = await _reissue(self.db, assignment)

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.key.reissue",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "provider": "bailian",
                "new_api_key_id": result.get("api_key_id"),
                "ram_user_name": result.get("ram_user_name"),
                "workspace_id": result.get("workspace_id"),
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return result

    # ─── Assignment: self-service AK/SK rotation ───

    async def rotate_key(
        self,
        assignment_id: UUID,
        *,
        grace_days: int,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Rotate a cloud-auto-granted API key's credential (dual-active).

        Mints a fresh credential that goes live immediately while the old
        one keeps working for ``grace_days`` (立即 / 3 / 7 / 14), after
        which the retire cron tears the old one down. Owner-only (or asset
        / IT admin). Does NOT return the new secret — the caller reveals
        it through the standard ``reveal`` flow, which keeps the existing
        credential-exfiltration gates (rate-limit, browser-session for
        admin-on-behalf) as the single choke point for plaintext.

        See :mod:`app.domains.apikey.rotation` for the per-provider mint /
        retire primitives and the full state machine.
        """
        from app.domains.apikey import rotation as rot

        if grace_days not in rot.ALLOWED_GRACE_DAYS:
            raise AppException(
                code="ROTATION_INVALID_GRACE",
                message=(
                    "grace_days must be one of "
                    f"{list(rot.ALLOWED_GRACE_DAYS)} (0 = retire "
                    "immediately)."
                ),
                status_code=422,
            )
        is_admin = can_manage_assets(current_user.role_names)
        try:
            assignment = await self._get_assignment(assignment_id, lock=True)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot rotate an assignment you don't own",
            ) from None
        if assignment.assigned_to != current_user.id and not is_admin:
            raise ForbiddenError(
                "rotate is owner-only (or asset / IT admin).",
            )
        effective_status = effective_assignment_status(
            assignment.status, assignment.valid_until, today=expiry_today(),
        )
        if effective_status != STATUS_ACTIVE:
            raise AppException(
                code="INVALID_STATUS_FOR_ROTATION",
                message=(
                    "Only an active key can be rotated (status is "
                    f"'{effective_status}'). Extend an expired key first, "
                    "or wait for a pending release to resolve."
                ),
                status_code=400,
            )
        kind = rot.detect_rotation_provider(assignment)
        if kind is None:
            raise AppException(
                code="ROTATION_NOT_SUPPORTED",
                message=(
                    "Only auto-granted cloud keys (AWS Bedrock, Aliyun "
                    "Bailian, Azure OpenAI) can be rotated. Manually "
                    "filled or external keys cannot."
                ),
                status_code=400,
            )
        existing_kind, _existing = rot.get_active_rotation(assignment)
        if existing_kind is not None:
            raise AppException(
                code="ROTATION_IN_FLIGHT",
                message=(
                    "A rotation is already in progress for this key. "
                    "Complete it (confirm you've switched) or cancel it "
                    "before starting another."
                ),
                status_code=409,
            )
        # Provider-specific feasibility (Bedrock 2-key limit).
        await rot.precheck_rotation(self.db, assignment, kind)

        # start_rotation durably commits the cloud credential + rotation
        # state (mirrors autogrant leak-safety) and refreshes the row.
        result = await rot.start_rotation(
            self.db, assignment,
            kind=kind, grace_days=grace_days,
            initiated_by=current_user.id,
        )
        retire_at = result["retire_at"]

        # grace_days == 0 → hard cut: retire the old credential now.
        # On a cloud-teardown failure retire_rotation keeps the row in
        # pending_retire (retire_ok=False) so the hourly sweep retries —
        # the "immediate" cut degrades to "cron retires shortly" rather
        # than stranding a live credential.
        retire_ok = True
        if grace_days <= 0:
            retire_summary = await rot.retire_rotation(
                self.db, assignment, kind=kind,
            )
            retire_ok = bool(retire_summary.get("retire_ok", True))

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.rotated",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success" if retire_ok else "partial",
            detail={
                "request_number": assignment.request_number,
                "kind": kind,
                "grace_days": grace_days,
                "retire_at": retire_at.isoformat(),
                "actor_role": "owner" if not is_admin else "admin",
                "immediate_retire": grace_days <= 0,
                "retire_ok": retire_ok,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        await self.db.commit()
        await self.db.refresh(assignment)

        # Notify the owner (post-commit; dispatch failure can't reverse
        # the durable rotation). For immediate-retire we skip the
        # "started" card in favour of the retired one so the employee
        # isn't told the old key is still valid when it isn't.
        if grace_days <= 0 and not retire_ok:
            # The immediate retire failed on the provider; the old
            # credential is still live and the sweep will retry. Send the
            # "started / still dual-active" card, NOT the retired one, so
            # the owner isn't told a live credential is dead.
            await rot.notify_rotation_started(
                self.db, assignment,
                retire_at=retire_at, grace_days=grace_days, trace_id=trace_id,
            )
        elif grace_days <= 0:
            await rot.notify_rotation_retired(
                self.db, assignment, trace_id=trace_id, forced=True,
            )
        else:
            await rot.notify_rotation_started(
                self.db, assignment,
                retire_at=retire_at, grace_days=grace_days, trace_id=trace_id,
            )
        await self.db.commit()
        return await self._to_response(assignment, current_user=current_user)

    async def force_complete_rotation(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Retire the old credential NOW ("我已切换,立即停用旧 key").

        Shortens the dual-active exposure window. Owner-only (or admin).
        """
        from app.domains.apikey import rotation as rot

        is_admin = can_manage_assets(current_user.role_names)
        try:
            assignment = await self._get_assignment(assignment_id, lock=True)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot manage a rotation on an assignment you don't own",
            ) from None
        if assignment.assigned_to != current_user.id and not is_admin:
            raise ForbiddenError(
                "rotation completion is owner-only (or asset / IT admin).",
            )
        kind, rotation = rot.get_active_rotation(assignment)
        if kind is None:
            raise AppException(
                code="NO_ACTIVE_ROTATION",
                message="This key has no rotation in progress to complete.",
                status_code=409,
            )
        # Only a fully-minted rotation can be completed. A row still in
        # ``minting`` means start_rotation's cloud mint is in flight (it
        # released the row lock at its tombstone commit) — acting now
        # would race the mint and no-op with empty handles, then the mint
        # would re-write pending_retire. Make the caller retry.
        if rotation.get("status") != rot.ROTATION_STATUS_PENDING_RETIRE:
            raise AppException(
                code="ROTATION_NOT_READY",
                message=(
                    "The rotation is still being provisioned; try again in "
                    "a moment."
                ),
                status_code=409,
            )
        retire_summary = await rot.retire_rotation(
            self.db, assignment, kind=kind,
        )
        retire_ok = bool(retire_summary.get("retire_ok", True))
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.rotation.completed",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            # The whole point of force-complete is retiring the old
            # credential — a failed cloud teardown means the action
            # failed, not "partially succeeded".
            status="success" if retire_ok else "failed",
            detail={
                "request_number": assignment.request_number,
                "kind": kind,
                "retire": retire_summary,
                "actor_role": "owner" if not is_admin else "admin",
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        await self.db.commit()
        await self.db.refresh(assignment)
        if not retire_ok:
            # Cloud teardown of the old credential failed — the old key
            # is still live. Two sub-cases, and the message MUST match
            # what actually happened to the rotation state:
            gave_up = bool(retire_summary.get("gave_up"))
            if gave_up:
                # retire_rotation exhausted MAX_RETIRE_ATTEMPTS and
                # CLEARED the rotation state (rotation_retire_at is now
                # NULL) — the hourly sweep will NOT rescan this row and a
                # manual retry here would 409 (no active rotation). The
                # old credential is stranded live and needs manual
                # cleanup. Unlike the cron give-up (which raises →
                # failure_notifier), this request path has no pager, so
                # emit a distinct failed audit row as the ops trail.
                await AuditService(self.db).log(
                    trace_id=trace_id,
                    actor_type="user",
                    actor_user_id=current_user.id,
                    action="apikey.rotation.retire_gave_up",
                    resource_type="api_key_assignment",
                    resource_id=str(assignment.id),
                    status="failed",
                    detail={
                        "request_number": assignment.request_number,
                        "kind": kind,
                        "retire": retire_summary,
                        "actor_role": "owner" if not is_admin else "admin",
                    },
                    ip_address=ip_address,
                    user_agent=user_agent,
                )
                await self.db.commit()
                raise AppException(
                    code="ROTATION_RETIRE_GAVE_UP",
                    message=(
                        "旧凭证多次云侧删除均失败,系统已停止自动重试;旧 key "
                        "可能仍然有效,已记录待 IT 手动清理,请联系 IT 处理。\n"
                        "Retiring the old credential failed repeatedly and "
                        "automatic retries have stopped — the old key may "
                        "still be live. This has been logged for IT to "
                        "clean up manually; please contact IT."
                    ),
                    status_code=502,
                )
            # Still within the retry budget — retire_rotation kept the
            # row in pending_retire (attempt counter persisted by the
            # commit above), so the hourly sweep — or the user, retrying
            # here — re-attempts. We must not claim it was retired.
            raise AppException(
                code="ROTATION_RETIRE_FAILED",
                message=(
                    "旧凭证的云侧删除暂时失败,旧 key 仍然有效;系统已保留"
                    "轮换态并每小时自动重试,你也可以稍后在此重试,并在 "
                    "/me/api-keys 确认旧凭证已停用。\n"
                    "Retiring the old credential failed on the provider "
                    "side — the old key is still live. The rotation is "
                    "kept and the hourly sweep retries automatically; you "
                    "can also retry here shortly."
                ),
                status_code=502,
            )
        await rot.notify_rotation_retired(
            self.db, assignment, trace_id=trace_id, forced=True,
        )
        await self.db.commit()
        return await self._to_response(assignment, current_user=current_user)

    async def cancel_rotation(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Abort a rotation and revert to the prior credential.

        Invalidates the freshly-minted credential and restores the old
        ``key_value`` so reveal hands back the original credential the
        employee is still using. Owner-only (or admin).
        """
        from app.domains.apikey import rotation as rot

        is_admin = can_manage_assets(current_user.role_names)
        try:
            assignment = await self._get_assignment(assignment_id, lock=True)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot manage a rotation on an assignment you don't own",
            ) from None
        if assignment.assigned_to != current_user.id and not is_admin:
            raise ForbiddenError(
                "rotation cancel is owner-only (or asset / IT admin).",
            )
        kind, rotation = rot.get_active_rotation(assignment)
        if kind is None:
            raise AppException(
                code="NO_ACTIVE_ROTATION",
                message="This key has no rotation in progress to cancel.",
                status_code=409,
            )
        # Same as complete: don't act on a still-``minting`` rotation —
        # the new credential handle isn't recorded yet, so cancel couldn't
        # invalidate it and would race the in-flight mint.
        if rotation.get("status") != rot.ROTATION_STATUS_PENDING_RETIRE:
            raise AppException(
                code="ROTATION_NOT_READY",
                message=(
                    "The rotation is still being provisioned; try again in "
                    "a moment."
                ),
                status_code=409,
            )
        cancel_summary = await rot.cancel_rotation(
            self.db, assignment, kind=kind,
        )
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.rotation.cancelled",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "kind": kind,
                "cancel": cancel_summary,
                "actor_role": "owner" if not is_admin else "admin",
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        await self.db.commit()
        await self.db.refresh(assignment)
        await rot.notify_rotation_cancelled(
            self.db, assignment, trace_id=trace_id,
        )
        await self.db.commit()
        return await self._to_response(assignment, current_user=current_user)

    # ─── Assignment: employee reveals key (audit logged) ───

    async def reveal_key(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> str:
        assignment = await self._get_assignment(assignment_id)
        audit = AuditService(self.db)

        is_owner = assignment.assigned_to == current_user.id
        is_admin = can_manage_assets(current_user.role_names)
        # Shared-access roster (migration 200): a teammate the Owner
        # added gets reveal access too. The whole point of the roster
        # is "small team shares one credential" — gating reveal at
        # Owner-only would force the team back to screenshot / chat-
        # paste smuggling. Admin and Owner paths are unchanged.
        is_member = (
            (not is_owner) and (not is_admin)
            and await self.is_member(assignment.id, current_user.id)
        )

        # Round-31 audit S3 (2026-05-18): audit FAILED reveal attempts
        # (forbidden + key-not-set) as well as successful ones, so
        # incident responders can answer "who tried to read AK-X
        # without authorisation?". Prior behaviour only audited the
        # success path, leaving probing attempts invisible.
        #
        # Round-4 verification fix: use ``log_denied_action`` (writes
        # the audit row in an independent committed session) instead
        # of the in-session ``audit.log(...)`` which would be rolled
        # back by ``get_db`` when the subsequent ``raise`` propagates.
        if not is_owner and not is_admin and not is_member:
            await log_denied_action(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.key.reveal_denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "reason": "not-owner-not-admin",
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            raise AppException(
                code="FORBIDDEN", message="Not allowed", status_code=403,
            )
        effective_status = effective_assignment_status(
            assignment.status, assignment.valid_until, today=expiry_today(),
        )
        if effective_status == STATUS_EXPIRED:
            await log_denied_action(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.key.reveal_denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "reason": "expired",
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            raise AppException(
                code="KEY_EXPIRED",
                message="Cannot reveal an expired API key. Extend it first.",
                status_code=409,
            )
        if not assignment.key_value:
            await log_denied_action(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.key.reveal_denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "reason": "key-not-set",
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            raise AppException(
                code="KEY_NOT_SET",
                message="Key has not been set yet",
                status_code=404,
            )

        # Block reveal while a release request is in flight. The user
        # has already declared they don't need this key (or, more
        # importantly, MIGHT not need it — the leader hasn't decided
        # yet), so the integrity argument behind the leader gate
        # extends to plaintext access: we don't want someone to grab
        # one last copy of the AK/SK while the approval is in flight.
        # Admins can still reveal (forensic / debugging path) but the
        # owner cannot.
        if assignment.status == STATUS_PENDING_RELEASE and not is_admin:
            await log_denied_action(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.key.reveal_denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "reason": "pending_release",
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            raise AppException(
                code="PENDING_RELEASE",
                message=(
                    "Reveal is blocked while a release request is "
                    "pending leader approval. Cancel the release "
                    "request to regain access."
                ),
                status_code=409,
            )

        # Decrypt FIRST, then audit success — Round-2 review BLOCKER:
        # the prior ordering (audit then decrypt) wrote a phantom
        # ``apikey.key.reveal status=success`` row when
        # ``_decrypt_key_value`` raised ``KEY_DECRYPT_FAILED`` (tampered
        # row / AAD mismatch / key rotation gap), AND the failure
        # carried no ``reveal_denied`` audit at all — exactly the case
        # the S3 control was supposed to catch. Now: decrypt → on
        # AppException write a denied row + re-raise → on success write
        # the success row.
        try:
            plaintext = _decrypt_key_value(
                assignment.key_value, assignment_id=assignment.id,
            )
        except AppException as exc:
            if exc.code == "KEY_DECRYPT_FAILED":
                # Independent-session audit so the row survives the
                # request rollback triggered by the re-raise below.
                await log_denied_action(
                    trace_id=trace_id,
                    actor_type="user",
                    actor_user_id=current_user.id,
                    action="apikey.key.reveal_denied",
                    resource_type="api_key_assignment",
                    resource_id=str(assignment.id),
                    detail={
                        "request_number": assignment.request_number,
                        "reason": "decrypt-failed",
                    },
                    ip_address=ip_address,
                    user_agent=user_agent,
                )
            raise
        # Tag the actor role into the success audit so forensic
        # replay can distinguish Owner self-reveal from a member
        # exercising shared-access from an admin's on-behalf-of
        # reveal. The shape is additive — existing dashboards that
        # only read ``request_number`` keep working.
        if is_owner:
            actor_role_for_audit = "owner"
        elif is_member:
            actor_role_for_audit = "member"
        else:
            actor_role_for_audit = "admin"
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.key.reveal",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "actor_role": actor_role_for_audit,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return plaintext

    # ─── Assignment: non-secret "how to use this key" hints ───

    async def usage_hints(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
    ) -> dict[str, Any]:
        """Return non-secret "how to use this key" hints for the UI.

        Safe to call WITHOUT revealing the secret — it never returns key
        material. It resolves the vendor kind, the adopted-provenance
        flag (存量纳管 keys whose plaintext Helix never held), and, for
        Azure OpenAI, the **live** list of model deployments IT created
        on the Foundry account so the employee's "如何使用" panel can show
        the real deployment name to call instead of the old (wrong)
        "go deploy a model yourself" instruction. When there is no usable
        deployment the ``contact_it`` flag tells the UI to route the
        employee to IT rather than to the Azure portal.

        Caller (router) is responsible for owner / member / admin
        authorization — same surface as ``reveal_key`` minus the
        browser-session step-up (this returns no secret).
        """
        assignment = await self._get_assignment(assignment_id)
        provider = await self._get_provider(assignment.provider_id)
        detail = assignment.detail or {}
        az = detail.get("azure_openai") or {}
        bedrock_d = detail.get("bedrock") or {}
        bailian_d = detail.get("bailian") or {}

        # Vendor kind — prefer the structured ``detail.<vendor>`` blob the
        # auto-grant wrote (the reliable discriminator), fall back to the
        # provider display name (mirrors the *_autogrant.py matchers).
        name = (provider.name if provider else "").lower()
        if az:
            kind = "azure_openai"
        elif bedrock_d:
            kind = "bedrock"
        elif (
            bailian_d
            or "bailian" in name
            or "百炼" in name
            or "dashscope" in name
            or "model studio" in name
        ):
            kind = "bailian"
        elif "bedrock" in name:
            kind = "bedrock"
        elif "azure" in name or "foundry" in name or "openai" in name:
            kind = "azure_openai"
        else:
            kind = "generic"

        adopted = detail.get("source") == "adopted"
        hints: dict[str, Any] = {
            "assignment_id": str(assignment.id),
            "vendor_kind": kind,
            "adopted": adopted,
            "has_key": bool(assignment.key_value),
            "provider_name": provider.name if provider else None,
            "contact_it": False,
        }

        if kind == "azure_openai":
            hints["deployments"] = []
            sub = az.get("subscription_id")
            rg = az.get("resource_group")
            account = az.get("foundry_account_name")
            integ_id = az.get("integration_id")
            # ``last_deployments`` is an optional stale fallback (last-known
            # deployment list, e.g. cached by a future sync). Used only
            # when the live ARM call is unavailable so the panel degrades
            # to "here's what we last saw" instead of a bare "contact IT".
            cached = [d for d in (az.get("last_deployments") or []) if d]
            if az.get("granted") and sub and rg and account and integ_id:
                try:
                    hints["deployments"] = await self._azure_list_deployments(
                        integration_id=integ_id,
                        subscription_id=sub,
                        resource_group=rg,
                        account_name=account,
                    )
                    hints["deployments_status"] = (
                        "ok" if hints["deployments"] else "none"
                    )
                except Exception as exc:  # live cloud call must never 500
                    logger.warning(
                        "usage_hints: azure list_foundry_deployments failed "
                        "for assignment %s: %s",
                        assignment.id, exc,
                    )
                    if cached:
                        hints["deployments"] = cached
                        hints["deployments_status"] = "stale"
                    else:
                        hints["deployments_status"] = "error"
            elif cached:
                hints["deployments"] = cached
                hints["deployments_status"] = "stale"
            else:
                # Grant hasn't provisioned a Foundry account (external /
                # admin-filled key, or grant still in flight).
                hints["deployments_status"] = "unconfigured"
            # Contact IT whenever there is no usable deployment to call.
            hints["contact_it"] = not hints["deployments"]
        elif adopted:
            # 存量纳管: Helix never held the console-created secret; the
            # employee must obtain / rotate it in the cloud console or via
            # IT. Reveal is genuinely unavailable (key_value is NULL).
            hints["contact_it"] = True

        return hints

    async def _azure_list_deployments(
        self,
        *,
        integration_id: str,
        subscription_id: str,
        resource_group: str,
        account_name: str,
    ) -> list[dict[str, str]]:
        """Build the Azure driver from the owning integration and list
        the Foundry account's model deployments. Read-only ARM call."""
        from app.domains.apikey.azure_openai_autogrant import _build_driver
        from app.integrations.models import Integration

        integ = await self.db.get(Integration, integration_id)
        if integ is None:
            raise RuntimeError(f"integration {integration_id} not found")
        driver = _build_driver(integ)
        return await driver.list_foundry_deployments(
            subscription_id=subscription_id,
            resource_group=resource_group,
            account_name=account_name,
        )

    # ─── Assignment: employee extends validity ───

    async def extend_validity(
        self,
        assignment_id: UUID,
        body: ApiKeyExtendRequest,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        # Lock the row so two concurrent extend calls can't both observe
        # extended_count=N and both flip to N+1 — that breaks the
        # "extensions performed" invariant the policy layer relies on.
        # Collapse "not found" + "found but not yours" into a single
        # ForbiddenError for non-admin callers so the extend endpoint
        # isn't an assignment-UUID enumeration oracle. Admins keep the
        # distinct 404 — same posture as the spend / reveal endpoints.
        is_admin = can_manage_assets(current_user.role_names)
        try:
            assignment = await self._get_assignment(assignment_id, lock=True)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot extend an assignment you don't own",
            ) from None
        if assignment.assigned_to != current_user.id and not is_admin:
            raise ForbiddenError(
                "Cannot extend an assignment you don't own",
            )
        # PR3-6 logic-scan fix (2026-05-19): give the
        # ``pending_release`` case a dedicated error code so the
        # frontend can render "release pending — cancel first" rather
        # than a generic "wrong status" toast. Same posture as the
        # reveal block during pending_release (PENDING_RELEASE 409).
        if assignment.status == STATUS_PENDING_RELEASE:
            raise AppException(
                code="PENDING_RELEASE_BLOCKS_EXTEND",
                message=(
                    "Cannot extend validity while a release request "
                    "is pending leader approval. Cancel the release "
                    "request first."
                ),
                status_code=409,
            )
        effective_status = effective_assignment_status(
            assignment.status, assignment.valid_until, today=expiry_today(),
        )
        if effective_status not in (STATUS_ACTIVE, STATUS_EXPIRED):
            raise AppException(
                code="INVALID_STATUS_FOR_EXTEND",
                message=f"Cannot extend assignment in status '{effective_status}'",
                status_code=400,
            )

        # Validate max 1 year from original valid_from. Round-20 audit
        # (A2F4): direct ``date(year+1, month, day)`` raises ValueError
        # on Feb 29 of a leap year + 1 (non-leap target). Fall back to
        # Feb 28 in the target year so the validity check produces a
        # meaningful error instead of a generic 500.
        if assignment.valid_from:
            try:
                max_date = date(
                    assignment.valid_from.year + 1,
                    assignment.valid_from.month,
                    assignment.valid_from.day,
                )
            except ValueError:
                # Feb 29 → Feb 28 of next year. Other month/day combos
                # don't have this issue (every month has at least 28
                # days). The 1-day shift toward the user (earlier
                # max_date) is the conservative choice.
                max_date = date(
                    assignment.valid_from.year + 1, 2, 28,
                )
            if body.new_valid_until > max_date:
                raise AppException(
                    code="EXTEND_EXCEEDS_MAX",
                    message="Extended date cannot exceed 1 year from the original start date",
                    status_code=400,
                )

        # Auth check BEFORE any mutation
        if effective_status == STATUS_EXPIRED:
            if not can_manage_assets(current_user.role_names):
                raise AppException(
                    code="ADMIN_REQUIRED_FOR_REACTIVATION",
                    message="Re-activating an expired assignment requires admin role. Please submit a new request.",
                    status_code=403,
                )

        # Round-46 (2026-05-20): leaver gate on the assignee. An admin
        # extending an expired row whose original assignee is now
        # ``left`` / ``leaver_pending`` would silently re-arm a
        # credential on a departed person — the cost-allocation rule
        # still points at their email, AWS Cost Explorer would
        # continue grouping spend under their name. Refuse loudly so
        # the admin instead routes the credential through transfer to
        # a current employee or revokes outright.
        assignee = await self.db.get(IdentityPerson, assignment.assigned_to)
        if assignee is None or assignee.deleted_at is not None:
            raise AppException(
                code="ASSIGNEE_NOT_FOUND",
                message=(
                    "Cannot extend: the assigned employee is deleted. "
                    "Revoke or transfer to a current employee first."
                ),
                status_code=410,
            )
        if (assignee.employment_status or "active").lower() in (
            "left", "leaver_pending",
        ):
            raise AppException(
                code="ASSIGNEE_NOT_ACTIVE",
                message=(
                    "Cannot extend: assigned employee employment_status "
                    f"is {assignee.employment_status!r}. Transfer the "
                    "key to a current employee or revoke."
                ),
                status_code=409,
            )

        # Only mutate after all auth checks pass
        old_until = assignment.valid_until
        assignment.valid_until = body.new_valid_until
        assignment.extended_count += 1
        if effective_status == STATUS_EXPIRED:
            assignment.status = STATUS_ACTIVE
        await self.db.flush()

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.key.extend",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "old_valid_until": str(old_until) if old_until else None,
                "new_valid_until": str(body.new_valid_until),
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    # ─── Assignment: admin revokes key ───

    async def revoke_key(
        self,
        assignment_id: UUID,
        reason: str | None,
        *,
        current_user: AuthUser | None,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
        reason_code: str | None = None,
        actor_role: str = ACTOR_ROLE_ADMIN,
    ) -> ApiKeyAssignmentResponse:
        """Revoke an assignment + tear down its cloud-side credential.

        Three callers / three actor_role values:

        * ``ACTOR_ROLE_ADMIN`` (default) — direct admin revoke from the
          HTTP route or MCP tool. The standard guard ``can_manage_assets``
          applies and ``current_user`` MUST be an admin.

        * ``ACTOR_ROLE_SELF_WITH_APPROVAL`` — entered exclusively from
          ``_finalize_release_approved`` after a leader-approved Feishu
          callback. The ``current_user`` here is the assignee themselves;
          we explicitly skip the ``can_manage_assets`` gate (the
          authorisation IS the leader's Feishu approval) but document
          via the actor_role column that this was NOT an admin action.

        * ``ACTOR_ROLE_SYSTEM_OFFBOARDING`` — reserved for PR2; the
          offboarding scanner invokes with a synthetic system actor
          when the leaver had no transferable owner.

        ``reason_code`` populates the new row column for governance
        reports; ``reason`` (the legacy free-text param) is stored as
        ``revoke_reason_note``. Either may be None; the schema layer
        validates the combination per actor_role (self requires
        reason_code; admin gets a default ``admin_revoke`` bucket).
        """
        # Validate actor_role first (cheap, no DB).
        if actor_role not in (
            ACTOR_ROLE_ADMIN,
            ACTOR_ROLE_SELF_WITH_APPROVAL,
            ACTOR_ROLE_SYSTEM_OFFBOARDING,
        ):
            raise AppException(
                code="INVALID_ACTOR_ROLE",
                message=f"unknown revoke actor_role: {actor_role!r}",
                status_code=500,
            )

        # Cheap role-only check for the admin path BEFORE any DB I/O —
        # avoids paying a row-lock just to 403 a non-admin caller.
        if actor_role == ACTOR_ROLE_ADMIN:
            if current_user is None or not can_manage_assets(
                current_user.role_names,
            ):
                raise AppException(
                    code="FORBIDDEN",
                    message="Revoking an API key requires asset / IT admin role.",
                    status_code=403,
                )
        elif actor_role == ACTOR_ROLE_SELF_WITH_APPROVAL:
            # Need a caller identity; ownership is re-checked on the
            # locked row below.
            if current_user is None:
                raise AppException(
                    code="FORBIDDEN",
                    message=(
                        "self_with_approval revoke requires a caller "
                        "identity."
                    ),
                    status_code=403,
                )

        # Lock the row so two concurrent admin revokes serialise — second
        # one observes STATUS_REVOKED and returns idempotently. Round-46
        # (2026-05-20): for ACTOR_ROLE_SELF_WITH_APPROVAL we run the
        # ownership check AFTER the lock — pre-fix used a separate
        # unlocked ``pre_target = _get_assignment(...)`` read, opening a
        # TOCTOU window where an apply_transfer between the unlocked
        # ownership check and the locked mutation let a former owner
        # drive a revoke on a key that had since been transferred to a
        # colleague.
        assignment = await self._get_assignment(assignment_id, lock=True)
        if actor_role == ACTOR_ROLE_SELF_WITH_APPROVAL:
            if assignment.assigned_to != current_user.id:
                raise AppException(
                    code="FORBIDDEN",
                    message=(
                        "self_with_approval revoke is only valid for the "
                        "assignee themselves; this caller does not own "
                        "the assignment."
                    ),
                    status_code=403,
                )
        # ACTOR_ROLE_SYSTEM_OFFBOARDING: no role-check (the calling
        # pipeline owns its own gate). The column writes below preserve
        # actor identity when it's available.
        # Idempotent on already-revoked rows: write a "noop" audit row so
        # forensics can still see the call attempt, then return current
        # state without double-clearing key_value / double-emitting the
        # primary revoke audit / re-firing cloud-side delete.
        if assignment.status == STATUS_REVOKED:
            await AuditService(self.db).log(
                trace_id=trace_id,
                actor_type="user" if current_user else "system",
                actor_user_id=current_user.id if current_user else None,
                action="apikey.key.revoke_noop",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="success",
                detail={
                    "request_number": assignment.request_number,
                    "reason": reason,
                    "reason_code": reason_code,
                    "actor_role": actor_role,
                    "note": "already revoked",
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
            return await self._to_response(assignment)

        # Cloud-side credential teardown BEFORE we clear the local row.
        # For Bedrock / Bailian auto-granted assignments the AK/SK pair
        # is live on AWS / Aliyun even though Helix knows about it —
        # anyone who already exfiltrated the plaintext (the assignee
        # themselves, or anyone they shared it with) keeps invoking the
        # provider on the company bill until the cloud-side keys are
        # actually deleted. Clearing ``key_value`` is bookkeeping, not
        # invalidation. For Azure OpenAI the credential is the AOAI
        # account key — DELETEing the cognitive-services account is
        # the only way to invalidate it AND stop the per-resource
        # billing meter.
        #
        # We do this BEFORE the status flip so a cloud-side failure
        # surfaces to the admin (409 / 500), the assignment stays in
        # its previous status, and a retry of the same revoke call
        # picks up where we left off. If the cloud call succeeds but
        # the subsequent flush / audit fails, the admin retry sees
        # ``already_absent=True`` from each driver — idempotent.
        cloud_revoke_summary = await self._revoke_cloud_resources(
            assignment, trace_id=trace_id,
        )

        # Disable any auto-created CmdbCostAllocationRule rows so the
        # cost-ingest cron stops attributing future facts to a revoked
        # AK's applicant. Done AFTER cloud teardown succeeds (cloud
        # failure raises and we don't reach here) and BEFORE the status
        # flip so the rule-disable + status flip + audit log all commit
        # in the same transaction — a partial state where the cloud
        # credential is gone but the rule keeps matching would silently
        # mis-charge a departed employee's email tag if any line item
        # ever carries it again (e.g. console-issued ad-hoc IAM user).
        # Disable, don't delete: forensics + the dashboard's "未分摊"
        # KPI rely on the rule row being inspectable.
        allocation_disable_summary = (
            await self._disable_allocation_rules_on_revoke(assignment)
        )

        assignment.status = STATUS_REVOKED
        assignment.key_value = None
        # A pending rotation is moot once the whole cloud identity is
        # torn down — clear the retire schedule + rotation blob so the
        # retire cron doesn't pick up a revoked row and the "轮换中"
        # badge disappears. (The status != 'revoked' cron filter already
        # skips it; this keeps the row honest.)
        if assignment.rotation_retire_at is not None or (
            assignment.detail
            and any(
                isinstance((assignment.detail.get(k) or {}), dict)
                and (assignment.detail.get(k) or {}).get("rotation")
                for k in ("bedrock", "bailian", "azure_openai")
            )
        ):
            _detail = dict(assignment.detail or {})
            for _k in ("bedrock", "bailian", "azure_openai"):
                _blob = _detail.get(_k)
                if isinstance(_blob, dict) and _blob.get("rotation"):
                    _blob = dict(_blob)
                    _blob.pop("rotation", None)
                    _detail[_k] = _blob
            assignment.detail = _detail
            assignment.rotation_retire_at = None
        # Forensic columns. ``revoked_by`` semantics:
        #   * admin → the admin who clicked the button
        #   * self_with_approval → the assignee themselves (the leader's
        #     id is in the Feishu approval payload, not on this column)
        #   * system_offboarding with a current_user → the offboarding
        #     admin who dispatched the auto-revoke provisioning task
        #   * system_offboarding without a current_user → NULL (pure
        #     cron sweep with no human actor)
        assignment.revoked_at = datetime.now(UTC)
        assignment.revoked_by = current_user.id if current_user else None
        # Code bucket: admin path defaults to ``admin_revoke`` when no
        # specific code was passed, so the column is never NULL on a
        # fresh-this-PR revoke row.
        if reason_code:
            assignment.revoke_reason_code = reason_code
        elif actor_role == ACTOR_ROLE_ADMIN:
            assignment.revoke_reason_code = "admin_revoke"
        elif actor_role == ACTOR_ROLE_SYSTEM_OFFBOARDING:
            assignment.revoke_reason_code = "offboarding"
        else:
            # self_with_approval requires reason_code at the schema
            # layer; this branch is defence-in-depth.
            assignment.revoke_reason_code = "other"
        assignment.revoke_reason_note = reason
        assignment.revoke_actor_role = actor_role
        await self.db.flush()

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user" if current_user else "system",
            actor_user_id=current_user.id if current_user else None,
            action="apikey.key.revoke",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "reason": reason,
                "reason_code": assignment.revoke_reason_code,
                "actor_role": actor_role,
                "cloud_revoke": cloud_revoke_summary,
                "allocation_rules_disabled": allocation_disable_summary,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        # Notify the assignee + their direct manager on the admin path
        # so the user finds out their key was pulled BEFORE the next
        # production request 401's. self_with_approval doesn't notify
        # (the user initiated it; they're already aware). system
        # path is silent here — PR2 emits a single aggregated leaver
        # summary instead of per-key spam.
        if actor_role == ACTOR_ROLE_ADMIN:
            try:
                await self._notify_admin_revoke(
                    assignment, reason=reason, trace_id=trace_id,
                )
            except Exception as notif_err:
                # Notification failure must not fail the revoke — cloud
                # teardown already succeeded, the audit row is written,
                # the row is in STATUS_REVOKED. A retry of the notifier
                # is wasted work because the next call sees STATUS_REVOKED
                # and short-circuits at the "already revoked" noop above.
                # Log loudly so the gap is visible in prod logs.
                logger.warning(
                    "apikey.revoke: notify failed for assignment %s "
                    "(revoke itself succeeded): %s",
                    assignment.id, notif_err,
                )

        return await self._to_response(assignment)

    async def _notify_admin_revoke(
        self,
        assignment: ApiKeyAssignment,
        *,
        reason: str | None,
        trace_id: str,
    ) -> None:
        """Send "your key was revoked by an admin" cards to the assignee
        and their direct manager.

        Why two recipients: the assignee needs the immediate "your
        production code is about to 401" signal; the manager needs the
        visibility because an admin pulling a subordinate's cloud
        credential is a managerial event (was it leak / compliance /
        mistake?). HRIS direct manager lookup follows the same
        ``manager_id`` field every other manager-cascade in the
        platform uses.

        Notification failure is swallowed by the caller (see revoke_key);
        this method does NOT need internal try/except.
        """
        from app.notifications.service import (
            NotificationEnvelope,
            NotificationService,
        )

        assignee = await self.db.get(IdentityPerson, assignment.assigned_to)
        if assignee is None:
            return

        provider = await self.db.get(ApiKeyProvider, assignment.provider_id)
        provider_name = provider.name if provider else "(unknown provider)"
        reason_text = (reason or "").strip() or "管理员未提供理由"

        notifier = NotificationService(self.db)

        # ── Assignee card ──
        await notifier.dispatch(
            NotificationEnvelope(
                recipient=assignee,
                title=f"API Key 已被管理员回收:{provider_name}",
                body_text=(
                    f"你的 API Key 申请单 {assignment.request_number}"
                    f"({provider_name})已被管理员回收,云端凭证已立即失效。\n"
                    f"理由:{reason_text}\n"
                    "如对此操作有疑问,请联系 IT 管理员。"
                ),
                template_key="apikey.revoked_by_admin.assignee",
                url="/me/api-keys",
                related_resource_type="api_key_assignment",
                related_resource_id=assignment.id,
                context={
                    "request_number": assignment.request_number,
                    "provider_name": provider_name,
                    "reason": reason_text,
                },
            ),
            trace_id=trace_id,
        )

        # ── Manager card ──
        manager_id = assignee.manager_id
        if manager_id is not None:
            manager = await self.db.get(IdentityPerson, manager_id)
            if manager is not None and manager.id != assignee.id:
                await notifier.dispatch(
                    NotificationEnvelope(
                        recipient=manager,
                        title=(
                            f"下属 API Key 被管理员回收:"
                            f"{assignee.display_name or assignee.work_email}"
                        ),
                        body_text=(
                            f"你的下属 {assignee.display_name or assignee.work_email} "
                            f"的 API Key 申请单 {assignment.request_number}"
                            f"({provider_name})已被管理员回收,云端凭证已立即失效。\n"
                            f"理由:{reason_text}"
                        ),
                        template_key="apikey.revoked_by_admin.manager",
                        related_resource_type="api_key_assignment",
                        related_resource_id=assignment.id,
                        context={
                            "request_number": assignment.request_number,
                            "provider_name": provider_name,
                            "reason": reason_text,
                            "assignee_name": (
                                assignee.display_name or assignee.work_email or ""
                            ),
                        },
                    ),
                    trace_id=trace_id,
                )

    async def _disable_allocation_rules_on_revoke(
        self,
        assignment: ApiKeyAssignment,
    ) -> list[dict[str, Any]]:
        """Disable (set ``enabled=False``) every CmdbCostAllocationRule
        the autogrant flow auto-created on behalf of this AK.

        Rule IDs are stored on ``assignment.detail.<bucket>.allocation_rule_id``
        by the autograny paths (bedrock_autogrant.py:518,
        azure_openai_autogrant.py:847; bailian autogrant currently doesn't
        emit a rule because per-key BSS cost isn't available — defensive
        lookup keeps the contract uniform).

        Idempotent: rules already ``enabled=False`` are detected via the
        ``already_disabled`` flag in the returned summary, and we don't
        re-stamp the description suffix to avoid unbounded suffix growth
        on retry. Rules referenced by detail but absent in the DB
        (admin manually deleted) appear in the summary with
        ``status='not_found'`` rather than raising — the revoke is
        already past the cloud-teardown point and shouldn't fail on
        stale book-keeping.

        Returns a list of ``{bucket, rule_id, status, ...}`` entries
        that gets stamped into the revoke audit row.
        """
        from app.domains.cmdb.models import CmdbCostAllocationRule

        detail = dict(assignment.detail or {})
        results: list[dict[str, Any]] = []
        for bucket_key in ("bedrock", "azure_openai", "bailian"):
            sub = detail.get(bucket_key) or {}
            rule_id = sub.get("allocation_rule_id")
            if not rule_id:
                continue
            rule = await self.db.get(CmdbCostAllocationRule, str(rule_id))
            if rule is None:
                results.append({
                    "bucket": bucket_key,
                    "rule_id": str(rule_id),
                    "status": "not_found",
                })
                continue
            if rule.enabled is False:
                results.append({
                    "bucket": bucket_key,
                    "rule_id": str(rule_id),
                    "status": "already_disabled",
                })
                continue
            rule.enabled = False
            # UTC for audit consistency — the audit_logs row this
            # disable lands in also stamps UTC, so an admin tracing a
            # revoke through the logs doesn't have to mentally bridge
            # two timezones.
            today_iso = datetime.now(UTC).date().isoformat()
            suffix = (
                f"\n\n[disabled by revoke of "
                f"{assignment.request_number} on {today_iso}]"
            )
            # Cap description at the column width (Text — no hard cap
            # in MySQL TEXT) but keep the suffix near the end so admins
            # see it at a glance.
            base_desc = rule.description or ""
            rule.description = (base_desc + suffix)[:65000]
            results.append({
                "bucket": bucket_key,
                "rule_id": str(rule_id),
                "status": "disabled",
                "rule_name": rule.name,
            })
        return results

    async def re_enable_allocation_rules(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None = None,
        user_agent: str | None = None,
    ) -> dict[str, Any]:
        """Re-enable the cost-allocation rules a prior revoke disabled.

        Exact inverse of :meth:`_disable_allocation_rules_on_revoke` —
        walks the same ``detail.<bucket>.allocation_rule_id`` map, flips
        ``enabled`` back to True, and stamps a re-enable suffix on the
        rule description. Surfaced by the reconciliation cost lens (a
        revoked key whose chargeback rule was disabled, then the
        assignment was re-activated / transferred so the rule should
        live again).

        Idempotent: rules already ``enabled=True`` are reported as
        ``already_enabled`` (count toward neither succeeded nor failed);
        rules referenced by detail but absent in the DB → ``not_found``.
        Returns ``{rules_re_enabled: [...ids], count, results}``.
        """
        from app.domains.cmdb.models import CmdbCostAllocationRule

        assignment = await self._get_assignment(assignment_id, lock=True)
        detail = dict(assignment.detail or {})
        results: list[dict[str, Any]] = []
        re_enabled_ids: list[str] = []
        for bucket_key in ("bedrock", "azure_openai", "bailian"):
            sub = detail.get(bucket_key) or {}
            rule_id = sub.get("allocation_rule_id")
            if not rule_id:
                continue
            rule = await self.db.get(CmdbCostAllocationRule, str(rule_id))
            if rule is None:
                results.append({
                    "bucket": bucket_key,
                    "rule_id": str(rule_id),
                    "status": "not_found",
                })
                continue
            if rule.enabled is True:
                results.append({
                    "bucket": bucket_key,
                    "rule_id": str(rule_id),
                    "status": "already_enabled",
                    "rule_name": rule.name,
                })
                continue
            rule.enabled = True
            today_iso = datetime.now(UTC).date().isoformat()
            suffix = (
                f"\n\n[re-enabled by admin for "
                f"{assignment.request_number} on {today_iso}]"
            )
            rule.description = ((rule.description or "") + suffix)[:65000]
            re_enabled_ids.append(str(rule_id))
            results.append({
                "bucket": bucket_key,
                "rule_id": str(rule_id),
                "status": "re_enabled",
                "rule_name": rule.name,
            })
        await self.db.flush()
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.cost_rules.re_enable",
            status="success",
            resource_type="apikey_assignment",
            resource_id=str(assignment_id),
            detail={
                "request_number": assignment.request_number,
                "count": len(re_enabled_ids),
                "results": results,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return {
            "rules_re_enabled": re_enabled_ids,
            "count": len(re_enabled_ids),
            "results": results,
        }

    async def transfer_assignment_owner(
        self,
        assignment_id: UUID,
        new_owner_id: UUID,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None = None,
        user_agent: str | None = None,
    ) -> dict[str, Any]:
        """Admin-direct ownership transfer (reconciliation fix path).

        Reuses :meth:`apply_transfer` — the same rewrite the offboarding
        peer-accept flow uses — so the cost-allocation rule swap + AWS
        owner-tag rewrite happen exactly as they would on a normal
        handover. ``apply_transfer`` enforces that the recipient is an
        active employee and that the row is in a transferable status, so
        this thin wrapper only adds the admin-initiated audit row and
        returns the refreshed assignment. ``transfer_request_id`` /
        ``offboarding_process_id`` are None (no peer request, no
        offboarding case backs an admin reconciliation fix).
        """
        await self.apply_transfer(
            assignment_id=assignment_id,
            new_owner_id=new_owner_id,
            transfer_request_id=None,
            offboarding_process_id=None,
        )
        assignment = await self._get_assignment(assignment_id, lock=False)
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.assignment.transfer_owner",
            status="success",
            resource_type="apikey_assignment",
            resource_id=str(assignment_id),
            detail={
                "request_number": assignment.request_number,
                "new_owner_id": str(new_owner_id),
                "via": "admin_reconciliation",
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    async def _revoke_cloud_resources(
        self,
        assignment: ApiKeyAssignment,
        *,
        trace_id: str,
    ) -> dict[str, Any]:
        """Tear down the cloud-side credential backing an auto-granted
        assignment.

        Dispatches by which auto-grant flow provisioned the row:

        * ``detail.bedrock.granted`` → AWS driver ``revoke_account`` on
          ``detail.bedrock.iam_user_name``. Deletes access keys, detaches
          policies, removes group memberships, then ``DeleteUser``.
        * ``detail.bailian.granted`` → Aliyun driver ``revoke_account``
          on ``detail.bailian.iam_user_name``. Same teardown shape.
        * ``detail.azure_openai.granted`` → Azure driver
          ``delete_azure_openai_account``. DELETEs the cognitive-services
          account, which both invalidates the keys and stops billing.

        Idempotent: each driver's revoke returns success on
        ``EntityNotExist`` / ``404``. A previously-failed revoke whose
        cloud-side delete actually completed surfaces ``already_absent``
        on retry without a hard error.

        Returns a structured summary that gets stamped into the revoke
        audit row so SREs can reconstruct exactly which cloud resources
        were touched.
        """
        detail = dict(assignment.detail or {})
        summary: dict[str, Any] = {"attempted": [], "errors": []}

        # Resolve the IAM-bound integration from the column first; fall
        # back to the per-grant detail blob for legacy rows that landed
        # before the column was populated. ``iam_integration_id`` is set
        # by every autograny path (bedrock_autogrant:379,
        # bailian_autogrant:~430, azure_openai_autogrant:~603).
        iam_integ_id = (
            str(assignment.iam_integration_id)
            if assignment.iam_integration_id else None
        )

        # Bedrock (AWS) -----------------------------------------------
        bedrock = detail.get("bedrock") or {}
        if bedrock.get("granted") and bedrock.get("iam_user_name"):
            await self._dispatch_cloud_revoke(
                kind="bedrock",
                iam_user_name=bedrock.get("iam_user_name"),
                integration_id_fallback=(
                    iam_integ_id or bedrock.get("integration_id")
                ),
                expected_provider="aws",
                summary=summary,
            )
        # Bailian (Aliyun) ---------------------------------------------
        # 1) Delete the sk- DashScope API key first (so an attacker
        #    holding the leaked sk- loses inference access immediately,
        #    even if the slower RAM teardown below fails). Idempotent —
        #    BailianClient.delete_api_key swallows NotFound on retry.
        # 2) Then tear down the companion RAM sub-account (legacy path
        #    via the Aliyun driver's revoke_account).
        bailian = detail.get("bailian") or {}
        # ``granted`` rows = autogrant-minted (dedicated RAM user + key).
        # ``adopted`` rows = console-created keys Helix纳管'd into the
        # registry (see bailian_adopt.py); they carry the real numeric
        # ModelStudio ``api_key_id`` so revoke can DeleteApiKey for-real —
        # BUT they intentionally have NO ``iam_user_name``, because the
        # RAM sub-account behind a console key is shared across many keys
        # and must never be torn down here. The RAM-teardown branch below
        # stays gated on ``granted`` so an adopted revoke only ever
        # deletes the single key, never the shared sub-account.
        if bailian.get("api_key_id") and (
            bailian.get("granted") or bailian.get("adopted")
        ):
            integ_id = (
                bailian.get("integration_id") or iam_integ_id
            )
            try:
                from app.domains.bailian.client import (
                    BailianClient,
                    resolve_credentials,
                )
                from app.integrations.models import Integration
                if not integ_id:
                    raise RuntimeError(
                        "bailian detail missing integration_id and "
                        "assignment.iam_integration_id is unset",
                    )
                integration = await self.db.get(Integration, UUID(integ_id))
                if integration is None:
                    raise RuntimeError(
                        f"integration {integ_id} not found for bailian "
                        "DeleteApiKey",
                    )
                client = BailianClient(resolve_credentials(integration))
                workspace_id = (
                    str(bailian.get("workspace_id"))
                    if bailian.get("workspace_id") else None
                )
                # Delete EVERY distinct sk- this assignment references, not
                # just the canonical one. During an in-flight rotation the
                # canonical ``api_key_id`` points at the NEW key while the
                # OLD (still-live) key id lives only in
                # ``detail.bailian.rotation.prev_handle.api_key_id``.
                # Deleting only the canonical id would orphan the OLD key —
                # live on Aliyun forever, still billing, still a valid
                # inference credential a leaver holds (the offboarding
                # revoke path made this a real hole). Sweep both.
                ids_to_delete: list[str] = [str(bailian["api_key_id"])]
                rot_prev = (
                    (bailian.get("rotation") or {}).get("prev_handle") or {}
                )
                prev_key_id = rot_prev.get("api_key_id")
                if prev_key_id and str(prev_key_id) not in ids_to_delete:
                    ids_to_delete.append(str(prev_key_id))
                # Mirror the cloud-side deletion into the local registry.
                # ``sync_api_keys`` deliberately never soft-deletes
                # assignment-linked rows, so without this the
                # bailian_api_keys row would linger active/undeleted
                # forever after the real key is gone. Same session/txn as
                # the status flip below, so registry-retire + revoke
                # commit atomically.
                from app.domains.bailian.service import BailianService
                _bailian_svc = BailianService(self.db)
                # ModelStudio 分区域:打错区域 delete 会 HTTP 200 但删不掉
                # (不报错)——撤销一把北京的 Key 会静默失效。
                _ws_region = await _bailian_svc.workspace_region(
                    integration_id=str(integration.id),
                    workspace_id=workspace_id,
                )
                for _key_id in ids_to_delete:
                    delete_result = await client.delete_api_key(
                        api_key_id=_key_id,
                        workspace_id=workspace_id,
                        region=_ws_region,
                    )
                    summary["attempted"].append({
                        "kind": "bailian.api_key",
                        "api_key_id": _key_id,
                        "result": delete_result,
                    })
                    registry_result = await _bailian_svc.mark_api_key_deleted_on_revoke(
                        integration_id=str(integ_id),
                        api_key_id=_key_id,
                    )
                    summary["attempted"].append({
                        "kind": "bailian.api_key.registry",
                        "api_key_id": _key_id,
                        "result": registry_result,
                    })
            except Exception as exc:  # noqa: BLE001
                logger.exception(
                    "apikey.revoke: Bailian DeleteApiKey failed "
                    "(assignment=%s api_key_id=%s trace=%s)",
                    assignment.id, bailian.get("api_key_id"), trace_id,
                )
                from app.integrations.service import _scrub_message
                safe_exc = _scrub_message(str(exc))[:200]
                summary["errors"].append({
                    "kind": "bailian.api_key",
                    "error": _scrub_message(str(exc))[:400],
                })
                raise AppException(
                    code="CLOUD_REVOKE_FAILED",
                    message=(
                        "Bailian DeleteApiKey failed: "
                        f"{safe_exc}. The assignment stays in its "
                        "previous status; retry the revoke after "
                        "verifying the Aliyun integration credentials. "
                        "The sk- key remains active until this call "
                        "succeeds."
                    ),
                    status_code=502,
                ) from exc
        if bailian.get("granted") and bailian.get("iam_user_name"):
            await self._dispatch_cloud_revoke(
                kind="bailian",
                iam_user_name=bailian.get("iam_user_name"),
                integration_id_fallback=(
                    bailian.get("integration_id") or iam_integ_id
                ),
                # Bailian autogrant only targets the 海外 (alicloud_intl)
                # tenant per ``app/domains/bailian/client.py:53,65``.
                # Legacy "aliyun" tag retained for forward-compat in case
                # a future variant lands.
                expected_provider=("alicloud_intl", "aliyun"),
                summary=summary,
            )
        # Azure OpenAI -------------------------------------------------
        # Round-46 (2026-05-20): the autograny path writes
        # ``foundry_account_name`` (azure_openai_autogrant.py:864,1012);
        # legacy ``account_name`` was intentionally removed and a test
        # asserts its ABSENCE. The pre-fix shape checked ``account_name``
        # so the entire AOAI cloud-side teardown was silently skipped on
        # every revoke — the cognitive-services account stayed live, the
        # KEY1 kept working, and the per-resource billing meter kept
        # running. Read the canonical key, with a one-tick fallback to
        # ``account_name`` to cover any pre-rename historical rows that
        # might still exist.
        aoai = detail.get("azure_openai") or {}
        foundry_account = (
            aoai.get("foundry_account_name") or aoai.get("account_name")
        )
        if (
            aoai.get("granted")
            and foundry_account
            and aoai.get("subscription_id")
            and aoai.get("resource_group")
        ):
            try:
                from app.domains.cmdb.lifecycle_service import _load_cloud_driver
                from app.integrations.models import Integration
                integ_id = (
                    aoai.get("integration_id")
                    or (str(assignment.iam_integration_id) if assignment.iam_integration_id else None)
                )
                if not integ_id:
                    raise RuntimeError(
                        "azure_openai detail missing integration_id and "
                        "assignment.iam_integration_id is unset",
                    )
                integration = await self.db.get(Integration, UUID(integ_id))
                if integration is None:
                    raise RuntimeError(
                        f"integration {integ_id} not found for azure_openai revoke",
                    )
                driver = await _load_cloud_driver(self.db, integration)
                try:
                    delete_result = await driver.delete_azure_openai_account(
                        subscription_id=str(aoai["subscription_id"]),
                        resource_group=str(aoai["resource_group"]),
                        account_name=str(foundry_account),
                    )
                finally:
                    try:
                        await driver.close()
                    except Exception:  # noqa: BLE001
                        pass
                summary["attempted"].append({
                    "kind": "azure_openai",
                    "account_name": foundry_account,
                    "result": delete_result,
                })
            except Exception as exc:  # noqa: BLE001
                logger.exception(
                    "apikey.revoke: azure_openai cloud-side delete failed "
                    "(assignment=%s account=%s trace=%s)",
                    assignment.id, foundry_account, trace_id,
                )
                # Round-20 post-fix: scrub the driver-supplied error
                # message before echoing it to the API caller. Azure /
                # boto / Aliyun SDK exception bodies can carry AKIA…
                # access keys, request signatures, or token fragments
                # inside ``SignatureDoesNotMatch`` / ``InvalidClient
                # TokenId`` messages.
                from app.integrations.service import _scrub_message
                safe_exc = _scrub_message(str(exc))[:200]
                summary["errors"].append({
                    "kind": "azure_openai",
                    "error": _scrub_message(str(exc))[:400],
                })
                raise AppException(
                    code="CLOUD_REVOKE_FAILED",
                    message=(
                        "Azure OpenAI cognitive-services account delete "
                        f"failed: {safe_exc}. The assignment stays "
                        "in its previous status; retry the revoke after "
                        "verifying the Azure subscription credentials."
                    ),
                    status_code=502,
                ) from exc

        return summary

    async def _dispatch_cloud_revoke(
        self,
        *,
        kind: str,
        iam_user_name: str | None,
        integration_id_fallback: str | None,
        expected_provider: str | tuple[str, ...],
        summary: dict[str, Any],
    ) -> None:
        """Shared dispatch for Bedrock (AWS) / Bailian (Aliyun) revoke.

        ``expected_provider`` accepts a single string or a tuple of
        acceptable provider slugs. Bailian only ever runs on
        ``alicloud_intl`` (see ``app/domains/bailian/client.py:53,65``),
        so a single ``"aliyun"`` value silently rejected EVERY Bailian
        RAM teardown and left orphan RAM sub-accounts forever. Accept
        a family of slugs so the autogrant ↔ revoke pair agrees on
        provider taxonomy.

        Raises ``AppException(code='CLOUD_REVOKE_FAILED', 502)`` on
        cloud-side failure so the outer revoke aborts without flipping
        the DB row to REVOKED — admin retries the same endpoint and
        each driver's ``revoke_account`` is idempotent on
        ``EntityNotExist``.
        """
        from app.domains.cmdb.lifecycle_service import _load_cloud_driver
        from app.integrations.driver_base import CloudAccountRevokeContext
        from app.integrations.models import Integration

        if not iam_user_name:
            return
        integ_id = integration_id_fallback
        if not integ_id:
            summary["errors"].append({
                "kind": kind,
                "error": "no integration_id on assignment detail",
            })
            raise AppException(
                code="CLOUD_REVOKE_FAILED",
                message=(
                    f"{kind} cloud revoke: integration_id missing on the "
                    "assignment; cannot resolve cloud account to revoke."
                ),
                status_code=502,
            )
        expected_providers: tuple[str, ...] = (
            (expected_provider,) if isinstance(expected_provider, str)
            else tuple(expected_provider)
        )
        try:
            integration = await self.db.get(Integration, UUID(str(integ_id)))
            if integration is None:
                raise RuntimeError(
                    f"integration {integ_id} not found for {kind} revoke",
                )
            if integration.provider not in expected_providers:
                # Defensive: the autogrant chose this integration; a
                # provider mismatch means the integrations table was
                # tampered with after the grant landed. Surface so an
                # admin investigates rather than mis-routing the revoke
                # to a different cloud account.
                raise RuntimeError(
                    f"integration {integ_id} provider="
                    f"{integration.provider!r} but {kind} expects one of "
                    f"{expected_providers!r}",
                )
            driver = await _load_cloud_driver(self.db, integration)
            try:
                result = await driver.revoke_account(
                    CloudAccountRevokeContext(
                        provider=integration.provider,
                        account_id=str(iam_user_name),
                    ),
                )
            finally:
                try:
                    await driver.close()
                except Exception:  # noqa: BLE001
                    pass
            if not result.ok:
                raise RuntimeError(
                    f"driver returned ok=False: {result.error_message}",
                )
            summary["attempted"].append({
                "kind": kind,
                "iam_user_name": iam_user_name,
                "ok": True,
                "external_ref": result.external_ref,
                "raw": (result.raw_response or {}),
            })
        except AppException:
            raise
        except Exception as exc:  # noqa: BLE001
            logger.exception(
                "apikey.revoke: %s cloud-side delete failed "
                "(iam_user_name=%s integration=%s)",
                kind, iam_user_name, integ_id,
            )
            # Round-20 post-fix: scrub the driver-supplied error
            # string before echoing back to the caller. boto / Aliyun
            # SDK error messages can embed AKIA… keys, request
            # signatures, or session tokens in
            # ``SignatureDoesNotMatch`` / ``InvalidAccessKeyId`` bodies.
            from app.integrations.service import _scrub_message
            safe_exc = _scrub_message(str(exc))[:200]
            summary["errors"].append({
                "kind": kind,
                "iam_user_name": iam_user_name,
                "error": _scrub_message(str(exc))[:400],
            })
            raise AppException(
                code="CLOUD_REVOKE_FAILED",
                message=(
                    f"{kind} cloud revoke failed: {safe_exc}. "
                    "Assignment stays in its previous status; retry "
                    "the revoke after verifying cloud credentials."
                ),
                status_code=502,
            ) from exc

    # ─── Assignment: admin re-binds IAM user ───

    async def rebind_iam_user(
        self,
        assignment_id: UUID,
        body: ApiKeyIamRebindRequest,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Re-target an assignment's IAM/RAM user binding.

        ``body.iam_user_id=None`` clears the binding (used when the
        cloud-side IAM user has been deleted and we want the row to
        reflect the orphan state). When set, the IAM user must exist
        in ``cmdb_iam_users`` and be non-deleted; we read its arn +
        integration_id off the row so the assignment stays consistent
        without the caller having to pass denormalised fields.

        Doesn't touch cloud state — pure Helix book-keeping. Use the
        CMDB grant / revoke flows for actual IAM lifecycle.
        """
        from app.domains.cmdb.models import CmdbIamUser

        # Row-lock so a concurrent revoke can't clear iam_* columns
        # underneath us — pre-fix the rebind would resurrect a stale
        # Bedrock binding on an already-revoked row.
        assignment = await self._get_assignment(assignment_id, lock=True)
        # Refuse to rebind a revoked row: the cloud IAM user backing it
        # was deleted by the revoke teardown chain. Stamping a fresh
        # binding now would just re-create the dangling pointer pattern.
        if assignment.status == STATUS_REVOKED:
            raise AppException(
                code="ASSIGNMENT_REVOKED",
                message=(
                    "Cannot rebind IAM binding on a revoked assignment — "
                    "the cloud-side IAM user was deleted by the revoke "
                    "teardown chain. Re-issue the key from the request "
                    "flow instead."
                ),
                status_code=400,
            )
        previous = {
            "iam_user_id": (
                str(assignment.iam_user_id) if assignment.iam_user_id else None
            ),
            "iam_user_arn": assignment.iam_user_arn,
            "iam_integration_id": assignment.iam_integration_id,
            "cloud_scope_id": assignment.cloud_scope_id,
        }

        # IAM user binding (existing behaviour). Always applied — the
        # caller passes the desired final state; None clears.
        if body.iam_user_id is None:
            assignment.iam_user_id = None
            assignment.iam_user_arn = None
            assignment.iam_integration_id = None
            new_user_name = None
        else:
            iam_row = await self.db.get(CmdbIamUser, body.iam_user_id)
            if iam_row is None or iam_row.deleted_at is not None:
                raise NotFoundError("CmdbIamUser", str(body.iam_user_id))
            # ARN shape guard (same posture as apply_transfer): reject
            # role / group / service-role ARNs that would otherwise pass
            # the `cmdb_iam_users` sync filter under structural drift.
            # ``cmdb_iam_users`` is populated by ListUsers paths only, so
            # this branch is defence-in-depth — the row IS structurally
            # a user. Loud check at the rebind boundary nonetheless.
            import re

            arn_value = iam_row.arn or ""
            aws_user = re.match(
                r"^arn:aws:iam::\d+:user/(?:[^/]+/)*[^/]+$", arn_value,
            )
            aliyun_user = re.match(
                r"^acs:ram::[\d*]+:user/(?:[^/]+/)*[^/]+$", arn_value,
            )
            if not (aws_user or aliyun_user):
                raise AppException(
                    code="INVALID_IAM_ARN",
                    message=(
                        "Selected CmdbIamUser row's ARN does not match "
                        "an IAM user shape (arn:aws:iam::<acct>:user/... "
                        "or acs:ram::*:user/...). Refusing to rebind to a "
                        f"role / group / service-role: {arn_value!r}"
                    ),
                    status_code=400,
                )
            assignment.iam_user_id = iam_row.id
            assignment.iam_user_arn = iam_row.arn
            assignment.iam_integration_id = str(iam_row.integration_id)
            new_user_name = iam_row.user_name

            # Mirror the new IAM binding into detail.bedrock when this
            # is a Bedrock assignment so future
            # ``maybe_auto_grant_bedrock`` short-circuits
            # ("already-granted") instead of trying to mint a second
            # IAM user. Use the canonical ``_is_bedrock_provider``
            # matcher (not a raw substring) — provider renames like
            # "Aliyun Bedrock-compatible" would otherwise false-match
            # and stamp ``bedrock.granted=True`` on a non-Bedrock row,
            # causing the next revoke to call AWS for a nonexistent
            # IAM user.
            from app.domains.apikey.bedrock_autogrant import (
                _is_bedrock_provider,
            )
            detail = dict(assignment.detail or {})
            bedrock_info = dict(detail.get("bedrock") or {})
            provider = await self._get_provider(assignment.provider_id)
            if bedrock_info or _is_bedrock_provider(provider):
                bedrock_info["granted"] = True
                bedrock_info["in_flight"] = False
                bedrock_info["iam_user_name"] = iam_row.user_name
                bedrock_info["user_arn"] = iam_row.arn
                bedrock_info["rebound_at"] = (
                    datetime.now(UTC).isoformat()
                )
                bedrock_info["rebound_by_user_id"] = str(current_user.id)
                detail["bedrock"] = bedrock_info
                assignment.detail = detail

        # Cloud-scope binding (new). Use ``model_fields_set`` to
        # distinguish "leave alone" (omit) vs "clear" (explicit null)
        # so legacy IAM-only callers don't accidentally clear an
        # admin's manual scope assignment. This block is independent
        # of the IAM block above — a caller can rebind only the
        # scope, only the IAM user, or both at once.
        if "cloud_scope_id" in body.model_fields_set:
            from app.domains.cmdb.models import CmdbCloudScope
            if body.cloud_scope_id is None:
                assignment.cloud_scope_id = None
            else:
                scope_row = await self.db.get(
                    CmdbCloudScope, body.cloud_scope_id,
                )
                if scope_row is None or scope_row.deleted_at is not None:
                    raise NotFoundError(
                        "CmdbCloudScope", str(body.cloud_scope_id),
                    )
                assignment.cloud_scope_id = str(scope_row.id)

        await self.db.flush()

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.iam_binding.rebind",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "previous": previous,
                "new_iam_user_id": (
                    str(assignment.iam_user_id)
                    if assignment.iam_user_id else None
                ),
                "new_iam_user_arn": assignment.iam_user_arn,
                "new_iam_user_name": new_user_name,
                "new_iam_integration_id": assignment.iam_integration_id,
                "new_cloud_scope_id": assignment.cloud_scope_id,
                "reason": body.reason,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        return await self._to_response(assignment)

    # ─── Manager-scope walk (list_assignments + notify-candidates) ───

    async def _resolve_subordinate_chain(
        self,
        manager_user_id: UUID,
        *,
        target_id: UUID,
        max_depth: int | None = None,
    ) -> bool:
        """Is ``target_id`` a (transitive) subordinate of ``manager_user_id``?

        Returns True iff a chain of ``manager_person_id`` links connects
        the target person up to the manager. Walks at most
        ``max_depth`` (default ``_MANAGER_CHAIN_MAX_DEPTH``) links so a
        pathological HRIS data state can't spin the loop. Used by
        ``list_assignments`` to gate non-admin views of someone else's
        API keys on the manager-scope contract from CLAUDE.md.

        Note: ``IdentityPerson.id`` and ``AuthUser.id`` are the same
        UUID by the auth model — ``manager_user_id`` is the auth row
        but lines up with the identity row's primary key.
        """
        # ``or`` would coerce a legitimate ``max_depth=0`` (caller
        # wants no walk) to the default 5. Use explicit None check.
        depth_cap = (
            self._MANAGER_CHAIN_MAX_DEPTH if max_depth is None else max_depth
        )
        # Resolve the target's identity row to start the walk.
        target = await self.db.get(IdentityPerson, target_id)
        if target is None or target.manager_person_id is None:
            return False
        seen: set[UUID] = {target.id}
        current_mgr_id = target.manager_person_id
        for _ in range(depth_cap):
            if not current_mgr_id:
                return False
            mgr_uuid = (
                current_mgr_id
                if isinstance(current_mgr_id, UUID)
                else UUID(str(current_mgr_id))
            )
            if mgr_uuid == manager_user_id:
                return True
            if mgr_uuid in seen:
                logger.warning(
                    "apikey.subordinate_chain: circular manager link "
                    "detected walking up from %s (repeated %s)",
                    target_id, mgr_uuid,
                )
                return False
            seen.add(mgr_uuid)
            mgr = await self.db.get(IdentityPerson, mgr_uuid)
            if mgr is None:
                return False
            current_mgr_id = mgr.manager_person_id
        return False

    # ─── Assignment: notify-recipient candidate list (admin + owner) ───

    # Walk at most this many manager links when building the
    # ``notify_recipients`` candidate set for a non-admin caller. Five
    # levels covers the deepest reporting line we ship today
    # (IC → manager → director → VP → CXO) with one extra for slack;
    # a circular-reference guard inside the loop is the real safety
    # belt against pathological HRIS data.
    _MANAGER_CHAIN_MAX_DEPTH = 5

    async def _resolve_manager_chain(
        self,
        applicant_id: UUID | str,
    ) -> list[IdentityPerson]:
        """Return [applicant, manager, skip-level, ...] up to chain depth.

        Used by both ``list_notify_candidates`` (frontend picker) and the
        ``update_assignment`` non-admin restriction. Skips persons with
        ``employment_status='left'`` so a deactivated manager doesn't
        anchor the candidate list, and bails on the first repeated id
        so a circular ``manager_person_id`` (HRIS data quality bug)
        can't spin the loop.
        """
        applicant = await self.db.get(
            IdentityPerson,
            applicant_id if isinstance(applicant_id, UUID) else UUID(str(applicant_id)),
        )
        if applicant is None:
            return []
        chain: list[IdentityPerson] = [applicant]
        seen: set[UUID] = {applicant.id}
        current_mgr_id = applicant.manager_person_id
        for _ in range(self._MANAGER_CHAIN_MAX_DEPTH):
            if not current_mgr_id:
                break
            mgr_uuid = (
                current_mgr_id
                if isinstance(current_mgr_id, UUID)
                else UUID(str(current_mgr_id))
            )
            if mgr_uuid in seen:
                logger.warning(
                    "apikey.manager_chain: circular manager_person_id "
                    "detected for applicant %s at depth %d (repeated %s); "
                    "stopping chain walk",
                    applicant.id, len(chain), mgr_uuid,
                )
                break
            seen.add(mgr_uuid)
            mgr = await self.db.get(IdentityPerson, mgr_uuid)
            if mgr is None:
                break
            # Round-3 security review HIGH: skip leavers BUT continue
            # walking up via their ``manager_person_id`` to the next
            # live ancestor. Prior behaviour ``break``-ed at the first
            # ``left`` manager, which produced a chain of just
            # ``[applicant]`` whenever the applicant's direct manager
            # had just offboarded. That collapsed the manager-blinding
            # guard's chain[1:] to empty and locked the owner out of
            # self-editing notify_recipients entirely. The skip-level
            # is a legitimate fallback recipient.
            if mgr.employment_status == "left":
                current_mgr_id = mgr.manager_person_id
                continue
            chain.append(mgr)
            current_mgr_id = mgr.manager_person_id
        return chain

    async def list_notify_candidates(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        q: str | None = None,
        trace_id: str | None = None,
    ) -> list[dict[str, Any]]:
        """Candidate recipient list for the API Key edit modal.

        Returns ``[{id, name, email, role: "applicant" | "manager" | "search"}, ...]``
        — a curated short list (applicant + manager chain) that the
        non-admin owner can pick from, optionally augmented with
        admin-only free-text search results when the caller is an
        asset/it admin AND passes ``q``. Without this endpoint the
        modal had to call ``identity-sync/lookups/persons`` which is
        ``require_admin``-gated — non-admin owners silently saw an
        empty Select (bug-class: dead self-service edit).
        """
        # Round-3 audit MEDIUM: collapse "not found" + "found but not
        # yours" to a single 403 for non-admin callers so the endpoint
        # isn't an enumeration oracle on assignment UUIDs. Mirrors the
        # reveal + spend pattern. Admins keep the distinct 404.
        # Read-side admin includes management + auditor.
        from app.core.rbac import (
            AUDITOR_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )
        is_admin = can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        )
        try:
            assignment = await self._get_assignment(assignment_id)
        except NotFoundError:
            if is_admin:
                raise
            await log_denied_action(
                trace_id=(
                    trace_id
                    or f"apikey.notify_candidates.{assignment_id}"
                ),
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.notify_candidates.denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment_id),
                detail={"reason": "not-found-non-admin-probe"},
            )
            raise ForbiddenError(
                "Only the assignment owner or an asset admin may list "
                "notification candidates for this assignment.",
            ) from None
        is_owner = assignment.assigned_to == current_user.id
        if not (is_admin or is_owner):
            await log_denied_action(
                trace_id=(
                    trace_id
                    or f"apikey.notify_candidates.{assignment.id}"
                ),
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.notify_candidates.denied",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "reason": "not-owner-not-admin",
                },
            )
            raise ForbiddenError(
                "Only the assignment owner or an asset admin may list "
                "notification candidates for this assignment.",
            )

        chain = await self._resolve_manager_chain(assignment.assigned_to)
        seen_ids: set[UUID] = set()
        result: list[dict[str, Any]] = []
        for idx, p in enumerate(chain):
            # Round-3 logic MEDIUM: drop the applicant row when the
            # applicant itself has left — admin editing a leaver-owned
            # assignment shouldn't be offered the leaver as a valid
            # recipient (budget-alert dispatcher would filter them out
            # at send-time anyway, producing the "every override is a
            # leaver → fallback to defaults" path the override-collapse
            # audit signal exists to catch). The chain ancestors stay
            # because the helper already filters live status on the
            # walk; only the index-0 applicant slot needs this guard.
            if idx == 0 and p.employment_status == "left":
                continue
            seen_ids.add(p.id)
            result.append({
                "id": str(p.id),
                "name": p.display_name or p.work_email or str(p.id),
                "email": p.work_email or "",
                "role": "applicant" if idx == 0 else "manager",
            })

        # Admin-only free-text search augment. Returns persons matching
        # ``q`` in display_name / work_email, excluding ``left`` and
        # anyone already in the chain (de-dupe). Cap at 20 to bound
        # payload size — same shape as identity-sync lookup but only
        # for admin callers, so non-admin owners can't enumerate.
        q_trimmed = (q or "").strip()
        if is_admin and len(q_trimmed) >= 2:
            ql = f"%{q_trimmed.lower()}%"
            from sqlalchemy import func as _func
            from sqlalchemy import or_
            rows = await self.db.execute(
                select(IdentityPerson)
                .where(
                    or_(
                        _func.lower(IdentityPerson.display_name).like(ql),
                        _func.lower(IdentityPerson.work_email).like(ql),
                    ),
                    IdentityPerson.employment_status != "left",
                )
                .order_by(IdentityPerson.display_name.asc())
                .limit(20)
            )
            for p in rows.scalars():
                if p.id in seen_ids:
                    continue
                seen_ids.add(p.id)
                result.append({
                    "id": str(p.id),
                    "name": p.display_name or p.work_email or str(p.id),
                    "email": p.work_email or "",
                    "role": "search",
                })
        return result

    # ─── Assignment: edit (budget + notify recipients) ───

    async def update_assignment(
        self,
        assignment_id: UUID,
        body: AssignmentUpdate,
        *,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyAssignmentResponse:
        """Edit ``monthly_budget_usd`` and / or ``notify_recipients`` on
        an existing assignment.

        Authz: ``current_user`` must be (a) an asset_admin / it_admin
        OR (b) the assignment's applicant (``assigned_to``). Anyone else
        gets 403. Same posture as ``get_assignment_spend`` —
        owner-self-service plus admin.

        Validation:

        * ``monthly_budget_usd`` is validated at the schema layer
          (``ge=20``); the service just writes the value through.
        * ``notify_recipients`` semantics are tri-state (see
          :class:`AssignmentUpdate` docstring). When a list is
          provided, every entry must be a well-formed UUID AND must
          exist in ``identity_persons``; we raise a
          :class:`ValidationError` listing the missing ids so the
          admin sees exactly which entries are bad rather than a
          vague "invalid recipients". An empty list ``[]`` is
          accepted as the explicit "no recipients / suppress alerts"
          override.

        Audit: a single ``apikey.assignment.updated`` row captures
        before / after for both fields (always emitted, even when
        only one field changed — the diff inside the detail dict
        shows which). The dedup row in ``apikey_budget_alerts_dedup``
        for the current month is intentionally NOT cleared on a
        budget edit: re-firing an 80% alert mid-month because an
        admin lowered the cap would mostly just spam finance. The
        next threshold crossing (or month rollover) takes the new
        cap.
        """
        # Round-2 review MEDIUM: cheap unlocked auth check FIRST, then
        # re-fetch with FOR UPDATE for the actual mutation. The prior
        # ordering grabbed a row-lock BEFORE the authz check, so a
        # scripted forbidden-PATCH burst held a lock per request until
        # the rollback fired — contention DoS surface against
        # legitimate admin edits. With the non-admin restriction now
        # exercising this path on every owner-self-edit, narrowing the
        # lock window is worth the extra SELECT.
        #
        # Round-41 (2026-05-19): collapse 404→403 on the peek for non-
        # admin callers — same enumeration-oracle posture as the spend,
        # reveal, extend, and notify_candidates endpoints. Admins keep
        # the distinct 404. Without this, a scripted PATCH against a
        # made-up UUID gives a non-admin caller a "row doesn't exist"
        # signal that a "row exists but isn't yours" signal would NOT
        # have leaked.
        is_admin = can_manage_assets(current_user.role_names)
        try:
            peek = await self._get_assignment(assignment_id)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Only the assignment owner or an asset admin may edit "
                "this assignment.",
            ) from None
        is_owner = peek.assigned_to == current_user.id
        if not (is_admin or is_owner):
            raise ForbiddenError(
                "Only the assignment owner or an asset admin may edit "
                "this assignment.",
            )
        assignment = await self._get_assignment(assignment_id, lock=True)
        # Round-4 verification MEDIUM: TOCTOU re-verify. An admin could
        # reassign the row between ``peek`` and the locked re-fetch;
        # we'd then apply the non-admin owner's PATCH to someone else's
        # assignment. Re-check ownership after the lock so the
        # mutation is bound to the row state the auth decision was
        # made on.
        if not is_admin and assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Assignment ownership changed during the edit; refresh "
                "and try again.",
            )

        previous = {
            "monthly_budget_usd": str(assignment.monthly_budget_usd),
            "notify_recipients": (
                list(assignment.notify_recipients)
                if assignment.notify_recipients is not None
                else None
            ),
        }

        changed: dict[str, Any] = {}

        # ── monthly_budget_usd ──
        if body.monthly_budget_usd is not None:
            # Pydantic already validated ``ge=20``. Coerce to Decimal
            # at the column's precision so equality on the audit row
            # diff is stable (no "1300.0" vs "1300.00" flapping).
            new_budget = Decimal(body.monthly_budget_usd).quantize(
                Decimal("0.01"),
            )
            if new_budget != assignment.monthly_budget_usd:
                assignment.monthly_budget_usd = new_budget
                changed["monthly_budget_usd"] = str(new_budget)

        # ── notify_recipients ──
        # ``model_fields_set`` distinguishes "field omitted" (caller
        # is only updating budget, leave recipients alone) from
        # "field explicitly null" (caller is resetting recipients
        # back to default applicant+manager). Both legitimate.
        if "notify_recipients" in body.model_fields_set:
            new_recipients: list[str] | None
            if body.notify_recipients is None:
                new_recipients = None
            else:
                # Canonicalise each id to a lowercase UUID string so
                # equality lookups against ``identity_persons.id`` are
                # stable and we don't store mixed-case duplicates.
                canon: list[str] = []
                bad: list[str] = []
                for raw in body.notify_recipients:
                    try:
                        canon.append(str(UUID(str(raw))))
                    except (ValueError, TypeError):
                        bad.append(str(raw))
                if bad:
                    raise ValidationError(
                        "notify_recipients contains non-UUID values: "
                        f"{bad!r}",
                    )

                # De-duplicate while preserving order — admins
                # pasting from a wiki list commonly produce dupes.
                seen: set[str] = set()
                unique: list[str] = []
                for cid in canon:
                    if cid not in seen:
                        seen.add(cid)
                        unique.append(cid)

                if unique:
                    found_rows = await self.db.execute(
                        select(IdentityPerson.id).where(
                            IdentityPerson.id.in_([UUID(x) for x in unique]),
                        )
                    )
                    found_ids = {str(r[0]) for r in found_rows.all()}
                    missing = [x for x in unique if x not in found_ids]
                    if missing:
                        raise ValidationError(
                            "notify_recipients references unknown "
                            f"identity_persons.id values: {missing!r}",
                        )

                # Non-admin restriction (security finding S2): an owner
                # editing their own assignment can ONLY pick recipients
                # from their manager chain (themselves + reporting line
                # up to ``_MANAGER_CHAIN_MAX_DEPTH``). Without this an
                # employee could ``PATCH notify_recipients=[<colleague>]``
                # to route their budget alerts at an unrelated person
                # (PII oversharing). Round-2 review HIGH: the immediate
                # manager (if the applicant has one) MUST appear in the
                # list — otherwise an owner about to blow their budget
                # could silently strip their manager off the alert
                # routing while still passing the chain check (e.g. set
                # ``notify_recipients=[applicant_only]``). That defeats
                # the chargeback signal which is the whole point of
                # keeping manager in the default path. Admin keeps the
                # unrestricted path. The empty-list suppression below is
                # also admin-gated.
                #
                # Round-3 audit MEDIUM: skip the chain validation when
                # the proposed value equals the current value
                # (frontend echoes the field on every PATCH even when
                # only ``monthly_budget_usd`` actually changed). The
                # intent of the restriction is "non-admin can't
                # INTRODUCE new off-chain recipients", not "non-admin
                # can't preserve a list an admin already saved". Without
                # this skip, an admin pre-curating notify_recipients to
                # include a legitimate off-chain finance person locks
                # the owner out of editing their own budget.
                #
                # Round-4 verification BLOCKER fix: distinguish NULL
                # (default-state) from [] (admin-suppressed). The prior
                # comparison ``sorted(unique) == sorted(assignment.notify_recipients or [])``
                # collapsed both to ``[]``, letting a non-admin owner
                # silently flip a default-NULL row to [] (suppress) —
                # exactly the attack S2 + the empty-list guard exist to
                # block. Require the CURRENT stored value to be a
                # non-None list AND match exactly for the bypass to
                # engage.
                current = assignment.notify_recipients
                value_unchanged = (
                    current is not None
                    and sorted(unique) == sorted(current)
                )
                if not is_admin and value_unchanged:
                    pass
                elif not is_admin:
                    if not unique:
                        raise ForbiddenError(
                            "Only an asset admin may suppress budget "
                            "alerts (notify_recipients=[]). Non-admin "
                            "owners must list at least themselves or "
                            "someone in their manager chain.",
                        )
                    chain = await self._resolve_manager_chain(
                        assignment.assigned_to,
                    )
                    allowed_ids = {str(p.id) for p in chain}
                    not_allowed = [
                        x for x in unique if x not in allowed_ids
                    ]
                    if not_allowed:
                        # Don't echo attacker-supplied UUIDs back into
                        # the user-visible message — count is enough
                        # for the UI to render a useful toast, and the
                        # service-side logger captures the IDs for
                        # forensics.
                        logger.warning(
                            "apikey.update_assignment: non-admin %s "
                            "tried to set notify_recipients to "
                            "off-chain ids %r (assignment %s)",
                            current_user.id, not_allowed, assignment.id,
                        )
                        raise ForbiddenError(
                            "Non-admin callers may only set "
                            "notify_recipients to themselves or "
                            "persons in their manager chain. "
                            f"{len(not_allowed)} of {len(unique)} "
                            "selected recipient(s) are outside that "
                            "scope.",
                        )
                    # Manager-blinding guard: if the applicant has a
                    # live manager, that manager (or any further-up
                    # chain entry) MUST be present. The applicant alone
                    # is not a sufficient list — the chargeback signal
                    # has to reach beyond the spender.
                    applicant_row = chain[0] if chain else None
                    # Round-4 verification MEDIUM: only enforce the
                    # guard when there IS at least one live ancestor.
                    # If every manager in the chain has left (mass
                    # layoff / dept dissolution / chain longer than
                    # _MANAGER_CHAIN_MAX_DEPTH with only leavers),
                    # chain[1:] is empty and the guard would 403 every
                    # owner submission with no recovery path. Allow
                    # self-only in that escape-hatch case — admin can
                    # still curate via the unrestricted path.
                    non_self_chain_ids = {str(p.id) for p in chain[1:]}
                    if (
                        applicant_row is not None
                        and applicant_row.manager_person_id is not None
                        and non_self_chain_ids
                    ):
                        # ``chain`` is [applicant, manager, skip, ...].
                        # Anyone past index 0 satisfies the requirement
                        # (covers the case where the immediate manager
                        # has left and the next live ancestor is the
                        # one the user picked).
                        if not (set(unique) & non_self_chain_ids):
                            # See comment above re: not echoing IDs in
                            # the error body. The frontend already
                            # fetches the candidate list separately and
                            # can render the available manager chain.
                            raise ForbiddenError(
                                "Non-admin callers MUST include at "
                                "least one reporting-line manager in "
                                "notify_recipients (cannot route "
                                "budget alerts only to yourself — "
                                "your manager is the chargeback "
                                "signal).",
                            )

                new_recipients = unique  # may be []

            if new_recipients != assignment.notify_recipients:
                assignment.notify_recipients = new_recipients
                changed["notify_recipients"] = new_recipients

        if changed:
            await self.db.flush()
            audit = AuditService(self.db)
            await audit.log(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=current_user.id,
                action="apikey.assignment.updated",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                status="success",
                detail={
                    "request_number": assignment.request_number,
                    "previous": previous,
                    "changed": changed,
                    "actor_was_admin": is_admin,
                    "actor_was_owner": is_owner,
                },
                ip_address=ip_address,
                user_agent=user_agent,
            )
        # If ``changed`` is empty we still return the assignment
        # (idempotent no-op); skip the audit row to avoid noise.

        return await self._to_response(assignment)

    # ─── Assignment: list ───

    async def list_assignments(
        self,
        *,
        current_user: AuthUser,
        page: int = 1,
        page_size: int = 20,
        status: str | None = None,
        provider_id: UUID | None = None,
        assigned_to: UUID | None = None,
        keyword: str | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> tuple[list[ApiKeyAssignmentResponse], int]:
        # Capture the effective date once for the whole page. Otherwise a
        # request crossing Shanghai midnight could select a row with the old
        # date and serialise it with the new one, violating the status filter
        # and pagination-total contract.
        effective_today = expiry_today()
        query = select(ApiKeyAssignment)
        count_query = select(func.count()).select_from(ApiKeyAssignment)

        conditions = []
        # Read-scope: write-side admins (global_admin + asset_admin)
        # plus the governance read roles (management + auditor) see
        # every row. Everyone else is scoped to self + subordinates
        # via the manager-chain walk below.
        from app.core.rbac import (
            AUDITOR_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )
        is_admin = can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        )
        if not is_admin:
            # Non-admin scope: caller may see their own assignments AND
            # — if the requested ``assigned_to`` resolves to a direct or
            # transitive subordinate via ``IdentityPerson.manager_person_id``
            # — that subordinate's assignments too. Matches the
            # CLAUDE.md Manager-scope contract ("Manager can see all
            # subordinates' assets, accounts, and permissions"). The
            # chain walk reuses the same helper as the notify-recipient
            # picker so the depth cap + circular-reference guard apply.
            #
            # Round-2 review HIGH: when the caller passes
            # ``assigned_to=<X>`` for someone who is NOT their
            # subordinate, raise 403 instead of silently coercing the
            # query to ``self``. The prior silent-coerce behaviour was
            # a footgun: a frontend / script asking "show me X's keys"
            # would receive the CALLER's keys with no signal that the
            # filter was ignored, leading to "X has these keys" being
            # misread. 403 is the honest answer.
            if assigned_to is None or assigned_to == current_user.id:
                # Shared-access roster augment (migration 200): a
                # member sees the rows their Owner sees. The OR-IN
                # subquery is cheap on the indexed (person_id) column
                # and bounds the row count by "keys this person is on
                # the roster of", which is small in practice.
                conditions.append(
                    or_(
                        ApiKeyAssignment.assigned_to == current_user.id,
                        ApiKeyAssignment.id.in_(
                            select(ApiKeyAssignmentMember.assignment_id).where(
                                ApiKeyAssignmentMember.person_id == current_user.id,
                            )
                        ),
                    )
                )
            else:
                is_subordinate = await self._resolve_subordinate_chain(
                    current_user.id, target_id=assigned_to,
                )
                if is_subordinate:
                    conditions.append(
                        ApiKeyAssignment.assigned_to == assigned_to,
                    )
                else:
                    raise ForbiddenError(
                        "You may only list your own API key "
                        "assignments or those of a direct / transitive "
                        "subordinate.",
                    )
        elif assigned_to:
            conditions.append(ApiKeyAssignment.assigned_to == assigned_to)
        if status:
            conditions.append(
                effective_status_condition(status, today=effective_today)
            )
        if provider_id:
            conditions.append(ApiKeyAssignment.provider_id == provider_id)
        if keyword:
            pattern = f"%{_esc(keyword)}%"
            # Match assignee name/email too — admins commonly look up
            # "whose key is this?" by typing the person's name. We add
            # the join on the existing query (and count_query) only when
            # keyword is set so the default list query stays cheap.
            query = query.join(
                IdentityPerson, IdentityPerson.id == ApiKeyAssignment.assigned_to,
            )
            count_query = count_query.join(
                IdentityPerson, IdentityPerson.id == ApiKeyAssignment.assigned_to,
            )
            conditions.append(
                or_(
                    ApiKeyAssignment.request_number.ilike(pattern, escape="\\"),
                    ApiKeyAssignment.project_name.ilike(pattern, escape="\\"),
                    IdentityPerson.display_name.ilike(pattern, escape="\\"),
                    IdentityPerson.work_email.ilike(pattern, escape="\\"),
                )
            )
        if start_date:
            try:
                start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time())
            except ValueError as exc:
                raise ValidationError(
                    f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}",
                ) from exc
            conditions.append(ApiKeyAssignment.created_at >= start_dt)
        if end_date:
            try:
                end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time())
            except ValueError as exc:
                raise ValidationError(
                    f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}",
                ) from exc
            conditions.append(ApiKeyAssignment.created_at <= end_dt)

        if conditions:
            query = query.where(and_(*conditions))
            count_query = count_query.where(and_(*conditions))

        total = (await self.db.execute(count_query)).scalar() or 0
        result = await self.db.execute(
            query.order_by(ApiKeyAssignment.created_at.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        assignments = list(result.scalars().all())
        # Round-42: thread ``current_user`` so admins see autogrant
        # error blobs in ``detail.*`` (needed for troubleshooting) while
        # non-admin readers get them stripped by ``_to_response``.
        return [
            await self._to_response(
                a, current_user=current_user, effective_today=effective_today,
            )
            for a in assignments
        ], total

    async def get_my_api_keys(self, current_user: AuthUser) -> list[ApiKeyAssignmentResponse]:
        """Return all non-rejected/revoked assignments the caller can see.

        Includes both keys the caller *owns* (``assigned_to == self``)
        AND keys the caller is a *member* of via the shared-access
        roster (migration 200) — the response's ``viewer_role`` field
        tells the frontend which Tag to render in the identity column.
        Members can act on the rows they see only to the extent the
        Owner panel exposes (reveal / spend); Owner-only mutations
        (transfer / release / extend / edit / manage members) stay
        gated server-side and the frontend hides those buttons based
        on ``viewer_role``.
        """
        result = await self.db.execute(
            select(ApiKeyAssignment).where(
                and_(
                    or_(
                        ApiKeyAssignment.assigned_to == current_user.id,
                        ApiKeyAssignment.id.in_(
                            select(ApiKeyAssignmentMember.assignment_id).where(
                                ApiKeyAssignmentMember.person_id == current_user.id,
                            )
                        ),
                    ),
                    ApiKeyAssignment.status.notin_([STATUS_REJECTED, STATUS_REVOKED]),
                )
            ).order_by(ApiKeyAssignment.created_at.desc())
        )
        assignments = list(result.scalars().all())
        effective_today = expiry_today()
        return [
            await self._to_response(
                a, current_user=current_user, effective_today=effective_today,
            )
            for a in assignments
        ]

    async def get_assignment_spend(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        month: str | None = None,
    ) -> dict[str, Any]:
        """Return month-to-date cloud spend for the assignment itself.

        Routes the query through :func:`compute_assignment_spend`, which
        dispatches per upstream product:

        * **Foundry** (Azure OpenAI) — ``cmdb_cost_facts`` in the AK's
          dedicated resource group filtered to Foundry-family services.
        * **Bedrock** (AWS) — ``cmdb_cost_facts`` filtered to Bedrock
          services whose principal-tag resource_id encodes the AK's
          IAM user.
        * **Vertex AI** (GCP) — ``cmdb_cost_facts`` in the AK's GCP
          project scope filtered to the Vertex/Gen-AI service family.
        * **Bailian** (Alibaba Cloud Model Studio) — per-key cost is
          NOT available from Aliyun BSS; the response surfaces
          ``data_available=False`` with a structured caveat so the
          panel renders an honest explanation rather than an
          erroneous "$0 / over-budget" bar.

        Replaces the legacy per-applicant ``cost_owner_person_id`` sum
        (PR #240) that conflated a single AK's spend with the
        applicant's entire scope-owner fallback attribution — the
        bug that surfaced as the NexusCRM AK-20260426-204D 127.5%
        false-positive over-budget alert in May 2026.

        Auth: applicant themselves OR an asset_admin / it_admin
        (``can_manage_assets`` on role names). Anyone else: 403.

        ``daily_series`` ends at ``last_fact_date`` — trailing days
        with no cost-facts rows are omitted, NOT zero-filled (2026-05-18
        Round-32). Without this, the sparkline rendered as misleading
        "$0 spend on those last days" whenever the cost-facts ingest
        cron lagged behind today. Mid-window gaps (a single day with
        no facts between days that have facts) still zero-fill since
        chart "holes" would be worse UX than one $0 dip. The dataLag
        banner on the panel surfaces the truncation explicitly.

        ``threshold_state`` matches the cron's threshold ladder
        (80% warning / 100% over) so the UI doesn't re-derive ranges
        and risk drifting from the notification copy. When
        ``data_available=False``, ``threshold_state`` is ``unknown`` —
        the UI must render the caveat instead of a percentage bar.
        """
        # Collapse "not found" + "found but not yours" into a single
        # 403 for non-admin callers so spend-endpoint UUIDs aren't an
        # enumeration oracle. Admin keeps the distinct 404 (same
        # rationale as the reveal endpoint — see router B12 fix).
        # Read-side admin: write-side admins + management (read-only
        # leadership view) + auditor — all see anyone's spend.
        from app.core.rbac import (
            AUDITOR_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )
        is_admin = can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        )
        try:
            assignment = await self._get_assignment(assignment_id)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot view spend for an assignment you don't own",
            ) from None
        if (
            not is_admin
            and not await self.is_owner_or_member(assignment, current_user.id)
        ):
            raise ForbiddenError(
                "Cannot view spend for an assignment you don't own",
            )

        # Month-window resolution. Anchored on Asia/Shanghai (the
        # business calendar — same TZ the ``apikey.budget_check``
        # cron uses), and the ``month_end`` is clamped to
        # ``today_shanghai`` so the current-month panel and the cron's
        # threshold compare see the same right-hand boundary. Pre-fix
        # the panel ran end=last_day_of_calendar_month against
        # ``datetime.now(UTC)`` so at month boundary the two surfaces
        # could disagree (CR2 #1). Past months keep the full
        # [1st, last_day] range because ``today_shanghai`` is past the
        # last day for any prior month.
        if month:
            # Strict ``YYYY-MM`` parse — single-digit month rejected
            # so a curl/MCP caller passing ``"2026-5"`` doesn't get
            # silently accepted with display drift on the response.
            if not re.fullmatch(r"\d{4}-\d{2}", month):
                raise ValidationError(
                    f"month must be YYYY-MM, got {month!r}",
                )
            try:
                year_str, month_str = month.split("-", 1)
                year = int(year_str)
                month_num = int(month_str)
                if not 1 <= month_num <= 12:
                    raise ValueError
            except ValueError as exc:
                raise ValidationError(
                    f"month must be YYYY-MM, got {month!r}",
                ) from exc
        else:
            today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
            year, month_num = today_sh.year, today_sh.month
        month_start = date(year, month_num, 1)
        last_day = calendar.monthrange(year, month_num)[1]
        last_day_of_target = date(year, month_num, last_day)
        today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
        # Clamp to today_shanghai so the current-month series stops
        # at today (not at the future end-of-month). Past months are
        # un-affected: today_sh > last_day_of_target.
        month_end = min(last_day_of_target, today_sh)

        applicant_id = str(assignment.assigned_to)

        # Per-AK dispatch (PR #251). The legacy per-applicant
        # ``cost_owner_person_id`` sum sat here pre-fix and is what
        # produced the false-positive over-budget alerts.
        from app.domains.apikey.spend import compute_assignment_spend

        spend = await compute_assignment_spend(
            self.db,
            assignment=assignment,
            month_start=month_start,
            month_end=month_end,
        )

        mtd_usd: Decimal = spend["mtd_usd"]
        budget_usd = Decimal(str(assignment.monthly_budget_usd or 0))

        # Threshold ladder mirrors ``apikey.budget_check`` cron — both
        # surfaces route through the same ``compute_assignment_spend``
        # helper so a 79.9% bar can never coexist with an "80% threshold
        # crossed" notification. When the bucket cannot resolve a
        # number (``data_available=False``), the state is ``unknown``
        # and the UI renders the caveat instead of a percent bar.
        if not spend["data_available"]:
            pct = 0.0
            state = "unknown"
        elif budget_usd <= 0:
            pct = 0.0
            state = "ok"
        else:
            pct_raw = float(mtd_usd / budget_usd * 100)
            pct = round(pct_raw, 2)
            if pct_raw < 80:
                state = "ok"
            elif pct_raw < 100:
                state = "warning"
            else:
                state = "over"

        person = await self.db.get(IdentityPerson, assignment.assigned_to)

        return {
            "assignment_id": str(assignment.id),
            "request_number": assignment.request_number,
            "status": assignment.status,
            "month": f"{year:04d}-{month_num:02d}",
            "applicant": {
                "id": applicant_id,
                "display_name": person.display_name if person else None,
                "work_email": person.work_email if person else None,
            },
            "bucket": spend["bucket"],
            "data_available": spend["data_available"],
            "data_caveat": spend["data_caveat"],
            "data_caveat_code": spend.get("data_caveat_code"),
            "data_caveat_params": spend.get("data_caveat_params") or {},
            "budget_usd": float(budget_usd),
            "mtd_usd": float(round(mtd_usd, 2)),
            # 折前(牌价)/ 折扣后 —— 目前只有百炼这一档有值,其余桶为
            # None。预算与 ``percent`` 仍按实付算(预算管的是真花出去的
            # 钱);折前纯展示,免得一把天天在用的百炼 Key 因为代金券把
            # 实付抵到 0 而在自助页上显示成「本月 $0.00」。
            "mtd_gross_usd": _opt_money(spend.get("mtd_gross_usd")),
            "percent": pct,
            "threshold_state": state,
            "daily_series": spend["daily_series"],
            "last_fact_date": spend["last_fact_date"],
        }

    async def get_assignment_spend_detail(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
        month: str | None = None,
    ) -> dict[str, Any]:
        """Detail-page payload — same shape as :meth:`get_assignment_spend`
        plus ``monthly_series`` (6-month bar chart) and ``by_model``
        (per-meter breakdown table) for the requested month.

        The two extra reductions reuse the bucket filter recipe
        :func:`compute_assignment_spend` builds, so the detail page's
        MTD card, the model table, and the existing panel all reconcile
        to the cent. Auth + month parsing are identical to the basic
        endpoint (applicant or admin; YYYY-MM clamped to current month
        in Asia/Shanghai); see :meth:`get_assignment_spend` for the
        rationale.
        """
        # Same 404→403 collapse as ``get_assignment_spend`` — the
        # detail endpoint must not become an assignment-UUID enumeration
        # oracle either. Admin keeps the distinct 404. Read-side admin
        # includes management + auditor.
        from app.core.rbac import (
            AUDITOR_ROLE,
            MANAGEMENT_ROLE,
            has_role,
        )
        is_admin = can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        )
        try:
            assignment = await self._get_assignment(assignment_id)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot view spend for an assignment you don't own",
            ) from None
        if (
            not is_admin
            and not await self.is_owner_or_member(assignment, current_user.id)
        ):
            raise ForbiddenError(
                "Cannot view spend for an assignment you don't own",
            )

        if month:
            if not re.fullmatch(r"\d{4}-\d{2}", month):
                raise ValidationError(
                    f"month must be YYYY-MM, got {month!r}",
                )
            try:
                year_str, month_str = month.split("-", 1)
                year = int(year_str)
                month_num = int(month_str)
                if not 1 <= month_num <= 12:
                    raise ValueError
            except ValueError as exc:
                raise ValidationError(
                    f"month must be YYYY-MM, got {month!r}",
                ) from exc
        else:
            today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
            year, month_num = today_sh.year, today_sh.month
        month_start = date(year, month_num, 1)
        last_day = calendar.monthrange(year, month_num)[1]
        last_day_of_target = date(year, month_num, last_day)
        today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
        month_end = min(last_day_of_target, today_sh)

        applicant_id = str(assignment.assigned_to)

        from app.domains.apikey.spend import compute_assignment_spend_detail

        spend = await compute_assignment_spend_detail(
            self.db,
            assignment=assignment,
            month_start=month_start,
            month_end=month_end,
        )

        mtd_usd: Decimal = spend["mtd_usd"]
        budget_usd = Decimal(str(assignment.monthly_budget_usd or 0))

        if not spend["data_available"]:
            pct = 0.0
            state = "unknown"
        elif budget_usd <= 0:
            pct = 0.0
            state = "ok"
        else:
            pct_raw = float(mtd_usd / budget_usd * 100)
            pct = round(pct_raw, 2)
            if pct_raw < 80:
                state = "ok"
            elif pct_raw < 100:
                state = "warning"
            else:
                state = "over"

        person = await self.db.get(IdentityPerson, assignment.assigned_to)
        # Fetch provider name explicitly — the relationship is not
        # eager-loaded on ``_get_assignment``, so an attribute access
        # under the async session would raise MissingGreenlet.
        provider_name: str | None = (
            await self.db.execute(
                select(ApiKeyProvider.name).where(
                    ApiKeyProvider.id == assignment.provider_id,
                )
            )
        ).scalar_one_or_none()

        # Resolve the viewer's relationship to the assignment so the
        # detail page can gate the "manage members" affordance without
        # a second round-trip. Mirrors :meth:`_to_response`'s
        # precedence: Owner > admin > member > none. Backend authz
        # already gated entry here, so the only branches we can hit
        # are owner / admin / member.
        from app.core.rbac import AUDITOR_ROLE, MANAGEMENT_ROLE, has_role
        if assignment.assigned_to == current_user.id:
            viewer_role = "owner"
        elif can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        ):
            viewer_role = "admin"
        elif await self.is_member(assignment.id, current_user.id):
            viewer_role = "member"
        else:
            viewer_role = "none"

        return {
            "assignment_id": str(assignment.id),
            "request_number": assignment.request_number,
            "status": assignment.status,
            "provider_name": provider_name,
            "project_name": assignment.project_name,
            "valid_until": (
                assignment.valid_until.isoformat()
                if assignment.valid_until else None
            ),
            "month": f"{year:04d}-{month_num:02d}",
            "applicant": {
                "id": applicant_id,
                "display_name": person.display_name if person else None,
                "work_email": person.work_email if person else None,
            },
            "bucket": spend["bucket"],
            "data_available": spend["data_available"],
            "data_caveat": spend["data_caveat"],
            "data_caveat_code": spend.get("data_caveat_code"),
            "data_caveat_params": spend.get("data_caveat_params") or {},
            "budget_usd": float(budget_usd),
            "mtd_usd": float(round(mtd_usd, 2)),
            # 折前(牌价)/ 折扣后 —— 目前只有百炼这一档有值,其余桶为
            # None。预算与 ``percent`` 仍按实付算(预算管的是真花出去的
            # 钱);折前纯展示,免得一把天天在用的百炼 Key 因为代金券把
            # 实付抵到 0 而在自助页上显示成「本月 $0.00」。
            "mtd_gross_usd": _opt_money(spend.get("mtd_gross_usd")),
            "percent": pct,
            "threshold_state": state,
            "daily_series": spend["daily_series"],
            "last_fact_date": spend["last_fact_date"],
            "monthly_series": spend["monthly_series"],
            "by_model": spend["by_model"],
            "by_profile": spend.get("by_profile", []),
            # Whose total ``by_profile`` divides, and against which
            # denominator its shares were computed. Without these the UI
            # can only guess, and the wrong guess (that the rows sum to
            # ``mtd_usd``) is exactly the double-count the Bedrock
            # per-key axis is designed to prevent.
            "by_profile_scope": spend.get("by_profile_scope"),
            "by_profile_total_usd": spend.get("by_profile_total_usd"),
            "attribution_axis": spend.get("attribution_axis"),
            "viewer_role": viewer_role,
        }

    # ─── Durable expiry transition (scheduled job only) ───

    async def check_and_update_expired(self, *, today: date | None = None) -> int:
        """Mark active assignments past their valid_until as expired.

        Round-31 audit B13 (2026-05-18): every flip writes an audit row
        (``apikey.assignment.expired``, actor_type=``system``) so the
        lifecycle trail captures "when did this key cross over". Prior
        behaviour silently mutated status with no audit, leaving
        forensic-replay reading "active → ??? → expired" with no
        timestamp or initiator. The audit detail records ``valid_until``
        and the day the flip was observed so a downstream sweep can
        reconcile clock skew at year boundaries.
        """
        today = today or expiry_today()
        # ``SELECT … FOR UPDATE SKIP LOCKED`` means two concurrent cron
        # executions cannot both observe
        # the same active+past-due row and both write the audit row.
        # ``skip_locked=True`` lets the second tx see the row as
        # already-being-handled and pass; the first tx flips it +
        # writes audit. After the first tx commits, the row's status
        # is no longer ``STATUS_ACTIVE`` so subsequent calls don't
        # re-enter the loop. Bounds the audit to one row per real
        # state change, preserving the "1 row per transition"
        # invariant other domains rely on.
        result = await self.db.execute(
            select(ApiKeyAssignment).where(
                and_(
                    ApiKeyAssignment.status == STATUS_ACTIVE,
                    ApiKeyAssignment.valid_until.is_not(None),
                    ApiKeyAssignment.valid_until < today,
                )
            ).with_for_update(skip_locked=True)
        )
        flipped = list(result.scalars().all())
        if not flipped:
            return 0
        audit = AuditService(self.db)
        expired_count = 0
        for a in flipped:
            # Defensive re-check inside the locked window: ``with_for_update``
            # acquires the row lock but if another tx already flipped
            # the row mid-fetch, our snapshot may show ``STATUS_ACTIVE``
            # even though the committed value is now ``STATUS_EXPIRED``.
            # Re-reading the live attr after lock is the canonical
            # double-check.
            if a.status != STATUS_ACTIVE:
                continue
            a.status = STATUS_EXPIRED
            expired_count += 1
            # The status transition and its audit row are one transaction.
            # If audit persistence fails, the cron rolls the transaction
            # back and retries next tick; an unaudited durable flip would
            # be impossible to repair because the row is no longer active.
            await audit.log(
                trace_id=f"apikey.expiry_sweep.{a.id}",
                actor_type="system",
                actor_user_id=None,
                action="apikey.assignment.expired",
                resource_type="api_key_assignment",
                resource_id=str(a.id),
                status="success",
                detail={
                    "request_number": a.request_number,
                    "valid_until": (
                        a.valid_until.isoformat()
                        if a.valid_until else None
                    ),
                    "observed_at": today.isoformat(),
                },
            )
        await self.db.flush()
        return expired_count

    # ─── Shared-access roster (migration 200) ───
    #
    # A key's *Owner* is ``ApiKeyAssignment.assigned_to`` (the applicant).
    # *Members* are extra persons the Owner has granted read access to
    # via :meth:`add_member` — they see the same spend / detail / reveal
    # surfaces the Owner sees, but cannot perform Owner-only mutations
    # (transfer, release, revoke, extend, edit budget, manage members).
    #
    # The roster lives in ``apikey_assignment_members``. The authz
    # helper :meth:`is_owner_or_member` is the single source of truth
    # the read-side endpoints (``get_assignment_spend``,
    # ``get_assignment_spend_detail``, ``reveal_key``, ``list_assignments``,
    # ``get_my_api_keys``, ``list_members``) consult — admins keep their
    # broader access via the existing ``can_manage_assets`` / management
    # / auditor checks.

    async def is_member(
        self, assignment_id: UUID, person_id: UUID,
    ) -> bool:
        """Cheap existence check on the roster.

        Filters out leavers AND ``leaver_pending`` — both terminal
        offboarding states must lose member rights immediately, not
        wait for the next sweep. The runtime leaver gate already
        blocks their mutations; this filter closes the read window
        between the HRIS flip to ``leaver_pending`` and the eventual
        ``cleanup_leaver_memberships`` sweep that fires on the
        subsequent flip to ``left``.

        Public (no underscore) because the router-layer reveal-gate
        consults it as a defence-in-depth check before requiring
        a browser session; making it private invited the kind of
        underscore-coupling that a router refactor can silently
        miss.
        """
        result = await self.db.execute(
            select(ApiKeyAssignmentMember.id)
            .join(
                IdentityPerson,
                IdentityPerson.id == ApiKeyAssignmentMember.person_id,
            )
            .where(
                ApiKeyAssignmentMember.assignment_id == assignment_id,
                ApiKeyAssignmentMember.person_id == person_id,
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status.notin_(
                    ("left", "leaver_pending"),
                ),
            )
            .limit(1)
        )
        return result.scalar_one_or_none() is not None

    # Backwards-compatible alias for the historical underscore name —
    # remove once every call site lands on the public spelling.
    _is_member = is_member

    async def is_owner_or_member(
        self, assignment: ApiKeyAssignment, person_id: UUID,
    ) -> bool:
        """``True`` if ``person_id`` is the Owner or an active roster member.

        Single source of truth for the read-side authz gate. Admin
        callers should fall through this helper — admin paths skip
        ownership checks entirely (see e.g.
        :meth:`get_assignment_spend` which short-circuits on
        ``is_admin`` BEFORE consulting this helper).
        """
        if assignment.assigned_to == person_id:
            return True
        return await self.is_member(assignment.id, person_id)

    async def list_members(
        self,
        assignment_id: UUID,
        *,
        current_user: AuthUser,
    ) -> list[ApiKeyMemberResponse]:
        """Roster of an API Key — Owner / member / admin can view.

        The same authz gate as the read-side spend endpoints: Owner,
        any roster member, or any admin (asset / IT / management /
        auditor). Non-admin non-owner non-member callers get a 403
        framed as "you don't own this assignment" to match the
        enumeration-oracle posture of the rest of the apikey domain
        (a not-found assignment + a foreign assignment both surface
        the same response shape).
        """
        from app.core.rbac import AUDITOR_ROLE, MANAGEMENT_ROLE, has_role

        is_admin = can_manage_assets(current_user.role_names) or has_role(
            current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE,
        )
        try:
            assignment = await self._get_assignment(assignment_id)
        except NotFoundError:
            if is_admin:
                raise
            raise ForbiddenError(
                "Cannot view members of an assignment you don't own.",
            ) from None
        if not is_admin and not await self.is_owner_or_member(
            assignment, current_user.id,
        ):
            raise ForbiddenError(
                "Cannot view members of an assignment you don't own.",
            )

        rows = (
            await self.db.execute(
                select(ApiKeyAssignmentMember)
                .where(
                    ApiKeyAssignmentMember.assignment_id == assignment.id,
                )
                .order_by(ApiKeyAssignmentMember.added_at.asc())
            )
        ).scalars().all()

        # Filter out leavers + hard-resolve display fields in bulk so the
        # response shape is honest (the panel won't show a leaver row
        # whose access is silently inert at the gate). Hard-deleted
        # persons are filtered by the FK CASCADE so the row wouldn't
        # exist; left/soft-deleted persons are filtered here.
        if not rows:
            return []
        person_ids = [r.person_id for r in rows]
        added_by_ids = [r.added_by for r in rows if r.added_by is not None]
        person_map: dict[UUID, IdentityPerson] = {}
        for batch in (person_ids, added_by_ids):
            if not batch:
                continue
            res = await self.db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(batch))
            )
            for p in res.scalars():
                person_map[p.id] = p

        out: list[ApiKeyMemberResponse] = []
        for r in rows:
            person = person_map.get(r.person_id)
            if (
                person is None
                or person.deleted_at is not None
                or (person.employment_status or "active").lower() == "left"
            ):
                continue
            adder = person_map.get(r.added_by) if r.added_by else None
            out.append(ApiKeyMemberResponse(
                id=r.id,
                assignment_id=r.assignment_id,
                person_id=r.person_id,
                display_name=person.display_name,
                work_email=person.work_email,
                added_by=r.added_by,
                added_by_name=(adder.display_name if adder else None),
                added_at=r.added_at,
            ))
        return out

    async def add_member(
        self,
        assignment_id: UUID,
        *,
        person_id: UUID,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> ApiKeyMemberResponse:
        """Owner adds a teammate to the roster.

        Authz: assignment Owner only — NOT admins. Admins manage keys
        through ``/admin/assets/api-keys`` which has its own audited
        rebind / revoke / IAM tooling; the member roster is the Owner's
        team-management surface and intentionally stays out of admin
        scope to keep the access-grant semantics auditable as "Owner-
        granted". The escape valve for "admin needs to grant access on
        behalf of a frozen Owner" is to transfer the key (which
        auto-promotes the recipient to Owner and adds the old Owner
        as a member — see :meth:`apply_transfer`).

        Idempotent: a second add for the same (assignment, person)
        returns the existing row without writing a duplicate audit row
        or re-firing the inbox notification. The UNIQUE constraint
        backstops a concurrent double-click race.

        Validates the target person:

        * Exists, soft-delete + leaver guard: ``deleted_at IS NULL`` +
          ``employment_status != 'left'``. Granting access to a leaver
          is exactly what the offboarding flow exists to prevent.
        * Not the Owner themselves: a self-add row would surface as a
          duplicate in the panel and confuse the "Owner / Member"
          identity Tag on ``/me/api-keys``.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment Owner may manage members.",
            )
        effective_status = effective_assignment_status(
            assignment.status, assignment.valid_until, today=expiry_today(),
        )
        if effective_status in (STATUS_REVOKED, STATUS_EXPIRED):
            raise AppException(
                code="INVALID_STATUS_FOR_MEMBER_ADD",
                message=(
                    f"Cannot add members to an {effective_status} API Key. "
                    "The credential is no longer live."
                ),
                status_code=400,
            )

        if person_id == current_user.id:
            raise AppException(
                code="CANNOT_ADD_SELF",
                message=(
                    "You are already the Owner of this key; no need to "
                    "add yourself as a member."
                ),
                status_code=400,
            )

        target = await self.db.get(IdentityPerson, person_id)
        if (
            target is None
            or target.deleted_at is not None
            or (target.employment_status or "active").lower() in (
                "left", "leaver_pending",
            )
        ):
            # Reject both terminal offboarding states up-front. A
            # person already in ``leaver_pending`` has the offboarding
            # provisioning queued; granting them member access between
            # the HRIS flip and the eventual flip-to-left would let
            # the leaver-sweep miss them. Mirrors the
            # ``apply_transfer`` recipient guard.
            raise AppException(
                code="MEMBER_NOT_ACTIVE",
                message=(
                    "Cannot add this person as a member: they are not an "
                    "active employee."
                ),
                status_code=400,
            )

        existing = (
            await self.db.execute(
                select(ApiKeyAssignmentMember).where(
                    ApiKeyAssignmentMember.assignment_id == assignment.id,
                    ApiKeyAssignmentMember.person_id == person_id,
                )
            )
        ).scalar_one_or_none()
        if existing is not None:
            # Idempotent — return existing row, no double-notify.
            adder = (
                await self.db.get(IdentityPerson, existing.added_by)
                if existing.added_by else None
            )
            return ApiKeyMemberResponse(
                id=existing.id,
                assignment_id=existing.assignment_id,
                person_id=existing.person_id,
                display_name=target.display_name,
                work_email=target.work_email,
                added_by=existing.added_by,
                added_by_name=adder.display_name if adder else None,
                added_at=existing.added_at,
            )

        # Concurrent-add race: two simultaneous POSTs both see
        # ``existing is None`` (the assignment row lock doesn't gap-
        # lock the missing child row on InnoDB / TiDB), both reach
        # the INSERT, and the second hits the UNIQUE constraint with
        # an IntegrityError that would surface as a 500 instead of
        # the documented "idempotent on duplicate" contract. Catch +
        # rollback + re-read so the second caller gets the
        # first call's row back.
        from sqlalchemy.exc import IntegrityError

        row = ApiKeyAssignmentMember(
            assignment_id=assignment.id,
            person_id=person_id,
            added_by=current_user.id,
            added_at=datetime.now(UTC),
        )
        self.db.add(row)
        try:
            await self.db.flush()
        except IntegrityError:
            # Concurrent insert won the race; re-read its row and
            # treat as the "idempotent" path. The audit row + notify
            # below will be skipped via the early-return below.
            await self.db.rollback()
            existing = (
                await self.db.execute(
                    select(ApiKeyAssignmentMember).where(
                        ApiKeyAssignmentMember.assignment_id == assignment.id,
                        ApiKeyAssignmentMember.person_id == person_id,
                    )
                )
            ).scalar_one_or_none()
            if existing is None:
                # Concurrent row gone between flush and re-read (e.g.
                # another concurrent DELETE) — surface a clean retry
                # rather than masking with a 500.
                raise AppException(
                    code="MEMBER_ADD_RACE",
                    message=(
                        "Concurrent membership change collided with this "
                        "add. Retry the request."
                    ),
                    status_code=409,
                ) from None
            adder = (
                await self.db.get(IdentityPerson, existing.added_by)
                if existing.added_by else None
            )
            return ApiKeyMemberResponse(
                id=existing.id,
                assignment_id=existing.assignment_id,
                person_id=existing.person_id,
                display_name=target.display_name,
                work_email=target.work_email,
                added_by=existing.added_by,
                added_by_name=adder.display_name if adder else None,
                added_at=existing.added_at,
            )

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.member.add",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "member_person_id": str(person_id),
                "member_email": target.work_email,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        # Best-effort inbox notification — never blocks the add. A
        # downstream notify outage must not roll back the membership
        # write; the row is already authoritative and the next access
        # attempt will succeed.
        try:
            await self._notify_member_added(
                assignment=assignment,
                target=target,
                owner_user=current_user,
                trace_id=trace_id,
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "apikey.member.add notify failed (%s → %s): %s",
                assignment.id, person_id, exc,
            )

        adder_self = current_user.display_name
        return ApiKeyMemberResponse(
            id=row.id,
            assignment_id=row.assignment_id,
            person_id=row.person_id,
            display_name=target.display_name,
            work_email=target.work_email,
            added_by=row.added_by,
            added_by_name=adder_self,
            added_at=row.added_at,
        )

    async def remove_member(
        self,
        assignment_id: UUID,
        *,
        person_id: UUID,
        current_user: AuthUser,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> None:
        """Owner removes a teammate from the roster.

        Authz mirror of :meth:`add_member`. Idempotent on a non-existent
        row — no audit, no notify; the panel just refreshes empty. The
        ex-member's next reveal / spend call will 403.
        """
        assignment = await self._get_assignment(assignment_id, lock=True)
        if assignment.assigned_to != current_user.id:
            raise ForbiddenError(
                "Only the assignment Owner may manage members.",
            )

        row = (
            await self.db.execute(
                select(ApiKeyAssignmentMember).where(
                    ApiKeyAssignmentMember.assignment_id == assignment.id,
                    ApiKeyAssignmentMember.person_id == person_id,
                )
            )
        ).scalar_one_or_none()
        if row is None:
            return  # idempotent

        target = await self.db.get(IdentityPerson, person_id)
        await self.db.delete(row)
        await self.db.flush()

        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.member.remove",
            resource_type="api_key_assignment",
            resource_id=str(assignment.id),
            status="success",
            detail={
                "request_number": assignment.request_number,
                "member_person_id": str(person_id),
                "member_email": target.work_email if target else None,
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )

        if target is not None and (
            target.deleted_at is None
            and (target.employment_status or "active").lower() != "left"
        ):
            try:
                await self._notify_member_removed(
                    assignment=assignment,
                    target=target,
                    owner_user=current_user,
                    trace_id=trace_id,
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "apikey.member.remove notify failed (%s ← %s): %s",
                    assignment.id, person_id, exc,
                )

    async def _add_member_internal(
        self,
        *,
        assignment: ApiKeyAssignment,
        person_id: UUID,
        added_by_id: UUID | None,
    ) -> None:
        """Insert a member row without the Owner-authz gate.

        Used by :meth:`apply_transfer` to auto-promote the previous
        Owner to member after a transfer accept. Skips inactive
        persons (leaver / deleted) so a transfer triggered by an
        offboarding flow doesn't leave a leaver on the new Owner's
        roster.

        Idempotent on duplicate insertion — silently skips on
        UNIQUE-violation rather than raising.
        """
        target = await self.db.get(IdentityPerson, person_id)
        if (
            target is None
            or target.deleted_at is not None
            or (target.employment_status or "active").lower() == "left"
        ):
            return
        existing = (
            await self.db.execute(
                select(ApiKeyAssignmentMember.id).where(
                    ApiKeyAssignmentMember.assignment_id == assignment.id,
                    ApiKeyAssignmentMember.person_id == person_id,
                )
            )
        ).scalar_one_or_none()
        if existing is not None:
            return
        row = ApiKeyAssignmentMember(
            assignment_id=assignment.id,
            person_id=person_id,
            added_by=added_by_id,
            added_at=datetime.now(UTC),
        )
        self.db.add(row)
        await self.db.flush()

    async def _notify_member_added(
        self,
        *,
        assignment: ApiKeyAssignment,
        target: IdentityPerson,
        owner_user: AuthUser,
        trace_id: str,
    ) -> None:
        """In-app + Feishu push: "you were added to API Key X as a member".

        Best-effort — caller catches and logs. Mirrors the existing
        release / transfer notification shape so the user's inbox is
        visually consistent.
        """
        from app.integrations.service import get_platform_setting
        from app.notifications.service import (
            NotificationEnvelope,
            NotificationService,
        )

        provider = await self.db.get(ApiKeyProvider, assignment.provider_id)
        provider_name = provider.name if provider else "(unknown provider)"
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="(未配置门户地址)",
            session=self.db,
        )
        detail_url = f"{portal_url}/me/api-keys/{assignment.id}"
        owner_name = owner_user.display_name or owner_user.work_email

        notifier = NotificationService(self.db)
        await notifier.dispatch(
            NotificationEnvelope(
                recipient=target,
                title=f"你被加入 API Key 共享名单:{provider_name}",
                body_text=(
                    f"{owner_name} 把你加入了 API Key 的共享访问名单"
                    f"(项目:{assignment.project_name},"
                    f"单号:{assignment.request_number})。\n"
                    "你现在可以查看该 Key 的本月用量、明细与密钥本身,"
                    "用于团队协作。\n"
                    f"查看详情:{detail_url}"
                ),
                channels=["feishu", "email"],
                template_key="apikey.member.added",
                url=detail_url,
                related_resource_type="api_key_assignment",
                related_resource_id=assignment.id,
                context={
                    "owner_name": owner_name,
                    "provider_name": provider_name,
                    "project_name": assignment.project_name,
                    "request_number": assignment.request_number,
                    "detail_url": detail_url,
                },
            ),
            trace_id=trace_id,
        )

    async def _notify_member_removed(
        self,
        *,
        assignment: ApiKeyAssignment,
        target: IdentityPerson,
        owner_user: AuthUser,
        trace_id: str,
    ) -> None:
        from app.integrations.service import get_platform_setting
        from app.notifications.service import (
            NotificationEnvelope,
            NotificationService,
        )

        provider = await self.db.get(ApiKeyProvider, assignment.provider_id)
        provider_name = provider.name if provider else "(unknown provider)"
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="(未配置门户地址)",
            session=self.db,
        )
        my_url = f"{portal_url}/me/api-keys"
        owner_name = owner_user.display_name or owner_user.work_email

        notifier = NotificationService(self.db)
        await notifier.dispatch(
            NotificationEnvelope(
                recipient=target,
                title=f"已退出 API Key 共享名单:{provider_name}",
                body_text=(
                    f"{owner_name} 把你从 API Key 的共享访问名单中移除"
                    f"(项目:{assignment.project_name},"
                    f"单号:{assignment.request_number})。\n"
                    "你将无法再查看该 Key 的用量与密钥。"
                    "如有问题请直接联系 Key 的负责人。\n"
                    f"查看我的 API Key:{my_url}"
                ),
                channels=["feishu", "email"],
                template_key="apikey.member.removed",
                url=my_url,
                related_resource_type="api_key_assignment",
                related_resource_id=assignment.id,
                context={
                    "owner_name": owner_name,
                    "provider_name": provider_name,
                    "project_name": assignment.project_name,
                    "request_number": assignment.request_number,
                    "my_url": my_url,
                },
            ),
            trace_id=trace_id,
        )

    async def cleanup_leaver_memberships(
        self,
        *,
        person_id: UUID,
        trace_id: str,
        actor_user_id: UUID | None,
        via: str,
    ) -> int:
        """Sweep every roster row for a person being flipped to ``left``.

        Called from :class:`LeaverBlockedGuard.flip_to_left` so the
        leaver loses every shared-access surface the moment HRIS marks
        them as gone. Mirrors the existing
        :func:`strip_person_from_alert_cc` sweep (cloud-billing cc
        scrub) in posture — best-effort, audited, idempotent.

        Returns the number of rows removed (zero on a clean leaver
        with no shared-access entries).
        """
        rows = (
            await self.db.execute(
                select(ApiKeyAssignmentMember).where(
                    ApiKeyAssignmentMember.person_id == person_id,
                )
            )
        ).scalars().all()
        if not rows:
            return 0

        # Audit-write per-assignment so a forensic query against
        # ``apikey.*`` actions on assignment X surfaces the leaver
        # removal. Keep the loop tight — the worst-case row count is
        # tens (a single person is rarely on > 20 keys).
        audit = AuditService(self.db)
        for r in rows:
            await audit.log(
                trace_id=trace_id,
                actor_type="user" if actor_user_id else "system",
                actor_user_id=actor_user_id,
                action="apikey.member.leaver_strip",
                resource_type="api_key_assignment",
                resource_id=str(r.assignment_id),
                status="success",
                detail={
                    "member_person_id": str(person_id),
                    "via": via,
                },
            )
            await self.db.delete(r)
        await self.db.flush()
        return len(rows)

    # ─── Private helpers ───

    async def _get_provider(self, provider_id: UUID) -> ApiKeyProvider:
        result = await self.db.execute(
            select(ApiKeyProvider).where(ApiKeyProvider.id == provider_id)
        )
        p = result.scalar_one_or_none()
        if not p:
            raise NotFoundError("ApiKeyProvider", str(provider_id))
        return p

    async def _get_assignment(
        self, assignment_id: UUID, *, lock: bool = False,
    ) -> ApiKeyAssignment:
        # ``lock=True`` adds SELECT ... FOR UPDATE so callers that mutate
        # status / extended_count / valid_until (revoke_key,
        # extend_validity) serialise concurrent calls and don't race the
        # status guard. Without it, two parallel extend calls both
        # observe extended_count=N and both flip to N+1, breaking the
        # "extends performed" invariant.
        stmt = select(ApiKeyAssignment).where(ApiKeyAssignment.id == assignment_id)
        if lock:
            # populate_existing refreshes from the locked row in case the
            # assignment is already cached in this session — otherwise the
            # FOR UPDATE locks but hands back the stale instance and the
            # status / extended_count guards race (identity-map trap).
            stmt = stmt.with_for_update().execution_options(populate_existing=True)
        result = await self.db.execute(stmt)
        a = result.scalar_one_or_none()
        if not a:
            raise NotFoundError("ApiKeyAssignment", str(assignment_id))
        return a

    async def _to_response(
        self,
        assignment: ApiKeyAssignment,
        *,
        current_user: AuthUser | None = None,
        provider: ApiKeyProvider | None = None,
        effective_today: date | None = None,
    ) -> ApiKeyAssignmentResponse:
        if not provider:
            provider = await self._get_provider(assignment.provider_id)

        # Load assignee info
        assignee_name = None
        assignee_email = None
        assignee_employee_no = None
        assignee_employment_status = None
        if current_user and current_user.id == assignment.assigned_to:
            assignee_name = current_user.display_name
            assignee_email = current_user.work_email
            assignee_employee_no = current_user.employee_no
            assignee_employment_status = current_user.employment_status
        else:
            from app.auth.models import AuthUser as AU
            result = await self.db.execute(select(AU).where(AU.id == assignment.assigned_to))
            user = result.scalar_one_or_none()
            if user:
                assignee_name = user.display_name
                assignee_email = user.work_email
                assignee_employee_no = user.employee_no
                assignee_employment_status = user.employment_status

        # Mask key
        #
        # Round-31 audit S5 (2026-05-18): tail-only masking. The prior
        # ``kv[:4] + "..." + kv[-4:]`` leaked the head of cloud-key
        # formats (AKIA / sk-ant- / sk-... prefixes are well-known and
        # combined with the 4-char tail give an attacker a recognisable
        # provider signal + a uniqueness fingerprint without ever
        # calling ``reveal``). Switch to ``**** + last-4`` (AWS console
        # / industry convention).
        #
        # S4 (decrypt fail-loud): a tampered or AAD-mismatched row
        # surfaces as ``key_masked=None`` (UI shows "key set but cannot
        # render"). Prior behaviour silently sliced the raw ciphertext
        # envelope into the mask field, leaking ``{"e..."`` into list
        # responses. The error is logged at WARNING for operator
        # follow-up — see ``_decrypt_key_value`` for the failure
        # taxonomy.
        key_masked = None
        has_key = bool(assignment.key_value)
        if assignment.key_value:
            try:
                kv = _decrypt_key_value(
                    assignment.key_value, assignment_id=assignment.id,
                )
            except AppException as exc:
                if exc.code == "KEY_DECRYPT_FAILED":
                    kv = None
                    logger.warning(
                        "apikey._to_response: rendering assignment %s "
                        "with key_masked=None (decrypt failed)",
                        assignment.id,
                    )
                else:
                    raise
            if kv is not None:
                if len(kv) >= 4:
                    key_masked = "****" + kv[-4:]
                else:
                    key_masked = "****"

        # Expiry alert
        days_until = None
        expiry_alert = None
        today = effective_today or expiry_today()
        if assignment.valid_until:
            days_until = (assignment.valid_until - today).days
            if days_until < 0:
                expiry_alert = "expired"
            elif days_until <= 10:
                expiry_alert = "10d"
            elif days_until <= 20:
                expiry_alert = "20d"
            elif days_until <= 30:
                expiry_alert = "30d"

        # Approval instance code from detail
        feishu = (assignment.detail or {}).get("feishu") or {}
        instance_code = feishu.get("instance_code")

        # ── IAM binding display fields ──
        # Pull denormalised label off the assignment row (set by
        # auto-grant + rebind) and JOIN the latest IAM user / integration
        # rows for the human-readable name. The assignment table already
        # caches user_arn so the JOIN can fail (IAM user deleted) without
        # losing context.
        iam_user_name: str | None = None
        iam_integration_name: str | None = None
        if assignment.iam_user_id is not None:
            from app.domains.cmdb.models import CmdbIamUser
            iam_row = await self.db.get(CmdbIamUser, assignment.iam_user_id)
            if iam_row is not None and iam_row.deleted_at is None:
                iam_user_name = iam_row.user_name
        if assignment.iam_integration_id is not None:
            from app.integrations.models import Integration
            integ = await self.db.get(Integration, assignment.iam_integration_id)
            if integ is not None:
                iam_integration_name = integ.name

        # Cloud-scope binding display field. Joined for label only;
        # missing scope (admin manually deleted post-binding) renders
        # as None which the UI treats as "binding present but scope
        # unresolved" — same pattern as iam_user above.
        cloud_scope_display: str | None = None
        if assignment.cloud_scope_id is not None:
            from app.domains.cmdb.models import CmdbCloudScope
            scope_row = await self.db.get(
                CmdbCloudScope, assignment.cloud_scope_id,
            )
            if scope_row is not None:
                cloud_scope_display = (
                    scope_row.display_name or scope_row.external_id
                )

        # Strip sensitive Feishu instance payload from the API
        # response. Round-20 audit (A2F7): ``feishu.latest_instance_detail``
        # carries every approver's name + comments + the full submitted
        # form JSON, which leaks through every list/get endpoint —
        # including the admin list which is the union of every
        # employee's payload, so an approver comment ("approving against
        # my better judgment ...") becomes visible to anyone who can
        # read /admin/assets/api-keys. The structured ``feishu.instance_code``
        # + ``feishu.instance_status`` are enough for the UI to render
        # status and the link to the Feishu approval card.
        sanitised_detail: dict[str, Any] | None = None
        if assignment.detail:
            sanitised_detail = dict(assignment.detail)
            feishu_blob = sanitised_detail.get("feishu")
            if isinstance(feishu_blob, dict):
                redacted_feishu = {
                    k: v for k, v in feishu_blob.items()
                    if k != "latest_instance_detail"
                }
                sanitised_detail["feishu"] = redacted_feishu
            # Round-42 audit: the autogrant blobs (``bedrock`` / ``bailian``
            # / ``azure_openai``) persist verbatim cloud-SDK error strings
            # into ``detail.*.error``. These leak through every list/get
            # endpoint to the assignment owner, who has no operational
            # use for them. SDK errors can include the master integration
            # SP ARN, internal integration UUIDs, ``InvalidTag`` hex
            # fragments from credential-decrypt failures, and tenant
            # identifiers. Strip the ``error`` (and the admin-resolved
            # ``*_attempted`` / integration-id companions) for non-admin
            # readers; admins still see the full payload because they
            # call ``_to_response`` with their own ``current_user`` and
            # the role check below short-circuits.
            #
            # Behaviour when ``current_user is None`` (background dispatch,
            # MCP tools that don't carry a user): redact by default —
            # safer to strip than to leak. The admin UI always passes
            # ``current_user``.
            # Strip the ENTIRE per-provider ``rotation`` sub-blob from
            # every response, for every reader (admin included). It holds
            # the stashed prior credential ciphertext (``prev_key_value``
            # — must never ride out) plus provider handles / initiator /
            # timestamps that no client needs: the UI reads the top-level
            # ``rotation_status`` + ``rotation_retire_at`` fields instead
            # (set below), and ``detect_rotation_provider`` keys off
            # ``granted`` not ``rotation``. Removing the whole sub-blob is
            # both safer (no secret/handle leakage to roster members) and
            # simpler than field-level filtering.
            for blob_key in ("bedrock", "bailian", "azure_openai"):
                blob = sanitised_detail.get(blob_key)
                if isinstance(blob, dict) and "rotation" in blob:
                    sanitised_detail[blob_key] = {
                        k: v for k, v in blob.items() if k != "rotation"
                    }
            is_admin_reader = (
                current_user is not None
                and can_manage_assets(current_user.role_names)
            )
            if not is_admin_reader:
                for blob_key in ("bedrock", "bailian", "azure_openai"):
                    blob = sanitised_detail.get(blob_key)
                    if isinstance(blob, dict):
                        sanitised_detail[blob_key] = {
                            k: v for k, v in blob.items()
                            if k not in (
                                "error",
                                "subscription_id_attempted",
                                "integration_id",
                                "integration_slug",
                            )
                        }
        # Shared-access roster (migration 200). Always denormalised
        # onto the response so the detail page / panel can render
        # without a second round-trip. Filters out leavers + deleted
        # — see :meth:`list_members` for the canonical version of
        # this filter; we replicate it inline (rather than call
        # ``list_members``) because the per-row admin / management /
        # auditor read paths land here with their own ``current_user``
        # and we don't want to re-do the authz gate.
        members_payload: list[dict[str, Any]] = []
        member_person_ids: set[UUID] = set()
        member_rows = (
            await self.db.execute(
                select(ApiKeyAssignmentMember)
                .where(ApiKeyAssignmentMember.assignment_id == assignment.id)
                .order_by(ApiKeyAssignmentMember.added_at.asc())
            )
        ).scalars().all()
        if member_rows:
            pids = [r.person_id for r in member_rows]
            pmap: dict[UUID, IdentityPerson] = {}
            res = await self.db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(pids))
            )
            for p in res.scalars():
                pmap[p.id] = p
            for r in member_rows:
                p = pmap.get(r.person_id)
                if (
                    p is None
                    or p.deleted_at is not None
                    or (p.employment_status or "active").lower() == "left"
                ):
                    continue
                member_person_ids.add(r.person_id)
                members_payload.append({
                    "id": str(r.id),
                    "person_id": str(r.person_id),
                    "display_name": p.display_name,
                    "work_email": p.work_email,
                    "added_at": (
                        r.added_at.isoformat() if r.added_at else None
                    ),
                })

        # ``viewer_role`` is server-side resolved so the frontend
        # doesn't have to re-derive it from the role list + the
        # member roster. Owner > admin > member > none (mutually
        # exclusive precedence; Owner who is also admin renders as
        # Owner because that's the most relevant identity for the
        # page's action gating).
        viewer_role: str | None = None
        if current_user is not None:
            if current_user.id == assignment.assigned_to:
                viewer_role = "owner"
            elif can_manage_assets(current_user.role_names):
                viewer_role = "admin"
            else:
                # Avoid a second roster query when we just materialised
                # the list above — check the in-memory set.
                if current_user.id in member_person_ids:
                    viewer_role = "member"
                else:
                    from app.core.rbac import (
                        AUDITOR_ROLE,
                        MANAGEMENT_ROLE,
                        has_role,
                    )
                    if has_role(
                        current_user.role_names,
                        MANAGEMENT_ROLE, AUDITOR_ROLE,
                    ):
                        viewer_role = "admin"
                    else:
                        viewer_role = "none"

        # ── Rotation state (migration 401) ──
        # Surface a compact rotation status + the retire timestamp so the
        # UI can render the "轮换中 · 旧 key X 停用" badge and the
        # complete / cancel actions without parsing the detail blob.
        from app.domains.apikey.rotation import get_active_rotation
        rotation_kind, rotation_blob = get_active_rotation(assignment)
        rotation_status = rotation_blob.get("status") if rotation_kind else None

        return ApiKeyAssignmentResponse(
            id=assignment.id,
            provider_id=assignment.provider_id,
            provider_name=provider.name,
            assigned_to=assignment.assigned_to,
            assigned_to_name=assignee_name,
            assigned_to_email=assignee_email,
            assigned_to_employee_no=assignee_employee_no,
            assigned_to_employment_status=assignee_employment_status,
            status=effective_assignment_status(
                assignment.status, assignment.valid_until, today=today,
            ),
            request_number=assignment.request_number,
            project_name=assignment.project_name,
            reason=assignment.reason,
            monthly_budget_usd=assignment.monthly_budget_usd,
            expected_usage=assignment.expected_usage,
            key_masked=key_masked,
            has_key=has_key,
            valid_from=assignment.valid_from,
            valid_until=assignment.valid_until,
            days_until_expiry=days_until,
            expiry_alert=expiry_alert,
            extended_count=assignment.extended_count,
            approval_instance_code=instance_code,
            detail=sanitised_detail,
            iam_user_id=assignment.iam_user_id,
            iam_user_arn=assignment.iam_user_arn,
            iam_user_name=iam_user_name,
            iam_integration_id=assignment.iam_integration_id,
            iam_integration_name=iam_integration_name,
            cloud_scope_id=assignment.cloud_scope_id,
            cloud_scope_display_name=cloud_scope_display,
            notify_recipients=(
                list(assignment.notify_recipients)
                if assignment.notify_recipients is not None
                else None
            ),
            revoked_at=assignment.revoked_at,
            revoked_by=assignment.revoked_by,
            revoked_by_name=await self._resolve_person_display_name(
                assignment.revoked_by,
            ),
            revoke_reason_code=assignment.revoke_reason_code,
            revoke_reason_note=assignment.revoke_reason_note,
            revoke_actor_role=assignment.revoke_actor_role,
            transferred_from=assignment.transferred_from,
            transferred_from_name=await self._resolve_person_display_name(
                assignment.transferred_from,
            ),
            members=members_payload,
            viewer_role=viewer_role,
            rotation_status=rotation_status,
            rotation_retire_at=assignment.rotation_retire_at,
            created_at=assignment.created_at,
            updated_at=assignment.updated_at,
        )

    async def _resolve_person_display_name(
        self, person_id: UUID | None,
    ) -> str | None:
        """Cheap lookup for the admin table's "revoked by" column.

        Returns None when the person row is missing (deleted) — the UI
        falls back to showing the raw UUID. We don't cache here; list
        responses already loop through ``_to_response`` once per row, so
        adding a batch-N lookup is a future micro-optimisation.
        """
        if person_id is None:
            return None
        person = await self.db.get(IdentityPerson, person_id)
        if person is None:
            return None
        return person.display_name or person.work_email

    @staticmethod
    def _generate_request_number() -> str:
        return f"AK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:4].upper()}"

    @staticmethod
    def _map_feishu_status(status: str | None) -> str:
        normalized = (status or "PENDING").lower()
        if normalized in {"approved", "pass", "approved_status"}:
            return "approved"
        if normalized in {"rejected", "reject", "refused"}:
            return "rejected"
        if normalized in {"canceled", "cancelled", "deleted"}:
            return "cancelled"
        return "pending"

    # ─── Feishu form helpers ───

    # Keyword fragments that map a Feishu widget name to the structured
    # form field. Chinese + English aliases — the Feishu approval form
    # usually uses Chinese names ("API Key 来源", "项目名称", etc.).
    _FIELD_KEYWORDS = {
        "source":  ["来源", "provider", "source", "供应商", "服务商"],
        "project": ["项目", "project"],
        "reason":  ["理由", "原因", "reason", "说明", "用途描述"],
        "budget":  ["预算", "额度", "budget", "monthly", "amount"],
        "usage":   ["用量", "用途", "usage", "使用场景", "场景"],
    }

    # 2026-05 → 2026-08-14 this was ``{"dateInterval"}``: Feishu's
    # /v4/instances rejected every shape tried back then, and the API
    # accepts submissions that simply omit the widget even when it's
    # marked ``required:true``. Cost: the ``AI API KEY Apply`` card's
    # required DateInterval widget (widget17763462677850001) reached
    # every approver EMPTY — the validity window the requester picked
    # existed only on the Helix row (valid_from / valid_until).
    #
    # The window now ships via
    # ``feishu_widget_values.date_interval_wire_value`` (shape read off
    # a real Feishu-native instance; see that module). Nothing is
    # unconditionally skipped any more — an unusable range still
    # degrades to "widget omitted" rather than blocking the submit,
    # and ``submit_with_durable_intent`` resubmits without the widget
    # if Feishu rejects the value.
    _SKIP_WIDGET_TYPES: set[str] = set()

    @classmethod
    def _auto_fill_answers(
        cls,
        answers: dict[str, Any],
        widgets: list[dict[str, Any]],
        provider: ApiKeyProvider,
        body: ApiKeyRequestSubmit,
    ) -> None:
        """Auto-fill widget values from the structured form data.

        Maps each required Feishu widget to a field on the submitted
        ``ApiKeyRequestSubmit`` by matching Chinese / English keyword
        fragments in the widget's display name. Users never see these
        widgets individually — they fill a single structured form on
        ``/my/requests`` and the backend reshapes the submission to fit
        whatever the current Feishu approval definition asks for.
        """

        def _match(name: str, category: str) -> bool:
            return any(k in name for k in cls._FIELD_KEYWORDS[category])

        for w in widgets:
            wid = w.get("id", "")
            wtype = w.get("type", "")
            raw_name = w.get("name") or ""
            name = raw_name.lower()
            if wtype == "dateInterval":
                # The key's validity window (2026-08-14). Keyword
                # matching can't reach this widget — the Feishu
                # template names it the literal string "DateInterval"
                # in both locales — so it is filled by TYPE, from the
                # same ``valid_from`` / ``valid_until`` the assignment
                # row stores. Serialisation happens in
                # ``_serialize_feishu_form``.
                #
                # **Unconditional overwrite**, above the "already
                # answered" guard, mirroring the Payment Information
                # trust boundary in the asset flow: ``body.answers`` is
                # client-controlled, and the window on the approval
                # card MUST be the window that will actually be
                # granted. A client-supplied interval would otherwise
                # show the approver "1 week" while the assignment row
                # (and the expiry cron) carried a year.
                if body.valid_from and body.valid_until:
                    answers[wid] = {
                        "start": body.valid_from,
                        "end": body.valid_until,
                    }
                else:
                    answers.pop(wid, None)
                continue
            if answers.get(wid) not in (None, "", [], {}):
                continue
            if wtype in cls._SKIP_WIDGET_TYPES:
                continue  # handled by _serialize_feishu_form (omitted)

            if _match(name, "source"):
                answers[wid] = provider.name
            elif _match(name, "project"):
                answers[wid] = body.project_name
            elif _match(name, "reason"):
                answers[wid] = body.reason
            elif _match(name, "budget") or wtype == "amount":
                # Feishu amount widget wants a plain number string
                # (not a {"amount":..,"currency":..} dict — that shape
                # is rejected with "控件值不合法或者为空"). The currency
                # is controlled by the widget's `currencyRange` option,
                # not the submitted value.
                answers[wid] = str(body.monthly_budget_usd)
            elif _match(name, "usage"):
                answers[wid] = body.expected_usage or ""

    @staticmethod
    def _parse_template_widgets(raw_form: str | list[dict[str, Any]]) -> list[dict[str, Any]]:
        if isinstance(raw_form, str):
            return json.loads(raw_form or "[]")
        return raw_form

    @staticmethod
    def _serialize_feishu_form(widgets: list[dict[str, Any]], answers: dict[str, Any]) -> str:
        # Honour Feishu ``display_condition`` so a conditional
        # ``required`` widget doesn't 400 a submit whose trigger isn't
        # met. Same shared evaluator the other Feishu-backed flows
        # use — see ``app/integrations/feishu_form_visibility.py``.
        # ``answers`` is widget-id keyed here (the auto-fill helper
        # writes by id), so trigger comparisons resolve correctly
        # against the actual submitted radio values.
        from app.integrations.feishu_form_visibility import (
            evaluate_display_conditions,
        )
        visibility = evaluate_display_conditions(widgets, answers)
        serialized: list[dict[str, Any]] = []
        for widget in widgets:
            wtype = widget.get("type", "input")
            if wtype == "text":
                continue
            # Widgets whose Feishu validation we can't satisfy reliably
            # are omitted; Feishu accepts the instance and the approver
            # can fill them from their side if needed.
            if wtype in ApiKeyService._SKIP_WIDGET_TYPES:
                continue
            wid = widget["id"]
            if not visibility.get(wid, True):
                continue
            if wtype == "dateInterval":
                # Native object, not a stringified one — see
                # ``feishu_widget_values.date_interval_wire_value``.
                # A half-filled / unparseable range serialises to ``""``
                # and the widget is dropped instead of raising
                # APPROVAL_FIELD_REQUIRED: Feishu accepts a submission
                # that omits it, and a request nobody can file is worse
                # than a card missing its dates.
                interval_value = date_interval_wire_value(answers.get(wid))
                if interval_value in (None, "", {}):
                    continue
                serialized.append({"id": wid, "type": wtype, "value": interval_value})
                continue
            answer = answers.get(wid)
            if answer in (None, "", [], {}):
                if widget.get("required"):
                    raise AppException(
                        code="APPROVAL_FIELD_REQUIRED",
                        message=f"Missing required approval field: {widget.get('name')}",
                        status_code=400,
                    )
                continue
            value = str(answer) if not isinstance(answer, (dict, list)) else json.dumps(answer, ensure_ascii=False)
            serialized.append({"id": wid, "type": wtype, "value": value})
        return json.dumps(serialized, ensure_ascii=False)