"""Tests for the inactivity sweep + grants-sync drift cleanup.

Covers PR #170 follow-up — three production bugs surfaced when the
500-employee rollout opened the cloud surface to real cloud admins:

  1. ``sweep_inactive_grants`` did NOT exempt ``cloud_admin`` /
     ``it_admin`` / ``global_admin``. Privileged-role holders received
     dozens of false "60 days no login" alerts on grants the role
     legitimately owns (Kidd held a Reader-equivalent on every Azure
     subscription so he could investigate billing — sweep flagged
     every one).

  2. ``_derive_grants_from_iam_users`` lifted ``IAMUserChangePassword``
     and other AWS self-service managed policies into discovered
     grants, so every IAM-mode AWS tenant emitted N noisy
     stale-grant alerts per leaver where N = (number of self-service
     policies AWS auto-attached × scopes).

  3. ``sync_grants_for_integration`` had no cleanup for discovered
     grants whose row stopped appearing in the driver's authoritative
     response (admin deleted IAM user, detached policy, …). Rows
     stayed ``active`` forever; the inactivity sweep kept paging them.

Each test asserts the post-fix behaviour: privileged people are
skipped, self-service policies are filtered before they ever become
grants, and stale rows close on the next successful sync.
"""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from unittest.mock import patch

import pytest
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUserRole
from app.domains.cmdb.models import (
    GRANT_NOTIFY_INACTIVE_IT,
    GRANT_NOTIFY_INACTIVE_OWNER,
    GRANT_NOTIFY_T1,
    GRANT_REVOKE_REASON_DRIFT_REMOVED,
    GRANT_STATUS_ACTIVE,
    GRANT_STATUS_EXPIRED,
    GRANT_STATUS_REVOKED,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbIamUser,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import (
    CloudScopeRef,
    DiscoveredGrant,
)
from app.integrations.models import Integration

# ── Test helpers (kept local so the file stays self-contained) ──


async def _get_role_id(db: AsyncSession, role_name: str) -> str:
    row = await db.execute(
        text("SELECT id FROM auth_roles WHERE name = :n"), {"n": role_name},
    )
    val = row.scalar_one()
    return str(val)


async def _make_person(
    db: AsyncSession, *, email: str, name: str,
    roles: tuple[str, ...] = (),
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    for role in roles:
        rid = await _get_role_id(db, role)
        db.add(AuthUserRole(user_id=p.id, role_id=uuid.UUID(rid)))
    await db.commit()
    return p


async def _make_integration(
    db: AsyncSession, *, provider: str = "aws",
) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-drift-test-{uuid.uuid4().hex[:6]}",
        slug=f"{provider}-drift-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(row)
    await db.commit()
    return row


async def _make_scope(
    db: AsyncSession, integration: Integration, *,
    external_id: str, owner_person_id: str | None = None,
    scope_type: str = "aws_account",
) -> CmdbCloudScope:
    s = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type=scope_type,
        external_id=external_id,
        display_name=external_id,
        status=SCOPE_STATUS_ACTIVE,
        owner_person_id=owner_person_id,
    )
    db.add(s)
    await db.commit()
    return s


def _patch_session_factory(db_session: AsyncSession):
    @asynccontextmanager
    async def _factory():
        yield db_session
    return patch(
        "app.domains.cmdb.lifecycle_service.async_session_factory",
        new=_factory,
    )


class _FakeDriver:
    provider = "aws"
    grants_scope_wide = True

    def __init__(self, grants_by_scope: dict[str, list[DiscoveredGrant]]) -> None:
        self._grants_by_scope = grants_by_scope
        self.owner_role_keys = frozenset()

    async def list_scope_grants(self, scope: CloudScopeRef):
        return list(self._grants_by_scope.get(scope.external_id, []))

    async def close(self) -> None:
        pass


def _patch_driver(driver: _FakeDriver):
    async def _loader(session, integration):
        return driver
    return patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        side_effect=_loader,
    )


# ── 1. Privileged-role exemption in sweep_inactive_grants ──


@pytest.mark.asyncio
async def test_sweep_skips_cloud_admin_grants(db_session: AsyncSession) -> None:
    """A grant owned by a ``cloud_admin`` MUST NOT advance to
    ``inactive_owner`` even when the grant has no last_login_at and
    was created long ago. Closes Kidd's Azure-grants false-alert flood.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    cloud_admin = await _make_person(
        db_session, email="cadmin@test.com", name="Cloud Admin",
        roles=("cloud_admin",),
    )
    employee = await _make_person(
        db_session, email="emp1@test.com", name="Employee One",
    )
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration, external_id="sub-A",
        owner_person_id=str(cloud_admin.id),
    )

    long_ago = datetime.now(UTC) - timedelta(days=400)
    # Privileged grant — should NOT be touched.
    privileged_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(cloud_admin.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    )
    # Regular employee grant — SHOULD be flagged.
    employee_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(employee.id),
        role_key="Contributor",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    )
    db_session.add_all([privileged_grant, employee_grant])
    await db_session.commit()

    privileged_id = privileged_grant.id
    employee_id = employee_grant.id

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    assert summary["skipped_privileged"] >= 1
    privileged_after = await db_session.get(CmdbCloudAccessGrant, privileged_id)
    employee_after = await db_session.get(CmdbCloudAccessGrant, employee_id)
    assert privileged_after is not None
    assert privileged_after.last_notified_stage is None, (
        "cloud_admin grant must not be advanced to inactive_owner stage"
    )
    assert employee_after is not None
    assert employee_after.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER, (
        "employee grant should still receive the normal first-stage flag"
    )


@pytest.mark.asyncio
async def test_sweep_skips_privileged_at_stages_1_and_2(
    db_session: AsyncSession,
) -> None:
    """Defense in depth: a privileged-role grant carrying a stale
    ``inactive_owner`` flag (e.g. assignment added AFTER migration 118
    cleared the prior backlog, or person promoted to ``cloud_admin``
    between Stage 0 and Stage 1 of the same run) MUST still be skipped
    at Stage 1 (escalate to IT) and Stage 2 (auto-disable). Without
    this guard the entire docstring promise of "skipped at every
    stage" silently breaks once a single race occurs.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    cloud_admin = await _make_person(
        db_session, email="cadmin2@test.com", name="Cloud Admin Two",
        roles=("cloud_admin",),
    )
    integration = await _make_integration(db_session, provider="azure")
    owner = await _make_person(
        db_session, email="owner2@test.com", name="Owner Two",
    )
    scope = await _make_scope(
        db_session, integration, external_id="sub-2",
        owner_person_id=str(owner.id),
    )

    long_ago_login = datetime.now(UTC) - timedelta(days=100)
    long_ago_update = datetime.now(UTC) - timedelta(days=20)
    # Stage 1 candidate — already at ``inactive_owner``.
    grant_stage1 = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(cloud_admin.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago_update,
        last_login_at=long_ago_login,
        last_notified_stage=GRANT_NOTIFY_INACTIVE_OWNER,
    )
    db_session.add(grant_stage1)
    await db_session.commit()
    # Force ``updated_at`` into the past so the Stage-1 selector
    # (``updated_at < escalation_cutoff``) picks it up.
    await db_session.execute(
        text("UPDATE cmdb_cloud_access_grants SET updated_at = :u WHERE id = :id"),
        {"u": long_ago_update, "id": str(grant_stage1.id)},
    )
    await db_session.commit()

    grant_stage1_id = grant_stage1.id

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    after = await db_session.get(CmdbCloudAccessGrant, grant_stage1_id)
    assert after is not None
    # Must still be at ``inactive_owner`` — the privileged-role guard
    # at Stage 1 refuses to advance it to ``inactive_it``.
    assert after.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER, (
        "Stage 1 must skip privileged-role grants even when they "
        "carry a stale ``inactive_owner`` flag"
    )
    assert after.status == GRANT_STATUS_ACTIVE
    assert summary["skipped_privileged"] >= 1


@pytest.mark.asyncio
async def test_sweep_skips_it_admin_and_global_admin(
    db_session: AsyncSession,
) -> None:
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    global_admin = await _make_person(
        db_session, email="gadmin@test.com", name="Global Admin",
        roles=("global_admin",),
    )
    integration = await _make_integration(db_session)
    owner = await _make_person(
        db_session, email="owner@test.com", name="Scope Owner",
    )
    scope = await _make_scope(
        db_session, integration, external_id="acct-1",
        owner_person_id=str(owner.id),
    )

    long_ago = datetime.now(UTC) - timedelta(days=400)
    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(global_admin.id),
        role_key="arn:aws:iam::aws:policy/AdministratorAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    )
    db_session.add(grant)
    await db_session.commit()
    grant_id = grant.id

    await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    after = await db_session.get(CmdbCloudAccessGrant, grant_id)
    assert after is not None
    assert after.last_notified_stage is None


# ── 2. Self-service policy filter in IAM-derived grants ──


@pytest.mark.asyncio
async def test_iam_derived_grants_skip_self_service_managed_policies(
    db_session: AsyncSession,
) -> None:
    """IAMUserChangePassword + siblings are auto-attached by AWS for
    every IAM user; ``_derive_grants_from_iam_users`` MUST drop them
    before they become tracked grants. Closes Lingtao's
    IAMUserChangePassword stale-alert per leaver.
    """
    from app.domains.cmdb.lifecycle_service import _derive_grants_from_iam_users

    person = await _make_person(
        db_session, email="lingtao@test.com", name="Lingtao Test",
    )
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-99",
    )

    iam_user = CmdbIamUser(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        user_name="lingtao@test.com",
        user_id_external="AIDA12345",
        owner_person_id=str(person.id),
        attached_policies=[
            {
                "name": "IAMUserChangePassword",
                "arn": "arn:aws:iam::aws:policy/IAMUserChangePassword",
            },
            {
                "name": "IAMUserSSHKeys",
                "arn": "arn:aws:iam::aws:policy/IAMUserSSHKeys",
            },
            {
                "name": "IAMSelfManageServiceSpecificCredentials",
                "arn": "arn:aws:iam::aws:policy/IAMSelfManageServiceSpecificCredentials",
            },
            # A real, business-grant policy — MUST still come through.
            {
                "name": "AmazonBedrockFullAccess",
                "arn": "arn:aws:iam::aws:policy/AmazonBedrockFullAccess",
            },
        ],
    )
    db_session.add(iam_user)
    await db_session.commit()

    derived = await _derive_grants_from_iam_users(
        db_session, str(integration.id), [scope],
    )
    role_keys = {g.role_key for g in derived}
    assert "arn:aws:iam::aws:policy/AmazonBedrockFullAccess" in role_keys
    assert "arn:aws:iam::aws:policy/IAMUserChangePassword" not in role_keys
    assert "arn:aws:iam::aws:policy/IAMUserSSHKeys" not in role_keys
    assert (
        "arn:aws:iam::aws:policy/IAMSelfManageServiceSpecificCredentials"
        not in role_keys
    )


@pytest.mark.asyncio
async def test_iam_derived_grants_filter_by_name_when_arn_missing(
    db_session: AsyncSession,
) -> None:
    """Tencent / RAM custom-policy fallback path stores ``name`` only
    (no ARN). The filter must still match by exact policy name."""
    from app.domains.cmdb.lifecycle_service import _derive_grants_from_iam_users

    person = await _make_person(
        db_session, email="lingtao2@test.com", name="Lingtao Two",
    )
    integration = await _make_integration(db_session, provider="tencent")
    scope = await _make_scope(
        db_session, integration, external_id="acct-tcl-1",
    )
    iam_user = CmdbIamUser(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        user_name="lingtao2@test.com",
        user_id_external="2222",
        owner_person_id=str(person.id),
        attached_policies=[
            {"name": "IAMUserChangePassword", "arn": ""},
            {"name": "QcloudReadOnly", "arn": ""},
        ],
    )
    db_session.add(iam_user)
    await db_session.commit()

    derived = await _derive_grants_from_iam_users(
        db_session, str(integration.id), [scope],
    )
    role_keys = {g.role_key for g in derived}
    assert "QcloudReadOnly" in role_keys
    assert "IAMUserChangePassword" not in role_keys


# ── 3. Drift sweep in sync_grants_for_integration ──


@pytest.mark.asyncio
async def test_grants_sync_closes_stale_discovered_grant(
    db_session: AsyncSession,
) -> None:
    """A previously-discovered grant whose policy was removed cloud-side
    must be marked revoked with reason ``drift_removed`` after the next
    successful sync. Closes Lingtao's "已经清理" still-alerting bug.
    """
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    person = await _make_person(
        db_session, email="alice@test.com", name="Alice",
    )
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-stale",
    )
    integration_id = str(integration.id)

    # Pre-existing discovered grant — never seen by this sync run.
    stale = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="arn:aws:iam::aws:policy/AmazonBedrockFullAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=30),
        last_notified_stage=GRANT_NOTIFY_INACTIVE_OWNER,
        source="discovered",
        last_seen_grants_sync_job_id="prior-run-uuid",
        last_seen_grants_sync_at=datetime.now(UTC) - timedelta(days=7),
    )
    db_session.add(stale)
    await db_session.commit()
    stale_id = stale.id

    # Driver returns a different grant set — the stale one is no longer
    # in the cloud's authoritative response.
    driver = _FakeDriver({
        "acct-stale": [DiscoveredGrant(
            principal_email="alice@test.com",
            principal_type="user",
            role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
            scope=CloudScopeRef(scope_type="aws_account", external_id="acct-stale"),
            external_ref="cloud-side-readonly",
        )],
    })
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["status"] == "success"
    assert result["drift_closed"] >= 1
    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, stale_id)
    assert after is not None
    assert after.status == GRANT_STATUS_REVOKED
    assert after.revoke_reason == GRANT_REVOKE_REASON_DRIFT_REMOVED
    assert after.revoked_at is not None
    assert after.last_notified_stage is None, (
        "drift sweep must clear inactive-stage marker so the inactivity "
        "sweep cannot pick this row up again"
    )


@pytest.mark.asyncio
async def test_grants_sync_does_not_close_helix_issued_rows(
    db_session: AsyncSession,
) -> None:
    """Helix-issued grants are the system-of-record for their access;
    cloud-side drift is a compliance signal, NOT a revoke trigger."""
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    person = await _make_person(
        db_session, email="helixuser@test.com", name="Helix User",
    )
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-helix",
    )
    integration_id = str(integration.id)

    helix_row = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="arn:aws:iam::aws:policy/AdministratorAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=10),
        source="helix",
        external_ref="helix-issued",
    )
    db_session.add(helix_row)
    await db_session.commit()
    helix_id = helix_row.id

    # Driver returns NOTHING for this scope — drift sweep should fire
    # but skip the helix row.
    driver = _FakeDriver({"acct-helix": []})
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["status"] == "success"
    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, helix_id)
    assert after is not None
    assert after.status == GRANT_STATUS_ACTIVE
    assert after.source == "helix"


@pytest.mark.asyncio
async def test_grants_sync_drift_sweep_skips_scopes_with_driver_errors(
    db_session: AsyncSession,
) -> None:
    """If the driver throws for a scope, that scope is NOT eligible for
    drift sweep — a transient outage must not auto-revoke discovered
    grants integration-wide."""
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    person = await _make_person(
        db_session, email="bob@test.com", name="Bob",
    )
    integration = await _make_integration(db_session)
    bad_scope = await _make_scope(
        db_session, integration, external_id="acct-bad",
    )
    integration_id = str(integration.id)

    pre = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(bad_scope.id),
        person_id=str(person.id),
        role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=10),
        source="discovered",
        last_seen_grants_sync_job_id="prior-run-uuid",
    )
    db_session.add(pre)
    await db_session.commit()
    pre_id = pre.id

    class _ExplodingDriver(_FakeDriver):
        async def list_scope_grants(self, scope: CloudScopeRef):
            raise RuntimeError("transient cloud outage")

    driver = _ExplodingDriver({})
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    # Status reports the failure; drift_closed is 0 because the only
    # scope had a driver error.
    assert result["drift_closed"] == 0
    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, pre_id)
    assert after is not None
    assert after.status == GRANT_STATUS_ACTIVE


@pytest.mark.asyncio
async def test_grants_sync_drift_sweep_refuses_below_safety_floor(
    db_session: AsyncSession,
) -> None:
    """50% match-rate floor: when a paginator silently truncates and
    the driver returns a small subset against a populated Helix side,
    the drift sweep must REFUSE the mass-revoke and surface an error
    so an operator can investigate. Mirrors the seat-sync handlers'
    ``safe_to_drift_revoke`` semantics.
    """
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    # Need >= 5 existing rows for the floor to engage.
    persons = []
    for i in range(6):
        persons.append(await _make_person(
            db_session, email=f"drift{i}@test.com", name=f"Drift {i}",
        ))
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-truncated",
    )
    integration_id = str(integration.id)

    # 6 pre-existing discovered active grants from a prior run.
    pre_ids = []
    for p in persons:
        row = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(p.id),
            role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
            status=GRANT_STATUS_ACTIVE,
            granted_at=datetime.now(UTC) - timedelta(days=10),
            source="discovered",
            last_seen_grants_sync_job_id="prior-run-uuid",
        )
        db_session.add(row)
        pre_ids.append(row.id)
    await db_session.commit()

    # Driver returns only ONE of the six (paginator truncation
    # simulating, e.g., AliCloud RAM page-2 silent failure). Ratio is
    # 1/6 = 17%, below the 50% floor → sweep MUST refuse.
    driver = _FakeDriver({
        "acct-truncated": [DiscoveredGrant(
            principal_email="drift0@test.com",
            principal_type="user",
            role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
            scope=CloudScopeRef(scope_type="aws_account", external_id="acct-truncated"),
            external_ref="single-survivor",
        )],
    })
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["drift_closed"] == 0, (
        "Safety floor must refuse mass-revoke when match-rate is "
        "below 50%"
    )
    # All pre-existing rows survive the run.
    db_session.expire_all()
    for grant_id in pre_ids:
        after = await db_session.get(CmdbCloudAccessGrant, grant_id)
        assert after is not None
        assert after.status == GRANT_STATUS_ACTIVE
    # Error is surfaced for operator visibility.
    assert any(
        "drift_sweep refused" in e for e in result.get("errors", [])
    ), f"errors={result.get('errors')}"


@pytest.mark.asyncio
async def test_grants_sync_drift_sweep_runs_below_floor_threshold(
    db_session: AsyncSession,
) -> None:
    """When existing-active count is below the 5-row sample floor,
    the ratio guard does NOT engage — small fresh tenants should still
    get drift cleanup."""
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    persons = []
    for i in range(2):
        persons.append(await _make_person(
            db_session, email=f"smalldrift{i}@test.com",
            name=f"Small Drift {i}",
        ))
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-small",
    )
    integration_id = str(integration.id)

    pre_ids = []
    for p in persons:
        row = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(p.id),
            role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
            status=GRANT_STATUS_ACTIVE,
            granted_at=datetime.now(UTC) - timedelta(days=10),
            source="discovered",
            last_seen_grants_sync_job_id="prior-run-uuid",
        )
        db_session.add(row)
        pre_ids.append(row.id)
    await db_session.commit()

    # Driver returns nothing; 0/2 = 0% but existing < 5 means the
    # floor doesn't engage and the sweep proceeds normally.
    driver = _FakeDriver({"acct-small": []})
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["drift_closed"] == 2


@pytest.mark.asyncio
async def test_grants_sync_drift_sweep_audit_records_grant_ids(
    db_session: AsyncSession,
) -> None:
    """The drift-sweep audit row must enumerate every closed grant id
    so a compliance auditor can answer "what specifically closed at
    HH:MM" without joining ``cmdb_cloud_access_grants`` back."""
    from app.audit.models import AuditLog
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    person = await _make_person(
        db_session, email="auditid@test.com", name="Audit Id",
    )
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-audit",
    )
    integration_id = str(integration.id)

    stale = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=10),
        source="discovered",
        last_seen_grants_sync_job_id="prior-run-uuid",
    )
    db_session.add(stale)
    await db_session.commit()
    stale_id = str(stale.id)

    driver = _FakeDriver({"acct-audit": []})
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["drift_closed"] == 1
    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "cloud.grant.drift_removed_sweep",
        )
    )).scalars().all()
    assert len(rows) == 1
    assert stale_id in rows[0].detail.get("closed_grant_ids", [])


@pytest.mark.asyncio
async def test_grants_sync_re_seen_grant_stays_active(
    db_session: AsyncSession,
) -> None:
    """A grant that re-appears in the driver's response keeps its
    active status (drift sweep refreshes the marker, not the status)."""
    from app.domains.cmdb.lifecycle_service import sync_grants_for_integration

    person = await _make_person(
        db_session, email="carol@test.com", name="Carol",
    )
    integration = await _make_integration(db_session)
    scope = await _make_scope(
        db_session, integration, external_id="acct-active",
    )
    integration_id = str(integration.id)

    pre = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=10),
        source="discovered",
        last_seen_grants_sync_job_id="prior-run-uuid",
    )
    db_session.add(pre)
    await db_session.commit()
    pre_id = pre.id

    driver = _FakeDriver({
        "acct-active": [DiscoveredGrant(
            principal_email="carol@test.com",
            principal_type="user",
            role_key="arn:aws:iam::aws:policy/ReadOnlyAccess",
            scope=CloudScopeRef(scope_type="aws_account", external_id="acct-active"),
            external_ref="reaffirmed",
        )],
    })
    with _patch_session_factory(db_session), _patch_driver(driver):
        result = await sync_grants_for_integration(integration_id)

    assert result["drift_closed"] == 0
    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, pre_id)
    assert after is not None
    assert after.status == GRANT_STATUS_ACTIVE


# ── 4. Helper unit-tests (fast, no DB) ──


def test_is_self_service_managed_policy_arn_match() -> None:
    from app.domains.cmdb.lifecycle_service import _is_self_service_managed_policy
    assert _is_self_service_managed_policy(
        "arn:aws:iam::aws:policy/IAMUserChangePassword", "",
    ) is True


def test_is_self_service_managed_policy_name_match() -> None:
    from app.domains.cmdb.lifecycle_service import _is_self_service_managed_policy
    assert _is_self_service_managed_policy("", "IAMUserSSHKeys") is True


def test_is_self_service_managed_policy_negative() -> None:
    from app.domains.cmdb.lifecycle_service import _is_self_service_managed_policy
    assert _is_self_service_managed_policy(
        "arn:aws:iam::aws:policy/AdministratorAccess", "",
    ) is False
    assert _is_self_service_managed_policy("", "AdministratorAccess") is False
    assert _is_self_service_managed_policy("", "") is False


def test_is_self_service_managed_policy_case_sensitive() -> None:
    """AWS managed-policy names are case-sensitive; the filter must be
    too so a custom policy named ``iamuserchangepassword`` (lowercase)
    is NOT incorrectly suppressed."""
    from app.domains.cmdb.lifecycle_service import _is_self_service_managed_policy
    assert _is_self_service_managed_policy("", "iamuserchangepassword") is False


# ── 5. Group-by (scope, person) sweep semantics (2026-05-05 fix) ──
#
# Bug: ``sweep_inactive_grants`` used to evaluate each grant
# independently. Cloud IAM tracks login at the IAM USER level
# (``PasswordLastUsed``, ``AccessKeyLastUsed`` on AWS), not per role
# / policy — so when refresh failed for one specific grant in a
# (scope, person) group, that single grant kept ``last_login_at =
# NULL`` while the rest of the group held a fresh timestamp. Sweep
# fired ``60+ 天无登录`` on, e.g., ``IAMUserChangePassword`` for a
# user who actively used the IAM identity 7 days ago. Reported
# 2026-05-05 (li.da / 达丽 on AWS SG).
#
# Post-fix the sweep groups by (scope_id, person_id), evaluates the
# group via ``MAX(last_login_at)``, and emits ONE notification + one
# stage advance per group. Tests below pin both halves: no false
# alarms when SOME grants in the group are fresh, and group-wide
# stage advance + single notification when the group is genuinely
# inactive.


@pytest.mark.asyncio
async def test_sweep_groups_by_scope_person_one_fresh_login_protects_all(
    db_session: AsyncSession,
) -> None:
    """A (scope, person) group with EVEN ONE grant carrying a fresh
    ``last_login_at`` MUST NOT trigger inactivity for any grant in
    the group. Pin for the li.da / IAMUserChangePassword false-alarm
    case: she had 7 grants on AWS SG; 6 had fresh last_login_at, the
    7th (IAMUserChangePassword) had NULL; the old per-grant sweep
    fired anyway."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    employee = await _make_person(
        db_session, email="li.da@test.com", name="Li Da",
    )
    owner = await _make_person(
        db_session, email="scope.owner@test.com", name="Scope Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-sg",
        owner_person_id=str(owner.id),
    )

    long_ago = datetime.now(UTC) - timedelta(days=400)
    fresh = datetime.now(UTC) - timedelta(days=7)
    # 6 grants with fresh last_login (representative of a person who
    # actually uses the AWS account daily).
    for rk in [
        "AdministratorAccess", "ReadOnlyAccess", "Billing",
        "IAMReadOnly", "S3FullAccess", "EC2FullAccess",
    ]:
        db_session.add(CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=long_ago,
            last_login_at=fresh,
            last_notified_stage=None,
            source="discovered",
        ))
    # The 7th grant — the one whose refresh failed in production.
    # last_login_at is NULL but every OTHER grant in the same (scope,
    # person) group has a fresh stamp, so the group's
    # MAX(last_login_at) = fresh and no notification should fire.
    suspect_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(employee.id),
        role_key="IAMUserChangePassword",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    )
    db_session.add(suspect_grant)
    await db_session.commit()

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    # Zero notifications: the group is fresh per the MAX rule.
    assert summary["notified_owner"] == 0, summary
    assert summary["groups_notified_owner"] == 0, summary

    # All 7 grants still active, no stage stamp.
    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.person_id == str(employee.id),
        )
    )).scalars().all())
    assert len(rows) == 7
    for r in rows:
        assert r.status == GRANT_STATUS_ACTIVE
        assert r.last_notified_stage is None


@pytest.mark.asyncio
async def test_sweep_emits_one_notification_per_group_not_per_grant(
    db_session: AsyncSession,
) -> None:
    """When a whole (scope, person) group is genuinely inactive, the
    sweep dispatches ONE notification — not N (one per role).
    Old behaviour: 7 roles → 7 emails on the same person; new
    behaviour: 1 email listing every role in the group.

    Asserted via the dispatch counter: ``groups_notified_owner == 1``,
    ``notified_owner == 1`` (currently 1:1 with groups for owner
    stage), and every grant in the group carries
    ``last_notified_stage = inactive_owner``.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    employee = await _make_person(
        db_session, email="quiet@test.com", name="Quiet User",
    )
    owner = await _make_person(
        db_session, email="owner@test.com", name="Scope Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-quiet",
        owner_person_id=str(owner.id),
    )

    very_old = datetime.now(UTC) - timedelta(days=120)
    role_keys = [
        "AdministratorAccess", "ReadOnlyAccess", "Billing",
        "EC2FullAccess",
    ]
    for rk in role_keys:
        db_session.add(CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=very_old,
            last_notified_stage=None,
            source="discovered",
        ))
    await db_session.commit()

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    assert summary["groups_notified_owner"] == 1, summary
    assert summary["notified_owner"] == 1, summary

    # All 4 grants in the group share the new stage stamp.
    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.person_id == str(employee.id),
        )
    )).scalars().all())
    assert len(rows) == 4
    for r in rows:
        assert r.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER


@pytest.mark.asyncio
async def test_sweep_fresh_grant_keeps_otherwise_old_group_safe(
    db_session: AsyncSession,
) -> None:
    """A brand-new grant added to a (scope, person) group whose other
    grants are old + never-used MUST keep the WHOLE group safe.

    Pre-fix per-grant logic fired on the old grants individually. The
    group-by fix correctly skips the group via MAX(last_login_at), but
    the all-NULL-last-login fallback originally checked the OLDEST
    grant_at — which would still fire when ANY grant in the group was
    old, making the brand-new grant useless as a group guard.

    Post-fix: the all-NULL branch uses the FRESHEST grant_at and only
    fires when EVERY grant in the group was issued before the cutoff.
    Pin: 4 grants from 100 days ago + 1 grant from today, all
    last_login=NULL → group is FRESH, no notification.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    employee = await _make_person(
        db_session, email="newuser@test.com", name="New User",
    )
    owner = await _make_person(
        db_session, email="owner4@test.com", name="Scope Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-fresh",
        owner_person_id=str(owner.id),
    )

    very_old = datetime.now(UTC) - timedelta(days=100)
    today = datetime.now(UTC)

    # 4 old grants — would individually trigger inactivity in the
    # pre-fix per-grant sweep.
    for rk in [
        "AdministratorAccess", "ReadOnlyAccess",
        "Billing", "EC2FullAccess",
    ]:
        db_session.add(CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=None,
            last_notified_stage=None,
            source="discovered",
        ))
    # 1 brand-new grant — the FRESHEST in the group.
    db_session.add(CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(employee.id),
        role_key="NewRoleAddedToday",
        status=GRANT_STATUS_ACTIVE,
        granted_at=today,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    ))
    await db_session.commit()

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    # Group is fresh because the freshest grant_at is today. No
    # notifications, no stage advance.
    assert summary["notified_owner"] == 0, summary
    assert summary["groups_notified_owner"] == 0, summary
    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.person_id == str(employee.id),
        )
    )).scalars().all())
    assert len(rows) == 5
    for r in rows:
        assert r.last_notified_stage is None


@pytest.mark.asyncio
async def test_sweep_disable_stage_disables_entire_group(
    db_session: AsyncSession,
) -> None:
    """Stage T+30 disable advances every grant in the group, not
    just the one that triggered. A leaver with N grants on a scope
    must lose all N IAM permissions atomically when the group hits
    ``inactive_it`` + disable_after."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants
    from app.domains.cmdb.models import (
        GRANT_NOTIFY_INACTIVE_IT,
        GRANT_STATUS_DISABLED,
    )

    employee = await _make_person(
        db_session, email="leaver@test.com", name="Leaver",
    )
    owner = await _make_person(
        db_session, email="owner3@test.com", name="Scope Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-disable",
        owner_person_id=str(owner.id),
    )

    very_old = datetime.now(UTC) - timedelta(days=400)
    very_old_login = datetime.now(UTC) - timedelta(days=120)
    role_keys = ["AdministratorAccess", "ReadOnlyAccess", "Billing"]
    for rk in role_keys:
        # last_notified_stage = inactive_it AND updated_at very old
        # — by-pass Stage 0/1, evaluate Stage 2 directly. Pre-stamp
        # updated_at via the model __init__ then push it back in
        # a follow-up UPDATE since the auto-bump would otherwise
        # touch it on commit.
        g = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=very_old_login,
            last_notified_stage=GRANT_NOTIFY_INACTIVE_IT,
            source="discovered",
        )
        db_session.add(g)
    await db_session.commit()
    # Force updated_at back so the disable-after threshold matches.
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants "
        "SET updated_at = :ts "
        "WHERE scope_id = :sid AND person_id = :pid",
    ), {
        "ts": datetime.now(UTC) - timedelta(days=45),
        "sid": str(scope.id),
        "pid": str(employee.id),
    })
    await db_session.commit()

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    assert summary["groups_disabled"] == 1, summary
    assert summary["disabled"] == len(role_keys), summary

    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.person_id == str(employee.id),
        )
    )).scalars().all())
    assert len(rows) == len(role_keys)
    for r in rows:
        assert r.status == GRANT_STATUS_DISABLED


# ── Stage-2 disable semantic split (review pass 2026-05-05) ──
#
# The original Stage 2 codepath called ``revoke_grant`` (which writes
# ``cloud.grant.revoke`` audit + stamps ``revoked_at`` /
# ``revoke_reason``) and then overwrote ``status`` to ``disabled``.
# This left audit / status column / revoked_* columns in three-way
# disagreement. The new ``disable_grant_due_to_inactivity`` helper is
# the single Stage-2 entry point and must:
#
#   * Set ``status='disabled'``
#   * Leave ``revoked_at`` / ``revoked_by_person_id`` /
#     ``revoke_reason`` NULL (the row is reversible policy-paused,
#     not terminally revoked)
#   * Write audit row ``cloud.grant.disable`` (NOT
#     ``cloud.grant.revoke``)
#   * Stash driver outcome in ``detail.disable_result``


@pytest.mark.asyncio
async def test_stage2_writes_disable_audit_not_revoke_audit(
    db_session: AsyncSession,
) -> None:
    """Stage 2 disable writes ``cloud.grant.disable`` audit and leaves
    ``revoked_*`` columns NULL. Pre-fix the audit was
    ``cloud.grant.revoke`` and ``revoked_at`` was filled in."""
    from app.audit.models import AuditLog
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants
    from app.domains.cmdb.models import (
        GRANT_NOTIFY_INACTIVE_IT,
        GRANT_STATUS_DISABLED,
    )

    employee = await _make_person(
        db_session, email="leaver-audit@test.com", name="Leaver Audit",
    )
    owner = await _make_person(
        db_session, email="owner-audit@test.com", name="Audit Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-audit",
        owner_person_id=str(owner.id),
    )

    very_old_login = datetime.now(UTC) - timedelta(days=120)
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(employee.id),
        role_key="ReadOnlyAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=400),
        last_login_at=very_old_login,
        last_notified_stage=GRANT_NOTIFY_INACTIVE_IT,
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants SET updated_at = :ts "
        "WHERE id = :gid",
    ), {"ts": datetime.now(UTC) - timedelta(days=45), "gid": str(g.id)})
    await db_session.commit()

    await sweep_inactive_grants(db_session)
    await db_session.commit()

    refreshed = (await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.id == g.id,
        )
    )).scalar_one()
    assert refreshed.status == GRANT_STATUS_DISABLED
    # Reversible policy stop — these stay NULL.
    assert refreshed.revoked_at is None
    assert refreshed.revoked_by_person_id is None
    assert refreshed.revoke_reason is None
    # Driver outcome stashed in detail.
    assert refreshed.detail is not None
    assert "disable_result" in refreshed.detail
    assert refreshed.detail["disable_result"]["reason"] == "inactive"
    assert "at" in refreshed.detail["disable_result"]

    # Audit row says disable, not revoke.
    audit_actions = [
        r.action for r in (await db_session.execute(
            select(AuditLog).where(
                AuditLog.resource_type == "cmdb_cloud_access_grant",
                AuditLog.resource_id == str(g.id),
            )
        )).scalars().all()
    ]
    assert "cloud.grant.disable" in audit_actions
    assert "cloud.grant.revoke" not in audit_actions


@pytest.mark.asyncio
async def test_stage1_short_circuits_when_no_admins_active(
    db_session: AsyncSession,
) -> None:
    """When no ``it_admin`` / ``global_admin`` user has a valid
    work_email + active employment_status, Stage 1 must skip with ONE
    operator-visible warning instead of one error per group."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants
    from app.domains.cmdb.models import GRANT_NOTIFY_INACTIVE_OWNER

    # Create three (scope, person) groups all sitting at
    # ``inactive_owner`` with stamps old enough to escalate.
    owner = await _make_person(
        db_session, email="owner-noadmin@test.com", name="Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    very_old = datetime.now(UTC) - timedelta(days=400)
    very_old_login = datetime.now(UTC) - timedelta(days=120)
    grant_ids: list[uuid.UUID] = []
    for i in range(3):
        person = await _make_person(
            db_session,
            email=f"emp-noadmin-{i}@test.com", name=f"Emp {i}",
        )
        scope = await _make_scope(
            db_session, integration, external_id=f"aws-noadmin-{i}",
            owner_person_id=str(owner.id),
        )
        g = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id), person_id=str(person.id),
            role_key="ReadOnlyAccess",
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=very_old_login,
            last_notified_stage=GRANT_NOTIFY_INACTIVE_OWNER,
            source="discovered",
        )
        db_session.add(g)
        grant_ids.append(g.id)
    await db_session.commit()
    # Push updated_at back so escalation_cutoff fires.
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants "
        "SET updated_at = :ts WHERE id IN ({ids})".format(
            ids=",".join(f"'{gid}'" for gid in grant_ids),
        ),
    ), {"ts": datetime.now(UTC) - timedelta(days=20)})
    await db_session.commit()

    # Seed a real global_admin then mark every admin in the test
    # session (including ones earlier tests may have left around) as
    # ``left`` — exercises the active-employment filter inside the
    # sweep, not just the "no rows in auth_user_roles" SELECT branch.
    leaver_admin = await _make_person(
        db_session, email="leaver-admin-noadmin@test.com",
        name="Leaver Admin", roles=("global_admin",),
    )
    await db_session.execute(text(
        "UPDATE identity_persons SET employment_status='left' "
        "WHERE id IN ("
        "  SELECT user_id FROM auth_user_roles WHERE role_id IN ("
        "    SELECT id FROM auth_roles WHERE name IN "
        "    ('it_admin','global_admin')"
        "  )"
        ")"
    ))
    await db_session.commit()
    # Expire identity-map caches — without this, the sweep's
    # ``session.get(IdentityPerson)`` returns the cached pre-UPDATE
    # ``employment_status="active"`` and the active-admin filter
    # passes anyway, defeating the test.
    db_session.expire_all()
    assert leaver_admin is not None  # ensure the var stays referenced

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    # Zero IT escalations dispatched, zero groups advanced.
    assert summary["notified_it"] == 0, summary
    assert summary["groups_notified_it"] == 0, summary
    # Exactly ONE operator-visible warning (not 3 — one per group).
    snf_msgs = [e for e in summary["errors"] if "no IT admins" in e]
    assert len(snf_msgs) == 1, summary["errors"]


@pytest.mark.asyncio
async def test_stage1_picks_up_concurrent_renewal_via_fresh_state(
    db_session: AsyncSession,
) -> None:
    """If a grant's ``updated_at`` has been bumped past the
    escalation cutoff (e.g. admin renewed during the run), Stage 1
    must NOT advance the group — the fresh re-fetch should override
    any pre-run snapshot."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants
    from app.domains.cmdb.models import GRANT_NOTIFY_INACTIVE_OWNER

    employee = await _make_person(
        db_session, email="emp-renew@test.com", name="Renew Emp",
    )
    owner = await _make_person(
        db_session, email="owner-renew@test.com", name="Renew Owner",
    )
    # Need at least one admin or Stage 1 short-circuits before the
    # fresh-state branch ever runs.
    await _make_person(
        db_session, email="admin-renew@test.com", name="Admin",
        roles=("global_admin",),
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-renew",
        owner_person_id=str(owner.id),
    )

    very_old = datetime.now(UTC) - timedelta(days=400)
    very_old_login = datetime.now(UTC) - timedelta(days=120)
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id), person_id=str(employee.id),
        role_key="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=very_old, last_login_at=very_old_login,
        last_notified_stage=GRANT_NOTIFY_INACTIVE_OWNER,
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()
    # Updated_at = NOW (simulating a fresh admin renewal that landed
    # during the sweep run). Even though last_login_at is old enough
    # to satisfy "still inactive?", the fresh updated_at should defer
    # the group out of escalation.
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants SET updated_at = :ts WHERE id = :gid",
    ), {"ts": datetime.now(UTC), "gid": str(g.id)})
    await db_session.commit()

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    assert summary["groups_notified_it"] == 0, summary
    refreshed = (await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.id == g.id,
        )
    )).scalar_one()
    # Stage flag stays at owner — no escalation.
    assert refreshed.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER


@pytest.mark.asyncio
async def test_disable_helper_aborts_on_concurrent_updated_at_bump(
    db_session: AsyncSession,
) -> None:
    """In-lock TOCTOU guard: ``disable_grant_due_to_inactivity``
    re-reads ``updated_at`` after acquiring the per-grant lock and
    refuses to disable when the value has moved past the caller's
    snapshot. Simulates an admin renewal landing in the
    sweep-snapshot → lock-acquire window."""
    from app.domains.cmdb.lifecycle_service import CloudLifecycleService
    from app.domains.cmdb.models import GRANT_STATUS_DISABLED

    employee = await _make_person(
        db_session, email="emp-toctou@test.com", name="TOCTOU Emp",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-toctou",
    )

    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id), person_id=str(employee.id),
        role_key="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=400),
        last_login_at=datetime.now(UTC) - timedelta(days=120),
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()

    # Caller's "I read updated_at at this point" snapshot. The grant
    # in DB has a fresher updated_at (the row was just inserted so its
    # server-side default ``updated_at`` is "now"). Use a 10-day
    # window to absorb any CI clock skew between the test runner and
    # the MySQL/TiDB container — 1 hour was the prior margin and
    # caused intermittent failures on overloaded CI agents.
    stale_snapshot = datetime.now(UTC) - timedelta(days=10)

    service = CloudLifecycleService(db_session)
    # Note: the TOCTOU guard fires BEFORE the driver is loaded
    # (line ordering in ``_disable_grant_due_to_inactivity_locked``).
    # We don't need to patch the driver loader — the helper aborts
    # at the in-lock fresh-read check and returns
    # ``did_disable=False`` without ever entering the driver path.
    result, did_disable = await service.disable_grant_due_to_inactivity(
        str(g.id),
        trace_id="test.toctou",
        expected_updated_at_le=stale_snapshot,
    )

    # Helper aborted — boolean is False, status stayed active.
    assert did_disable is False, (
        "helper should have reported did_disable=False on TOCTOU abort"
    )
    assert result.status == GRANT_STATUS_ACTIVE, (
        "helper should have aborted on stale snapshot"
    )
    refreshed = (await db_session.execute(
        select(CmdbCloudAccessGrant).where(CmdbCloudAccessGrant.id == g.id)
    )).scalar_one()
    assert refreshed.status == GRANT_STATUS_ACTIVE
    assert refreshed.status != GRANT_STATUS_DISABLED


@pytest.mark.asyncio
async def test_sweep_counters_only_count_real_disables(
    db_session: AsyncSession,
) -> None:
    """Sweep counter arithmetic must reflect what actually happened,
    not what was attempted. Construct a 3-grant group at
    ``inactive_it`` past the disable cutoff (all three would normally
    be disabled), then monkey-patch
    ``CloudLifecycleService.disable_grant_due_to_inactivity`` to
    return ``did_disable=False`` for two of the three grants and
    ``did_disable=True`` for the third — simulating the TOCTOU /
    LockBusy / terminal-status short-circuit paths.

    Expected:
    * ``summary["disabled"] == 1`` (only the real disable bumps)
    * ``summary["groups_disabled"] == 1`` (at-least-one-success per group)

    Pre-2026-05-05 review-pass-2 the counters bumped per call, so
    this scenario would have reported ``disabled=3, groups_disabled=1``
    while only one row actually changed."""
    from app.domains.cmdb.lifecycle_service import (
        CloudLifecycleService,
        sweep_inactive_grants,
    )
    from app.domains.cmdb.models import (
        GRANT_NOTIFY_INACTIVE_IT,
        GRANT_STATUS_DISABLED,
    )

    employee = await _make_person(
        db_session, email="emp-counter@test.com", name="Counter Emp",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-counter",
    )

    very_old = datetime.now(UTC) - timedelta(days=400)
    very_old_login = datetime.now(UTC) - timedelta(days=120)
    role_keys = ["AdministratorAccess", "ReadOnlyAccess", "Billing"]
    grant_ids: list[uuid.UUID] = []
    for rk in role_keys:
        g = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=very_old_login,
            last_notified_stage=GRANT_NOTIFY_INACTIVE_IT,
            source="discovered",
        )
        db_session.add(g)
        grant_ids.append(g.id)
    await db_session.commit()
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants SET updated_at = :ts "
        "WHERE scope_id = :sid AND person_id = :pid",
    ), {
        "ts": datetime.now(UTC) - timedelta(days=45),
        "sid": str(scope.id),
        "pid": str(employee.id),
    })
    await db_session.commit()

    # Patch the helper: first two calls return short-circuit; third
    # call dispatches to the real implementation so one row truly
    # flips to disabled. The order maps to the sweep's per-grant
    # iteration order on this group (Python dict preserves insert
    # order; group's `gs` is built from the initial active-grants
    # SELECT in id order).
    real_helper = CloudLifecycleService.disable_grant_due_to_inactivity
    call_count = {"n": 0}

    async def _patched_helper(self, grant_id, *, trace_id, expected_updated_at_le=None):
        call_count["n"] += 1
        if call_count["n"] <= 2:
            # Short-circuit: pretend TOCTOU / terminal-status / LockBusy.
            from app.domains.cmdb.models import CmdbCloudAccessGrant as _G
            grant = await self.db.get(_G, uuid.UUID(grant_id))
            return grant, False
        return await real_helper(
            self,
            grant_id,
            trace_id=trace_id,
            expected_updated_at_le=expected_updated_at_le,
        )

    with patch.object(
        CloudLifecycleService,
        "disable_grant_due_to_inactivity",
        new=_patched_helper,
    ):
        summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    assert call_count["n"] == 3, "helper should be called once per grant in the group"
    # Counter only ticks for the one real disable.
    assert summary["disabled"] == 1, summary
    # Group ticks because at-least-one-success.
    assert summary["groups_disabled"] == 1, summary

    # DB confirms: exactly one row is actually disabled.
    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.person_id == str(employee.id),
        )
    )).scalars().all())
    disabled_count = sum(1 for r in rows if r.status == GRANT_STATUS_DISABLED)
    assert disabled_count == 1, [r.status for r in rows]


@pytest.mark.asyncio
async def test_sweep_counters_skip_when_entire_group_short_circuits(
    db_session: AsyncSession,
) -> None:
    """``summary["groups_disabled"]`` must NOT increment when EVERY
    grant in the group short-circuits. Pre-fix it ticked
    unconditionally per group, so a sweep tick where every helper
    call deferred (e.g. concurrent admin renewal across the whole
    group) still claimed "1 group disabled" in cron telemetry."""
    from app.domains.cmdb.lifecycle_service import (
        CloudLifecycleService,
        sweep_inactive_grants,
    )
    from app.domains.cmdb.models import GRANT_NOTIFY_INACTIVE_IT

    employee = await _make_person(
        db_session, email="emp-allshort@test.com", name="All Short",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="aws-allshort",
    )

    very_old = datetime.now(UTC) - timedelta(days=400)
    very_old_login = datetime.now(UTC) - timedelta(days=120)
    role_keys = ["ReadOnlyAccess", "Billing"]
    for rk in role_keys:
        g = CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(employee.id),
            role_key=rk,
            status=GRANT_STATUS_ACTIVE,
            granted_at=very_old,
            last_login_at=very_old_login,
            last_notified_stage=GRANT_NOTIFY_INACTIVE_IT,
            source="discovered",
        )
        db_session.add(g)
    await db_session.commit()
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants SET updated_at = :ts "
        "WHERE scope_id = :sid AND person_id = :pid",
    ), {
        "ts": datetime.now(UTC) - timedelta(days=45),
        "sid": str(scope.id),
        "pid": str(employee.id),
    })
    await db_session.commit()

    async def _always_short_circuit(self, grant_id, *, trace_id, expected_updated_at_le=None):
        from app.domains.cmdb.models import CmdbCloudAccessGrant as _G
        grant = await self.db.get(_G, uuid.UUID(grant_id))
        return grant, False

    with patch.object(
        CloudLifecycleService,
        "disable_grant_due_to_inactivity",
        new=_always_short_circuit,
    ):
        summary = await sweep_inactive_grants(db_session)
    await db_session.commit()

    assert summary["disabled"] == 0, summary
    assert summary["groups_disabled"] == 0, summary


# ── 4. Scope-owner exemption (2026-05-12) ─────────────────────────
#
# The scope's owner (``CmdbCloudScope.owner_person_id``) is the human
# responsible for approving every member's grant on the scope.
# Auto-revoking the owner's OWN grant on either schedule (90-day
# expires_at sweep or 60+14+30 inactivity sweep) creates a deadlock:
# the owner loses their access, can no longer approve renewals (their
# own or anyone else's), and re-requesting access to a scope you own
# is busywork. Codified after a user pointed at the gcp-digital-bi
# scope where the owner held 4 grants with 72 days remaining and
# asked why they were on the same clock as every other member.
#
# Symmetric with the existing privileged-role exemption — but
# evaluated at the (scope, person) pair level rather than globally.
# A person who is owner of scope A and a regular member on scope B
# still has their scope-B grant subject to the normal sweep cycles.


@pytest.mark.asyncio
async def test_expiry_sweep_skips_scope_owners_own_grant(
    db_session: AsyncSession,
) -> None:
    """T+0 revoke pass MUST skip a grant where
    ``grant.person_id == scope.owner_person_id``. Member grants on
    the same scope still revoke normally."""
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants

    owner = await _make_person(
        db_session, email="owner@test.com", name="Scope Owner",
    )
    member = await _make_person(
        db_session, email="member@test.com", name="Member",
    )
    integration = await _make_integration(db_session, provider="gcp")
    scope = await _make_scope(
        db_session, integration, external_id="gcp-bi",
        owner_person_id=str(owner.id), scope_type="gcp_project",
    )

    past = datetime.now(UTC) - timedelta(days=1)
    owner_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(owner.id),
        role_key="roles/owner",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=past,
    )
    member_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(member.id),
        role_key="roles/bigquery.dataViewer",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=past,
    )
    db_session.add_all([owner_grant, member_grant])
    await db_session.commit()

    owner_id = owner_grant.id
    member_id = member_grant.id

    summary = await sweep_expiring_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    assert summary["owner_skipped"] >= 1, summary
    owner_after = await db_session.get(CmdbCloudAccessGrant, owner_id)
    member_after = await db_session.get(CmdbCloudAccessGrant, member_id)
    assert owner_after is not None
    assert owner_after.status == GRANT_STATUS_ACTIVE, (
        "scope owner's own grant must not be auto-revoked on expiry"
    )
    assert member_after is not None
    assert member_after.status == GRANT_STATUS_EXPIRED, (
        "regular member grant on the same scope still expires"
    )


@pytest.mark.asyncio
async def test_expiry_sweep_skips_owner_reminder_notifications(
    db_session: AsyncSession,
) -> None:
    """T-14/T-7/T-1 reminder pass MUST also skip the scope owner.
    Without this guard the owner gets a "your access expires in N
    days" Feishu+email every reminder window for a grant the sweep
    will never actually revoke — pure spam.
    """
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants

    owner = await _make_person(
        db_session, email="owner-r@test.com", name="Owner R",
    )
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration, external_id="sub-r",
        owner_person_id=str(owner.id), scope_type="azure_subscription",
    )

    in_one_day = datetime.now(UTC) + timedelta(hours=12)
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(owner.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=89),
        expires_at=in_one_day,
    )
    db_session.add(g)
    await db_session.commit()
    g_id = g.id

    summary = await sweep_expiring_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    after = await db_session.get(CmdbCloudAccessGrant, g_id)
    assert after is not None
    assert after.last_notified_stage != GRANT_NOTIFY_T1, (
        "scope owner must not be flagged with the T-1 reminder stage"
    )
    assert summary["notified"] == 0, summary
    assert summary["owner_skipped"] >= 1, summary


@pytest.mark.asyncio
async def test_expiry_sweep_owner_exemption_is_per_pair_not_global(
    db_session: AsyncSession,
) -> None:
    """A person who owns scope A and holds a regular grant on scope B
    keeps the scope-B grant on the normal expiry clock. Exemption is
    keyed on the (scope, person) pair, NOT on the person alone.
    Same person, different scopes — different fate.
    """
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants

    person = await _make_person(
        db_session, email="dual@test.com", name="Dual Role",
    )
    other_owner = await _make_person(
        db_session, email="otherowner@test.com", name="Other Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope_owned = await _make_scope(
        db_session, integration, external_id="acct-owned",
        owner_person_id=str(person.id),
    )
    scope_member = await _make_scope(
        db_session, integration, external_id="acct-member",
        owner_person_id=str(other_owner.id),
    )

    past = datetime.now(UTC) - timedelta(days=1)
    g_owned = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope_owned.id),
        person_id=str(person.id),
        role_key="AdministratorAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=past,
    )
    g_member = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope_member.id),
        person_id=str(person.id),
        role_key="ReadOnlyAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=past,
    )
    db_session.add_all([g_owned, g_member])
    await db_session.commit()

    owned_id = g_owned.id
    member_id = g_member.id

    await sweep_expiring_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    owned_after = await db_session.get(CmdbCloudAccessGrant, owned_id)
    member_after = await db_session.get(CmdbCloudAccessGrant, member_id)
    assert owned_after is not None and owned_after.status == GRANT_STATUS_ACTIVE
    assert member_after is not None and member_after.status == GRANT_STATUS_EXPIRED


@pytest.mark.asyncio
async def test_expiry_sweep_leaves_prod_grants_fully_unmanaged(
    db_session: AsyncSession,
) -> None:
    """Production scopes are not managed by Helix's lifecycle sweep.

    A prod grant in a reminder window (T-14) gets NO reminder card, and
    a prod grant already past expiry is NEITHER auto-revoked NOR
    notified (member or owner). Regression for the 2026-07-16 prod
    feedback: a "一键续期" card was sent for ``dbaas-prod-clinic`` but
    clicking it always 403'd on the prod write lock ("链接无效或已过期").
    The fix stops sending prod reminders entirely and keeps prod out of
    every notification path.
    """
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants

    owner = await _make_person(
        db_session, email="prod-owner@test.com", name="Prod Owner",
    )
    member = await _make_person(
        db_session, email="prod-member@test.com", name="Prod Member",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="dbaas-prod-clinic",
        owner_person_id=str(owner.id),
    )
    scope.environment = "prod"
    await db_session.commit()

    # One grant in the T-14 reminder window, one already past expiry.
    reminder_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(member.id),
        role_key="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=76),
        expires_at=datetime.now(UTC) + timedelta(days=13, hours=12),
    )
    overdue_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(member.id),
        role_key="AdministratorAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=datetime.now(UTC) - timedelta(days=1),
    )
    db_session.add_all([reminder_grant, overdue_grant])
    await db_session.commit()
    reminder_id, overdue_id = reminder_grant.id, overdue_grant.id

    summary = await sweep_expiring_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    # No prod card sent; both counted as prod-skipped, none revoked.
    assert summary["notified"] == 0, summary
    assert summary["expired_revoked"] == 0, summary
    assert summary["prod_reminder_skipped"] >= 1, summary
    assert summary["prod_skipped"] >= 1, summary

    reminder_after = await db_session.get(CmdbCloudAccessGrant, reminder_id)
    overdue_after = await db_session.get(CmdbCloudAccessGrant, overdue_id)
    # Reminder-window grant untouched (no stage stamp = no card).
    assert reminder_after is not None
    assert reminder_after.status == GRANT_STATUS_ACTIVE
    assert reminder_after.last_notified_stage is None
    # Past-expiry grant held active (prod is never auto-recycled) and NOT
    # marked with any prod-overdue notification stage.
    assert overdue_after is not None
    assert overdue_after.status == GRANT_STATUS_ACTIVE
    assert overdue_after.last_notified_stage is None


@pytest.mark.asyncio
async def test_expiry_revoke_pass_derives_prod_from_constant(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The T+0 revoke pass must derive its prod exclusion from
    ``PROD_ENVIRONMENTS`` (via ``_is_prod_grant``), not a hardcoded
    ``env == "prod"``. Regression for the 2026-07-16 audit finding:
    growing the constant (e.g. adding ``"production"``) made the
    reminder pass skip the new env while the revoke pass still
    auto-revoked it at T+0 — the worst combination (no warning card,
    then silent revocation of production access).
    """
    import app.domains.cmdb.lifecycle_service as lifecycle_service
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants

    monkeypatch.setattr(
        lifecycle_service, "PROD_ENVIRONMENTS",
        frozenset({"prod", "production"}),
    )

    owner = await _make_person(
        db_session, email="prod2-owner@test.com", name="Prod2 Owner",
    )
    member = await _make_person(
        db_session, email="prod2-member@test.com", name="Prod2 Member",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="acct-production-env",
        owner_person_id=str(owner.id),
    )
    scope.environment = "production"
    await db_session.commit()

    overdue = CmdbCloudAccessGrant(
        id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(member.id),
        role_key="AdministratorAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=91),
        expires_at=datetime.now(UTC) - timedelta(days=1),
    )
    db_session.add(overdue)
    await db_session.commit()
    overdue_id = overdue.id

    summary = await sweep_expiring_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    assert summary["expired_revoked"] == 0, summary
    assert summary["prod_skipped"] >= 1, summary
    after = await db_session.get(CmdbCloudAccessGrant, overdue_id)
    assert after is not None
    assert after.status == GRANT_STATUS_ACTIVE, (
        "an env in PROD_ENVIRONMENTS must be held past expiry by the "
        "revoke pass, exactly like the reminder pass skips it"
    )
    assert after.last_notified_stage is None


# ── Prod exclusion in the INACTIVITY sweep (2026-07-16 audit) ───────
#
# PR #1961 established "production cloud access is NOT managed by
# Helix's lifecycle sweeps" and taught the EXPIRY sweep to honor it.
# The inactivity pipeline had zero prod exclusion: it sent owner/IT
# reminder cards about prod grants and at T+30 called
# ``driver.revoke_access`` against a production account — bypassing
# the prod write lock, which is router/MCP-scoped. These tests pin
# the fix: prod groups are skipped at every stage, and the disable
# helper itself refuses prod grants so no future caller regresses.


@pytest.mark.asyncio
async def test_inactivity_sweep_skips_prod_grants_at_stage0(
    db_session: AsyncSession,
) -> None:
    """A prod grant idle 100+ days gets NO inactivity reminder and no
    stage stamp; the SAME person's equally-idle non-prod grant on
    another scope keeps the normal Stage-0 handling (grouping is
    per-scope, so a mixed prod+nonprod person is not over- or
    under-swept)."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    owner = await _make_person(
        db_session, email="inact-prod-owner@test.com", name="IP Owner",
    )
    member = await _make_person(
        db_session, email="inact-prod-member@test.com", name="IP Member",
    )
    integration = await _make_integration(db_session, provider="aws")
    prod_scope = await _make_scope(
        db_session, integration, external_id="acct-prod-inact",
        owner_person_id=str(owner.id),
    )
    prod_scope.environment = "prod"
    nonprod_scope = await _make_scope(
        db_session, integration, external_id="acct-dev-inact",
        owner_person_id=str(owner.id),
    )
    await db_session.commit()

    long_ago = datetime.now(UTC) - timedelta(days=400)
    idle_login = datetime.now(UTC) - timedelta(days=120)
    prod_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(prod_scope.id), person_id=str(member.id),
        role_key="AdministratorAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago, last_login_at=idle_login,
        last_notified_stage=None, source="discovered",
    )
    nonprod_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(nonprod_scope.id), person_id=str(member.id),
        role_key="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago, last_login_at=idle_login,
        last_notified_stage=None, source="discovered",
    )
    db_session.add_all([prod_grant, nonprod_grant])
    await db_session.commit()
    prod_id, nonprod_id = prod_grant.id, nonprod_grant.id

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    prod_after = await db_session.get(CmdbCloudAccessGrant, prod_id)
    nonprod_after = await db_session.get(CmdbCloudAccessGrant, nonprod_id)
    assert prod_after is not None
    assert prod_after.last_notified_stage is None, (
        "prod grant must not receive an inactivity owner-reminder stamp"
    )
    assert prod_after.status == GRANT_STATUS_ACTIVE
    assert nonprod_after is not None
    assert nonprod_after.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER, (
        "the same person's non-prod group must keep normal Stage-0 handling"
    )
    assert summary["skipped_prod"] >= 1, summary
    assert summary["groups_notified_owner"] == 1, summary


@pytest.mark.asyncio
async def test_inactivity_sweep_never_disables_prod_grant(
    db_session: AsyncSession,
) -> None:
    """A prod grant carrying a stale ``inactive_it`` flag past the
    disable cutoff must NOT be disabled and the cloud driver must
    NEVER be loaded — pre-fix Stage 2 called ``driver.revoke_access``
    against the production account (bypassing the router/MCP-scoped
    prod write lock)."""
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    owner = await _make_person(
        db_session, email="inact-prod-owner2@test.com", name="IP Owner 2",
    )
    member = await _make_person(
        db_session, email="inact-prod-member2@test.com", name="IP Member 2",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="acct-prod-disable",
        owner_person_id=str(owner.id),
    )
    scope.environment = "prod"
    await db_session.commit()

    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id), person_id=str(member.id),
        role_key="AdministratorAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=400),
        last_login_at=datetime.now(UTC) - timedelta(days=120),
        last_notified_stage=GRANT_NOTIFY_INACTIVE_IT,
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()
    await db_session.execute(text(
        "UPDATE cmdb_cloud_access_grants SET updated_at = :ts "
        "WHERE id = :gid",
    ), {"ts": datetime.now(UTC) - timedelta(days=45), "gid": str(g.id)})
    await db_session.commit()
    g_id = g.id

    async def _forbidden_loader(session, scope):  # noqa: ARG001
        raise AssertionError(
            "driver must never be loaded for a prod grant"
        )

    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        side_effect=_forbidden_loader,
    ) as loader:
        summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    # Driver errors are suppressed inside the disable helper, so the
    # loader NOT being called is the assertion that matters — a
    # raised AssertionError alone would have been swallowed and the
    # row disabled anyway.
    assert loader.call_count == 0, (
        "inactivity sweep must not touch the cloud driver for prod grants"
    )
    assert summary["disabled"] == 0, summary
    assert summary["groups_disabled"] == 0, summary
    assert summary["skipped_prod"] >= 1, summary

    after = await db_session.get(CmdbCloudAccessGrant, g_id)
    assert after is not None
    assert after.status == GRANT_STATUS_ACTIVE, (
        "prod grant must be held active — never auto-disabled"
    )
    assert after.last_notified_stage == GRANT_NOTIFY_INACTIVE_IT


@pytest.mark.asyncio
async def test_disable_helper_refuses_prod_grant(
    db_session: AsyncSession,
) -> None:
    """Belt-and-braces: ``disable_grant_due_to_inactivity`` itself
    raises ``CLOUD_SCOPE_PROD_BLOCKED`` on a prod grant, so a future
    caller that bypasses the sweep's group-level skip still cannot
    auto-disable production access."""
    from app.core.exceptions import AppException
    from app.domains.cmdb.lifecycle_service import CloudLifecycleService

    member = await _make_person(
        db_session, email="prod-refusal@test.com", name="Prod Refusal",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="acct-prod-refusal",
    )
    scope.environment = "prod"
    await db_session.commit()

    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id), person_id=str(member.id),
        role_key="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=400),
        last_login_at=datetime.now(UTC) - timedelta(days=120),
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()
    g_id = g.id

    async def _forbidden_loader(session, scope):  # noqa: ARG001
        raise AssertionError(
            "driver must never be loaded for a prod grant"
        )

    service = CloudLifecycleService(db_session)
    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        side_effect=_forbidden_loader,
    ) as loader:
        with pytest.raises(AppException) as exc_info:
            await service.disable_grant_due_to_inactivity(
                str(g_id), trace_id="test.prod-refusal",
            )

    assert exc_info.value.code == "CLOUD_SCOPE_PROD_BLOCKED"
    assert exc_info.value.status_code == 403
    assert loader.call_count == 0

    refreshed = (await db_session.execute(
        select(CmdbCloudAccessGrant).where(CmdbCloudAccessGrant.id == g_id)
    )).scalar_one()
    assert refreshed.status == GRANT_STATUS_ACTIVE, (
        "refusal must leave the prod grant untouched"
    )


@pytest.mark.asyncio
async def test_inactivity_sweep_skips_scope_owner_at_stage0(
    db_session: AsyncSession,
) -> None:
    """Stage 0 (T+60 → ``inactive_owner``) MUST skip a (scope, person)
    pair where ``person_id == scope.owner_person_id`` even when the
    group's last_login_at is well past the inactivity cutoff. Same
    rationale as the privileged-role exemption — the owner manages
    their own access, the sweep doesn't.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    owner = await _make_person(
        db_session, email="iowner@test.com", name="Inactive Owner",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="acct-iown",
        owner_person_id=str(owner.id),
    )

    long_ago = datetime.now(UTC) - timedelta(days=400)
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(owner.id),
        role_key="AdministratorAccess",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago,
        last_login_at=None,
        last_notified_stage=None,
        source="discovered",
    )
    db_session.add(g)
    await db_session.commit()
    g_id = g.id

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    after = await db_session.get(CmdbCloudAccessGrant, g_id)
    assert after is not None
    assert after.last_notified_stage is None, (
        "scope owner's own grant must not advance to inactive_owner stage"
    )
    assert summary["skipped_owner"] >= 1, summary


@pytest.mark.asyncio
async def test_inactivity_sweep_skips_owner_at_stages_1_and_2(
    db_session: AsyncSession,
) -> None:
    """Defense in depth — same case as the privileged-role
    test_sweep_skips_privileged_at_stages_1_and_2: a (scope, person)
    pair that ENTERED inactive_owner / inactive_it before becoming
    the scope's owner (e.g. owner was rebound to this person while
    their group sat at stage 1) must not advance/disable at Stage 1
    or Stage 2.
    """
    from app.domains.cmdb.lifecycle_service import sweep_inactive_grants

    rebound_owner = await _make_person(
        db_session, email="rebound@test.com", name="Rebound Owner",
    )
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration, external_id="sub-rebound",
        owner_person_id=str(rebound_owner.id),
        scope_type="azure_subscription",
    )

    long_ago_login = datetime.now(UTC) - timedelta(days=100)
    long_ago_update = datetime.now(UTC) - timedelta(days=20)
    g_stage1 = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(rebound_owner.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=long_ago_update,
        last_login_at=long_ago_login,
        last_notified_stage=GRANT_NOTIFY_INACTIVE_OWNER,
    )
    db_session.add(g_stage1)
    await db_session.commit()
    await db_session.execute(
        text("UPDATE cmdb_cloud_access_grants SET updated_at = :u WHERE id = :id"),
        {"u": long_ago_update, "id": str(g_stage1.id)},
    )
    await db_session.commit()
    g_id = g_stage1.id

    summary = await sweep_inactive_grants(db_session)
    await db_session.commit()
    db_session.expire_all()

    after = await db_session.get(CmdbCloudAccessGrant, g_id)
    assert after is not None
    assert after.last_notified_stage == GRANT_NOTIFY_INACTIVE_OWNER, (
        "scope owner must not advance from inactive_owner → inactive_it"
    )
    assert after.last_notified_stage != GRANT_NOTIFY_INACTIVE_IT
    assert after.status == GRANT_STATUS_ACTIVE
    assert summary["skipped_owner"] >= 1, summary


# ── 5. Per-item commit (2026-07-10 login-outage regression) ────────
#
# ``notification_outbox.recipient_person_id`` FK-references
# ``identity_persons.id``; TiDB's FK check locks the parent row until
# COMMIT. The expiry sweep used to run its whole pass in ONE
# transaction — on 2026-07-10 a T-14 reminder storm (821
# notifications, 20-minute run) held FK locks on every notified
# employee's ``identity_persons`` row, so their login's
# ``UPDATE identity_persons SET last_login_at`` hit 1205 lock-wait
# timeouts and Google/Feishu login 500'd platform-wide for the whole
# run. The fix commits after every processed grant so a lock is held
# only for that item's dispatch, not the full sweep.


@pytest.mark.asyncio
async def test_expiry_sweep_commits_each_grant_before_moving_on(
    db_session: AsyncSession,
) -> None:
    """Processed grants must already be COMMITTED when the sweep
    returns — a session.rollback() afterwards (simulating a crash
    before the cron handler's final commit) must not undo them. On the
    pre-fix single-transaction sweep this rollback wiped every
    ``last_notified_stage`` write, proving the locks were held to the
    very end."""
    from app.domains.cmdb.lifecycle_service import sweep_expiring_grants
    from app.domains.cmdb.models import GRANT_NOTIFY_T14

    owner = await _make_person(
        db_session, email="owner-c@test.com", name="Owner C",
    )
    member_a = await _make_person(
        db_session, email="member-a@test.com", name="Member A",
    )
    member_b = await _make_person(
        db_session, email="member-b@test.com", name="Member B",
    )
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(
        db_session, integration, external_id="acct-commit",
        owner_person_id=str(owner.id),
    )

    # Both grants fall in the T-14 reminder window.
    expires_t14 = datetime.now(UTC) + timedelta(days=13, hours=12)
    grants = [
        CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(p.id),
            role_key="ReadOnlyAccess",
            status=GRANT_STATUS_ACTIVE,
            granted_at=datetime.now(UTC) - timedelta(days=76),
            expires_at=expires_t14,
        )
        for p in (member_a, member_b)
    ]
    db_session.add_all(grants)
    await db_session.commit()
    grant_ids = [g.id for g in grants]

    summary = await sweep_expiring_grants(db_session)
    assert summary["notified"] == 2, summary

    # Simulate the cron handler dying before its final commit.
    await db_session.rollback()
    db_session.expire_all()

    for gid in grant_ids:
        after = await db_session.get(CmdbCloudAccessGrant, gid)
        assert after is not None
        assert after.last_notified_stage == GRANT_NOTIFY_T14, (
            "per-grant work must be committed inside the sweep — "
            "holding it for one end-of-run commit keeps FK locks on "
            "identity_persons and blocks those employees' logins "
            "(2026-07-10 outage)"
        )