from __future__ import annotations

import os
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import get_protocol_members
from uuid import uuid4

import pytest
from sqlalchemy import event
from sqlalchemy.orm import Session

from cloud.database import CloudDatabase
from cloud.db_models import TaskAttemptRow
from cloud.plugins import PluginManifest
from cloud.pool import PooledDevice
from cloud.repository import CloudRepository, LeasedAssignment, TaskAttemptRecord
from cloud.scheduler import ScheduledTask, TaskConstraints


@pytest.fixture(
    params=[
        pytest.param("sqlite", id="sqlite"),
        pytest.param("postgresql", id="postgresql", marks=pytest.mark.integration),
    ]
)
def database_url(request: pytest.FixtureRequest, tmp_path: Path) -> str:
    if request.param == "sqlite":
        return f"sqlite:///{(tmp_path / 'contract.sqlite3').as_posix()}"

    url = os.getenv("TEST_POSTGRES_URL")
    if not url:
        pytest.skip("TEST_POSTGRES_URL is required for PostgreSQL contract tests")
    return url


def _unique_id(prefix: str) -> str:
    return f"{prefix}-{uuid4().hex}"


def _device(device_id: str, host_id: str) -> PooledDevice:
    return PooledDevice(
        device_id=device_id,
        host_id=host_id,
        driver_type="wda",
        status="idle",
        capability_tags=["ios", "physical"],
        synced_at=datetime(2026, 7, 12, tzinfo=UTC),
    )


def test_cloud_repository_exposes_crud_and_atomic_lease_operations() -> None:
    members = get_protocol_members(CloudRepository)

    assert {
        "upsert_host",
        "replace_host_devices",
        "list_hosts",
        "list_devices",
        "enqueue_task",
        "get_task",
        "save_plugin",
        "assign_task",
        "claim_assignment",
        "renew_lease",
        "record_task_result",
        "reap_expired_leases",
        "list_task_attempts",
        "health_check",
        "close",
    } <= members


def test_repository_transfer_records_are_immutable() -> None:
    assert TaskAttemptRecord.__dataclass_params__.frozen is True
    assert LeasedAssignment.__dataclass_params__.frozen is True


def test_repository_crud_contract(database_url: str) -> None:
    database = CloudDatabase(database_url)
    repository = database.repository
    host_id = _unique_id("host")
    device_id = _unique_id("device")
    task_id = _unique_id("task")
    plugin_name = _unique_id("plugin")
    seen_at = datetime(2026, 7, 12, 1, 2, 3, tzinfo=UTC)

    try:
        repository.upsert_host(
            host_id,
            address="127.0.0.1:9000",
            last_seen_at=seen_at,
        )
        repository.replace_host_devices(host_id, [_device(device_id, host_id)])
        repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="open settings",
                workflow_definition_id=None,
                constraints=TaskConstraints(
                    driver_type="wda",
                    capability_tags=["ios"],
                ),
                created_at=seen_at,
            )
        )
        repository.save_plugin(
            PluginManifest(
                name=plugin_name,
                version="1.0.0",
                entry_point_kind="tool",
                target="cloud.store:CloudStore",
            ),
            wired=False,
        )

        assert repository.get_host(host_id) is not None
        assert repository.get_device(device_id) == _device(device_id, host_id)
        assert repository.get_task(task_id) is not None
        assert repository.get_plugin(plugin_name) is not None
        repository.health_check()
    finally:
        database.close()


def test_repository_can_atomically_transfer_explicit_device_takeover(
    database_url: str,
) -> None:
    database = CloudDatabase(database_url)
    repository = database.repository
    old_host_id = _unique_id("old-host")
    new_host_id = _unique_id("new-host")
    device_id = _unique_id("shared-device")

    try:
        repository.replace_host_devices(
            old_host_id,
            [_device(device_id, old_host_id)],
        )
        repository.replace_host_devices(
            new_host_id,
            [_device(device_id, new_host_id)],
            allow_device_takeover=True,
        )

        matching = [
            device
            for device in repository.list_devices()
            if device.device_id == device_id
        ]
        assert len(matching) == 1
        assert matching[0].host_id == new_host_id
    finally:
        database.close()


def test_failed_snapshot_transaction_rolls_back(database_url: str) -> None:
    database = CloudDatabase(database_url)
    repository = database.repository
    host_id = _unique_id("rollback-host")
    original_device_id = _unique_id("original-device")
    duplicate_device_id = _unique_id("duplicate-device")

    try:
        repository.upsert_host(
            host_id,
            address=None,
            last_seen_at=datetime(2026, 7, 12, tzinfo=UTC),
        )
        repository.replace_host_devices(
            host_id,
            [_device(original_device_id, host_id)],
        )

        def fail_before_insert(
            _connection: object,
            _cursor: object,
            statement: str,
            _parameters: object,
            _context: object,
            _executemany: bool,
        ) -> None:
            if statement.startswith("INSERT INTO pooled_devices"):
                raise RuntimeError("injected snapshot write failure")

        event.listen(database.engine, "before_cursor_execute", fail_before_insert)
        try:
            with pytest.raises(RuntimeError, match="injected snapshot write failure"):
                repository.replace_host_devices(
                    host_id,
                    [_device(duplicate_device_id, host_id)],
                )
        finally:
            event.remove(database.engine, "before_cursor_execute", fail_before_insert)

        devices = [
            device for device in repository.list_devices() if device.host_id == host_id
        ]
        assert [device.device_id for device in devices] == [original_device_id]
        repository.health_check()
    finally:
        database.close()


def test_repository_state_survives_application_restart(database_url: str) -> None:
    host_id = _unique_id("restart-host")
    device_id = _unique_id("restart-device")
    first_process = CloudDatabase(database_url)
    try:
        first_process.repository.upsert_host(
            host_id,
            address="host.internal",
            last_seen_at=datetime(2026, 7, 12, tzinfo=UTC),
        )
        first_process.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
    finally:
        first_process.close()

    restarted_process = CloudDatabase(database_url)
    try:
        host = restarted_process.repository.get_host(host_id)
        device = restarted_process.repository.get_device(device_id)

        assert host is not None
        assert host.address == "host.internal"
        assert device == _device(device_id, host_id)
    finally:
        restarted_process.close()


def test_task_lease_and_terminal_fields_round_trip(database_url: str) -> None:
    database = CloudDatabase(database_url)
    task_id = _unique_id("lease-task")
    created_at = datetime(2026, 7, 12, 1, 0, tzinfo=UTC)
    lease_expires_at = datetime(2026, 7, 12, 1, 5, tzinfo=UTC)
    updated_at = datetime(2026, 7, 12, 1, 1, tzinfo=UTC)
    task = ScheduledTask(
        id=task_id,
        goal="capture diagnostics",
        workflow_definition_id=None,
        constraints=TaskConstraints(),
        status="failed",
        assigned_device_id="device-a",
        assigned_host_id="host-a",
        attempt_count=2,
        lease_id="lease-a",
        lease_expires_at=lease_expires_at,
        terminal_result={"steps": 3, "status": "failed"},
        failure_reason="device disconnected",
        updated_at=updated_at,
        created_at=created_at,
    )

    try:
        database.repository.enqueue_task(task)

        assert database.repository.get_task(task_id) == task
    finally:
        database.close()


def test_task_attempt_history_is_ordered_and_complete(database_url: str) -> None:
    database = CloudDatabase(database_url)
    task_id = _unique_id("attempt-task")
    created_at = datetime(2026, 7, 12, 1, 0, tzinfo=UTC)
    lease_expires_at = datetime(2026, 7, 12, 1, 5, tzinfo=UTC)

    try:
        with Session(database.engine) as session, session.begin():
            session.add_all(
                [
                    TaskAttemptRow(
                        task_id=task_id,
                        attempt=2,
                        lease_id="lease-2",
                        host_id="host-b",
                        device_id="device-b",
                        status="failed",
                        lease_expires_at=lease_expires_at.isoformat(),
                        created_at=created_at.isoformat(),
                        completed_at=lease_expires_at.isoformat(),
                        failure_reason="execution failed",
                        result_json='{"exit_code": 1}',
                    ),
                    TaskAttemptRow(
                        task_id=task_id,
                        attempt=1,
                        lease_id="lease-1",
                        host_id="host-a",
                        device_id="device-a",
                        status="expired",
                        lease_expires_at=lease_expires_at.isoformat(),
                        created_at=created_at.isoformat(),
                        completed_at=lease_expires_at.isoformat(),
                        failure_reason="lease expired",
                        result_json=None,
                    ),
                ]
            )

        attempts = database.repository.list_task_attempts(task_id)

        assert [attempt.attempt for attempt in attempts] == [1, 2]
        assert attempts[0].status == "expired"
        assert attempts[1].terminal_result == {"exit_code": 1}
        assert attempts[1].failure_reason == "execution failed"
    finally:
        database.close()


def test_atomic_assignment_creates_lease_attempt_and_reservation(
    database_url: str,
) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("assignment-host")
    device_id = _unique_id("assignment-device")
    task_id = _unique_id("assignment-task")
    now = datetime(2026, 7, 12, 2, 0, tzinfo=UTC)
    lease_expires_at = now + timedelta(minutes=1)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="open settings",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )

        assignment = database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="lease-1",
            lease_expires_at=lease_expires_at,
            now=now,
        )

        assert assignment is not None
        assert assignment.attempt == 1
        assert assignment.lease_id == "lease-1"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "assigned"
        assert task.attempt_count == 1
        assert task.lease_expires_at == lease_expires_at
        assert device_id in database.repository.list_reserved_device_ids(now=now)
        attempts = database.repository.list_task_attempts(task_id)
        assert len(attempts) == 1
        assert attempts[0].status == "assigned"
        assert attempts[0].lease_id == "lease-1"
    finally:
        database.close()


def test_active_assignment_blocks_reuse_of_stale_idle_snapshot(
    database_url: str,
) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("reservation-host")
    device_id = _unique_id("reservation-device")
    first_task_id = _unique_id("reservation-task")
    second_task_id = _unique_id("reservation-task")
    now = datetime(2026, 7, 12, 3, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        for task_id in (first_task_id, second_task_id):
            database.repository.enqueue_task(
                ScheduledTask(
                    id=task_id,
                    goal="run task",
                    workflow_definition_id=None,
                    constraints=TaskConstraints(),
                    created_at=now,
                )
            )

        first = database.repository.assign_task(
            task_id=first_task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="lease-active",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )
        second = database.repository.assign_task(
            task_id=second_task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="lease-blocked",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )

        assert first is not None
        assert second is None
        blocked_task = database.repository.get_task(second_task_id)
        assert blocked_task is not None
        assert blocked_task.status == "queued"
        assert database.repository.list_task_attempts(second_task_id) == []
    finally:
        database.close()


def test_expired_assignment_no_longer_reserves_device(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("expired-host")
    device_id = _unique_id("expired-device")
    first_task_id = _unique_id("expired-task")
    second_task_id = _unique_id("expired-task")
    assigned_at = datetime(2026, 7, 12, 4, 0, tzinfo=UTC)
    after_expiry = assigned_at + timedelta(minutes=2)

    try:
        database.repository.upsert_host(
            host_id,
            address=None,
            last_seen_at=assigned_at,
        )
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        for task_id in (first_task_id, second_task_id):
            database.repository.enqueue_task(
                ScheduledTask(
                    id=task_id,
                    goal="run task",
                    workflow_definition_id=None,
                    constraints=TaskConstraints(),
                    created_at=assigned_at,
                )
            )
        assert database.repository.assign_task(
            task_id=first_task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="lease-expired",
            lease_expires_at=assigned_at + timedelta(minutes=1),
            now=assigned_at,
        )

        assert device_id not in database.repository.list_reserved_device_ids(
            now=after_expiry
        )
        assert database.repository.assign_task(
            task_id=second_task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="lease-new",
            lease_expires_at=after_expiry + timedelta(minutes=1),
            now=after_expiry,
        )
    finally:
        database.close()


def test_owning_host_claims_one_active_assignment(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("claim-host")
    device_id = _unique_id("claim-device")
    task_id = _unique_id("claim-task")
    now = datetime(2026, 7, 12, 5, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="claim me",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="claim-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )

        assignment = database.repository.claim_assignment(
            host_id=host_id,
            now=now + timedelta(seconds=1),
        )

        assert assignment is not None
        assert assignment.task_id == task_id
        assert assignment.lease_id == "claim-lease"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "dispatched"
        attempts = database.repository.list_task_attempts(task_id)
        assert len(attempts) == 1
        assert attempts[0].status == "dispatched"
        assert database.repository.claim_assignment(host_id=host_id, now=now) is None
    finally:
        database.close()


def test_foreign_host_cannot_claim_assignment(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("owner-host")
    device_id = _unique_id("owner-device")
    task_id = _unique_id("owner-task")
    now = datetime(2026, 7, 12, 6, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="owner only",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="owner-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )

        assert (
            database.repository.claim_assignment(
                host_id=_unique_id("foreign-host"),
                now=now,
            )
            is None
        )
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "assigned"
    finally:
        database.close()


def test_expired_assignment_cannot_be_claimed(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("expired-claim-host")
    device_id = _unique_id("expired-claim-device")
    task_id = _unique_id("expired-claim-task")
    now = datetime(2026, 7, 12, 7, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="too late",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="expired-claim-lease",
            lease_expires_at=now + timedelta(seconds=1),
            now=now,
        )

        assert (
            database.repository.claim_assignment(
                host_id=host_id,
                now=now + timedelta(seconds=2),
            )
            is None
        )
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "assigned"
    finally:
        database.close()


def test_active_lease_renews_for_owning_host(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("renew-host")
    device_id = _unique_id("renew-device")
    task_id = _unique_id("renew-task")
    now = datetime(2026, 7, 12, 8, 0, tzinfo=UTC)
    initial_expiry = now + timedelta(minutes=1)
    renewed_expiry = now + timedelta(minutes=2)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="renew me",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="renew-lease",
            lease_expires_at=initial_expiry,
            now=now,
        )
        database.repository.claim_assignment(host_id=host_id, now=now)

        status = database.repository.renew_lease(
            task_id=task_id,
            attempt=1,
            lease_id="renew-lease",
            host_id=host_id,
            lease_expires_at=renewed_expiry,
            now=now + timedelta(seconds=30),
        )

        assert status == "renewed"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.lease_expires_at == renewed_expiry
        attempts = database.repository.list_task_attempts(task_id)
        assert attempts[0].lease_expires_at == renewed_expiry
    finally:
        database.close()


@pytest.mark.parametrize(
    ("attempt", "lease_id", "host_id"),
    [
        (2, "lease-current", "owner"),
        (1, "lease-stale", "owner"),
        (1, "lease-current", "foreign"),
    ],
)
def test_stale_or_foreign_lease_renewal_conflicts(
    database_url: str,
    attempt: int,
    lease_id: str,
    host_id: str,
) -> None:
    database = CloudDatabase(database_url)
    owner_id = _unique_id("renew-owner")
    device_id = _unique_id("renew-conflict-device")
    task_id = _unique_id("renew-conflict-task")
    now = datetime(2026, 7, 12, 9, 0, tzinfo=UTC)
    initial_expiry = now + timedelta(minutes=1)

    try:
        database.repository.upsert_host(owner_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            owner_id,
            [_device(device_id, owner_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="do not renew",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=owner_id,
            device_id=device_id,
            lease_id="lease-current",
            lease_expires_at=initial_expiry,
            now=now,
        )

        status = database.repository.renew_lease(
            task_id=task_id,
            attempt=attempt,
            lease_id=lease_id,
            host_id=owner_id if host_id == "owner" else _unique_id("foreign"),
            lease_expires_at=now + timedelta(minutes=2),
            now=now + timedelta(seconds=30),
        )

        assert status == "conflict"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.lease_expires_at == initial_expiry
    finally:
        database.close()


def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("expired-renew-host")
    device_id = _unique_id("expired-renew-device")
    task_id = _unique_id("expired-renew-task")
    now = datetime(2026, 7, 12, 10, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="expired renewal",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="expired-renew-lease",
            lease_expires_at=now + timedelta(seconds=1),
            now=now,
        )

        assert (
            database.repository.renew_lease(
                task_id=task_id,
                attempt=1,
                lease_id="expired-renew-lease",
                host_id=host_id,
                lease_expires_at=now + timedelta(minutes=2),
                now=now + timedelta(seconds=2),
            )
            == "expired"
        )
        assert (
            database.repository.renew_lease(
                task_id=_unique_id("missing-task"),
                attempt=1,
                lease_id="missing-lease",
                host_id=host_id,
                lease_expires_at=now + timedelta(minutes=2),
                now=now,
            )
            == "not_found"
        )
    finally:
        database.close()


def test_terminal_result_is_recorded_idempotently_and_releases_reservation(
    database_url: str,
) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("result-host")
    device_id = _unique_id("result-device")
    task_id = _unique_id("result-task")
    now = datetime(2026, 7, 12, 11, 0, tzinfo=UTC)
    completed_at = now + timedelta(seconds=30)
    result = {"steps": 4, "summary": "completed"}

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="complete me",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="result-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )
        database.repository.claim_assignment(host_id=host_id, now=now)

        first_status = database.repository.record_task_result(
            task_id=task_id,
            attempt=1,
            lease_id="result-lease",
            host_id=host_id,
            status="done",
            failure_reason=None,
            terminal_result=result,
            completed_at=completed_at,
        )
        repeated_status = database.repository.record_task_result(
            task_id=task_id,
            attempt=1,
            lease_id="result-lease",
            host_id=host_id,
            status="done",
            failure_reason=None,
            terminal_result=result,
            completed_at=completed_at + timedelta(seconds=1),
        )

        assert first_status == "recorded"
        assert repeated_status == "already_recorded"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "done"
        assert task.terminal_result == result
        attempts = database.repository.list_task_attempts(task_id)
        assert attempts[0].status == "done"
        assert attempts[0].terminal_result == result
        assert attempts[0].completed_at == completed_at
        assert device_id not in database.repository.list_reserved_device_ids(
            now=completed_at
        )
    finally:
        database.close()


def test_conflicting_terminal_result_cannot_overwrite_recorded_outcome(
    database_url: str,
) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("result-conflict-host")
    device_id = _unique_id("result-conflict-device")
    task_id = _unique_id("result-conflict-task")
    now = datetime(2026, 7, 12, 12, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="stable outcome",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="stable-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )
        assert (
            database.repository.record_task_result(
                task_id=task_id,
                attempt=1,
                lease_id="stable-lease",
                host_id=host_id,
                status="failed",
                failure_reason="device offline",
                terminal_result={"retryable": True},
                completed_at=now + timedelta(seconds=10),
            )
            == "recorded"
        )

        assert (
            database.repository.record_task_result(
                task_id=task_id,
                attempt=1,
                lease_id="stable-lease",
                host_id=host_id,
                status="done",
                failure_reason=None,
                terminal_result={"retryable": False},
                completed_at=now + timedelta(seconds=20),
            )
            == "conflict"
        )
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "failed"
        assert task.failure_reason == "device offline"
        assert task.terminal_result == {"retryable": True}
    finally:
        database.close()


@pytest.mark.parametrize(
    ("attempt", "lease_id", "host_kind", "report_delay"),
    [
        (2, "active-lease", "owner", 10),
        (1, "stale-lease", "owner", 10),
        (1, "active-lease", "foreign", 10),
        (1, "active-lease", "owner", 61),
    ],
)
def test_stale_foreign_or_expired_result_is_rejected(
    database_url: str,
    attempt: int,
    lease_id: str,
    host_kind: str,
    report_delay: int,
) -> None:
    database = CloudDatabase(database_url)
    owner_id = _unique_id("result-owner")
    device_id = _unique_id("result-stale-device")
    task_id = _unique_id("result-stale-task")
    now = datetime(2026, 7, 12, 13, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(owner_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            owner_id,
            [_device(device_id, owner_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="reject stale result",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=owner_id,
            device_id=device_id,
            lease_id="active-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )

        status = database.repository.record_task_result(
            task_id=task_id,
            attempt=attempt,
            lease_id=lease_id,
            host_id=owner_id if host_kind == "owner" else _unique_id("foreign"),
            status="done",
            failure_reason=None,
            terminal_result={"ignored": True},
            completed_at=now + timedelta(seconds=report_delay),
        )

        assert status == "conflict"
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "assigned"
        assert task.terminal_result is None
    finally:
        database.close()


@pytest.mark.parametrize("claimed", [False, True], ids=["assigned", "dispatched"])
def test_expired_lease_requeues_with_auditable_history(
    database_url: str,
    claimed: bool,
) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("requeue-host")
    device_id = _unique_id("requeue-device")
    task_id = _unique_id("requeue-task")
    now = datetime(2026, 7, 12, 14, 0, tzinfo=UTC)
    expired_at = now + timedelta(seconds=10)
    reaped_at = expired_at + timedelta(seconds=1)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="retry after expiry",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="expired-attempt-1",
            lease_expires_at=expired_at,
            now=now,
        )
        if claimed:
            database.repository.claim_assignment(host_id=host_id, now=now)

        reaped_task_ids = database.repository.reap_expired_leases(
            now=reaped_at,
            max_attempts=2,
        )
        assert task_id in reaped_task_ids

        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "queued"
        assert task.attempt_count == 1
        assert task.assigned_host_id is None
        assert task.assigned_device_id is None
        assert task.lease_id is None
        assert task.lease_expires_at is None
        assert task.failure_reason is None
        attempts = database.repository.list_task_attempts(task_id)
        assert len(attempts) == 1
        assert attempts[0].status == "expired"
        assert attempts[0].failure_reason == "lease expired"
        assert attempts[0].completed_at == reaped_at
        assert device_id not in database.repository.list_reserved_device_ids(
            now=reaped_at
        )

        second_assignment = database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="active-attempt-2",
            lease_expires_at=reaped_at + timedelta(minutes=1),
            now=reaped_at,
        )
        assert second_assignment is not None
        assert second_assignment.attempt == 2
        assert [
            attempt.status
            for attempt in database.repository.list_task_attempts(task_id)
        ] == ["expired", "assigned"]
    finally:
        database.close()


def test_expired_lease_fails_at_attempt_limit(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("limit-host")
    device_id = _unique_id("limit-device")
    task_id = _unique_id("limit-task")
    now = datetime(2026, 7, 12, 15, 0, tzinfo=UTC)
    expired_at = now + timedelta(seconds=10)
    reaped_at = expired_at + timedelta(seconds=1)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="fail after expiry",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="final-attempt",
            lease_expires_at=expired_at,
            now=now,
        )

        reaped_task_ids = database.repository.reap_expired_leases(
            now=reaped_at,
            max_attempts=1,
        )
        assert task_id in reaped_task_ids

        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "failed"
        assert task.failure_reason == "lease expired after 1 attempts"
        assert task.lease_id is None
        assert task.lease_expires_at is None
        attempts = database.repository.list_task_attempts(task_id)
        assert attempts[0].status == "expired"
        assert device_id not in database.repository.list_reserved_device_ids(
            now=reaped_at
        )
        assert (
            database.repository.reap_expired_leases(
                now=reaped_at,
                max_attempts=1,
            )
            == []
        )
    finally:
        database.close()


def test_unexpired_lease_is_not_reaped(database_url: str) -> None:
    database = CloudDatabase(database_url)
    host_id = _unique_id("active-reaper-host")
    device_id = _unique_id("active-reaper-device")
    task_id = _unique_id("active-reaper-task")
    now = datetime(2026, 7, 12, 16, 0, tzinfo=UTC)

    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="stay active",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="active-reaper-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )

        assert (
            database.repository.reap_expired_leases(
                now=now + timedelta(seconds=30),
                max_attempts=2,
            )
            == []
        )
        task = database.repository.get_task(task_id)
        assert task is not None
        assert task.status == "assigned"
        assert device_id in database.repository.list_reserved_device_ids(now=now)
    finally:
        database.close()


def test_task_lifecycle_logs_structured_identifiers(
    database_url: str, monkeypatch
) -> None:
    import cloud.sql_repository as repository_module

    database = CloudDatabase(database_url)
    now = datetime(2026, 7, 12, 20, 0, tzinfo=UTC)
    host_id = _unique_id("log-host")
    device_id = _unique_id("log-device")
    task_id = _unique_id("log-task")
    events: list[dict[str, object]] = []

    def record_info(_message: str, *, extra: dict[str, object]) -> None:
        events.append(extra)

    monkeypatch.setattr(repository_module.logger, "info", record_info)
    try:
        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
        database.repository.replace_host_devices(
            host_id,
            [_device(device_id, host_id)],
        )
        database.repository.enqueue_task(
            ScheduledTask(
                id=task_id,
                goal="sensitive typed text",
                workflow_definition_id=None,
                constraints=TaskConstraints(),
                created_at=now,
            )
        )
        database.repository.assign_task(
            task_id=task_id,
            host_id=host_id,
            device_id=device_id,
            lease_id="log-lease",
            lease_expires_at=now + timedelta(minutes=1),
            now=now,
        )
        database.repository.claim_assignment(host_id=host_id, now=now)
        database.repository.record_task_result(
            task_id=task_id,
            attempt=1,
            lease_id="log-lease",
            host_id=host_id,
            status="done",
            failure_reason=None,
            terminal_result={"screenshot": "secret-image"},
            completed_at=now + timedelta(seconds=1),
        )

        assert [event["event"] for event in events] == [
            "assigned",
            "claimed",
            "completed",
        ]
        assert all(event["task_id"] == task_id for event in events)
        assert all(event["host_id"] == host_id for event in events)
        assert all(event["device_id"] == device_id for event in events)
        assert all(event["correlation_id"] for event in events)
        assert "sensitive typed text" not in repr(events)
        assert "secret-image" not in repr(events)
    finally:
        database.close()