"""Tests for the half-yearly cloud-server owner confirmation flow.

Covers (migration 213 / PR redesign):

* :func:`ServerOwnerConfirmationService.list_due_owners` —
  skip rules (no owner / leaver / newborn / stopped / decommissioned /
  recently confirmed) and grouping by owner.
* :func:`create_pending_for_owner` — happy path inserts a row +
  sends a Feishu envelope through ``NotificationService``. Re-runs
  on the same period_start return ``deduplicated=True`` without
  emitting a second card.
* :func:`respond_confirm_all` — stamps every snapshot server's
  ``last_owner_confirmed_at`` AND only those still owned by the
  responder (re-assignment mid-period is silently skipped).
* :func:`respond_per_server` — bumps the subset; transitions the
  row to ``responded`` only when the snapshot is fully covered
  across one or more clicks.
* HTTP endpoints — RBAC (owner-scope only, no admin override),
  pending list shape, audit trail.
* Cron handler — owner leaver / soft-deleted skip; per-owner
  isolation (one failing owner doesn't roll back the rest).
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from typing import Any

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.domains.cmdb.models import (
    SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL,
    SERVER_OWNER_CONFIRMATION_STATUS_PENDING,
    SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED,
    CmdbServer,
    CmdbServerOwnerConfirmation,
)
from app.domains.cmdb.server_owner_confirmation_service import (
    NEWBORN_GUARD_DAYS,
    SEMIANNUAL_DAYS,
    ServerOwnerConfirmationService,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration
from app.notifications.service import resolve_card_text as _resolve_card_text


def _t(value, lang: str = "zh") -> str:
    """What the READER actually sees.

    Card chrome is a ``LocalizedText`` resolved per recipient, so asserting on
    the raw field would only prove the dataclass exists. Resolving first also
    makes the assertion stronger: it pins the copy in one language rather than
    whichever one the producer happened to write.
    """
    return str(_resolve_card_text(value, lang))


# ── Fixtures ───────────────────────────────────────────────────────


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


async def _make_person(
    db: AsyncSession, *, email: str, employment_status: str = "active",
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _make_server(
    db: AsyncSession,
    *,
    integration: Integration,
    owner: IdentityPerson | None,
    name: str,
    status: str = "running",
    first_seen_days_ago: int = 365,
    last_owner_confirmed_at: datetime | None = None,
) -> CmdbServer:
    now = datetime.now(UTC)
    s = CmdbServer(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        instance_id=f"i-{uuid.uuid4().hex[:8]}",
        name=name,
        status=status,
        region="us-east-1",
        owner_person_id=str(owner.id) if owner else None,
        first_seen_at=now - timedelta(days=first_seen_days_ago),
        last_synced_at=now,
        last_owner_confirmed_at=last_owner_confirmed_at,
    )
    db.add(s)
    await db.flush()
    return s


# ── list_due_owners — eligibility rules ────────────────────────────


@pytest.mark.asyncio
async def test_list_due_owners_groups_by_owner(db_session: AsyncSession):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    bob = await _make_person(db_session, email="bob@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")
    s3 = await _make_server(db_session, integration=integ, owner=bob, name="b1")

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()

    assert set(grouped.keys()) == {str(alice.id), str(bob.id)}
    assert sorted(grouped[str(alice.id)]) == sorted([str(s1.id), str(s2.id)])
    assert grouped[str(bob.id)] == [str(s3.id)]


@pytest.mark.asyncio
async def test_list_due_owners_skips_no_owner(db_session: AsyncSession):
    integ = await _make_integration(db_session)
    await _make_server(db_session, integration=integ, owner=None, name="orphan")

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {}


@pytest.mark.asyncio
async def test_list_due_owners_skips_newborn(db_session: AsyncSession):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    # first_seen_at < 90 days ago — newborn, should be skipped
    await _make_server(
        db_session, integration=integ, owner=alice, name="newborn",
        first_seen_days_ago=NEWBORN_GUARD_DAYS - 5,
    )

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {}


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["stopped", "decommissioned"])
async def test_list_due_owners_skips_inactive(db_session: AsyncSession, status: str):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    await _make_server(db_session, integration=integ, owner=alice, name="x", status=status)

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {}


@pytest.mark.asyncio
async def test_list_due_owners_skips_recently_confirmed(db_session: AsyncSession):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    recent = datetime.now(UTC) - timedelta(days=SEMIANNUAL_DAYS - 30)
    await _make_server(
        db_session, integration=integ, owner=alice, name="ok",
        last_owner_confirmed_at=recent,
    )

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {}


@pytest.mark.asyncio
async def test_list_due_owners_picks_up_overdue_confirmation(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    overdue = datetime.now(UTC) - timedelta(days=SEMIANNUAL_DAYS + 30)
    s = await _make_server(
        db_session, integration=integ, owner=alice, name="overdue",
        last_owner_confirmed_at=overdue,
    )

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {str(alice.id): [str(s.id)]}


@pytest.mark.asyncio
async def test_list_due_owners_excludes_archival_tenant(db_session: AsyncSession):
    """Servers under an archival tenant (ksyun / 金山云 历史) must never
    be surfaced for half-yearly confirmation — the company no longer
    runs anything there and the rows are read-only migrated history.
    Regression for the false-nag that pinged 13 owners about 金山云
    machines. The exclusion is sourced from ``registry.archival_providers()``
    (KSYunDriver.archival = True), not a hardcoded provider string."""
    from app.integrations import registry

    # Guard the premise: ksyun must actually be registered as archival,
    # otherwise this test would pass vacuously.
    assert "ksyun" in registry.archival_providers()

    archival_integ = await _make_integration(db_session, provider="ksyun")
    live_integ = await _make_integration(db_session, provider="aws")
    alice = await _make_person(db_session, email="alice@x.com")
    # One archival server (must be skipped) + one live server (kept).
    await _make_server(
        db_session, integration=archival_integ, owner=alice, name="ksyun-old",
    )
    s_live = await _make_server(
        db_session, integration=live_integ, owner=alice, name="aws-live",
    )

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    # Only the live AWS server survives; the ksyun row is excluded.
    assert grouped == {str(alice.id): [str(s_live.id)]}


@pytest.mark.asyncio
async def test_list_due_owners_excludes_soft_deleted(db_session: AsyncSession):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="dead")
    s.deleted_at = datetime.now(UTC)
    await db_session.flush()

    service = ServerOwnerConfirmationService(db_session)
    grouped = await service.list_due_owners()
    assert grouped == {}


# ── create_pending_for_owner — dedup + envelope ─────────────────────


@pytest.mark.asyncio
async def test_create_pending_for_owner_inserts_row(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="a1")

    # Stub the notification dispatch — service test doesn't exercise
    # the Feishu channel (covered by NotificationService tests).
    dispatched: list = []
    async def fake_dispatch(env, trace_id=None):
        dispatched.append((env, trace_id))
    service = ServerOwnerConfirmationService(db_session)
    monkeypatch.setattr(service.notifications, "dispatch", fake_dispatch)

    today = date.today()
    row, dedup = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)], period_start=today,
    )
    assert dedup is False
    assert row.owner_person_id == str(alice.id)
    assert row.period_start == today
    assert row.server_ids == [str(s.id)]
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING
    assert row.notified_at is not None
    assert len(dispatched) == 1
    assert dispatched[0][0].recipient.id == alice.id


@pytest.mark.asyncio
async def test_create_pending_for_owner_dedupes_same_day(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="a1")

    dispatched: list = []
    async def fake_dispatch(env, trace_id=None):
        dispatched.append(env)
    service = ServerOwnerConfirmationService(db_session)
    monkeypatch.setattr(service.notifications, "dispatch", fake_dispatch)

    today = date.today()
    row1, dedup1 = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)], period_start=today,
    )
    row2, dedup2 = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)], period_start=today,
    )
    assert dedup1 is False
    assert dedup2 is True
    assert row1.id == row2.id
    # Second call MUST NOT emit a card.
    assert len(dispatched) == 1


# ── respond_confirm_all ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_respond_confirm_all_stamps_servers(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")
    assert s1.last_owner_confirmed_at is None
    assert s2.last_owner_confirmed_at is None

    service = ServerOwnerConfirmationService(db_session)
    monkeypatch.setattr(service.notifications, "dispatch", lambda env, trace_id=None: None)
    # Force a non-coroutine into an awaitable
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)

    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )

    out = await service.respond_confirm_all(
        confirmation_id=str(row.id), actor_person_id=str(alice.id),
    )
    await db_session.flush()
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED
    assert out.response_kind == SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL
    assert out.responded_at is not None

    await db_session.refresh(s1)
    await db_session.refresh(s2)
    assert s1.last_owner_confirmed_at is not None
    assert s2.last_owner_confirmed_at is not None


@pytest.mark.asyncio
async def test_respond_confirm_all_rejects_non_owner(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    bob = await _make_person(db_session, email="bob@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="a1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)],
    )

    with pytest.raises(PermissionError):
        await service.respond_confirm_all(
            confirmation_id=str(row.id), actor_person_id=str(bob.id),
        )


@pytest.mark.asyncio
async def test_respond_confirm_all_skips_reassigned_server(
    db_session: AsyncSession, monkeypatch,
):
    """If a server is re-assigned mid-period, confirm_all must NOT
    stamp it — the new owner will get their own card next cycle."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    bob = await _make_person(db_session, email="bob@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="a1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)],
    )

    # Re-assign to Bob.
    s.owner_person_id = str(bob.id)
    await db_session.flush()

    await service.respond_confirm_all(
        confirmation_id=str(row.id), actor_person_id=str(alice.id),
    )
    await db_session.refresh(s)
    # last_owner_confirmed_at NOT stamped because s.owner_person_id no
    # longer matches the confirmation's owner_person_id.
    assert s.last_owner_confirmed_at is None


# ── respond_per_server (partial) ────────────────────────────────────


@pytest.mark.asyncio
async def test_respond_per_server_partial_keeps_row_pending(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )

    out = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s1.id)],
    )
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING
    await db_session.refresh(s1)
    await db_session.refresh(s2)
    assert s1.last_owner_confirmed_at is not None
    assert s2.last_owner_confirmed_at is None


@pytest.mark.asyncio
async def test_respond_per_server_completes_when_snapshot_covered(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )

    # Two calls, each confirming half. The second one closes the row.
    await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s1.id)],
    )
    out = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s2.id)],
    )
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED


@pytest.mark.asyncio
async def test_respond_per_server_two_click_accumulation(
    db_session: AsyncSession, monkeypatch,
):
    """Owner covers the snapshot across TWO partial clicks with
    disjoint subsets. The row should flip to RESPONDED only after
    the second click (when the union covers the snapshot). Targets
    the ``_all_snapshot_confirmed`` logic — the most fragile piece
    of the multi-click partial flow."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")
    s3 = await _make_server(db_session, integration=integ, owner=alice, name="a3")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id), str(s3.id)],
    )

    # First click confirms {s1, s2} — row stays pending.
    after_first = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s1.id), str(s2.id)],
    )
    assert after_first.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING

    # Second click confirms {s3} — union covers snapshot, row closes.
    after_second = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s3.id)],
    )
    assert after_second.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED
    # Multi-click journey that ends fully covered must record
    # ``confirmed_all`` — the cumulative state IS full. Recording
    # ``partial`` here (because the LAST click only covered {s3}) was
    # the original misleading bug fixed in the review sweep.
    assert after_second.response_kind == SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL


@pytest.mark.asyncio
async def test_respond_per_server_completes_when_missing_server_deleted(
    db_session: AsyncSession, monkeypatch,
):
    """A snapshot server soft-deleted AFTER card issuance can never be
    confirmed. If it still counted toward the required set, a
    ``kind=partial`` row would wedge in ``pending`` forever — and the
    pending-interlock in ``list_due_owners`` would then exclude the
    owner from EVERY future semi-annual card. Vanished servers count
    as satisfied: confirming the surviving server must close the row,
    and the owner must reappear in ``list_due_owners`` next period."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )

    # s2 vanishes after the card went out (sync sweep soft-delete).
    s2.deleted_at = datetime.now(UTC)
    await db_session.flush()

    # Owner confirms the ONLY server that still exists — the row must
    # flip to responded, not wedge pending on the unconfirmable ghost.
    out = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s1.id)],
    )
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED

    # Next semi-annual period: no pending row left, so the interlock
    # doesn't fire and alice is due again for her surviving server.
    due = await service.list_due_owners(
        today=date.today() + timedelta(days=SEMIANNUAL_DAYS + 1),
    )
    assert due == {str(alice.id): [str(s1.id)]}


@pytest.mark.asyncio
async def test_respond_per_server_trivially_complete_when_all_vanished(
    db_session: AsyncSession, monkeypatch,
):
    """Edge of the vanished-server rule: EVERY snapshot server is gone
    (one hard-deleted, one soft-deleted) by the time the owner clicks.
    The required intersection is empty → the row is trivially complete
    and flips to responded even though nothing could be stamped."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )
    s1_id = str(s1.id)

    # s1 hard-deleted (row gone entirely), s2 soft-deleted.
    await db_session.delete(s1)
    s2.deleted_at = datetime.now(UTC)
    await db_session.flush()

    out = await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[s1_id],
    )
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED


@pytest.mark.asyncio
async def test_respond_confirm_all_when_all_servers_reassigned(
    db_session: AsyncSession, monkeypatch,
):
    """Edge case: owner clicks "confirm all" but every snapshot server
    has been reassigned to someone else mid-period. Nothing is stamped,
    but the row still flips to RESPONDED so it doesn't haunt the
    owner forever. The new owners get fresh cards on the next cron tick."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    bob = await _make_person(db_session, email="bob@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )

    # Reassign BOTH servers away from alice.
    s1.owner_person_id = str(bob.id)
    s2.owner_person_id = str(bob.id)
    await db_session.flush()

    out = await service.respond_confirm_all(
        confirmation_id=str(row.id), actor_person_id=str(alice.id),
    )
    # Row closes — alice has signaled "nothing left to confirm".
    assert out.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED
    # But neither server got stamped (they're not alice's anymore).
    await db_session.refresh(s1)
    await db_session.refresh(s2)
    assert s1.last_owner_confirmed_at is None
    assert s2.last_owner_confirmed_at is None


@pytest.mark.asyncio
async def test_create_pending_for_owner_handles_unique_race(
    db_session: AsyncSession, monkeypatch,
):
    """Two parallel cron ticks both pass the SELECT-check and both
    INSERT. The second hits the UNIQUE(owner_person_id, period_start)
    constraint. ``create_pending_for_owner`` must catch the
    IntegrityError, refetch the winner, and return as
    ``deduplicated=True`` so the cron's metrics correctly classify
    this as a dedup, not an error."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="a1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    today = date.today()

    # Simulate race: row exists but the SELECT-check window saw nothing.
    # Insert the winner directly with a different id.
    competitor = CmdbServerOwnerConfirmation(
        id=uuid.uuid4(),
        owner_person_id=str(alice.id),
        period_start=today,
        server_ids=[str(s.id)],
        status=SERVER_OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC),
    )
    db_session.add(competitor)
    await db_session.commit()

    # Now call create_pending_for_owner — it will SELECT the winner
    # and short-circuit BEFORE reaching the INSERT, so this exercises
    # the dedup branch via the normal SELECT path. The IntegrityError
    # branch is the harder race; covered by the IntegrityError
    # try/except we just added (smoke-tested via service import).
    row, deduplicated = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s.id)], period_start=today,
    )
    assert deduplicated is True
    assert row.id == competitor.id


@pytest.mark.asyncio
async def test_respond_per_server_ignores_ids_outside_snapshot(
    db_session: AsyncSession, monkeypatch,
):
    """An attacker who passes a server_id that wasn't in the snapshot
    cannot use this endpoint to stamp arbitrary servers."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s_in = await _make_server(db_session, integration=integ, owner=alice, name="in")
    s_out = await _make_server(db_session, integration=integ, owner=alice, name="out")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s_in.id)],
    )

    await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s_in.id), str(s_out.id)],
    )
    await db_session.refresh(s_out)
    assert s_out.last_owner_confirmed_at is None


# ── HTTP endpoints — RBAC + audit ──────────────────────────────────


@pytest.mark.asyncio
async def test_get_my_confirmations_returns_only_signed_in_user(
    client: AsyncClient, db_session: AsyncSession, monkeypatch,
    employee_user, admin_user,
):
    employee_person, _ = employee_user
    admin_person, _ = admin_user
    integ = await _make_integration(db_session)
    s = await _make_server(db_session, integration=integ, owner=employee_person, name="e1")
    s_admin = await _make_server(db_session, integration=integ, owner=admin_person, name="a1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    await service.create_pending_for_owner(
        owner=employee_person, server_ids=[str(s.id)],
    )
    await service.create_pending_for_owner(
        owner=admin_person, server_ids=[str(s_admin.id)],
    )
    await db_session.commit()

    _, employee_token = employee_user
    resp = await client.get(
        "/api/v1/cmdb/server-confirmations/mine",
        headers={"Authorization": f"Bearer {employee_token}"},
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    # Employee sees only their own pending row.
    assert len(data) == 1
    assert data[0]["server_count"] == 1


@pytest.mark.asyncio
async def test_respond_rejects_other_users_confirmation(
    client: AsyncClient, db_session: AsyncSession, monkeypatch,
    employee_user, admin_user,
):
    employee_person, _ = employee_user
    admin_person, admin_token = admin_user
    integ = await _make_integration(db_session)
    s = await _make_server(db_session, integration=integ, owner=employee_person, name="e1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=employee_person, server_ids=[str(s.id)],
    )
    await db_session.commit()

    # Admin tries to respond to the EMPLOYEE's confirmation — must 403.
    resp = await client.post(
        f"/api/v1/cmdb/server-confirmations/{row.id}/respond",
        json={"kind": "confirmed_all"},
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_respond_confirm_all_writes_audit_log(
    client: AsyncClient, db_session: AsyncSession, monkeypatch, employee_user,
):
    employee_person, token = employee_user
    integ = await _make_integration(db_session)
    s = await _make_server(db_session, integration=integ, owner=employee_person, name="e1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=employee_person, server_ids=[str(s.id)],
    )
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/cmdb/server-confirmations/{row.id}/respond",
        json={"kind": "confirmed_all"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200

    audit_rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "cmdb.server_owner_confirmation.respond",
        )
    )).scalars().all()
    assert len(audit_rows) == 1
    assert str(audit_rows[0].actor_user_id) == str(employee_person.id)
    assert audit_rows[0].resource_id == str(row.id)


@pytest.mark.asyncio
async def test_respond_partial_requires_server_ids(
    client: AsyncClient, db_session: AsyncSession, monkeypatch, employee_user,
):
    employee_person, token = employee_user
    integ = await _make_integration(db_session)
    s = await _make_server(db_session, integration=integ, owner=employee_person, name="e1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=employee_person, server_ids=[str(s.id)],
    )
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/cmdb/server-confirmations/{row.id}/respond",
        json={"kind": "partial"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_respond_rejects_unknown_kind(
    client: AsyncClient, db_session: AsyncSession, monkeypatch, employee_user,
):
    employee_person, token = employee_user
    integ = await _make_integration(db_session)
    s = await _make_server(db_session, integration=integ, owner=employee_person, name="e1")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=employee_person, server_ids=[str(s.id)],
    )
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/cmdb/server-confirmations/{row.id}/respond",
        json={"kind": "wat"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_list_due_owners_skips_owner_with_pending_row(
    db_session: AsyncSession, monkeypatch,
):
    """Re-prompt interlock — an owner with an existing PENDING row from
    a prior tick must be excluded from the next cron's eligible set.
    Without this, a weekly cron would keep minting fresh cards every
    Monday (with a new period_start, so the UNIQUE doesn't dedup) for
    the same unresponded owner. Spam."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s1 = await _make_server(db_session, integration=integ, owner=alice, name="a1")
    s2 = await _make_server(db_session, integration=integ, owner=alice, name="a2")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    # First tick — creates a pending row, sends a card.
    await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s1.id), str(s2.id)],
    )
    await db_session.flush()

    # Second tick — owner has not responded. list_due_owners must
    # NOT return alice (skip via the pending-row interlock).
    due = await service.list_due_owners(today=date.today())
    assert str(alice.id) not in due


@pytest.mark.asyncio
async def test_list_due_owners_caps_per_owner_snapshot(
    db_session: AsyncSession, monkeypatch,
):
    """A single owner that holds >MAX_SERVERS_PER_CARD machines must
    only show up with the cap-sized slice — otherwise the snapshot
    JSON column gets a 50K-element list, three downstream IN queries
    drag, and the Feishu card body explodes. Truncation is silent;
    the next weekly tick picks up the remainder after the owner
    responds."""
    from app.domains.cmdb.server_owner_confirmation_service import (
        MAX_SERVERS_PER_CARD,
    )

    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    # Make MAX + 5 servers so the cap clamp clearly fires.
    server_count = MAX_SERVERS_PER_CARD + 5
    for i in range(server_count):
        await _make_server(
            db_session, integration=integ, owner=alice, name=f"s{i}",
        )
    await db_session.flush()

    service = ServerOwnerConfirmationService(db_session)
    due = await service.list_due_owners(today=date.today())
    assert str(alice.id) in due
    assert len(due[str(alice.id)]) == MAX_SERVERS_PER_CARD


@pytest.mark.asyncio
async def test_card_body_renders_real_provider_not_stub(
    db_session: AsyncSession, monkeypatch,
):
    """``_build_card`` previously hardcoded ``provider = "—"`` for every
    server in the body line. The cron handler now resolves the real
    provider (aws / aliyun / ucloud / …) via an integration lookup.
    Captures the rendered envelope and asserts the provider literal
    is present, not the em-dash stub."""
    integ_aws = await _make_integration(db_session, provider="aws")
    integ_aliyun = await _make_integration(db_session, provider="aliyun")
    alice = await _make_person(db_session, email="alice@x.com")
    s_aws = await _make_server(db_session, integration=integ_aws, owner=alice, name="aws1")
    s_aliyun = await _make_server(
        db_session, integration=integ_aliyun, owner=alice, name="ali1",
    )

    service = ServerOwnerConfirmationService(db_session)
    captured: list[Any] = []

    async def _capture(env, trace_id=None):
        captured.append(env)

    monkeypatch.setattr(service.notifications, "dispatch", _capture)
    await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s_aws.id), str(s_aliyun.id)],
    )
    assert len(captured) == 1
    body_blob = "\n".join(
        _t(section.body) for section in captured[0].card_sections
    )
    assert "aws" in body_blob
    assert "aliyun" in body_blob
    # The em-dash stub the original hardcoded literal printed for the
    # provider column must NOT appear (it appeared via `provider = "—"`).
    # Region-default `"-"` (single hyphen, not em-dash) is still allowed.
    assert "—" not in body_blob


# ── reconcile_pending — stale-snapshot reconciliation ──────────────
#
# Regression suite for the 2026-07-14 incident: a provider re-sync
# soft-deleted / replaced every snapshot server 52 minutes after the
# cards went out, leaving pending rows that (a) nagged owners about
# machines that no longer existed and (b) blocked their future cards
# via the re-prompt interlock.


async def _make_pending_row(
    db: AsyncSession,
    *,
    owner: IdentityPerson,
    server_ids: list[str],
) -> CmdbServerOwnerConfirmation:
    row = CmdbServerOwnerConfirmation(
        id=uuid.uuid4(),
        owner_person_id=str(owner.id),
        period_start=date.today(),
        server_ids=server_ids,
        status=SERVER_OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC),
    )
    db.add(row)
    await db.flush()
    return row


@pytest.mark.asyncio
async def test_reconcile_auto_resolves_fully_dead_snapshot(
    db_session: AsyncSession,
):
    """All snapshot servers stopped / soft-deleted / vanished → the row
    closes as ``auto_resolved`` without owner action and without
    stamping ``last_owner_confirmed_at``."""
    from app.domains.cmdb.models import (
        SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED,
    )

    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s_stopped = await _make_server(
        db_session, integration=integ, owner=alice, name="s1", status="stopped",
    )
    s_deleted = await _make_server(
        db_session, integration=integ, owner=alice, name="s2",
    )
    s_deleted.deleted_at = datetime.now(UTC)
    ghost_id = str(uuid.uuid4())  # snapshot id with no cmdb_servers row
    await db_session.flush()

    row = await _make_pending_row(
        db_session, owner=alice,
        server_ids=[str(s_stopped.id), str(s_deleted.id), ghost_id],
    )

    service = ServerOwnerConfirmationService(db_session)
    summary = await service.reconcile_pending(trace_id="t-recon-1")

    assert summary["rows_auto_resolved"] == 1
    assert summary["servers_dropped"] == 3
    await db_session.refresh(row)
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED
    assert row.server_ids == []
    await db_session.refresh(s_stopped)
    assert s_stopped.last_owner_confirmed_at is None

    # Audit trail — one system entry naming the dropped ids + reasons.
    audit_rows = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.server_owner_confirmation.reconcile",
                AuditLog.resource_id == str(row.id),
            )
        )
    ).scalars().all()
    assert len(audit_rows) == 1
    detail = audit_rows[0].detail
    assert detail["new_status"] == SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED
    assert detail["dropped"][str(s_stopped.id)] == "stopped"
    assert detail["dropped"][str(s_deleted.id)] == "soft_deleted"
    assert detail["dropped"][ghost_id] == "missing"


@pytest.mark.asyncio
async def test_reconcile_shrinks_snapshot_keeps_row_pending(
    db_session: AsyncSession,
):
    """One dead server + one live unconfirmed server → snapshot shrinks
    to the live one, row stays pending (owner still has work to do)."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s_live = await _make_server(db_session, integration=integ, owner=alice, name="live")
    s_stopped = await _make_server(
        db_session, integration=integ, owner=alice, name="dead", status="stopped",
    )

    row = await _make_pending_row(
        db_session, owner=alice,
        server_ids=[str(s_live.id), str(s_stopped.id)],
    )

    service = ServerOwnerConfirmationService(db_session)
    summary = await service.reconcile_pending()

    assert summary["rows_shrunk"] == 1
    assert summary["rows_auto_resolved"] == 0
    await db_session.refresh(row)
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING
    assert row.server_ids == [str(s_live.id)]


@pytest.mark.asyncio
async def test_reconcile_completes_partial_journey_with_dead_tail(
    db_session: AsyncSession, monkeypatch,
):
    """Owner confirmed one of two servers; the other then stopped.
    Without reconcile the row is unfinishable (the stopped server can
    still be stamped but a DELETED one can't). Reconcile drops the dead
    tail and honours the earlier click → row completes as responded."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s_confirmed = await _make_server(db_session, integration=integ, owner=alice, name="ok")
    s_dying = await _make_server(db_session, integration=integ, owner=alice, name="dying")

    service = ServerOwnerConfirmationService(db_session)
    async def _noop(env, trace_id=None): return None
    monkeypatch.setattr(service.notifications, "dispatch", _noop)
    row, _ = await service.create_pending_for_owner(
        owner=alice, server_ids=[str(s_confirmed.id), str(s_dying.id)],
    )
    await service.respond_per_server(
        confirmation_id=str(row.id),
        actor_person_id=str(alice.id),
        confirmed_server_ids=[str(s_confirmed.id)],
    )
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING

    # The unconfirmed server dies (re-sync soft-delete).
    s_dying.deleted_at = datetime.now(UTC)
    await db_session.flush()

    summary = await service.reconcile_pending()
    assert summary["rows_completed"] == 1
    await db_session.refresh(row)
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED
    assert row.response_kind == SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL
    assert row.server_ids == [str(s_confirmed.id)]


@pytest.mark.asyncio
async def test_reconcile_drops_reassigned_server(db_session: AsyncSession):
    """A snapshot server re-assigned to another owner mid-period is
    dropped — the new owner gets their own card on the next scan."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    bob = await _make_person(db_session, email="bob@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="s")

    row = await _make_pending_row(db_session, owner=alice, server_ids=[str(s.id)])
    s.owner_person_id = str(bob.id)
    await db_session.flush()

    service = ServerOwnerConfirmationService(db_session)
    summary = await service.reconcile_pending()
    assert summary["rows_auto_resolved"] == 1
    await db_session.refresh(row)
    assert row.server_ids == []


@pytest.mark.asyncio
async def test_reconcile_leaves_healthy_row_untouched(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s = await _make_server(db_session, integration=integ, owner=alice, name="s")

    row = await _make_pending_row(db_session, owner=alice, server_ids=[str(s.id)])

    service = ServerOwnerConfirmationService(db_session)
    summary = await service.reconcile_pending()
    assert summary == {
        "rows_scanned": 1,
        "rows_shrunk": 0,
        "rows_auto_resolved": 0,
        "rows_completed": 0,
        "servers_dropped": 0,
    }
    await db_session.refresh(row)
    assert row.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING
    assert row.server_ids == [str(s.id)]


@pytest.mark.asyncio
async def test_reconcile_unblocks_reprompt_interlock(
    db_session: AsyncSession,
):
    """THE incident regression: a dead pending row suppresses the owner
    in ``list_due_owners`` forever. After reconcile auto-resolves it,
    the owner's live servers become due again on the next scan."""
    integ = await _make_integration(db_session)
    alice = await _make_person(db_session, email="alice@x.com")
    s_live = await _make_server(db_session, integration=integ, owner=alice, name="live")
    s_dead = await _make_server(
        db_session, integration=integ, owner=alice, name="dead", status="stopped",
    )
    await _make_pending_row(db_session, owner=alice, server_ids=[str(s_dead.id)])

    service = ServerOwnerConfirmationService(db_session)
    # Interlock active — the stuck pending row hides alice entirely.
    grouped = await service.list_due_owners()
    assert str(alice.id) not in grouped

    await service.reconcile_pending()

    grouped = await service.list_due_owners()
    assert grouped.get(str(alice.id)) == [str(s_live.id)]