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

Covers the three providers (Bedrock IAM / Bailian sk- / Azure OpenAI
key slot), the grace-period state machine, force-complete, cancel,
the retire cron sweep, and the guardrails (in-flight lock, non-active
status, non-rotatable provider, grace validation, owner-only).

No cloud is hit — the boto AWSProvider methods, the BailianClient, and
the Azure driver are monkeypatched to canned responses.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException, ForbiddenError
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.apikey.service import (
    ApiKeyService,
    _decrypt_key_value,
    _encrypt_key_value,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

# ── Shared fixtures / helpers ────────────────────────────────────────

def _auth(person: IdentityPerson, *, admin: bool = False) -> SimpleNamespace:
    return SimpleNamespace(
        id=person.id,
        role_names=["global_admin"] if admin else ["employee"],
        work_email=person.work_email,
        display_name=person.display_name,
        employee_no=None,
        employment_status="active",
    )


async def _make_person(db_session: AsyncSession) -> IdentityPerson:
    tag = uuid.uuid4().hex[:8]
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{tag}",
        display_name="Shuai Feng",
        work_email=f"shuai.feng+{tag}@pingcap.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.commit()
    return person


async def _make_provider(
    db_session: AsyncSession, name: str,
) -> ApiKeyProvider:
    provider = ApiKeyProvider(
        id=uuid.uuid4(), name=name, description=name,
        is_active=True, sort_order=0,
    )
    db_session.add(provider)
    await db_session.commit()
    return provider


async def _make_assignment(
    db_session: AsyncSession,
    *,
    provider_id: uuid.UUID,
    person_id: uuid.UUID,
    integration_id: uuid.UUID,
    detail: dict,
    key_bundle: str,
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider_id,
        assigned_to=person_id,
        status="active",
        request_number=f"AK-20260716-{uuid.uuid4().hex[:4].upper()}",
        project_name="LLM eval",
        reason="rotation test",
        monthly_budget_usd=Decimal("200.00"),
        iam_integration_id=str(integration_id),
        detail=detail,
    )
    # Encrypt AFTER we have the row id (AAD binding).
    db_session.add(a)
    await db_session.flush()
    a.key_value = _encrypt_key_value(key_bundle, assignment_id=a.id)
    await db_session.commit()
    return a


# ── Bedrock ──────────────────────────────────────────────────────────

async def _bedrock_integ(db_session: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws",
        name="AWS SG", slug="aws-sg", enabled=True, is_primary=False,
        capabilities=["iam"],
        # Put creds in config_public so _build_aws_provider skips the
        # decrypt path (secrets_encrypted None) and boto Session gets
        # real-looking (offline) creds; all IAM calls are monkeypatched.
        config_public={
            "access_key_id": "AKIATEST",
            "secret_access_key": "shh",
            "default_region": "ap-southeast-1",
        },
        secrets_encrypted=None,
        secrets_nonce=None,
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


def _patch_bedrock(monkeypatch, *, existing_keys, created):
    """Patch AWSProvider IAM methods. Returns a dict recording calls."""
    from app.domains.cmdb.providers.aws import AWSProvider

    calls: dict = {"deleted": [], "created": 0}

    async def fake_list(self, user_name):
        return list(existing_keys)

    async def fake_create(self, user_name):
        calls["created"] += 1
        return dict(created)

    async def fake_delete(self, user_name, access_key_id):
        calls["deleted"].append(access_key_id)
        return {"deleted": True, "already_absent": False}

    monkeypatch.setattr(AWSProvider, "list_access_keys", fake_list)
    monkeypatch.setattr(AWSProvider, "create_access_key", fake_create)
    monkeypatch.setattr(AWSProvider, "delete_access_key", fake_delete)
    monkeypatch.setattr(AWSProvider, "close", AsyncMock())
    return calls


async def _bedrock_setup(db_session, monkeypatch, *, existing_keys=None):
    integ = await _bedrock_integ(db_session)
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "AWS Bedrock")
    detail = {"bedrock": {
        "granted": True,
        "iam_user_name": "bedrock-llm-20260716-AAAA",
        "user_arn": "arn:aws:iam::000:user/bedrock-llm-20260716-AAAA",
        "region": "ap-southeast-1",
        "integration_id": str(integ.id),
    }}
    old_bundle = (
        "ACCESS_KEY_ID=AKIAOLD\nSECRET_ACCESS_KEY=oldsecret\n"
        "REGION=ap-southeast-1\nUSER_ARN=arn:aws:iam::000:user/x"
    )
    a = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=detail, key_bundle=old_bundle,
    )
    calls = _patch_bedrock(
        monkeypatch,
        existing_keys=existing_keys or [{"access_key_id": "AKIAOLD", "status": "Active", "create_date": ""}],
        created={"access_key_id": "AKIANEW", "secret_access_key": "newsecret"},
    )
    return integ, person, a, calls


@pytest.mark.asyncio
async def test_bedrock_rotate_dual_active(db_session, monkeypatch):
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    resp = await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert resp.status == "active"
    assert resp.rotation_status == "pending_retire"
    assert resp.rotation_retire_at is not None
    assert calls["created"] == 1
    # key_value now holds the NEW credential; old key NOT deleted yet.
    assert calls["deleted"] == []
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    plaintext = _decrypt_key_value(fresh.key_value, assignment_id=a.id)
    assert "AKIANEW" in plaintext
    rot = fresh.detail["bedrock"]["rotation"]
    assert rot["status"] == "pending_retire"
    assert rot["prev_handle"]["access_key_id"] == "AKIAOLD"
    assert rot["new_handle"]["access_key_id"] == "AKIANEW"
    # prev_key_value stash present in DB but the ENTIRE rotation sub-blob
    # is stripped from the API response (R1 review #6): no prev_key_value,
    # no prev/new handle, no initiator/timestamps leak to any reader.
    assert "prev_key_value" in rot
    resp_bedrock = (resp.detail or {}).get("bedrock", {})
    assert "rotation" not in resp_bedrock


@pytest.mark.asyncio
async def test_bedrock_rotate_immediate_retire(db_session, monkeypatch):
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    resp = await svc.rotate_key(
        a.id, grace_days=0, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    # grace 0 = hard cut: old key deleted immediately, rotation cleared.
    assert calls["deleted"] == ["AKIAOLD"]
    assert resp.rotation_status is None
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    assert fresh.rotation_retire_at is None
    assert "rotation" not in (fresh.detail.get("bedrock") or {})


@pytest.mark.asyncio
async def test_bedrock_force_complete(db_session, monkeypatch):
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert calls["deleted"] == []
    resp = await svc.force_complete_rotation(
        a.id, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert calls["deleted"] == ["AKIAOLD"]
    assert resp.rotation_status is None
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    assert fresh.rotation_retire_at is None


@pytest.mark.asyncio
async def test_bedrock_cancel_restores_old(db_session, monkeypatch):
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    resp = await svc.cancel_rotation(
        a.id, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    # Cancel deletes the NEW key, keeps the old, restores key_value.
    assert calls["deleted"] == ["AKIANEW"]
    assert resp.rotation_status is None
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    plaintext = _decrypt_key_value(fresh.key_value, assignment_id=a.id)
    assert "AKIAOLD" in plaintext
    assert "AKIANEW" not in plaintext


@pytest.mark.asyncio
async def test_bedrock_retire_sweep(db_session, monkeypatch):
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    # Force the retire timestamp into the past and run the sweep.
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) - timedelta(hours=1)
    await db_session.commit()
    from app.domains.apikey.rotation import sweep_rotation_retire
    summary = await sweep_rotation_retire(db_session)
    assert summary["retired"] == 1
    assert calls["deleted"] == ["AKIAOLD"]
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert after.rotation_retire_at is None
    assert "rotation" not in (after.detail.get("bedrock") or {})


@pytest.mark.asyncio
async def test_bedrock_retire_sweep_reminder(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    # retire in ~12h → inside the 24h reminder window, not yet due.
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) + timedelta(hours=12)
    det = dict(fresh.detail)
    det["bedrock"]["rotation"]["retire_at"] = fresh.rotation_retire_at.isoformat()
    fresh.detail = det
    await db_session.commit()
    from app.domains.apikey.rotation import sweep_rotation_retire
    summary = await sweep_rotation_retire(db_session)
    assert summary["reminded"] == 1
    assert summary["retired"] == 0
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert after.detail["bedrock"]["rotation"]["reminded_at"] is not None
    # Second sweep does not re-remind.
    summary2 = await sweep_rotation_retire(db_session)
    assert summary2["reminded"] == 0


@pytest.mark.asyncio
async def test_retire_failure_keeps_rotation_and_counts_failed(db_session, monkeypatch):
    """R1-P2: a failed cloud teardown of the OLD credential must NOT be
    reported as retired.

    The old key is still live on the provider, so the row must stay in
    pending_retire (blob + timestamp kept) for the hourly sweep to retry,
    the failure must be counted (``retire_failed``) so the handler's
    systemic-outage alarm can fire, and the owner must not be told it was
    retired. Previously ``retire_rotation`` swallowed the exception,
    cleared the blob, and the sweep reported ``retired`` — orphaning a
    live credential while claiming success.
    """
    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) - timedelta(hours=1)
    await db_session.commit()

    from app.domains.apikey.rotation import sweep_rotation_retire
    from app.domains.cmdb.providers.aws import AWSProvider

    # Cloud delete of the OLD access key fails (IAM outage / throttle).
    async def boom(self, user_name, access_key_id):
        raise RuntimeError("IAM throttled")
    monkeypatch.setattr(AWSProvider, "delete_access_key", boom)

    summary = await sweep_rotation_retire(db_session)
    assert summary["retired"] == 0
    assert summary["retire_attempted"] == 1
    assert summary["retire_failed"] == 1

    after = await db_session.get(ApiKeyAssignment, a.id)
    await db_session.refresh(after)
    # Row stays in pending_retire — blob + timestamp kept for the retry.
    assert after.rotation_retire_at is not None
    assert "rotation" in (after.detail.get("bedrock") or {})

    # Cloud recovers → the next sweep retries and actually retires.
    async def ok_delete(self, user_name, access_key_id):
        calls["deleted"].append(access_key_id)
        return {"deleted": True, "already_absent": False}
    monkeypatch.setattr(AWSProvider, "delete_access_key", ok_delete)

    summary2 = await sweep_rotation_retire(db_session)
    assert summary2["retired"] == 1
    assert calls["deleted"] == ["AKIAOLD"]
    after2 = await db_session.get(ApiKeyAssignment, a.id)
    await db_session.refresh(after2)
    assert after2.rotation_retire_at is None
    assert "rotation" not in (after2.detail.get("bedrock") or {})


@pytest.mark.asyncio
async def test_bedrock_precheck_two_key_limit(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(
        db_session, monkeypatch,
        existing_keys=[
            {"access_key_id": "AKIA1", "status": "Active", "create_date": ""},
            {"access_key_id": "AKIA2", "status": "Active", "create_date": ""},
        ],
    )
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_KEY_LIMIT"


@pytest.mark.asyncio
async def test_rotate_in_flight_blocked(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_IN_FLIGHT"


@pytest.mark.asyncio
async def test_rotate_owner_only(db_session, monkeypatch):
    _integ, _person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    other = await _make_person(db_session)
    svc = ApiKeyService(db_session)
    with pytest.raises(ForbiddenError):
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(other),
            trace_id="t", ip_address=None, user_agent=None,
        )


@pytest.mark.asyncio
async def test_rotate_invalid_grace(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=5, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_INVALID_GRACE"


@pytest.mark.asyncio
async def test_rotate_non_rotatable_provider(db_session, monkeypatch):
    integ = await _bedrock_integ(db_session)
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "OpenAI")
    a = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail={}, key_bundle="sk-openai-xxx",
    )
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_NOT_SUPPORTED"


@pytest.mark.asyncio
async def test_rotate_non_active_status(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    a.status = "expired"
    await db_session.commit()
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "INVALID_STATUS_FOR_ROTATION"


@pytest.mark.asyncio
async def test_revoke_clears_rotation(db_session, monkeypatch):
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    assert fresh.rotation_retire_at is not None
    # Revoke tears down the whole IAM user → clears rotation state.
    # Patch the cloud revoke to a no-op so we exercise only the clearing.
    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources",
        AsyncMock(return_value={"attempted": [], "errors": []}),
    )
    admin = await _make_person(db_session)
    await svc.revoke_key(
        a.id, reason="test", current_user=_auth(admin, admin=True),
        trace_id="t", ip_address=None, user_agent=None,
    )
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert after.status == "revoked"
    assert after.rotation_retire_at is None
    assert "rotation" not in (after.detail.get("bedrock") or {})


# ── Bailian ──────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_bailian_rotate_and_retire(db_session, monkeypatch):
    from app.domains.bailian import client as bailian_client
    from app.domains.bailian.models import BailianApiKey, BailianWorkspace

    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="alicloud_intl",
        name="Aliyun Intl", slug="aliyun-intl", enabled=True,
        is_primary=False, capabilities=["iam"],
        config_public={"region": "ap-southeast-1"},
        secrets_encrypted=None, secrets_nonce=None,
    )
    db_session.add(integ)
    await db_session.commit()
    ws = BailianWorkspace(
        id=uuid.uuid4(), integration_id=str(integ.id),
        workspace_id="ws-123", name="default", is_default=True,
    )
    db_session.add(ws)
    await db_session.commit()
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "Aliyun Bailian")
    detail = {"bailian": {
        "granted": True,
        "workspace_id": "ws-123",
        "ram_user_name": "bailian-x",
        "iam_user_name": "bailian-x",
        "api_key_id": "OLDID",
        "integration_id": str(integ.id),
    }}
    a = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=detail,
        key_bundle="API_KEY=sk-old\nAPI_KEY_ID=OLDID",
    )

    calls: dict = {"deleted": []}

    def fake_init(self, creds):
        pass

    async def fake_create(
        self, *, workspace_id, description=None, region=None,
    ):
        return {"api_key_id": "NEWID", "api_key_value": "sk-new", "created_by": "u"}

    async def fake_delete(
        self, *, api_key_id, workspace_id=None, region=None,
    ):
        calls["deleted"].append(api_key_id)
        return {"deleted": True, "already_absent": False}

    monkeypatch.setattr(bailian_client.BailianClient, "__init__", fake_init)
    monkeypatch.setattr(bailian_client.BailianClient, "create_api_key", fake_create)
    monkeypatch.setattr(bailian_client.BailianClient, "delete_api_key", fake_delete)
    monkeypatch.setattr(
        bailian_client, "resolve_credentials", lambda integration: {},
    )

    svc = ApiKeyService(db_session)
    resp = await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert resp.rotation_status == "pending_retire"
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    assert fresh.detail["bailian"]["api_key_id"] == "NEWID"
    plaintext = _decrypt_key_value(fresh.key_value, assignment_id=a.id)
    assert "sk-new" in plaintext
    rot = fresh.detail["bailian"]["rotation"]
    assert rot["prev_handle"]["api_key_id"] == "OLDID"
    # A new registry row for the new key exists (active).
    from sqlalchemy import select
    rows = (await db_session.execute(
        select(BailianApiKey).where(BailianApiKey.assignment_id == a.id)
    )).scalars().all()
    assert any(r.api_key_id == "NEWID" and r.deleted_at is None for r in rows)

    # Retire → delete OLD key.
    await svc.force_complete_rotation(
        a.id, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert calls["deleted"] == ["OLDID"]


# ── Azure OpenAI ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_azure_rotate_and_retire(db_session, monkeypatch):
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="azure",
        name="Azure", slug="azure-main", enabled=True, is_primary=False,
        capabilities=["iam"], config_public={}, secrets_encrypted=None,
        secrets_nonce=None,
    )
    db_session.add(integ)
    await db_session.commit()
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "Azure OpenAI")
    detail = {"azure_openai": {
        "granted": True,
        "subscription_id": "sub-1",
        "resource_group": "helix-aoai-x",
        "foundry_account_name": "aoai-x",
        "foundry_account_endpoint": "https://aoai-x.services.ai.azure.com/",
        "foundry_project_endpoint": "https://aoai-x.services.ai.azure.com/api/projects/p",
        "foundry_project_name": "p",
        "active_key_slot": "key1",
        "integration_id": str(integ.id),
    }}
    a = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=detail,
        key_bundle="AZURE_OPENAI_API_KEY=K1OLD",
    )

    regen_calls: list = []

    async def fake_regen(*, subscription_id, resource_group, account_name, key_name):
        regen_calls.append(key_name)
        # regenerating a slot returns fresh values for that slot.
        return {"key1": "K1NEW", "key2": "K2NEW"}

    fake_driver = SimpleNamespace(
        regenerate_foundry_account_key=fake_regen,
        close=AsyncMock(),
    )

    async def fake_load(db, integration):
        return fake_driver

    monkeypatch.setattr(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver", fake_load,
    )

    svc = ApiKeyService(db_session)
    resp = await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert resp.rotation_status == "pending_retire"
    # Rotation regenerated the OTHER slot (key2) and handed it out.
    assert regen_calls == ["key2"]
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    assert fresh.detail["azure_openai"]["active_key_slot"] == "key2"
    plaintext = _decrypt_key_value(fresh.key_value, assignment_id=a.id)
    assert "K2NEW" in plaintext
    rot = fresh.detail["azure_openai"]["rotation"]
    assert rot["prev_handle"]["slot"] == "key1"

    # Retire → regenerate the OLD slot (key1) to invalidate it.
    await svc.force_complete_rotation(
        a.id, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    assert regen_calls == ["key2", "key1"]


# ── Regression tests for the R1 adversarial review fixes ─────────────

@pytest.mark.asyncio
async def test_sweep_survives_row_error_and_keeps_going(db_session, monkeypatch):
    """R1 #1: a per-row failure must not abort the whole sweep.

    The pre-fix sweep iterated pre-loaded ORM instances and, after a
    per-row ``session.rollback()``, touched an expired attribute →
    ``MissingGreenlet`` that aborted the entire sweep. The fix re-fetches
    each row fresh per iteration. Here we make the FIRST due row's
    ``retire_rotation`` raise, and assert the sweep records the error AND
    still retires the SECOND row (didn't abort).
    """
    integ = await _bedrock_integ(db_session)
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "AWS Bedrock")

    async def _mk_inflight(rn: str) -> ApiKeyAssignment:
        detail = {"bedrock": {
            "granted": True,
            "iam_user_name": f"bedrock-{rn}",
            "region": "ap-southeast-1",
            "integration_id": str(integ.id),
            "rotation": {
                "status": "pending_retire",
                "prev_handle": {"access_key_id": "AKIAOLD"},
                "new_handle": {"access_key_id": "AKIANEW"},
                "retire_at": (datetime.now(UTC) - timedelta(hours=1)).isoformat(),
                "grace_days": 7,
                "reminded_at": None,
                "prev_key_value": _encrypt_key_value("old", assignment_id=uuid.uuid4()),
            },
        }}
        a = ApiKeyAssignment(
            id=uuid.uuid4(), provider_id=provider.id, assigned_to=person.id,
            status="active", request_number=rn, project_name="p", reason="r",
            monthly_budget_usd=Decimal("10"), iam_integration_id=str(integ.id),
            detail=detail,
            rotation_retire_at=datetime.now(UTC) - timedelta(hours=1),
        )
        db_session.add(a)
        await db_session.flush()
        a.key_value = _encrypt_key_value("k", assignment_id=a.id)
        await db_session.commit()
        return a

    a1 = await _mk_inflight("AK-20260716-SWEEP1")
    a2 = await _mk_inflight("AK-20260716-SWEEP2")

    from app.domains.apikey import rotation as rot

    real_retire = rot.retire_rotation

    async def flaky_retire(session, assignment, *, kind):
        if assignment.id == a1.id:
            raise RuntimeError("simulated transient failure on row 1")
        return await real_retire(session, assignment, kind=kind)

    monkeypatch.setattr(rot, "retire_rotation", flaky_retire)
    # Patch AWS delete so the second row's real retire succeeds.
    _patch_bedrock(
        monkeypatch,
        existing_keys=[{"access_key_id": "AKIAOLD", "status": "Active", "create_date": ""}],
        created={"access_key_id": "AKIANEW", "secret_access_key": "s"},
    )

    summary = await rot.sweep_rotation_retire(db_session)
    # Sweep did NOT abort: it scanned both, errored on one, retired the other.
    assert summary["scanned"] == 2
    assert len(summary["errors"]) == 1
    assert summary["retired"] == 1
    # Retire-specific counters (finding A): both rows were due (attempted),
    # one failed. The cron handler gates its systemic-failure raise on
    # these, not on scanned==errors.
    assert summary["retire_attempted"] == 2
    assert summary["retire_failed"] == 1
    # The failed row keeps its rotation state (retry next tick); the other cleared.
    a1_fresh = await db_session.get(ApiKeyAssignment, a1.id)
    a2_fresh = await db_session.get(ApiKeyAssignment, a2.id)
    assert a1_fresh.rotation_retire_at is not None
    assert a2_fresh.rotation_retire_at is None


@pytest.mark.asyncio
async def test_bailian_revoke_deletes_old_rotation_key(db_session, monkeypatch):
    """R1 #2: revoking a Bailian key mid-rotation must delete BOTH the new
    canonical sk- AND the old (prev_handle) sk-, else the old one is
    orphaned live on Aliyun (billing + valid credential a leaver holds).
    """
    from app.domains.bailian import client as bailian_client

    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="alicloud_intl",
        name="Aliyun Intl", slug="aliyun-intl-r", enabled=True,
        is_primary=False, capabilities=["iam"], config_public={},
        secrets_encrypted=None, secrets_nonce=None,
    )
    db_session.add(integ)
    await db_session.commit()
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "Aliyun Bailian")
    # Mid-rotation: canonical api_key_id = NEW, prev_handle = OLD.
    detail = {"bailian": {
        "granted": True,
        "workspace_id": "ws-1",
        "ram_user_name": "bailian-x",
        "iam_user_name": "bailian-x",
        "api_key_id": "NEWID",
        "integration_id": str(integ.id),
        "rotation": {
            "status": "pending_retire",
            "prev_handle": {"api_key_id": "OLDID", "workspace_id": "ws-1"},
            "new_handle": {"api_key_id": "NEWID", "workspace_id": "ws-1"},
            "retire_at": (datetime.now(UTC) + timedelta(days=7)).isoformat(),
            "grace_days": 7,
        },
    }}
    a = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=detail, key_bundle="API_KEY=sk-new",
    )
    a.rotation_retire_at = datetime.now(UTC) + timedelta(days=7)
    await db_session.commit()

    deleted: list = []

    def fake_init(self, creds):
        pass

    async def fake_delete(
        self, *, api_key_id, workspace_id=None, region=None,
    ):
        deleted.append(api_key_id)
        return {"deleted": True, "already_absent": False}

    monkeypatch.setattr(bailian_client.BailianClient, "__init__", fake_init)
    monkeypatch.setattr(bailian_client.BailianClient, "delete_api_key", fake_delete)
    monkeypatch.setattr(bailian_client, "resolve_credentials", lambda integration: {})
    # RAM teardown dispatch is orthogonal; stub it out.
    monkeypatch.setattr(
        ApiKeyService, "_dispatch_cloud_revoke", AsyncMock(return_value=None),
    )

    admin = await _make_person(db_session)
    await ApiKeyService(db_session).revoke_key(
        a.id, reason="offboarding", current_user=_auth(admin, admin=True),
        trace_id="t", ip_address=None, user_agent=None,
    )
    # BOTH keys deleted — the orphan hole is closed.
    assert set(deleted) == {"NEWID", "OLDID"}


@pytest.mark.asyncio
async def test_complete_rejects_minting(db_session, monkeypatch):
    """R1 #5: complete/cancel must not act on a still-``minting`` rotation
    (handles not yet written; would race the in-flight mint)."""
    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    # Force the row into the minting state (as if start_rotation's cloud
    # mint were still running).
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    det = dict(fresh.detail)
    det["bedrock"] = {**det["bedrock"], "rotation": {"status": "minting"}}
    fresh.detail = det
    await db_session.commit()
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.force_complete_rotation(
            a.id, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_NOT_READY"
    with pytest.raises(AppException) as ei2:
        await svc.cancel_rotation(
            a.id, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei2.value.code == "ROTATION_NOT_READY"


# ── Retire-failure hardening (fix: retire failures no longer ─────────
# ── masquerade as success; minting self-heals) ───────────────────────

def _patch_bedrock_delete_raises(monkeypatch):
    """Re-patch delete_access_key to a raising stub (cloud retire down)."""
    from app.domains.cmdb.providers.aws import AWSProvider

    async def boom(self, user_name, access_key_id):
        raise RuntimeError("iam is down")

    monkeypatch.setattr(AWSProvider, "delete_access_key", boom)


def _seed_retire_attempts(assignment, *, kind: str, attempts: int) -> None:
    """Stamp ``retire_attempts`` into the rotation blob (prior failures).

    Deep-copies each nested layer: mutating the nested dicts in place
    would also mutate the OLD attribute value, so SQLAlchemy would see
    no net change and skip the UPDATE — and the sweep's FOR-UPDATE
    re-read (populate_existing) would then discard the seed.
    """
    det = dict(assignment.detail)
    blob = dict(det[kind])
    rotation = dict(blob["rotation"])
    rotation["retire_attempts"] = attempts
    blob["rotation"] = rotation
    det[kind] = blob
    assignment.detail = det


@pytest.mark.asyncio
async def test_retire_sweep_cloud_failure_keeps_state(db_session, monkeypatch):
    """A failed cloud retire must NOT clear the rotation: the old
    credential is still live. State kept (attempt counter recorded),
    failed audit row landed, no "retired" owner notification."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) - timedelta(hours=1)
    await db_session.commit()
    _patch_bedrock_delete_raises(monkeypatch)
    notify_spy = AsyncMock()
    monkeypatch.setattr(rot, "notify_rotation_retired", notify_spy)

    summary = await rot.sweep_rotation_retire(db_session)

    # Never counted as retired; the row lands in errors with the
    # retire_failed action in details.
    assert summary["retired"] == 0
    assert len(summary["errors"]) == 1
    err = summary["errors"][0]
    assert err["request_number"] == a.request_number
    assert err["attempts"] == 1
    assert err["gave_up"] is False
    assert {"request_number": a.request_number, "action": "retire_failed"} in (
        summary["details"]
    )
    # Rotation state KEPT so the hourly cron retries.
    after = await db_session.get(ApiKeyAssignment, a.id)
    rot_blob = (after.detail.get("bedrock") or {}).get("rotation") or {}
    assert rot_blob.get("status") == "pending_retire"
    assert rot_blob.get("retire_attempts") == 1
    assert rot_blob.get("last_retire_error")
    assert after.rotation_retire_at is not None
    # Failed audit row landed.
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.retire_failed",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "failed"
    # Owner never told the old credential is gone.
    notify_spy.assert_not_awaited()


@pytest.mark.asyncio
async def test_retire_sweep_gives_up_after_max_attempts(db_session, monkeypatch):
    """After MAX_RETIRE_ATTEMPTS consecutive failures the state is
    cleared (row doesn't strand forever) but the summary carries
    gave_up=True and the audit row is status=failed."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) - timedelta(hours=1)
    # 5 prior failures → this sweep's failure is attempt 6 == the cap.
    _seed_retire_attempts(fresh, kind="bedrock", attempts=rot.MAX_RETIRE_ATTEMPTS - 1)
    await db_session.commit()
    _patch_bedrock_delete_raises(monkeypatch)

    summary = await rot.sweep_rotation_retire(db_session)

    assert summary["retired"] == 0
    assert len(summary["errors"]) == 1
    err = summary["errors"][0]
    assert err["attempts"] == rot.MAX_RETIRE_ATTEMPTS
    assert err["gave_up"] is True
    # State cleared so the row stops being scanned forever.
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert "rotation" not in (after.detail.get("bedrock") or {})
    assert after.rotation_retire_at is None
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.retire_failed",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "failed"


@pytest.mark.asyncio
async def test_retire_sweep_succeeds_after_prior_failures(db_session, monkeypatch):
    """Prior retire failures don't poison a later success: the retry
    that lands clears state, audits success, and notifies the owner."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    _integ, person, a, calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    fresh.rotation_retire_at = datetime.now(UTC) - timedelta(hours=1)
    _seed_retire_attempts(fresh, kind="bedrock", attempts=2)
    await db_session.commit()
    notify_spy = AsyncMock()
    monkeypatch.setattr(rot, "notify_rotation_retired", notify_spy)

    summary = await rot.sweep_rotation_retire(db_session)

    assert summary["retired"] == 1
    assert summary["errors"] == []
    assert calls["deleted"] == ["AKIAOLD"]
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert "rotation" not in (after.detail.get("bedrock") or {})
    assert after.rotation_retire_at is None
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.retired",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "success"
    notify_spy.assert_awaited_once()


@pytest.mark.asyncio
async def test_retire_sweep_repairs_stale_minting(db_session, monkeypatch):
    """A crashed mid-mint tombstone (status=minting, older than the
    stale threshold) is cleared by the sweep with a failed
    mint_interrupted audit row; a FRESH minting row is left alone."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    integ = await _bedrock_integ(db_session)
    person = await _make_person(db_session)
    provider = await _make_provider(db_session, "AWS Bedrock")
    now = datetime.now(UTC)

    def _detail(minted_at: datetime) -> dict:
        return {"bedrock": {
            "granted": True,
            "iam_user_name": "bedrock-llm-20260716-AAAA",
            "region": "ap-southeast-1",
            "integration_id": str(integ.id),
            "rotation": {
                "status": "minting",
                "initiated_by": str(person.id),
                "grace_days": 7,
                "minted_at": minted_at.isoformat(),
                "prev_key_value": None,
            },
        }}

    stale = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=_detail(now - timedelta(hours=3)),
        key_bundle="ACCESS_KEY_ID=AKIAOLD",
    )
    fresh = await _make_assignment(
        db_session, provider_id=provider.id, person_id=person.id,
        integration_id=integ.id, detail=_detail(now - timedelta(minutes=5)),
        key_bundle="ACCESS_KEY_ID=AKIAOLD2",
    )
    stale.rotation_retire_at = now + timedelta(days=7)
    fresh.rotation_retire_at = now + timedelta(days=7)
    await db_session.commit()

    summary = await rot.sweep_rotation_retire(db_session)

    # The stale tombstone was cleared and surfaced loudly.
    assert summary["errors"], "stale minting repair must land in errors"
    assert summary["errors"][0]["request_number"] == stale.request_number
    assert {"request_number": stale.request_number, "action": "mint_interrupted"} in (
        summary["details"]
    )
    after_stale = await db_session.get(ApiKeyAssignment, stale.id)
    assert "rotation" not in (after_stale.detail.get("bedrock") or {})
    assert after_stale.rotation_retire_at is None
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.mint_interrupted",
            AuditLog.resource_id == str(stale.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "failed"
    # The fresh mid-mint row is untouched (its worker may still land).
    after_fresh = await db_session.get(ApiKeyAssignment, fresh.id)
    fresh_rot = (after_fresh.detail.get("bedrock") or {}).get("rotation") or {}
    assert fresh_rot.get("status") == "minting"
    assert after_fresh.rotation_retire_at is not None
    fresh_audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.mint_interrupted",
            AuditLog.resource_id == str(fresh.id),
        )
    )).scalars().all()
    assert fresh_audits == []


@pytest.mark.asyncio
async def test_force_complete_retire_failure_surfaces_502(db_session, monkeypatch):
    """Force-complete with a failing cloud retire must surface a 502
    ROTATION_RETIRE_FAILED (not report success over a live old
    credential): rotation stays pending_retire, failed audit row lands."""
    from sqlalchemy import select

    from app.audit.models import AuditLog

    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    _patch_bedrock_delete_raises(monkeypatch)

    with pytest.raises(AppException) as ei:
        await svc.force_complete_rotation(
            a.id, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_RETIRE_FAILED"
    assert ei.value.status_code == 502
    # Rotation stays pending_retire so the hourly cron retries.
    after = await db_session.get(ApiKeyAssignment, a.id)
    rot_blob = (after.detail.get("bedrock") or {}).get("rotation") or {}
    assert rot_blob.get("status") == "pending_retire"
    assert rot_blob.get("retire_attempts") == 1
    assert after.rotation_retire_at is not None
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.completed",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "failed"


@pytest.mark.asyncio
async def test_force_complete_gave_up_surfaces_distinct_error(db_session, monkeypatch):
    """When force-complete's retire hits MAX_RETIRE_ATTEMPTS it CLEARS
    the rotation state — so the 502 must NOT promise auto-retry (there is
    no rotation_retire_at left to rescan). It raises the distinct
    ROTATION_RETIRE_GAVE_UP and lands a retire_gave_up failed audit row
    as the ops trail (this request path has no cron failure notifier)."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    svc = ApiKeyService(db_session)
    await svc.rotate_key(
        a.id, grace_days=7, current_user=_auth(person),
        trace_id="t", ip_address=None, user_agent=None,
    )
    fresh = await db_session.get(ApiKeyAssignment, a.id)
    # 5 prior failures → this force-complete is attempt 6 == the cap.
    _seed_retire_attempts(fresh, kind="bedrock", attempts=rot.MAX_RETIRE_ATTEMPTS - 1)
    await db_session.commit()
    _patch_bedrock_delete_raises(monkeypatch)

    with pytest.raises(AppException) as ei:
        await svc.force_complete_rotation(
            a.id, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_RETIRE_GAVE_UP"
    assert ei.value.status_code == 502
    # State cleared — nothing left for the sweep to retry.
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert "rotation" not in (after.detail.get("bedrock") or {})
    assert after.rotation_retire_at is None
    # A distinct failed audit row is the ops trail.
    gave_up_audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.retire_gave_up",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(gave_up_audits) == 1
    assert gave_up_audits[0].status == "failed"


@pytest.mark.asyncio
async def test_rotation_retire_handler_raises_on_any_error(db_session, monkeypatch):
    """The cron handler must land the run FAILED (raise) when ANY row
    errored — previously it only raised when errors == scanned, so a
    partial failure masqueraded as a green run and never reached the
    failure notifier."""
    from app.scheduled_jobs.handlers import run_apikey_rotation_retire

    summary = {
        "now": datetime.now(UTC).isoformat(),
        "scanned": 5,
        "retired": 4,
        "reminded": 0,
        "errors": [{"request_number": "AK-20260716-ZZZZ", "error": "iam is down"}],
        "details": [],
    }
    monkeypatch.setattr(
        "app.domains.apikey.rotation.sweep_rotation_retire",
        AsyncMock(return_value=summary),
    )
    with pytest.raises(RuntimeError, match=r"1/5 rotations"):
        await run_apikey_rotation_retire(db_session, {})


@pytest.mark.asyncio
async def test_start_rotation_stamps_retire_at_before_mint(db_session, monkeypatch):
    """The minting tombstone must carry rotation_retire_at BEFORE the
    cloud call (crashed mints stay sweep-visible), and a mint failure
    clears the blob + lands a failed mint_failed audit row (durable
    marker for a possible vendor-side orphan)."""
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.domains.apikey import rotation as rot

    _integ, person, a, _calls = await _bedrock_setup(db_session, monkeypatch)
    observed: dict = {}

    async def crashing_minter(db, assignment, integration):
        # Mid-mint: the tombstone commit already happened — the sweep
        # scheduling column must be stamped alongside it.
        observed["retire_at_mid_mint"] = assignment.rotation_retire_at
        rot_blob = (assignment.detail or {}).get("bedrock", {}).get("rotation") or {}
        observed["status_mid_mint"] = rot_blob.get("status")
        raise RuntimeError("cloud died mid-mint")

    monkeypatch.setitem(rot._MINTERS, "bedrock", crashing_minter)
    svc = ApiKeyService(db_session)
    with pytest.raises(AppException) as ei:
        await svc.rotate_key(
            a.id, grace_days=7, current_user=_auth(person),
            trace_id="t", ip_address=None, user_agent=None,
        )
    assert ei.value.code == "ROTATION_CLOUD_FAILED"
    # The tombstone was sweep-visible while the mint was in flight.
    assert observed["retire_at_mid_mint"] is not None
    assert observed["status_mid_mint"] == "minting"
    # Failure cleared the blob (re-rotation unblocked) …
    after = await db_session.get(ApiKeyAssignment, a.id)
    assert "rotation" not in (after.detail.get("bedrock") or {})
    assert after.rotation_retire_at is None
    # … and left the durable failed audit marker.
    audits = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.rotation.mint_failed",
            AuditLog.resource_id == str(a.id),
        )
    )).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "failed"