"""Handler tests for ``jamf.device_sync`` — the device-centric MDM ledger.

The job this replaced handed out per-person seat allocations, which is
how a managed Mac came to look like a licence somebody holds until they
resign. These tests pin the corrected model:

  * every enrolled device gets a ledger row keyed by the tenant's own
    identifiers, with the holder as a derived attribute;
  * NO ``software_seat_allocations`` are written — that was the bug, and
    a test that tolerates them would let it back in;
  * the licence-utilisation snapshot still balances at the device count,
    because "we pay for 60 devices and 45 are enrolled" is a real
    question even after the person-level model is gone;
  * a leaver holding two Macs is two ``leaver_residual`` rows keyed by
    serial, not one collapsed on the shared email;
  * a device disappearing from the tenant is the ground truth for "the
    administrator removed it", and closes an open cleanup todo.
"""
from __future__ import annotations

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

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

from app.domains.asset.models import (
    MDM_MATCH_EXTERNAL,
    MDM_MATCH_LEAVER_RESIDUAL,
    MDM_MATCH_MATCHED,
    MDM_MATCH_UNOWNED,
    MDM_STATE_ENROLLED,
    MDM_STATE_GONE,
    MDM_STATE_PENDING_CLEANUP,
    AssetItem,
    AssetSoftwareCatalog,
    MdmManagedDevice,
    SoftwareExternalHolder,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.domains.provisioning.models import (
    ACTION_REVOKE,
    EXEC_MANUAL_NOTIFY,
    SOURCE_ASSET_DISPOSAL,
    STATUS_CANCELLED,
    STATUS_COMPLETED,
    STATUS_NOTIFIED,
    TARGET_HARDWARE,
    ProvisioningTask,
    SoftwareConnector,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import MdmDevice
from app.integrations.models import Integration
from app.scheduled_jobs.mdm_handlers import run_jamf_device_sync

_LOAD = "app.core.coordinators.jamf_seat_coordinator.load_jamf_driver"
_PLANS = "app.core.coordinators.jamf_seat_coordinator.get_jamf_reconciliation_plans"


async def _jamf_integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(), kind="mdm", provider="jamf",
        name="Jamf test", slug=f"jamf-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=True,
        config_public={"base_url": "https://x.jamfcloud.com", "client_id": "c"},
    )
    db.add(integ)
    await db.flush()
    return integ


async def _plan(db: AsyncSession, integ: Integration) -> SoftwareLicensePlan:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jamf", asset_type="SaaS",
        status="active", offboarding_policy="auto_revoke",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="Jamf Seat",
        price=Decimal("0.00"), billing_cycle="yearly", status="active",
        max_seats=60,
    )
    db.add(plan)
    await db.flush()
    db.add(SoftwareConnector(
        software_id=sw.id, integration_id=integ.id, is_active=True,
        binding_config={},
    ))
    await db.flush()
    return plan


async def _mdm_software(
    db: AsyncSession, integ: Integration, *, owner,
) -> AssetSoftwareCatalog:
    """An MDM-bound catalogue row WITH an owner — the cleanup recipient."""
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jamf", asset_type="SaaS", status="active",
        owner_person_id=owner.id,
    )
    db.add(sw)
    await db.flush()
    db.add(SoftwareConnector(
        software_id=sw.id, integration_id=integ.id, is_active=True,
        binding_config={},
    ))
    await db.flush()
    return sw


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


def _dev(serial: str, *, email: str | None, dtype: str = "computer") -> MdmDevice:
    return MdmDevice(
        external_id=f"id-{serial}", device_type=dtype,
        management_id=f"m-{serial}", name=f"MAC-{serial}",
        serial_number=serial, model="MacBook Pro", os_version="14.5",
        managed=True, assigned_user_email=email,
    )


def _stub_driver(devices: list[MdmDevice], integration_id: uuid.UUID) -> AsyncMock:
    driver = AsyncMock()
    driver.integration_id = str(integration_id)
    driver.list_devices = AsyncMock(return_value=devices)
    return driver


async def _run(db: AsyncSession, devices, integ, plans) -> dict:
    driver = _stub_driver(devices, integ.id)
    with patch(_LOAD, return_value=driver), patch(_PLANS, return_value=plans):
        return await run_jamf_device_sync(db, {})


@pytest.mark.asyncio
async def test_ledger_is_the_write_and_no_seats_are_allocated(
    db_session: AsyncSession,
) -> None:
    """One row per device, four buckets that balance, zero allocations.

    The allocation assertion is the point of this whole change: an
    enrolment is not a seat a person holds, so nothing here may create a
    row that the offboarding scanner would later read as one.
    """
    integ = await _jamf_integration(db_session)
    plan = await _plan(db_session, integ)
    alice = await _person(db_session, "alice@pingcap.com", "active")
    bob = await _person(db_session, "bob@pingcap.com", "left")

    devices = [
        _dev("A1", email="alice@pingcap.com"),
        _dev("A2", email="alice@pingcap.com"),   # multi-device employee
        _dev("B1", email="bob@pingcap.com"),     # leaver, device 1
        _dev("B2", email="bob@pingcap.com"),     # leaver, device 2
        _dev("U1", email=None),                  # pool / shared
        _dev("X1", email="contractor@vendor.io"),  # not a Helix person
    ]
    summary = await _run(db_session, devices, integ, [plan])

    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert len(rows) == 6
    by_serial = {r.serial_key: r for r in rows}
    assert by_serial["a1"].assigned_person_id == alice.id
    assert by_serial["a1"].match_category == MDM_MATCH_MATCHED
    assert by_serial["a1"].owner_source == "mdm"
    assert by_serial["b1"].assigned_person_id == bob.id
    assert by_serial["b1"].match_category == MDM_MATCH_LEAVER_RESIDUAL
    assert by_serial["u1"].match_category == MDM_MATCH_UNOWNED
    assert by_serial["x1"].match_category == MDM_MATCH_EXTERNAL
    assert all(r.enrollment_state == MDM_STATE_ENROLLED for r in rows)
    assert all(r.provider == "jamf" for r in rows)

    # Licence utilisation still reported, still balancing at the device
    # count — 6 == 2 matched + 2 leaver + 1 external + 1 unowned.
    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat).where(
            SoftwareSeatSyncStat.plan_id == plan.id,
        )
    )).scalar_one()
    assert stat.vendor_license_holders == 6
    assert stat.holders_matched == 2
    assert stat.holders_leaver_residual == 2
    assert stat.holders_external == 1
    assert stat.holders_no_email == 1

    # THE regression guard: no person-keyed seat rows, ever.
    allocs = (await db_session.execute(
        select(SoftwareSeatAllocation).where(
            SoftwareSeatAllocation.plan_id == plan.id,
        )
    )).scalars().all()
    assert allocs == []

    # A leaver's two Macs stay two rows keyed by serial rather than
    # collapsing onto the one email they share.
    leaver_rows = (await db_session.execute(
        select(SoftwareExternalHolder).where(
            SoftwareExternalHolder.plan_id == plan.id,
            SoftwareExternalHolder.category == "leaver_residual",
        )
    )).scalars().all()
    # Keyed by the tenant's device identity rather than the serial — see
    # ``test_duplicate_serials_do_not_break_the_reconciliation_identity``.
    # The shared recorder lowercases holder_key on write.
    assert {r.holder_key for r in leaver_rows} == {
        "computer:id-b1", "computer:id-b2",
    }
    assert all(r.email == "bob@pingcap.com" for r in leaver_rows)

    assert summary["devices"] == 6
    assert summary["ledger_created"] == 6
    assert summary["plans_reconciled"] == 1


@pytest.mark.asyncio
async def test_asset_ledger_supplies_the_holder_jamf_lacks(
    db_session: AsyncSession,
) -> None:
    """A blank User-and-Location is recovered from the asset by serial.

    ``owner_source`` must say ``asset``: the share of the fleet the MDM
    cannot attribute on its own is the number that says whether its
    records are worth trusting, and collapsing the two sources hides it.
    """
    integ = await _jamf_integration(db_session)
    plan = await _plan(db_session, integ)
    carol = await _person(db_session, "carol@pingcap.com", "active")
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-S9", serial_number="S9",
        status="assigned", category="laptop", assigned_to=carol.id,
    )
    db_session.add(asset)
    await db_session.flush()

    await _run(db_session, [_dev("S9", email=None)], integ, [plan])

    row = (await db_session.execute(select(MdmManagedDevice))).scalar_one()
    assert row.assigned_person_id == carol.id
    assert row.owner_source == "asset"
    assert row.match_category == MDM_MATCH_MATCHED
    assert row.asset_id == asset.id

    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat).where(
            SoftwareSeatSyncStat.plan_id == plan.id,
        )
    )).scalar_one()
    assert stat.holders_matched == 1
    assert stat.holders_no_email == 0


@pytest.mark.asyncio
async def test_ledger_written_even_without_a_bound_licence_plan(
    db_session: AsyncSession,
) -> None:
    """The inventory is the job; the licence snapshot is an extra.

    An admin who never wired a software catalogue row to the integration
    must still get a device ledger — otherwise the disposal-cleanup loop
    silently has nothing to match against.
    """
    integ = await _jamf_integration(db_session)
    summary = await _run(db_session, [_dev("Z1", email=None)], integ, [])

    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert len(rows) == 1
    assert summary["plans_reconciled"] == 0
    assert not summary["errors"]
    assert "licence snapshot skipped" in summary["reconciliation"]


@pytest.mark.asyncio
async def test_second_run_updates_in_place(db_session: AsyncSession) -> None:
    """Re-syncing refreshes a row rather than duplicating the device."""
    integ = await _jamf_integration(db_session)
    await _person(db_session, "dana@pingcap.com", "active")

    await _run(db_session, [_dev("K1", email=None)], integ, [])
    await _run(db_session, [_dev("K1", email="dana@pingcap.com")], integ, [])

    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert len(rows) == 1
    assert rows[0].mdm_assigned_email == "dana@pingcap.com"
    assert rows[0].match_category == MDM_MATCH_MATCHED
    assert rows[0].first_seen_at is not None


@pytest.mark.asyncio
async def test_duplicate_device_in_roster_counted_once(
    db_session: AsyncSession,
) -> None:
    """A tenant returning one device twice must not inflate every bucket.

    Without the guard the second copy overwrites the first, the reported
    device total counts it twice, and the reconciliation identity — which
    is fail-loud for this provider — blows up the whole run.
    """
    integ = await _jamf_integration(db_session)
    plan = await _plan(db_session, integ)
    dupe = _dev("D1", email=None)

    summary = await _run(db_session, [dupe, dupe], integ, [plan])

    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert len(rows) == 1
    assert summary["unowned"] == 1
    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat).where(
            SoftwareSeatSyncStat.plan_id == plan.id,
        )
    )).scalar_one()
    # ``vendor_license_holders`` is the raw roster length (2); the buckets
    # count the deduped device (1). The helper's identity check would
    # reject a mismatch, so this pins that the handler reports the deduped
    # total, not the raw one.
    assert stat.vendor_license_holders == 1


@pytest.mark.asyncio
async def test_device_gone_from_tenant_closes_the_cleanup_todo(
    db_session: AsyncSession,
) -> None:
    """Disappearance is the proof of removal, and it closes the loop.

    An administrator who deletes the record in Jamf should not also have
    to tick a box in Helix. The next sync not finding the device is
    stronger evidence than the box anyway, and ``cleanup_confirmed_via``
    records which of the two happened.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "erin@pingcap.com", "active")
    # A fleet, not one device: retiring from a one-device roster is an
    # empty roster, which the retirement floor refuses on purpose.
    fleet = [_dev(f"G{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "g0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_NOTIFIED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_reason = "buyback"
    await db_session.flush()

    summary = await _run(db_session, fleet[1:], integ, [])  # G0 dropped

    await db_session.refresh(device)
    await db_session.refresh(task)
    assert device.enrollment_state == MDM_STATE_GONE
    assert device.gone_at is not None
    assert device.cleanup_confirmed_via == "sync"
    assert task.status == STATUS_COMPLETED
    assert summary["marked_gone"] == 1
    assert summary["cleanup_auto_confirmed"] == 1
    assert summary["cleanup_tasks_closed"] == 1


@pytest.mark.asyncio
async def test_returning_device_clears_the_gone_stamp(
    db_session: AsyncSession,
) -> None:
    """A device that reappears is enrolled again, not permanently retired.

    Re-enrolment and a transient inventory gap look identical from here,
    and both mean the same thing: the tenant has it, so the ledger must
    say so rather than keep a stale tombstone.
    """
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"R{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])
    await _run(db_session, fleet[1:], integ, [])
    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "r0")
    )
    assert device.enrollment_state == MDM_STATE_GONE

    summary = await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_ENROLLED
    assert device.gone_at is None
    assert summary["returned"] == 1


@pytest.mark.asyncio
async def test_empty_roster_never_retires_the_fleet(
    db_session: AsyncSession,
) -> None:
    """A vendor returning nothing is a fault, not an emptied tenant.

    Marking a device gone stamps ``cleanup_confirmed_via='sync'`` and
    closes its todo with the words "confirmed by Jamf inventory" — a
    factual claim about the vendor. A re-scoped API client or a paginator
    that 200s with no results arrives here as an empty roster, and without
    a floor it would retire the whole fleet and file that claim against
    every open cleanup.
    """
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"E{i}", email=None) for i in range(6)]
    await _run(db_session, fleet, integ, [])

    summary = await _run(db_session, [], integ, [])

    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert all(r.enrollment_state == MDM_STATE_ENROLLED for r in rows)
    assert summary["marked_gone"] == 0
    assert "empty roster" in summary["retirement_refused"]
    # Surfaced in ``errors`` so it reaches the job row an operator reads —
    # a refusal that lives only in a summary field is a tenant frozen
    # behind a green job. It carries the ``[retire-refused]`` marker, which
    # the shared helper recognises as a defensive refusal, so the run is
    # still a success and nobody gets paged.
    assert any("[retire-refused]" in e for e in summary["errors"])


@pytest.mark.asyncio
async def test_halved_roster_never_retires_the_fleet(
    db_session: AsyncSession,
) -> None:
    """A truncated page looks like attrition. Fleets do not halve overnight."""
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"H{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    summary = await _run(db_session, fleet[:2], integ, [])

    gone = [
        r for r in (
            await db_session.execute(select(MdmManagedDevice))
        ).scalars().all()
        if r.enrollment_state == MDM_STATE_GONE
    ]
    assert gone == []
    assert summary["marked_gone"] == 0
    assert "below the" in summary["retirement_refused"]


@pytest.mark.asyncio
async def test_ordinary_attrition_still_retires(
    db_session: AsyncSession,
) -> None:
    """The floor must not be so cautious that real removals never land."""
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"A{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    summary = await _run(db_session, fleet[:7], integ, [])

    assert summary["marked_gone"] == 1
    assert summary.get("retirement_refused") is None


@pytest.mark.asyncio
async def test_a_device_that_comes_back_reopens_its_cleanup(
    db_session: AsyncSession,
) -> None:
    """A disappearance that did not last must not lose the request.

    One night's roster omits the device, so the sync closes its cleanup on
    the strength of that. The next night it is back — the asset is still
    disposed and the tenant still has the machine, so the request is live
    again. Leaving the todo closed loses it permanently: nothing re-scans
    historical disposals.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "fin@pingcap.com", "active")
    # Enough devices that removing one clears the retirement floor.
    fleet = [_dev(f"B{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "b0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_NOTIFIED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=3)
    device.cleanup_reason = "buyback"
    device.cleanup_last_reminded_at = datetime.now(UTC)
    await db_session.flush()

    # Night 1: the tenant omits it → closed as vendor-confirmed.
    await _run(db_session, fleet[1:], integ, [])
    await db_session.refresh(device)
    await db_session.refresh(task)
    assert device.enrollment_state == MDM_STATE_GONE
    assert task.status == STATUS_COMPLETED

    # Night 2: it is back.
    summary = await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    await db_session.refresh(task)
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP
    assert device.gone_at is None
    # The earlier confirmation was based on evidence that did not hold.
    assert device.cleanup_confirmed_at is None
    assert device.cleanup_confirmed_via is None
    # Cleared so the notify loop speaks now, not after the rest of a
    # reminder week counted against a closed request.
    assert device.cleanup_last_reminded_at is None
    # Superseded, not resurrected: a recycled task id would leave the card
    # that closed this able to close it again from somebody's chat history.
    assert task.status == STATUS_CANCELLED
    assert device.cleanup_task_id != task.id
    fresh = await db_session.get(ProvisioningTask, device.cleanup_task_id)
    assert fresh.status == STATUS_NOTIFIED
    assert fresh.completed_at is None
    assert "did not last" in fresh.context_data["reopened_reason"]
    assert summary["cleanup_tasks_reopened"] == 1


@pytest.mark.asyncio
async def test_returning_device_without_a_cleanup_just_re_enrolls(
    db_session: AsyncSession,
) -> None:
    """Only a device that had an open request comes back as pending."""
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"C{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])
    await _run(db_session, fleet[1:], integ, [])

    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "c0")
    )
    assert device.enrollment_state == MDM_STATE_ENROLLED
    assert device.cleanup_task_id is None


@pytest.mark.asyncio
async def test_over_long_vendor_strings_do_not_abort_the_sync(
    db_session: AsyncSession,
) -> None:
    """Jamf's free-text fields have no guaranteed cap; the columns do.

    Without clipping, one over-long realname raises 1406 and — because
    the upsert loop has no per-device savepoint — takes the whole night's
    sync down with it.
    """
    integ = await _jamf_integration(db_session)
    long_device = MdmDevice(
        external_id="id-LONG", device_type="computer",
        management_id="m" * 200,
        name="N" * 600, serial_number="LONGSERIAL",
        model="M" * 600, os_version="v" * 200, managed=True,
        assigned_user_email=None, assigned_user_name="R" * 600,
    )

    summary = await _run(db_session, [long_device], integ, [])

    assert not summary["errors"]
    row = (await db_session.execute(select(MdmManagedDevice))).scalar_one()
    assert len(row.device_name) == 255
    assert len(row.model) == 255
    assert len(row.mdm_assigned_name) == 255
    assert len(row.management_id) == 64
    assert len(row.os_version) == 64


@pytest.mark.asyncio
async def test_duplicate_serials_do_not_break_the_reconciliation_identity(
    db_session: AsyncSession,
) -> None:
    """Two tenant records for one serial must not roll back the ledger.

    The shared recorder dedups external-holder rows by ``holder_key`` and
    then asserts the deduped leaver count equals the reported one, raising
    if they differ. Keying on the serial would collapse a re-enrolment's
    two records into one, fail that identity, and — because this provider
    is fail-loud — discard the whole ledger write.
    """
    integ = await _jamf_integration(db_session)
    plan = await _plan(db_session, integ)
    await _person(db_session, "gone@pingcap.com", "left")
    # Same serial, two tenant records — exactly what a re-enrolment leaves.
    d1 = _dev("DUPSERIAL", email="gone@pingcap.com")
    d2 = _dev("DUPSERIAL", email="gone@pingcap.com")
    d2.external_id = "id-DUPSERIAL-2"

    summary = await _run(db_session, [d1, d2], integ, [plan])

    assert not summary["errors"]
    rows = (await db_session.execute(select(MdmManagedDevice))).scalars().all()
    assert len(rows) == 2
    holders = (await db_session.execute(
        select(SoftwareExternalHolder).where(
            SoftwareExternalHolder.plan_id == plan.id,
        )
    )).scalars().all()
    assert len(holders) == 2
    assert len({h.holder_key for h in holders}) == 2
    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat).where(
            SoftwareSeatSyncStat.plan_id == plan.id,
        )
    )).scalar_one()
    assert stat.holders_leaver_residual == 2


@pytest.mark.asyncio
async def test_unconfigured_integration_raises(db_session: AsyncSession) -> None:
    with patch(_LOAD, return_value=None):
        with pytest.raises(RuntimeError, match="not configured"):
            await run_jamf_device_sync(db_session, {})


@pytest.mark.asyncio
async def test_a_claim_the_tenant_contradicts_is_put_back_on_the_queue(
    db_session: AsyncSession,
) -> None:
    """A tick-box is a claim; the roster is the fact, and the fact wins.

    Somebody says they removed the device and the next sync still finds
    it, so the removal did not take. Before this, the claim closed the
    todo, the reminder query skipped terminal todos, and nothing else was
    looking — one wrong tick silenced a sold machine's cleanup for good.

    The replacement todo is a NEW task on purpose: the one-click confirm
    link is a signed token bound to a task id, valid for two weeks and not
    single-use, so recycling the id would leave the card that closed this
    the first time able to close it again from somebody's chat history.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "claim@pingcap.com", "active")
    fleet = [_dev(f"Q{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "q0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_COMPLETED,
        context_data={"serial_number": "q0"},
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=5)
    device.cleanup_reason = "sold"
    # An admin ticked it as done two days ago.
    device.cleanup_confirmed_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_confirmed_via = "admin"
    device.cleanup_last_reminded_at = datetime.now(UTC) - timedelta(days=4)
    await db_session.flush()
    old_task_id = task.id

    summary = await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    await db_session.refresh(task)
    assert summary["cleanup_claims_contradicted"] == 1
    # Still pending, claim withdrawn.
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP
    assert device.cleanup_confirmed_at is None
    assert device.cleanup_confirmed_via is None
    # Clearing the stamp is the ONE-SHOT that makes the next tick speak,
    # and it self-limits because the notify loop re-writes it. The card's
    # wording comes from the replacement task's ``reopened_from_task_id``
    # instead — see ``test_a_reopened_todo_is_chased_once_not_every_tick``
    # for why that marker must never decide timing.
    assert device.cleanup_last_reminded_at is None
    # A fresh todo, and the old one superseded rather than resurrected —
    # so a stale one-click token addresses a terminal task and does nothing.
    assert device.cleanup_task_id is not None
    assert device.cleanup_task_id != old_task_id
    assert task.status == STATUS_CANCELLED
    fresh = await db_session.get(ProvisioningTask, device.cleanup_task_id)
    assert fresh.status == STATUS_NOTIFIED
    assert fresh.context_data["reopened_from_task_id"] == str(old_task_id)
    assert "did not take" in fresh.context_data["reopened_reason"]
    # The superseded task is CANCELLED and the device now points at the
    # fresh one — which is the only reason ``_release_cancelled_cleanups``
    # (which runs later in the SAME tick, joining on the device's current
    # pointer) does not immediately release the cleanup we just re-raised.
    assert summary["cleanup_orphans_released"] == 0
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP


@pytest.mark.asyncio
async def test_a_confirmed_and_actually_removed_device_stays_closed(
    db_session: AsyncSession,
) -> None:
    """The contradiction must only fire when the tenant still has it.

    An admin who really did delete the record must not be chased again —
    that would make the confirm button feel broken and train people to
    ignore the card.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "ok@pingcap.com", "active")
    fleet = [_dev(f"W{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "w0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_COMPLETED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=5)
    device.cleanup_confirmed_at = datetime.now(UTC) - timedelta(days=1)
    device.cleanup_confirmed_via = "admin"
    await db_session.flush()

    # The device really is gone from the tenant this run.
    summary = await _run(db_session, fleet[1:], integ, [])

    await db_session.refresh(device)
    assert summary["cleanup_claims_contradicted"] == 0
    assert device.enrollment_state == MDM_STATE_GONE
    # The vendor's confirmation supersedes the admin's claim.
    assert device.cleanup_confirmed_via == "sync"


@pytest.mark.asyncio
async def test_a_todo_cancelled_on_the_generic_queue_releases_the_device(
    db_session: AsyncSession,
) -> None:
    """Two domains must not disagree in silence.

    ``POST /provisioning/tasks/{id}/cancel`` knows nothing about this
    ledger. Cancelling a hardware cleanup there used to leave the device
    ``pending_cleanup`` against a terminal task: never chased (the
    reminder query skips terminal), never re-opened (a deliberate cancel
    is respected), and rendered on the page as an open request nobody is
    working on.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "cancel@pingcap.com", "active")
    fleet = [_dev(f"V{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "v0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_CANCELLED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_reason = "sold"
    await db_session.flush()

    summary = await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    assert summary["cleanup_orphans_released"] == 1
    assert device.enrollment_state == MDM_STATE_ENROLLED
    assert device.cleanup_task_id is None
    assert device.cleanup_reason is None


@pytest.mark.asyncio
async def test_a_just_made_claim_is_not_called_a_lie_by_a_stale_roster(
    db_session: AsyncSession,
) -> None:
    """The grace window closes a race, not a doubt.

    This run's roster was fetched seconds before the loop walks it. An
    admin who deleted the record and confirmed inside that gap would be
    contradicted by a snapshot taken before they acted. Deferring costs
    one night; calling someone a liar over a stale read costs their trust
    in the card.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "race@pingcap.com", "active")
    fleet = [_dev(f"T{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "t0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_COMPLETED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=3)
    # Confirmed one minute ago — inside the grace window.
    device.cleanup_confirmed_at = datetime.now(UTC) - timedelta(minutes=1)
    device.cleanup_confirmed_via = "admin"
    await db_session.flush()

    summary = await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    assert summary["cleanup_claims_contradicted"] == 0
    assert device.cleanup_confirmed_via == "admin"
    assert device.cleanup_task_id == task.id


@pytest.mark.asyncio
async def test_a_reopened_todo_is_chased_once_not_every_tick(
    db_session: AsyncSession,
) -> None:
    """The one-flag-two-questions trap, pinned.

    ``reopened_from_task_id`` is permanent, so it can decide the card's
    VOICE but never its timing. It once did both: because the marker never
    expires, the interval was waived on every run — ten cards a working
    day, escalated to every admin, until the device left the tenant.
    Timing belongs to ``cleanup_last_reminded_at``, which the notify loop
    re-writes and which therefore self-limits.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        due_cleanup_notifications,
    )

    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "tick@pingcap.com", "active")
    fleet = [_dev(f"S{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "s0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_COMPLETED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=40)
    device.cleanup_confirmed_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_confirmed_via = "admin"
    device.cleanup_last_reminded_at = datetime.now(UTC) - timedelta(days=3)
    await db_session.flush()

    await _run(db_session, fleet, integ, [])
    await db_session.refresh(device)

    # Due immediately after the contradiction, in the reminder's voice.
    due = await due_cleanup_notifications(db_session)
    assert len(due) == 1
    assert due[0][2] is True

    # Once the notify loop has spoken, the interval applies again — this is
    # the assertion the storm slipped past.
    device.cleanup_last_reminded_at = datetime.now(UTC)
    await db_session.flush()
    assert await due_cleanup_notifications(db_session) == []
    # Still silent half a week later; only the weekly cadence reopens it.
    device.cleanup_last_reminded_at = datetime.now(UTC) - timedelta(days=3)
    await db_session.flush()
    assert await due_cleanup_notifications(db_session) == []


@pytest.mark.asyncio
async def test_retired_history_does_not_freeze_future_retirements(
    db_session: AsyncSession,
) -> None:
    """The guard must compare like with like.

    Nothing purges ``gone`` rows. Counting them as "known" made the floor
    tighten as history accumulated: once the cumulative retired count
    reached the live fleet size, any single retirement was refused —
    permanently, silently, with the job still green.
    """
    integ = await _jamf_integration(db_session)
    # A long history: 20 devices already retired, 8 live.
    history = [_dev(f"HIST{i}", email=None) for i in range(20)]
    live = [_dev(f"LIVE{i}", email=None) for i in range(8)]
    await _run(db_session, history + live, integ, [])
    # Retire the history in chunks the floor accepts.
    for keep in (14, 8, 2, 0):
        await _run(db_session, history[:keep] + live, integ, [])
    gone = [
        r for r in (
            await db_session.execute(select(MdmManagedDevice))
        ).scalars().all()
        if r.enrollment_state == MDM_STATE_GONE
    ]
    assert len(gone) == 20

    # One more ordinary retirement, against 8 live rows and 20 retired.
    summary = await _run(db_session, live[:7], integ, [])

    assert summary.get("retirement_refused") is None
    assert summary["marked_gone"] == 1


@pytest.mark.asyncio
async def test_a_real_mass_retirement_has_a_way_through(
    db_session: AsyncSession,
) -> None:
    """An office closes and thirty machines are scrapped in one night.

    The floor refuses that, correctly — and then refuses it again every
    night after, because the refusal is precisely what stops those
    machines from ever being marked gone. A safety floor whose only exit
    is a DB data fix is a wall, so the ratio is overridable for one run.
    """
    integ = await _jamf_integration(db_session)
    fleet = [_dev(f"O{i}", email=None) for i in range(20)]
    await _run(db_session, fleet, integ, [])

    # 16 of 20 scrapped — well under the 50% floor.
    refused = await _run(db_session, fleet[:4], integ, [])
    assert refused["marked_gone"] == 0
    assert "[retire-refused]" in refused["retirement_refused"]
    # Surfaced where an operator looks, and recognised as a defensive
    # refusal rather than a failed unit of work.
    assert any("[retire-refused]" in e for e in refused["errors"])

    # A lowered floor with no expiry is ignored outright — job config is
    # a persisted row, so that is not a one-run exception, it is a guard
    # quietly weakened forever.
    driver = _stub_driver(fleet[:4], integ.id)
    with patch(_LOAD, return_value=driver), patch(_PLANS, return_value=[]):
        no_expiry = await run_jamf_device_sync(
            db_session, {"retire_min_ratio": 0.1},
        )
    assert no_expiry["marked_gone"] == 0
    assert "retire_override_until" in no_expiry["retire_ratio_note"]

    # With an expiry in the future, the override applies.
    until = (datetime.now(UTC) + timedelta(days=1)).isoformat()
    driver = _stub_driver(fleet[:4], integ.id)
    with patch(_LOAD, return_value=driver), patch(_PLANS, return_value=[]):
        allowed = await run_jamf_device_sync(
            db_session,
            {"retire_min_ratio": 0.1, "retire_override_until": until},
        )

    assert allowed.get("retirement_refused") is None
    assert allowed["marked_gone"] == 16

    # And once the window passes the default returns on its own, so
    # forgetting to remove the key costs nothing.
    stale = (datetime.now(UTC) - timedelta(days=1)).isoformat()
    more = [_dev(f"P{i}", email=None) for i in range(20)]
    driver = _stub_driver(more, integ.id)
    with patch(_LOAD, return_value=driver), patch(_PLANS, return_value=[]):
        await run_jamf_device_sync(db_session, {})
    driver = _stub_driver(more[:4], integ.id)
    with patch(_LOAD, return_value=driver), patch(_PLANS, return_value=[]):
        expired = await run_jamf_device_sync(
            db_session,
            {"retire_min_ratio": 0.1, "retire_override_until": stale},
        )
    assert expired["marked_gone"] == 0
    assert "expired" in expired["retire_ratio_note"]


@pytest.mark.asyncio
async def test_a_tightening_override_does_not_unlock_the_fleet_bypass(
    db_session: AsyncSession,
) -> None:
    """Two questions, two knobs — the trap this project keeps falling into.

    "How strict is the ratio" and "may this run retire devices the tenant no
    longer mentions at all" are different authorisations. Deriving the second
    from the first (``min_ratio != DEFAULT``) said YES to an operator who
    *tightened* the floor after a scare, unlocking the whole-fleet bypass the
    guard exists to hold shut — and said NO to a deliberate override that
    happened to equal the default, while announcing it as active.
    """
    from datetime import timedelta as _td

    from app.core.coordinators.mdm_device_coordinator import _safe_to_retire
    from app.scheduled_jobs.mdm_handlers import _resolve_retire_overrides

    now = datetime.now(UTC)
    soon = (now + _td(days=1)).isoformat()

    # Tightening: the ratio moves, the fleet-swap authorisation must not.
    ratio, allow, _note = _resolve_retire_overrides(
        {"retire_min_ratio": 0.9, "retire_override_until": soon}, now=now,
    )
    assert ratio == 0.9
    assert allow is False
    ok, _ = _safe_to_retire(
        reported=0, known=45, roster_size=45,
        min_ratio=ratio, allow_zero_overlap=allow,
    )
    assert ok is False, "a tightening override must not authorise a fleet swap"

    # The fleet-swap authorisation is its own key, and works without
    # touching the ratio at all.
    ratio, allow, _note = _resolve_retire_overrides(
        {"retire_allow_zero_overlap": True, "retire_override_until": soon},
        now=now,
    )
    assert allow is True
    ok, _ = _safe_to_retire(
        reported=0, known=45, roster_size=45,
        min_ratio=ratio, allow_zero_overlap=allow,
    )
    assert ok is True

    # Both expire on the one shared window, and neither is honoured without it.
    for cfg in (
        {"retire_allow_zero_overlap": True},
        {
            "retire_allow_zero_overlap": True,
            "retire_override_until": (now - _td(days=1)).isoformat(),
        },
        {
            "retire_min_ratio": 5,
            "retire_allow_zero_overlap": True,
            "retire_override_until": soon,
        },
    ):
        _r, allow, note = _resolve_retire_overrides(cfg, now=now)
        assert allow is False, cfg
        assert note


@pytest.mark.asyncio
async def test_a_contradiction_is_recorded_not_inferred(
    db_session: AsyncSession,
) -> None:
    """The UI needs a fact that survives the clearing the contradiction does.

    Overturning a claim nulls ``cleanup_confirmed_at``, so anything derived
    from that field is true exactly when a contradiction has NOT happened
    (inside the grace window) and false once one has. Both surfaces read
    ``cleanup_contradicted_at`` instead.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "rec@pingcap.com", "active")
    fleet = [_dev(f"N{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "n0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_COMPLETED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=5)
    device.cleanup_confirmed_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_confirmed_via = "admin"
    await db_session.flush()
    assert device.cleanup_contradicted_at is None

    await _run(db_session, fleet, integ, [])

    await db_session.refresh(device)
    # The claim is gone AND the refutation is on record — that pairing is the
    # whole point: one field alone cannot express it.
    assert device.cleanup_confirmed_at is None
    assert device.cleanup_contradicted_at is not None


@pytest.mark.asyncio
async def test_resolving_a_cleanup_clears_the_contradiction_mark(
    db_session: AsyncSession,
) -> None:
    """A "was refuted once" on a device with no open request is a claim
    about nothing — it would keep a warning on the page for ever."""
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        withdraw_stale_cleanups,
    )

    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "clr@pingcap.com", "active")
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-CLR", serial_number="CLR1",
        status="disposed", category="laptop",
    )
    db_session.add(asset)
    await db_session.flush()
    await _run(db_session, [_dev("CLR1", email=None)], integ, [])
    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "clr1")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_NOTIFIED,
    )
    db_session.add(task)
    await db_session.flush()
    device.asset_id = asset.id
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=2)
    device.cleanup_contradicted_at = datetime.now(UTC) - timedelta(days=1)
    await db_session.flush()

    # The asset stops being disposed, so the request is withdrawn.
    asset.status = "idle"
    await db_session.flush()
    assert await withdraw_stale_cleanups(db_session) == 1

    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_ENROLLED
    assert device.cleanup_contradicted_at is None


@pytest.mark.asyncio
async def test_the_success_path_also_clears_the_contradiction_mark(
    db_session: AsyncSession,
) -> None:
    """A refutation is a live fact only while the request is open.

    Three resolution paths cleared it (withdraw / release / re-raise) and
    the fourth — the one where the device actually leaves the tenant, i.e.
    the success path — did not. That row would keep reporting "you said
    this was done and it wasn't" about a device where it demonstrably now
    is.
    """
    integ = await _jamf_integration(db_session)
    person = await _person(db_session, "done@pingcap.com", "active")
    fleet = [_dev(f"Z{i}", email=None) for i in range(8)]
    await _run(db_session, fleet, integ, [])

    device = await db_session.scalar(
        select(MdmManagedDevice).where(MdmManagedDevice.serial_key == "z0")
    )
    task = ProvisioningTask(
        user_id=person.id, action=ACTION_REVOKE,
        target_type=TARGET_HARDWARE, target_id=device.id,
        source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
        status=STATUS_NOTIFIED,
    )
    db_session.add(task)
    await db_session.flush()
    device.enrollment_state = MDM_STATE_PENDING_CLEANUP
    device.cleanup_task_id = task.id
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=9)
    # It was claimed done once, and the sync caught that out.
    device.cleanup_contradicted_at = datetime.now(UTC) - timedelta(days=2)
    await db_session.flush()

    # This time it really is gone from the tenant.
    await _run(db_session, fleet[1:], integ, [])

    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_GONE
    assert device.cleanup_confirmed_via == "sync"
    assert device.cleanup_contradicted_at is None


@pytest.mark.asyncio
async def test_the_zero_overlap_remedy_names_the_key_that_works(
    db_session: AsyncSession,
) -> None:
    """A remedy that does not work rebuilds the wall it was meant to remove.

    The zero-overlap refusal told the operator to set ``retire_min_ratio``,
    which that branch stopped reading when the two authorisations were
    split. Following it produces the same refusal with nothing to explain
    why — the wall, reconstructed out of a sentence.
    """
    from datetime import timedelta as _td

    from app.core.coordinators.mdm_device_coordinator import _safe_to_retire
    from app.scheduled_jobs.mdm_handlers import _resolve_retire_overrides

    now = datetime.now(UTC)
    soon = (now + _td(days=1)).isoformat()
    _ok, refusal = _safe_to_retire(
        reported=0, known=20, roster_size=20, allow_zero_overlap=False,
    )
    assert "retire_allow_zero_overlap" in refusal
    assert "NOT read `retire_min_ratio`" in refusal

    # And the key it names actually clears the refusal.
    named = {"retire_allow_zero_overlap": True, "retire_override_until": soon}
    ratio, allow, _n = _resolve_retire_overrides(named, now=now)
    ok, _ = _safe_to_retire(
        reported=0, known=20, roster_size=20,
        min_ratio=ratio, allow_zero_overlap=allow,
    )
    assert ok is True

    # A string that is not a yes must not authorise a fleet retirement —
    # ``bool("no")`` is True, and this was the looser of the two keys.
    for falsy in ("no", "false", "off", ""):
        _r, allow, _n = _resolve_retire_overrides(
            {"retire_allow_zero_overlap": falsy,
             "retire_override_until": soon}, now=now,
        )
        assert allow is False, falsy


@pytest.mark.asyncio
async def test_a_backlog_is_drained_at_a_bounded_rate_and_says_so(
    db_session: AsyncSession,
) -> None:
    """A burst of cleanups must not become a burst of cards.

    Two real events produce one: the reconciler discovering a pre-feature
    backlog on the first night, and an office closure disposing thirty
    laptops in an afternoon. Thirty cards in one minute to one person is
    how a card stops being read — and these cards are the only thing
    between a sold Mac and a tenant that still manages it.

    Three things are asserted together, because a cap without the other
    two is worse than none: only ``cap`` go out, the run REPORTS what it
    held back (a silent cap reads as "5 due, 5 sent, healthy" forever),
    and the next tick reaches the ones it deferred.
    """
    from app.scheduled_jobs.mdm_handlers import _run_mdm_cleanup_notify_once

    integ = await _jamf_integration(db_session)
    owner = await _person(db_session, "burst-owner@pingcap.com", "active")
    await _mdm_software(db_session, integ, owner=owner)
    fleet = [_dev(f"B{i}", email=None) for i in range(9)]
    await _run(db_session, fleet, integ, [])

    now = datetime.now(UTC)
    devices = (
        await db_session.execute(
            select(MdmManagedDevice).order_by(MdmManagedDevice.serial_key)
        )
    ).scalars().all()
    for i, device in enumerate(devices):
        task = ProvisioningTask(
            user_id=owner.id, action=ACTION_REVOKE,
            target_type=TARGET_HARDWARE, target_id=device.id,
            source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
            status=STATUS_NOTIFIED, assignee_id=owner.id,
        )
        db_session.add(task)
        await db_session.flush()
        device.enrollment_state = MDM_STATE_PENDING_CLEANUP
        device.cleanup_task_id = task.id
        # Distinct ages so "oldest first" is a real ordering, not a tie.
        device.cleanup_requested_at = now - timedelta(days=30 - i)
    await db_session.commit()

    first = await _run_mdm_cleanup_notify_once(db_session, {})

    assert first["due"] == 9
    assert first["first_notice"] == 5
    assert first["deferred"] == 4
    # Not a counter alone: the run has to SAY it, on the row an operator
    # reads, or a permanently-capped queue looks like a healthy one.
    assert "deferring 4" in first.get("note", "")
    assert "notify_burst_cap" in first.get("note", "")

    spoken = set(
        (
            await db_session.execute(
                select(MdmManagedDevice.serial_key).where(
                    MdmManagedDevice.cleanup_last_reminded_at.is_not(None),
                )
            )
        ).scalars().all()
    )
    assert len(spoken) == 5
    # Oldest-waiting went first — B0 is the oldest by construction.
    assert "b0" in spoken

    # The deferred four are reached on the next tick, not stranded.
    second = await _run_mdm_cleanup_notify_once(db_session, {})
    assert second["due"] == 4
    assert second["first_notice"] == 4
    assert second["deferred"] == 0
    assert "note" not in second

    remaining = (
        await db_session.execute(
            select(MdmManagedDevice).where(
                MdmManagedDevice.cleanup_last_reminded_at.is_(None),
                MdmManagedDevice.enrollment_state == MDM_STATE_PENDING_CLEANUP,
            )
        )
    ).scalars().all()
    assert remaining == []


@pytest.mark.asyncio
async def test_the_burst_cap_is_operator_adjustable_but_not_breakable(
    db_session: AsyncSession,
) -> None:
    """``job.config`` is a row a human edits by hand.

    A widened cap must take effect; a nonsense value must fall back to the
    default rather than silently stopping every cleanup card. ``0`` is the
    dangerous one — it reads as a legitimate integer and would freeze the
    queue while every run reported success.
    """
    from app.scheduled_jobs.mdm_handlers import _run_mdm_cleanup_notify_once

    integ = await _jamf_integration(db_session)
    owner = await _person(db_session, "burst-cfg@pingcap.com", "active")
    await _mdm_software(db_session, integ, owner=owner)
    fleet = [_dev(f"C{i}", email=None) for i in range(7)]
    await _run(db_session, fleet, integ, [])

    now = datetime.now(UTC)
    devices = (
        await db_session.execute(select(MdmManagedDevice))
    ).scalars().all()
    for i, device in enumerate(devices):
        task = ProvisioningTask(
            user_id=owner.id, action=ACTION_REVOKE,
            target_type=TARGET_HARDWARE, target_id=device.id,
            source=SOURCE_ASSET_DISPOSAL, execution_mode=EXEC_MANUAL_NOTIFY,
            status=STATUS_NOTIFIED, assignee_id=owner.id,
        )
        db_session.add(task)
        await db_session.flush()
        device.enrollment_state = MDM_STATE_PENDING_CLEANUP
        device.cleanup_task_id = task.id
        device.cleanup_requested_at = now - timedelta(days=20 - i)
    await db_session.commit()

    # A zero cap must NOT be honoured.
    out = await _run_mdm_cleanup_notify_once(db_session, {"notify_burst_cap": 0})
    assert out["first_notice"] == 5, "a 0 cap froze the queue"

    # A widened cap clears the rest in one tick.
    out2 = await _run_mdm_cleanup_notify_once(
        db_session, {"notify_burst_cap": "50"},
    )
    assert out2["first_notice"] == 2
    assert out2["deferred"] == 0