"""Round-13 hardening tests (P01121 walkthrough, 2026-05-08).

Each test pins one bug from the audit so a future refactor can't
silently re-introduce it.

Bug catalogue:
  * Q1a — force-transfer crashed with IntegrityError on missing
    ``asset_label`` for the synthetic AssetTransferRequest. Once one
    blocker tripped this, the session was poisoned and all remaining
    blockers in the same tick failed PendingRollbackError.
  * Q1b — software_seat / software_permission blockers must NOT be
    iterated by force_transfer (their offboarding path is the
    auto_revoke provisioning pipeline; transferring "ownership" of
    a seat to the manager is a no-op that creates a misleading
    audit trail).
  * Q1c — per-blocker SAVEPOINT isolates failures so an exception
    on blocker N does not abort blockers N+1, N+2, ...
  * Q5  — the day a process is first seen, ``_advance_single`` must
    NOT run force-transfer; instead push IT admins and defer the
    actual force-transfer 24 hours so an HRIS-late detection (P01121:
    discovered T+7) doesn't skip the human window.
  * Q6b — hardware force-transfer prefers the local office asset
    admin (asset.location → asset_office_addresses.city, strict
    equality) over the per-leaver fallback. Cross-geo offboarding
    is the dominant flow.
"""
from __future__ import annotations

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

import pytest
from sqlalchemy import select

from app.auth.models import AuthRole, AuthUserRole
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.asset.models import (
    AssetItem,
    AssetLocation,
    AssetLocationAdmin,
    AssetOfficeAddress,
    AssetOfficeLocalAdmin,
)
from app.domains.offboarding.models import (
    PROC_BLOCKED,
    REMINDER_IT_GRACE,
    RES_API_KEY,
    RES_HARDWARE,
    RES_SOFTWARE_PERMISSION,
    RES_SOFTWARE_SEAT,
    RESO_PENDING,
    RESO_TRANSFERRED,
    AssetTransferRequest,
    OffboardingBlocker,
    OffboardingProcess,
    OffboardingReminderLog,
)
from app.hr_lifecycle.advance_service import ProcessAdvanceService
from app.identity.person_models import IdentityPerson

# ─────────────────────── helpers ───────────────────────

async def _make_person(
    db_session,
    *,
    status: str = "active",
    manager_id: uuid.UUID | None = None,
    work_email: str | None = None,
    display_name: str | None = None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        source_system="test",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=work_email or f"u{uuid.uuid4().hex[:8]}@x.test",
        display_name=display_name or "Test Person",
        employment_status=status,
        manager_person_id=manager_id,
    )
    db_session.add(p)
    await db_session.flush()
    return p


async def _make_leaver_with_proc(
    db_session,
    *,
    discovery_offset_hours: float = 30.0,
    manager: IdentityPerson | None = None,
) -> tuple[IdentityPerson, OffboardingProcess]:
    """Create a leaver_pending person + offboarding_process.

    ``discovery_offset_hours`` controls ``proc.created_at`` relative
    to now: 30h means "discovered yesterday-ish, past the grace
    window". 1h means "discovered just now, inside the grace window".
    """
    leaver = await _make_person(
        db_session,
        status="leaver_pending",
        manager_id=manager.id if manager else None,
    )
    leaver.termination_date = date.today() - timedelta(days=1)
    proc = OffboardingProcess(
        id=uuid.uuid4(),
        person_id=leaver.id,
        leaver_date=leaver.termination_date,
        status=PROC_BLOCKED,
        created_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(
            hours=discovery_offset_hours,
        ),
    )
    db_session.add(proc)
    await db_session.flush()
    return leaver, proc


async def _ensure_global_admin(db_session) -> IdentityPerson:
    """Make a global_admin user so the grace-window IT push has a target.

    AuthUser is an alias for IdentityPerson — the same row IS both,
    so we just attach a global_admin role to a regular person.
    """
    admin_person = await _make_person(
        db_session, work_email="it.admin@x.test", display_name="IT Admin",
    )
    role_id = await db_session.scalar(
        select(AuthRole.id).where(AuthRole.name == "global_admin"),
    )
    db_session.add(AuthUserRole(user_id=admin_person.id, role_id=role_id))
    await db_session.flush()
    return admin_person


# ─────────────────────── Q1a / Q1b / Q1c ───────────────────────

@pytest.mark.asyncio
async def test_force_transfer_synthetic_xfer_has_asset_label(db_session):
    """Round-13 Q1a: every synthetic AssetTransferRequest INSERT must
    carry ``asset_label`` (NOT NULL on the column). The previous
    version omitted it for force-transfer and crashed every T+0
    tick on the first iteration.

    2026-06-29: a real synthetic transfer only happens for the explicit
    ``transfer_to_person`` disposition now (``keep`` / ``recycle_to_office``
    no longer mint a person-transfer — see
    ``test_hardware_no_force_transfer_to_lead_2026_06_29``). Use that
    disposition here so the asset_label invariant stays exercised.
    """
    manager = await _make_person(db_session, work_email="mgr@x.test")
    successor = await _make_person(db_session, work_email="successor@x.test")
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    # Hardware blocker — leaver owns the asset.
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
        location=None,
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook 16",
        resolution=RESO_PENDING,
    ))
    # Explicit successor → real force-transfer → synthetic xfer row.
    proc.hardware_disposition = {
        str(asset.id): {
            "action": "transfer_to_person",
            "person_id": str(successor.id),
        },
    }
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues
    svc.offboarding.suspend_sso = AsyncMock(  # type: ignore[assignment]
        return_value=type("R", (), {
            "sso_suspended_at": datetime.now(UTC),
        })(),
    )
    svc.leaver_guard.flip_to_left = AsyncMock()  # type: ignore[assignment]

    result = await svc._advance_single(
        leaver, fallback_email=manager.work_email,
    )
    assert result["force_transferred"] == 1

    # The synthetic transfer must exist with a non-null asset_label.
    xfer = await db_session.scalar(
        select(AssetTransferRequest).where(
            AssetTransferRequest.offboarding_process_id == proc.id,
        ),
    )
    assert xfer is not None
    assert xfer.asset_label, "asset_label must be set (NOT NULL column)"
    assert xfer.asset_label == "MacBook 16"


@pytest.mark.asyncio
async def test_force_transfer_skips_software_seat_and_permission(db_session):
    """Round-13 Q1b: force_transfer must NOT iterate software_seat /
    software_permission blockers. Those types are auto_revoke
    territory — handing the seat's user_id to the manager in our DB
    without touching the vendor is a misleading no-op.
    """
    manager = await _make_person(db_session, work_email="mgr2@x.test")
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    seat_blocker_id = uuid.uuid4()
    perm_blocker_id = uuid.uuid4()
    db_session.add(OffboardingBlocker(
        id=seat_blocker_id,
        process_id=proc.id,
        resource_type=RES_SOFTWARE_SEAT,
        resource_id=uuid.uuid4(),
        label="Zoom — Workplace",
        resolution=RESO_PENDING,
    ))
    db_session.add(OffboardingBlocker(
        id=perm_blocker_id,
        process_id=proc.id,
        resource_type=RES_SOFTWARE_PERMISSION,
        resource_id=uuid.uuid4(),
        label="Confluence — Editor",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues

    count = await svc._force_transfer_all(
        person=leaver, process=proc,
        blockers=[
            await db_session.get(OffboardingBlocker, seat_blocker_id),
            await db_session.get(OffboardingBlocker, perm_blocker_id),
        ],
        new_owner=manager,
    )
    # Both blockers SKIPPED — 0 force-transferred.
    assert count == 0
    # No synthetic transfer rows created for either type.
    xfer_count = await db_session.scalar(
        select(AssetTransferRequest).where(
            AssetTransferRequest.offboarding_process_id == proc.id,
        ),
    )
    assert xfer_count is None


@pytest.mark.asyncio
async def test_force_transfer_savepoint_isolates_failure(db_session):
    """Round-13 Q1c: a failure on blocker N must not stop blockers
    N+1..M from being processed. Pre-round-13 a single IntegrityError
    on the first iteration left the session poisoned and dead-
    lettered every subsequent blocker with PendingRollbackError.
    """
    manager = await _make_person(db_session, work_email="mgr3@x.test")
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    # Two API key blockers — first one points at a non-existent
    # resource_id (returns moved=False, no transfer); second one is
    # valid and should still proceed.
    bad_id = uuid.uuid4()
    provider = ApiKeyProvider(
        id=uuid.uuid4(),
        name=f"prov-{uuid.uuid4().hex[:6]}",
    )
    db_session.add(provider)
    await db_session.flush()
    good_assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        request_number="REQ-OK",
        project_name="proj",
        reason="testing",
        monthly_budget_usd=Decimal("0"),
        assigned_to=leaver.id,
        status="active",
    )
    db_session.add(good_assignment)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_API_KEY,
        resource_id=bad_id,
        label="API Key MISSING",
        resolution=RESO_PENDING,
    ))
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_API_KEY,
        resource_id=good_assignment.id,
        label="API Key GOOD",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues
    blockers = list((
        await db_session.execute(
            select(OffboardingBlocker).where(
                OffboardingBlocker.process_id == proc.id,
            ),
        )
    ).scalars().all())
    count = await svc._force_transfer_all(
        person=leaver, process=proc, blockers=blockers, new_owner=manager,
    )
    # The valid one transferred.
    assert count == 1
    # Good blocker resolution flipped.
    refreshed = await db_session.scalar(
        select(OffboardingBlocker).where(
            OffboardingBlocker.resource_id == good_assignment.id,
        ),
    )
    assert refreshed.resolution == RESO_TRANSFERRED
    assert refreshed.transfer_request_id is not None
    # ApiKeyAssignment.assigned_to flipped.
    refreshed_ak = await db_session.scalar(
        select(ApiKeyAssignment).where(
            ApiKeyAssignment.id == good_assignment.id,
        ),
    )
    assert refreshed_ak.assigned_to == manager.id


# ─────────────────────── Q5: T+0 grace ───────────────────────

@pytest.mark.asyncio
async def test_grace_window_pushes_it_and_skips_force_transfer(db_session):
    """Round-13 Q5: a leaver discovered <24h ago must NOT be
    force-transferred; instead IT admins receive a one-time push
    naming the pending blockers and the system waits a day. P01121
    (HRIS-late detection) was the trigger.
    """
    await _ensure_global_admin(db_session)
    manager = await _make_person(db_session, work_email="mgr4@x.test")
    leaver, proc = await _make_leaver_with_proc(
        db_session, manager=manager, discovery_offset_hours=1.0,
    )
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook 16",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    # Grace-window push now ENQUEUEs durable outbox rows (FK-lock login-outage
    # fix); assert against enqueue.
    enqueue_mock = AsyncMock()
    svc.notify.enqueue = enqueue_mock  # type: ignore[assignment]

    result = await svc._advance_single(
        leaver, fallback_email=manager.work_email,
    )
    # Held — no force-transfer, IT push fired.
    assert result["force_transferred"] == 0
    assert result["transitioned"] is False
    assert result["grace_held"] is True
    # IT push enqueued.
    assert enqueue_mock.await_count >= 1
    # Reminder log entry — second tick must be a no-op.
    log = await db_session.scalar(
        select(OffboardingReminderLog).where(
            OffboardingReminderLog.process_id == proc.id,
            OffboardingReminderLog.reminder_kind == REMINDER_IT_GRACE,
        ),
    )
    assert log is not None
    # Asset NOT transferred — leaver still owns it.
    refreshed_asset = await db_session.scalar(
        select(AssetItem).where(AssetItem.id == asset.id),
    )
    assert refreshed_asset.assigned_to == leaver.id


@pytest.mark.asyncio
async def test_grace_window_idempotent(db_session):
    """Two ticks within the same grace window must result in only
    ONE IT push (idempotency via offboarding_reminder_log)."""
    await _ensure_global_admin(db_session)
    manager = await _make_person(db_session, work_email="mgr5@x.test")
    leaver, proc = await _make_leaver_with_proc(
        db_session, manager=manager, discovery_offset_hours=2.0,
    )
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_API_KEY,
        resource_id=uuid.uuid4(),
        label="key",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    enqueue_mock = AsyncMock()
    svc.notify.enqueue = enqueue_mock  # type: ignore[assignment]

    await svc._advance_single(leaver, fallback_email=manager.work_email)
    first_calls = enqueue_mock.await_count
    await svc._advance_single(leaver, fallback_email=manager.work_email)
    # Second tick: same await count (no extra push).
    assert enqueue_mock.await_count == first_calls


@pytest.mark.asyncio
async def test_post_grace_runs_force_transfer(db_session):
    """A leaver discovered >24h ago is past the grace window —
    force-transfer should run normally on the next tick.
    """
    manager = await _make_person(db_session, work_email="mgr6@x.test")
    leaver, proc = await _make_leaver_with_proc(
        db_session, manager=manager, discovery_offset_hours=30.0,
    )
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues
    svc.offboarding.suspend_sso = AsyncMock(  # type: ignore[assignment]
        return_value=type("R", (), {
            "sso_suspended_at": datetime.now(UTC),
        })(),
    )
    svc.leaver_guard.flip_to_left = AsyncMock()  # type: ignore[assignment]

    result = await svc._advance_single(
        leaver, fallback_email=manager.work_email,
    )
    assert result["grace_held"] is False
    assert result["force_transferred"] == 1
    refreshed_asset = await db_session.scalar(
        select(AssetItem).where(AssetItem.id == asset.id),
    )
    # 2026-05-20 contract change: T+0 default disposition is ``keep``
    # — the asset stays with the leaver as held_by_leaver, the IT same-
    # day notification fires, and the offboarding process is not blocked.
    # (Pre-2026-05-20 default was ``recycle_to_office`` to the leaver's
    # home office, degrading to held_by_leaver when no office matched.
    # The end state was the same in this leaver's case but via a longer
    # code path.)
    assert refreshed_asset.assigned_to == leaver.id
    assert refreshed_asset.status == "held_by_leaver"


# ─────────────────────── Q6b: office-admin lookup ───────────────────────

@pytest.mark.asyncio
async def test_hardware_force_transfer_prefers_location_admin(db_session):
    """Round-13 Q6b (rewritten 2026-05-19 for PR-#406 disposition
    contract): hardware whose leaver's ``work_location_name`` matches
    a configured office's ``city`` must recycle to that office —
    ``holder_kind='office'`` + ``status='held_by_office'`` +
    ``assigned_to=None``, with the inventory routing surfacing the
    administrator of the location that office sits at. Old contract was
    a force-transfer that rewrote ``assigned_to`` to the office admin —
    exactly the "stack on Kristine" pattern this PR removes.

    The recipient moved from the office's 当地 Admin to the location
    administrator on 2026-08-25; the disposition contract this case
    actually asserts is unchanged, so it only gains the managed-location
    row the resolver needs.
    """
    manager = await _make_person(db_session, work_email="mgr7@x.test")
    office_admin = await _make_person(
        db_session,
        work_email="kristine@x.test",
        display_name="Kristine Ye",
    )
    office_id = uuid.uuid4()
    db_session.add(AssetOfficeAddress(
        id=office_id,
        city="Sunnyvale",
        office_name="美国 Silicon Valley, USA",
        is_active=True,
    ))
    await db_session.flush()
    db_session.add(AssetOfficeLocalAdmin(
        office_id=office_id, person_id=office_admin.id, is_primary=True,
    ))
    _sv_location = AssetLocation(
        id=uuid.uuid4(), code=f"SV{uuid.uuid4().hex[:8]}", name="Sunnyvale",
    )
    db_session.add(_sv_location)
    await db_session.flush()
    db_session.add(AssetLocationAdmin(
        location_id=_sv_location.id, person_id=office_admin.id, is_primary=True,
    ))
    await db_session.flush()
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    leaver.work_location_name = "Sunnyvale"  # PR-#406: home office
    await db_session.flush()
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-US-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
        location="Sunnyvale",  # exact match
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook 16",
        resolution=RESO_PENDING,
    ))
    # 2026-05-20 contract change: default disposition is ``keep``
    # (held_by_leaver). To exercise the office-admin lookup, an
    # explicit ``recycle_to_office`` disposition must be set.
    proc.hardware_disposition = {
        str(asset.id): {"action": "recycle_to_office"},
    }
    await db_session.flush()

    sent: list[tuple[uuid.UUID, str]] = []

    async def _capture(envelope, **_kw):
        sent.append((envelope.recipient.id, envelope.template_key))

        class _R:
            delivered_channels = ["feishu"]

        return _R()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock(side_effect=_capture)  # type: ignore[assignment]
    svc.offboarding.suspend_sso = AsyncMock(  # type: ignore[assignment]
        return_value=type("R", (), {
            "sso_suspended_at": datetime.now(UTC),
        })(),
    )
    svc.leaver_guard.flip_to_left = AsyncMock()  # type: ignore[assignment]

    await svc._advance_single(leaver, fallback_email=manager.work_email)
    refreshed_asset = await db_session.scalar(
        select(AssetItem).where(AssetItem.id == asset.id),
    )
    # Office-held, no Person assignee.
    assert refreshed_asset.assigned_to is None
    assert refreshed_asset.holder_kind == "office"
    assert refreshed_asset.holder_office_id == office_id
    assert refreshed_asset.status == "held_by_office"

    # 2026-06-29: office recycle is inventory, NOT a person-transfer — no
    # synthetic AssetTransferRequest (which would surface as a phantom
    # incoming transfer in the office admin's /my/todos). The office admin
    # is reached via the dedicated ``offboarding.recycled_to_office`` push.
    xfer = await db_session.scalar(
        select(AssetTransferRequest).where(
            AssetTransferRequest.asset_ref_id == asset.id,
        ),
    )
    assert xfer is None
    assert (office_admin.id, "offboarding.recycled_to_office") in sent
    assert (office_admin.id, "force_transfer_received") not in sent


@pytest.mark.asyncio
async def test_hardware_force_transfer_falls_back_when_location_doesnt_match(
    db_session,
):
    """PR-#406 contract: when the leaver has no resolvable home office
    AND no admin override, disposition degrades to ``held_by_leaver``
    (NOT a blind manager-transfer). The data-quality issue lives in
    the leaver row (HRIS missing work_location_name), and the safe
    default is to freeze the asset for manual cleanup rather than
    forge an unwanted handoff.
    """
    manager = await _make_person(db_session, work_email="mgr8@x.test")
    office_admin = await _make_person(
        db_session, work_email="kristine2@x.test",
    )
    _office_2_id = uuid.uuid4()
    db_session.add(AssetOfficeAddress(
        id=_office_2_id,
        city="Sunnyvale",
        is_active=True,
    ))
    await db_session.flush()
    db_session.add(AssetOfficeLocalAdmin(
        office_id=_office_2_id, person_id=office_admin.id, is_primary=True,
    ))
    await db_session.flush()
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
        location="北美",  # close-but-not-equal — and leaver has no
                          # work_location_name, so home office can't
                          # be inferred either.
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues
    svc.offboarding.suspend_sso = AsyncMock(  # type: ignore[assignment]
        return_value=type("R", (), {
            "sso_suspended_at": datetime.now(UTC),
        })(),
    )
    svc.leaver_guard.flip_to_left = AsyncMock()  # type: ignore[assignment]

    await svc._advance_single(leaver, fallback_email=manager.work_email)
    refreshed_asset = await db_session.scalar(
        select(AssetItem).where(AssetItem.id == asset.id),
    )
    # No office match → degrade to held_by_leaver; assignee preserved.
    assert refreshed_asset.assigned_to == leaver.id
    assert refreshed_asset.status == "held_by_leaver"


@pytest.mark.asyncio
async def test_hardware_force_transfer_skips_inactive_office_admin(
    db_session,
):
    """PR-#406 contract: when the leaver's home office resolves but the
    office's admin is inactive, the recycle still routes to the
    office (held_by_office); inventory + offboarding handoff fan-out
    then falls through to IT fallback at notification time. The
    asset is NOT transferred to a person — the bug class this PR
    removes is "rewrite assigned_to to someone who didn't ask for
    it".
    """
    manager = await _make_person(db_session, work_email="mgr9@x.test")
    inactive_admin = await _make_person(
        db_session, work_email="ex@x.test", status="left",
    )
    office_id = uuid.uuid4()
    db_session.add(AssetOfficeAddress(
        id=office_id,
        city="Sunnyvale",
        is_active=True,
    ))
    await db_session.flush()
    # Seeded through the join as well, so the departed-contact filter in
    # ``get_office_primary_local_admin`` is what makes this fall back —
    # not simply an empty list.
    db_session.add(AssetOfficeLocalAdmin(
        office_id=office_id, person_id=inactive_admin.id, is_primary=True,
    ))
    await db_session.flush()
    leaver, proc = await _make_leaver_with_proc(db_session, manager=manager)
    leaver.work_location_name = "Sunnyvale"  # PR-#406: home office set
    await db_session.flush()
    asset = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"P-{uuid.uuid4().hex[:6]}",
        asset_name="MacBook",
        category="laptop",
        status="claimed",
        assigned_to=leaver.id,
        location="Sunnyvale",
    )
    db_session.add(asset)
    db_session.add(OffboardingBlocker(
        id=uuid.uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=asset.id,
        label="MacBook",
        resolution=RESO_PENDING,
    ))
    # 2026-05-20 contract change: default disposition is ``keep``;
    # exercise the explicit office-recycle path.
    proc.hardware_disposition = {
        str(asset.id): {"action": "recycle_to_office"},
    }
    await db_session.flush()

    svc = ProcessAdvanceService(db_session)
    svc.notify.dispatch = AsyncMock()  # type: ignore[assignment]
    svc.notify.enqueue = AsyncMock()  # type: ignore[assignment]  # grace push enqueues
    svc.offboarding.suspend_sso = AsyncMock(  # type: ignore[assignment]
        return_value=type("R", (), {
            "sso_suspended_at": datetime.now(UTC),
        })(),
    )
    svc.leaver_guard.flip_to_left = AsyncMock()  # type: ignore[assignment]

    await svc._advance_single(leaver, fallback_email=manager.work_email)
    refreshed_asset = await db_session.scalar(
        select(AssetItem).where(AssetItem.id == asset.id),
    )
    # Office is still the holder; the inactive-admin signal is handled
    # at notification-fanout time (recipient falls through to IT).
    assert refreshed_asset.assigned_to is None
    assert refreshed_asset.holder_kind == "office"
    assert refreshed_asset.holder_office_id == office_id
    assert refreshed_asset.status == "held_by_office"