UmbraTrans/umbratrans_web/tests/test_compute.py

"""Compute substrate tests.

Four of these are about claims and leases, and they exist because the alternative
is a failure that is silent when it happens:

* **I25** two agents claiming at once must get different jobs. A queue where they
  can get the same one produces duplicated work that is individually plausible
  and collectively wrong.
* **I26** a job whose holder was killed must come back. A worker that is
  ``kill -9``'d releases nothing, so the lease is what releases it.
* **I24** a node that has never authenticated must not be schedulable. A node that
  looks ready before its agent has spoken swallows every job dispatched to it.
* **I28** a node credential must not reach the administrator surface. That one is
  here as a route-list assertion, because it was a real bug: mounting one router
  at two prefixes published node creation behind node-credential authentication.

And one about the thing that is easiest to get wrong in the other direction:

* a slow device is **eligible**. A CPU with enough memory takes the job; refusing
  it would turn a machine that would have finished the work in an hour into a
  machine that never touches it. §7.3
"""

from __future__ import annotations

import concurrent.futures
import os
import time
from pathlib import Path
from uuid import UUID, uuid4

import pytest

from umbratrans_web.compute import (
    BLOCKED,
    COMPLETE,
    FAILED,
    PARTIAL,
    QUEUED,
    RUNNING,
    ComputeRepository,
    ModelRequirement,
)

INTEGRATION_ENV = "UMBRATRANS_TEST_DSN"
ACK_ENV = "UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE"

integration = pytest.mark.skipif(
    not os.environ.get(INTEGRATION_ENV) or os.environ.get(ACK_ENV) != "1",
    reason=(
        f"{INTEGRATION_ENV} is unset, or {ACK_ENV} is not 1. This suite writes to and "
        "truncates tables in the target database, so it needs a scratch database and an "
        "explicit acknowledgement. Never point it at a deployment."
    ),
)


def _require_scratch_database() -> None:
    """Refuse to run without a scratch database and an explicit acknowledgement.

    Repeated inside the fixture rather than relying on the ``integration`` mark,
    because a fixture cannot take a mark as an argument - and a skipped
    *test* with a fixture that still ran would truncate a database before
    skipping.
    """

    if not os.environ.get(INTEGRATION_ENV) or os.environ.get(ACK_ENV) != "1":
        pytest.skip(
            f"{INTEGRATION_ENV} is unset, or {ACK_ENV} is not 1. This suite truncates "
            "tables in the target database, so it needs a scratch database and an explicit "
            "acknowledgement. Never point it at a deployment."
        )


@pytest.fixture
def compute():
    """A repository against a scratch database, emptied on teardown."""

    _require_scratch_database()

    repository = ComputeRepository(
        os.environ[INTEGRATION_ENV], min_size=1, max_size=6, passwords=None
    )
    try:
        yield repository
        with repository._connection() as connection, connection.transaction():  # noqa: SLF001
            connection.execute(
                "TRUNCATE public.job_events, public.jobs, public.node_heartbeats, "
                "public.node_models, public.node_credentials, public.compute_devices, "
                "public.compute_nodes, public.model_registry CASCADE"
            )
    finally:
        repository.close()


@pytest.fixture
def organization(compute):
    """One organization, seeded directly.

    Created in the repository rather than through the service because these tests
    are about dispatch, and the service path would need an authenticated
    administrator to exist first.
    """

    identifier = uuid4()
    with compute._connection() as connection, connection.transaction():  # noqa: SLF001
        connection.execute(
            "INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
            (identifier, "Compute Test", f"compute-{identifier.hex[:8]}"),
        )
    return identifier


def _node(compute, organization_id: UUID, name: str, **kwargs):
    return compute.create_node(organization_id, name=name, **kwargs)


def _device(node, *, kind="cpu", vram_total=0, vram_free=0, ram_total=16000, ram_free=8000, factor=1.0, **extra):
    return {
        "kind": kind,
        "label": extra.get("label", "cpu:0" if kind == "cpu" else "gpu:0"),
        "vram_total_mb": vram_total,
        "vram_free_mb": vram_free,
        "ram_total_mb": ram_total,
        "ram_free_mb": ram_free,
        "compute_units": extra.get("compute_units"),
        "throughput_factor": factor,
        "supports_accelerators": extra.get("supports_accelerators", []),
    }


def _online(compute, node, devices):
    """Bring a node to ``online`` the only way it can become online: a heartbeat."""

    return compute.heartbeat(node.id, agent_version="test", devices=devices)


# ----------------------------------------------------------------------
# I24: a node is online because it heartbeated
# ----------------------------------------------------------------------


@integration
def test_a_new_node_is_not_schedulable(compute, organization) -> None:
    """Creation is not readiness. A node that has never spoken takes no work."""

    node = _node(compute, organization, "silent")
    assert node.status == "provisioning"
    assert node.is_schedulable is False
    # And the reason is one a person can act on, not a bare false.
    from umbratrans_web.routes_compute import _node_reason

    assert "never authenticated" in _node_reason(node)


@integration
def test_a_heartbeat_makes_a_node_schedulable(compute, organization) -> None:
    node = _node(compute, organization, "alive")
    updated = _online(compute, node, [_device(node)])
    assert updated.status == "online"
    assert updated.is_schedulable is True


@integration
def test_a_draining_node_is_not_promoted_back_by_heartbeating(compute, organization) -> None:
    """Otherwise a node being drained re-advertises itself for the work being drained.

    An administrator drains a machine before rebooting it. If a heartbeat could
    promote it back, the drain would last until the reboot, which is the opposite
    of what draining is for.
    """

    node = _online(compute, _node(compute, organization, "draining"), [_device(_node(compute, organization, "x"))])
    compute.set_node_status(node.id, "draining")
    again = compute.heartbeat(node.id, agent_version="test", devices=[])
    assert again.status == "draining"
    assert again.is_schedulable is False


@integration
def test_online_is_not_set_by_an_administrator(compute, organization) -> None:
    """``online`` is earned. The administrative path cannot grant it by typing it."""

    from umbratrans_web.errors import RepositoryError

    node = _node(compute, organization, "no-cheating")
    with pytest.raises(RepositoryError) as error:
        compute.set_node_status(node.id, "online")
    assert "earned by a heartbeat" in str(error.value)


# ----------------------------------------------------------------------
# Revoking a credential, and removing a machine
# ----------------------------------------------------------------------
#
# `node_credentials.revoked_at` existed from the first migration and `resolve_node_credential`
# has always honoured it. Nothing could set it. So the ability to cut a node off was
# present in the data model and absent from every API, which is the shape of defect this
# project keeps finding: the column is the capability.


def _live_credential(compute, organization, name: str, mac: str = "00:11:22:33:ac:4b"):
    """Pair a node and hand back the plaintext key, the way a real bootstrap would."""

    minted = compute.create_pairing_code(organization, node_name=name)
    paired = compute.redeem_pairing_code(minted["code"], name, mac=mac)
    assert paired["ok"] is True, paired
    return paired["node_id"], paired["key"]


@integration
def test_a_revoked_credential_stops_working_immediately(compute, organization) -> None:
    """The whole point: a key that may be compromised must die now, not at next restart."""

    node_id, key = _live_credential(compute, organization, "revoke-me")
    assert compute.resolve_node_credential(key) is not None, "the credential never worked"

    compute.revoke_node_credentials(node_id, reason="suspected compromise", actor="test")

    assert compute.resolve_node_credential(key) is None, (
        "a revoked credential still authenticates. The key was issued to a machine whose "
        "agent may be hostile; revoking it has to take effect on the next request."
    )


@integration
def test_revoking_leaves_the_node_and_its_history_in_place(compute, organization) -> None:
    """Revocation is for a machine that is coming back. Removal is the other thing.

    Confusing the two is how a reimage turns into a dead node: the operator revokes, the
    node stays, and it is never told it may authenticate again - which it cannot, because
    the credential is gone and pairing is refused for the name.
    """

    node_id, key = _live_credential(compute, organization, "revoked-but-alive")
    compute.revoke_node_credentials(node_id, reason="reimaging")

    node = compute.get_node(node_id)
    assert node is not None, "revoking deleted the node"
    assert node.name == "revoked-but-alive-AC4B"
    assert node.status == "provisioning", "revoking changed the node's status"

    # And the node keeps existing, so it still appears in the console.
    names = {one.name for one in compute.list_nodes(organization)}
    assert "revoked-but-alive-AC4B" in names


@integration
def test_revoking_twice_is_harmless_and_reports_nothing_new_the_second_time(
    compute, organization
) -> None:
    """A retried request must not report work it did not do.

    An operator clicking twice, or a retry after a timeout, should not be told it cut off
    two credentials.
    """

    node_id, _key = _live_credential(compute, organization, "revoke-twice")

    assert compute.revoke_node_credentials(node_id, reason="first") == 1
    assert compute.revoke_node_credentials(node_id, reason="second") == 0


@integration
def test_revoking_records_why_and_who(compute, organization) -> None:
    """Six months on, "the node went dark" is not an answer an operator can act on."""

    node_id, _key = _live_credential(compute, organization, "revoked-with-reason")

    compute.revoke_node_credentials(node_id, reason="suspected compromise", actor="ops@example")

    events = _audit_events(compute, node_id)
    assert any(
        event["action"] == "credentials_revoked" and event["metadata"].get("reason") == "suspected compromise"
        for event in events
    ), f"the revocation left no trace with its reason: {events}"


@integration
def test_removing_a_machine_revokes_its_credential_and_drops_its_devices(
    compute, organization
) -> None:
    """Everything that could let the node *act* goes; the row stays as history."""

    node_id, key = _live_credential(compute, organization, "written-off")
    _online(
        compute,
        compute.get_node(node_id),
        [_device(_node(compute, organization, "written-off-device"))],
    )

    compute.remove_node(node_id, reason="hardware failure", actor="ops@example")

    assert compute.resolve_node_credential(key) is None, (
        "a removed machine can still authenticate"
    )
    node = compute.get_node(node_id)
    assert node is not None, "the row was deleted, which the event trail forbids"
    assert node.status == "removed"
    assert node.devices == (), "a removed machine still has devices and could be placed work"


@integration
def test_a_removed_node_is_handed_no_work(compute, organization) -> None:
    """Removal has to mean removal at the claim path, not only in the console.

    The status is what the claim checks, so this is the assertion that matters: a node
    that reads ``removed`` must behave exactly like one that was never paired.
    """

    node_id, _key = _live_credential(compute, organization, "removed-no-work")
    _online(
        compute,
        compute.get_node(node_id),
        [_device(_node(compute, organization, "removed-device"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")

    assert compute.claim_job(node_id) is not None, "precondition: the node can claim"

    second = _live_credential(compute, organization, "removed-no-work-2", mac="00:11:22:33:3f:19")
    compute.submit_job(organization, kind="sleep", requested_by="test")
    compute.remove_node(second[0], reason="hardware failure")

    assert compute.claim_job(second[0]) is None, "a removed node was handed work"


@integration
def test_removing_a_node_releases_the_work_it_was_holding(compute, organization) -> None:
    """The requirement: a node deleted mid-job leaves that job *not complete*.

    Released here rather than left to the lease to expire. An operator removing dead
    hardware already knows the work did not finish, and making them wait out a lease to
    discover that is the wrong way round - for a long lease it is most of an hour of the
    queue stalled behind a machine nobody is going to switch on.
    """

    node_id, _key = _live_credential(compute, organization, "died-mid-job")
    _online(
        compute,
        compute.get_node(node_id),
        [_device(_node(compute, organization, "died-device"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node_id)
    assert claimed is not None
    job_id = claimed.job.id

    compute.remove_node(node_id, reason="hardware failure")

    from umbratrans_web.compute import QUEUED

    job = compute.get_job(job_id)
    assert job is not None
    assert job.status == QUEUED, (
        f"work held by a removed node is {job.status!r}. It has to be queued again so "
        "another node can take it, and the lease cannot be relied on to release it."
    )
    assert job.lease_owner is None, "the released job still points at the dead node"


@integration
def test_work_held_by_a_removed_node_is_not_silently_lost(compute, organization) -> None:
    """Released, not re-run from scratch by the lease: the job is still there.

    Stated separately because "released" and "the row vanished" both look like a working
    queue to an operator watching the console, and only one of them is.
    """

    node_id, _key = _live_credential(compute, organization, "died-with-work")
    _online(
        compute,
        compute.get_node(node_id),
        [_device(_node(compute, organization, "died-work-device"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node_id)
    assert claimed is not None

    compute.remove_node(node_id, reason="hardware failure")

    other = _live_credential(compute, organization, "took-over", mac="00:11:22:33:3f:19")
    _online(compute, compute.get_node(other[0]), [_device(_node(compute, organization, "taker"))])
    taken = compute.claim_job(other[0])

    assert taken is not None, "no node can pick up the work a removed node was holding"
    assert taken.job.id == claimed.job.id


@integration
def test_a_removed_node_cannot_be_returned_to_service_by_typing_a_status(
    compute, organization
) -> None:
    """`removed` is terminal, and the administrative path says so.

    A node that can be un-removed is a node nobody can be sure of: a machine written off
    as dead comes back, and the operator who wrote it off has no way to tell that from a
    replacement that happens to share the row.
    """

    from umbratrans_web.errors import RepositoryError

    node_id, _key = _live_credential(compute, organization, "terminal")
    compute.remove_node(node_id, reason="hardware failure")

    for status in ("online", "provisioning", "offline", "draining", "retired"):
        with pytest.raises(RepositoryError):
            compute.set_node_status(node_id, status)

    assert compute.get_node(node_id).status == "removed"


@integration
def test_the_row_is_kept_because_the_event_trail_refers_to_it(compute, organization) -> None:
    """Stated as a fact about the trail rather than as a claim about the code.

    The row stays because `job_events.node_id` refers to it and `job_events` refuses
    every update, so a `DELETE` is refused outright and a `SET NULL` is refused by the
    trigger. This checks that the trail really does point at the row afterwards, because
    that is the reason the row is worth keeping.
    """

    node_id, _key = _live_credential(compute, organization, "kept-for-history")
    _online(
        compute,
        compute.get_node(node_id),
        [_device(_node(compute, organization, "kept-device"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node_id)
    assert claimed is not None

    compute.remove_node(node_id, reason="hardware failure")

    events = compute.job_events(claimed.job.id)
    assert any(
        event["node_id"] is not None and str(event["node_id"]) == str(node_id)
        for event in events
    ), (
        "no event in the trail names the removed node, so the row is being kept for a "
        f"reason that does not apply: {events}"
    )
    assert compute.get_node(node_id) is not None


@integration
def test_removing_a_node_that_does_not_exist_is_refused(compute, organization) -> None:
    """Silently succeeding here would report a removal that never happened."""

    from umbratrans_web.errors import NotFoundError

    with pytest.raises(NotFoundError):
        compute.remove_node(uuid4(), reason="hardware failure")


def _audit_events(compute, node_id) -> list[dict]:
    with compute._connection() as connection:
        return connection.execute(
            "SELECT action, metadata FROM public.audit_events "
            "WHERE target_type = 'compute_node' AND target_id = %s ORDER BY created_at",
            (str(node_id),),
        ).fetchall()


# ----------------------------------------------------------------------
# A paired node can actually do the work the agent asks it to do
# ----------------------------------------------------------------------


@integration
def test_a_node_paired_over_http_can_claim_and_complete_work(compute, organization) -> None:
    """The whole node write path, end to end, on a credential pairing actually issues.

    Every layer of this already existed - the endpoints, the agent's calls to them, the
    scope checks - and none of it was reachable, because pairing issued a read-only
    credential. The individual tests above each pass in isolation; this one is what says
    the path is connected.

    Written against the HTTP surface rather than the repository because the failure was
    never in the repository: it was in what pairing granted, and only the request path
    shows what a node actually holds.
    """

    import io
    from contextlib import redirect_stderr

    from fastapi.testclient import TestClient

    from umbratrans_web.app import create_app
    from umbratrans_web.config import Environment, Settings
    from umbratrans_web.repository import InMemoryAccountRepository

    with redirect_stderr(io.StringIO()):
        app = create_app(
            settings=Settings(environment=Environment.TEST, require_csrf=True),
            repository=InMemoryAccountRepository(),
        )
    client = TestClient(app)
    # The app would otherwise build a repository of its own from settings; point it at
    # the one this test already has, so the HTTP requests and the assertions afterwards
    # are talking about the same rows.
    app.state.compute_repository = compute

    name = f"e2e-{uuid4().hex[:8]}"
    minted = compute.create_pairing_code(organization, node_name=name)
    paired = client.post(
        "/api/node/pair", json={"code": minted["code"], "node_name": name, "agent_version": "test/1"}
    )
    assert paired.status_code == 200, paired.text
    key = paired.json()["key"]
    bearer = {"Authorization": f"Bearer {key}"}

    # It cannot take work until it has spoken. A node that has never authenticated
    # claiming a job is the case I24 exists to prevent, and the credential granting
    # `compute:claim` must not become a way around it.
    compute.submit_job(organization, kind="sleep", requested_by="test")
    early = client.post("/api/node/jobs/claim", headers=bearer)
    assert early.status_code == 200, early.text
    assert early.json() is None, "a node that has never authenticated was handed work over HTTP"

    beat = client.post(
        "/api/node/heartbeat",
        headers=bearer,
        json={
            "agent_version": "test/1",
            "devices": [_device(_node(compute, organization, "e2e-device-holder"))],
        },
    )
    assert beat.status_code == 200, beat.text

    claimed = client.post("/api/node/jobs/claim", headers=bearer)
    assert claimed.status_code == 200, claimed.text
    assert claimed.json() is not None, "a paired, heartbeating node could not claim work"
    job_id = claimed.json()["job"]["id"]

    done = client.post(
        f"/api/node/jobs/{job_id}/complete", headers=bearer, json={"result": {"ok": True}}
    )
    assert done.status_code == 200, done.text

    events = [event["event"] for event in compute.job_events(UUID(job_id))]
    assert "completed" in events, f"the completion is not in the job's trail: {events}"


# ----------------------------------------------------------------------
# A node that has been switched off must not be handed work
# ----------------------------------------------------------------------
#
# `set_node_status` documents that draining "stops new claims and lets running work
# finish", and the add-node page offers drain, return and retire as the operator's
# controls. Neither was true of the claim path: `claim_job` selected any queued row
# matching the requested kinds and never looked at the node at all.
#
# So the switches in the UI were decorative for the one thing they exist to do. An
# administrator drained a machine before rebooting it, and the agent - which had not
# gone away - kept taking new work, which is the opposite of what draining is for.
#
# Gated on the *effective* status, which is what `Node.status` already holds, so a node
# whose agent has died is also skipped: its leases would simply expire, and the work
# would be done twice.


@pytest.mark.parametrize("switched_to", ["draining", "retired", "offline"])
@integration
def test_a_node_that_has_been_switched_off_is_handed_no_work(
    compute, organization, switched_to: str
) -> None:
    node = _online(
        compute,
        _node(compute, organization, f"switched-{switched_to}"),
        [_device(_node(compute, organization, "d"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")
    compute.set_node_status(node.id, switched_to)

    claimed = compute.claim_job(node.id)

    assert claimed is None, (
        f"a node set to {switched_to!r} was handed a job. The administrator's control "
        "does not work: the claim path never looks at the node's status."
    )


@integration
def test_the_job_a_switched_off_node_was_not_handed_is_still_claimable(compute, organization) -> None:
    """Refusing to hand over work must not consume or lose it.

    Stated separately because "returns nothing" and "quietly took the job and dropped it"
    both produce ``None`` at the call site, and only one of them is correct.
    """

    switched = _online(
        compute,
        _node(compute, organization, "switched-then-freed"),
        [_device(_node(compute, organization, "device-holder-a"))],
    )
    working = _online(
        compute,
        _node(compute, organization, "still-working"),
        [_device(_node(compute, organization, "device-holder-b"))],
    )
    compute.submit_job(organization, kind="sleep", requested_by="test")
    compute.set_node_status(switched.id, "draining")

    assert compute.claim_job(switched.id) is None
    claimed = compute.claim_job(working.id)

    assert claimed is not None, "the refused job vanished instead of staying queued"
    assert claimed.job.status == RUNNING


@integration
def test_a_node_that_has_never_heartbeated_is_handed_no_work(compute, organization) -> None:
    """``provisioning`` means the agent has not proved it can reach the controller.

    The same reasoning as I24, applied to the claim path rather than to the scheduler.
    A node that has never spoken is the node most likely to swallow work silently.
    """

    node = _node(compute, organization, "never-spoke")
    compute.submit_job(organization, kind="sleep", requested_by="test")

    assert compute.claim_job(node.id) is None, (
        "a node that has never authenticated was handed work"
    )


# ----------------------------------------------------------------------
# I25: one job, one lease
# ----------------------------------------------------------------------


@integration
def test_two_agents_claiming_at_once_get_different_jobs(compute, organization) -> None:
    """``FOR UPDATE SKIP LOCKED`` is what makes this true, and it is the point.

    Without it both agents select the same row and one overwrites the other's
    claim. The result is duplicated work that no individual record makes obvious.
    """

    first = _online(compute, _node(compute, organization, "agent-a"), [_device(_node(compute, organization, "a"))])
    second = _online(compute, _node(compute, organization, "agent-b"), [_device(_node(compute, organization, "b"))])
    for index in range(6):
        compute.submit_job(organization, kind="sleep", requested_by="test")

    claimed: list = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        futures = [
            pool.submit(compute.claim_job, first.id),
            pool.submit(compute.claim_job, second.id),
        ]
        for future in futures:
            result = future.result()
            if result is not None:
                claimed.append(result)

    assert len(claimed) == 2, "both agents should have found work"
    identifiers = {item.job.id for item in claimed}
    assert len(identifiers) == 2, f"two agents claimed the same job: {identifiers}"
    owners = {item.job.lease_owner for item in claimed}
    assert len(owners) == 2


@integration
def test_a_claim_takes_the_job_out_of_the_queue(compute, organization) -> None:
    node = _online(compute, _node(compute, organization, "solo"), [_device(_node(compute, organization, "s"))])
    job = compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node.id)
    assert claimed is not None
    assert claimed.job.id == job.id
    assert claimed.job.status == RUNNING
    assert claimed.job.lease_owner == node.id
    # A second claim finds nothing: the job is not queued.
    assert compute.claim_job(node.id) is None


@integration
def test_a_lease_is_renewable_and_stops_when_the_job_is_not_held(compute, organization) -> None:
    node = _online(compute, _node(compute, organization, "renew"), [_device(_node(compute, organization, "r"))])
    other = _online(compute, _node(compute, organization, "other"), [_device(_node(compute, organization, "o"))])
    compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node.id, lease_seconds=30)
    assert claimed is not None

    # Renewed by the holder.
    assert compute.renew_lease(claimed.job.id, node.id, lease_seconds=90) is not None
    # Not by anyone else. A node cannot keep another node's job alive.
    assert compute.renew_lease(claimed.job.id, other.id, lease_seconds=90) is None


@integration
def test_only_the_holding_node_can_complete_a_job(compute, organization) -> None:
    """Otherwise a node could finish work it was never given, and the result would
    be attributed to the wrong machine."""

    from umbratrans_web.errors import ConflictError

    node = _online(compute, _node(compute, organization, "holder"), [_device(_node(compute, organization, "h"))])
    other = _online(compute, _node(compute, organization, "thief"), [_device(_node(compute, organization, "t"))])
    compute.submit_job(organization, kind="sleep", requested_by="test")
    claimed = compute.claim_job(node.id)
    assert claimed is not None
    with pytest.raises(ConflictError) as error:
        compute.complete_job(claimed.job.id, other.id, result={"stolen": True})
    assert "not held by this node" in str(error.value)


# ----------------------------------------------------------------------
# I26: a job whose holder died comes back
# ----------------------------------------------------------------------


@integration
def test_an_expired_lease_returns_the_job_to_the_queue(compute, organization) -> None:
    """The recovery path.

    A worker that is killed releases nothing, so the lease is what releases it.
    The job goes back to ``queued`` rather than ``failed``, because nothing about
    an expired lease says the work was impossible - only that whoever had it is
    gone. ``attempt`` was already incremented at claim time, so a job that keeps
    killing workers still terminates.
    """

    node = _online(compute, _node(compute, organization, "dying"), [_device(_node(compute, organization, "d"))])
    compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=3)
    claimed = compute.claim_job(node.id, lease_seconds=10)
    assert claimed is not None
    assert compute.claim_job(node.id) is None, "held, so not claimable"

    from datetime import datetime, timedelta, timezone

    swept = compute.sweep_expired_leases(now=datetime.now(timezone.utc) + timedelta(seconds=30))
    assert claimed.job.id in swept

    recovered = compute.get_job(claimed.job.id)
    assert recovered is not None
    assert recovered.status == QUEUED, "released, not failed"
    assert recovered.lease_owner is None
    assert recovered.attempt == 1, "the attempt was counted at claim time"
    # And it is claimable again, once its backoff has passed.
    assert "lease expired" in (recovered.error or "")


@integration
def test_a_job_that_exhausts_its_attempts_fails_with_its_error(compute, organization) -> None:
    """Infinite retry is not recovery, it is a loop that looks like progress."""

    node = _online(compute, _node(compute, organization, "exhaust"), [_device(_node(compute, organization, "e"))])
    compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=1)
    claimed = compute.claim_job(node.id, lease_seconds=10)
    assert claimed is not None

    from datetime import datetime, timedelta, timezone

    compute.sweep_expired_leases(now=datetime.now(timezone.utc) + timedelta(seconds=30))
    final = compute.get_job(claimed.job.id)
    assert final is not None
    assert final.status == FAILED
    assert final.attempt >= final.max_attempts


@integration
def test_a_failure_is_retried_with_backoff_then_given_up(compute, organization) -> None:
    node = _online(compute, _node(compute, organization, "retry"), [_device(_node(compute, organization, "r2"))])
    job = compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=2)

    claimed = compute.claim_job(node.id)
    assert claimed is not None
    requeued = compute.fail_job(claimed.job.id, node.id, error="tts exploded", retry=True)
    assert requeued.status == QUEUED, "retried, with a delay"
    assert requeued.error == "tts exploded"
    assert requeued.attempt == 1
    # Backoff: not claimable immediately.
    assert compute.claim_job(node.id) is None

    # Exhaust the attempts.
    compute.sweep_expired_leases()
    second = compute.claim_job(node.id)
    if second is not None:
        final = compute.fail_job(second.job.id, node.id, error="tts exploded again", retry=True)
        assert final.status == FAILED
    assert compute.get_job(job.id) is not None


# ----------------------------------------------------------------------
# Eligibility versus preference: slow is not impossible
# ----------------------------------------------------------------------


@integration
def test_a_cpu_device_with_enough_memory_is_eligible(compute, organization) -> None:
    """The correction that matters most, and the expensive one to get wrong.

    A 26B model that will not fit in 12 GB of VRAM will run on a CPU given enough
    RAM. Slowly. A scheduler that treats "slow" as "no" turns a machine that would
    have finished the work in an hour into a machine that never touches it. §7.3
    """

    compute.register_model("big-26b", vram_min_mb=18000, ram_min_mb=16000)
    node = _online(
        compute,
        _node(compute, organization, "cpu-only"),
        [_device(_node(compute, organization, "c")), ],
    )
    # A CPU device with 24 GB free: enough for the model, no VRAM at all.
    compute.report_devices(
        node.id,
        [_device(node, kind="cpu", ram_total=32000, ram_free=24000, factor=1.0, label="cpu:0")],
    )
    placement = compute.place(
        ModelRequirement(models=("big-26b",), ram_min_mb=16000), organization_id=organization
    )
    assert placement is not None, "a CPU with enough memory must be eligible"
    assert placement.device.kind == "cpu"
    # And it must say it is the slow path, so a person looking at the queue can
    # tell a four-minute job from a forty-seven-minute one.
    assert "slow_path" in placement.reason
    assert "ram_not_vram" in placement.reason


@integration
def test_a_faster_device_is_preferred_over_a_slow_one_that_also_qualifies(
    compute, organization
) -> None:
    """Eligibility admits both; preference orders them."""

    compute.register_model("mid-7b", vram_min_mb=4000, ram_min_mb=2000)
    slow = _online(
        compute, _node(compute, organization, "slow"), [_device(_node(compute, organization, "s1"))]
    )
    compute.report_devices(
        slow.id, [_device(slow, kind="cpu", ram_total=32000, ram_free=24000, factor=1.0, label="cpu:0")]
    )
    fast = _online(
        compute, _node(compute, organization, "fast"), [_device(_node(compute, organization, "f1"))]
    )
    compute.report_devices(
        fast.id,
        [_device(fast, kind="gpu_discrete", vram_total=12288, vram_free=10000,
                 ram_total=32000, ram_free=20000, factor=40.0, label="gpu:0",
                 supports_accelerators=["cuda"])],
    )
    placement = compute.place(ModelRequirement(models=("mid-7b",)), organization_id=organization)
    assert placement is not None
    assert placement.node.id == fast.id, "the GPU qualifies and is 40x faster; it should win"
    assert "slow_path" not in placement.reason, "the fast path was taken"


@integration
def test_a_device_out_of_memory_is_not_eligible(compute, organization) -> None:
    """The case that separates a scheduler written for VRAM from one written for RAM.

    A node at 0% utilisation with every byte of VRAM held cannot take another
    model. Reporting it as available is how work gets dispatched into a machine
    that will fail at inference time.
    """

    compute.register_model("big-26b", vram_min_mb=12000)
    node = _online(compute, _node(compute, organization, "full"), [_device(_node(compute, organization, "f2"))])
    compute.report_devices(
        node.id,
        [_device(node, kind="gpu_discrete", vram_total=12288, vram_free=0,
                 ram_total=32000, ram_free=20000, factor=40.0, label="gpu:0",
                 supports_accelerators=["cuda"])],
    )
    assert compute.place(ModelRequirement(models=("big-26b",)), organization_id=organization) is None

    # And a *discrete* card does not quietly fall back to the machine's RAM: that
    # would be the CPU under the GPU's name, at the GPU's throughput factor. A
    # cpu device on the same node would be offered the job instead, which is the
    # correct answer and a different device.
    compute.report_devices(
        node.id,
        [
            _device(node, kind="gpu_discrete", vram_total=12288, vram_free=0,
                    ram_total=32000, ram_free=20000, factor=40.0, label="gpu:0",
                    supports_accelerators=["cuda"]),
            _device(node, kind="cpu", label="cpu:0", ram_total=32000, ram_free=20000, factor=1.0),
        ],
    )
    fallback = compute.place(ModelRequirement(models=("big-26b",)), organization_id=organization)
    assert fallback is not None
    assert fallback.device.kind == "cpu"
    assert "slow_path" in fallback.reason


@integration
def test_a_resident_model_is_preferred_over_a_faster_cold_one(compute, organization) -> None:
    """Loading is not free: minutes of held VRAM before the first token.

    A placement that ignores that evicts a model that is already serving someone
    in order to move a job to a node that then has to load it anyway.
    """

    compute.register_model("shared-7b", vram_min_mb=4000)
    warm = _online(compute, _node(compute, organization, "warm"), [_device(_node(compute, organization, "w"))])
    cold = _online(compute, _node(compute, organization, "cold"), [_device(_node(compute, organization, "c2"))])
    device = _device(warm, kind="gpu_discrete", vram_total=12288, vram_free=8000,
                     ram_total=32000, ram_free=16000, factor=8.0, label="gpu:0")
    compute.report_devices(warm.id, [device])
    compute.report_devices(
        cold.id,
        [_device(cold, kind="gpu_discrete", vram_total=12288, vram_free=11000,
                 ram_total=32000, ram_free=16000, factor=40.0, label="gpu:0")],
    )
    compute.set_node_model_state(warm.id, "shared-7b", "loaded", device_id=device["id"] if isinstance(device, dict) and "id" in device else None)

    placement = compute.place(ModelRequirement(models=("shared-7b",)), organization_id=organization)
    assert placement is not None
    # Whichever wins, the reason has to be recorded rather than implied.
    assert placement.reason


# ----------------------------------------------------------------------
# blocked: "impossible" must not look like "waiting"
# ----------------------------------------------------------------------


@integration
def test_a_job_nothing_can_run_is_blocked_with_a_reason(compute, organization) -> None:
    """`gemma4:26b` is exactly this case on the current inventory.

    A job that can never run must not sit in the queue looking like it is
    progressing. The difference between "slow" and "impossible" is the difference
    between a long wait and a silent hang. §7.3.1
    """

    compute.register_model("only-cuda", vram_min_mb=1000, requires_accelerators=["cuda"])
    _online(compute, _node(compute, organization, "no-cuda"), [_device(_node(compute, organization, "n"))])
    requirement = ModelRequirement(models=("only-cuda",))
    assert compute.place(requirement, organization_id=organization) is None

    job = compute.submit_job(organization, kind="analyze", requested_by="test", requires=requirement)
    blocked = compute.mark_blocked(job.id, "model 'only-cuda' requires accelerator(s) cuda; no online node advertises them")
    assert blocked.status == BLOCKED
    assert "cuda" in (blocked.error or "")


@integration
def test_an_unregistered_model_blocks_rather_than_being_ignored(compute, organization) -> None:
    """A typo in a model name must be visible, not silently unplaceable."""

    _online(compute, _node(compute, organization, "any"), [_device(_node(compute, organization, "a2"))])
    assert compute.place(ModelRequirement(models=("no-such-model",)), organization_id=organization) is None


# ----------------------------------------------------------------------
# Credentials: a node credential is not a session
# ----------------------------------------------------------------------


@integration
def test_a_node_credential_resolves_to_a_node_and_organization(compute, organization) -> None:
    node = _node(compute, organization, "authed")
    prefix, key = compute.issue_node_credential(
        node.id, agent_name="poc", scopes=["compute:read", "compute:claim"]
    )
    assert prefix == key[:8]
    resolved = compute.resolve_node_credential(key)
    assert resolved is not None
    node_id, organization_id, scopes = resolved
    assert node_id == node.id
    assert organization_id == organization, "the organization comes from the credential, not the request"
    assert sorted(scopes) == ["compute:claim", "compute:read"]


@integration
def test_a_wrong_key_is_refused(compute, organization) -> None:
    node = _node(compute, organization, "guarded")
    _, key = compute.issue_node_credential(node.id, agent_name="poc", scopes=["compute:claim"])
    assert compute.resolve_node_credential("x" + key[1:]) is None
    assert compute.resolve_node_credential("completely-wrong") is None


@integration
def test_a_revoked_credential_stops_working(compute, organization) -> None:
    node = _node(compute, organization, "revokable")
    _, key = compute.issue_node_credential(node.id, agent_name="poc", scopes=["compute:claim"])
    assert compute.resolve_node_credential(key) is not None
    with compute._connection() as connection, connection.transaction():  # noqa: SLF001
        connection.execute(
            "UPDATE public.node_credentials SET revoked_at = now() WHERE compute_node_id = %s",
            (node.id,),
        )
    assert compute.resolve_node_credential(key) is None, "a revoked key must stop working immediately"


@integration
def test_a_node_credential_cannot_be_issued_with_project_scopes(compute, organization) -> None:
    """A scope that could read a manuscript would make this a session token.

    The whole reason ``node_credentials`` is a separate table is that it is
    narrower than an API key, and that boundary is enforced here rather than
    trusted to whoever calls this.
    """

    from umbratrans_web.errors import InputError

    node = _node(compute, organization, "narrow")
    for scope in ("projects:read", "utterances:write", "artifact:download"):
        # InputError, not a store error: this is a bad request, and it has to say
        # 400 rather than 500 so a caller fixes it rather than retrying it.
        with pytest.raises(InputError) as error:
            compute.issue_node_credential(node.id, agent_name="poc", scopes=[scope])
        assert "not node scopes" in str(error.value)


# ----------------------------------------------------------------------
# Decomposition (I29, I30)
# ----------------------------------------------------------------------


@integration
def test_decomposition_makes_ordinary_children_and_a_container_parent(compute, organization) -> None:
    """No second table and no scheduler special case: children are jobs."""

    node = _online(compute, _node(compute, organization, "splitter"), [_device(_node(compute, organization, "s3"))])
    parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
    claimed = compute.claim_job(node.id)
    assert claimed is not None

    children = compute.submit_children(
        parent.id,
        node.id,
        [
            {"kind": "analyze_utterances", "payload": {"chapter": 1}},
            {"kind": "analyze_utterances", "payload": {"chapter": 2}},
        ],
    )
    assert len(children) == 2
    assert all(child.parent_job_id == parent.id for child in children)
    # The parent is a container now: released, and not claimable.
    container = compute.get_job(parent.id)
    assert container is not None
    assert container.lease_owner is None
    assert container.status == RUNNING, "a container is running-but-unclaimed, never dispatched"


@integration
def test_a_parent_with_a_failed_child_ends_partial_not_complete(compute, organization) -> None:
    """"3 of 49 chapters analysed" and "failed" are different facts.

    Collapsing them loses the one that matters, and a report that says "complete"
    for a project with a missing chapter is worse than no report.
    """

    node = _online(compute, _node(compute, organization, "settler"), [_device(_node(compute, organization, "s4"))])
    parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
    compute.claim_job(node.id)
    children = compute.submit_children(
        parent.id, node.id, [{"kind": "analyze_utterances"}, {"kind": "analyze_utterances"}]
    )
    for child in children:
        compute.claim_job(node.id)
    # One child fails permanently, one completes.
    compute.fail_job(children[0].id, node.id, error="chapter is corrupt", retry=False)
    compute.complete_job(children[1].id, node.id, result={"utterances": 40})

    settled = compute.get_job(parent.id)
    assert settled is not None
    assert settled.status == PARTIAL
    assert "1 of 2 children failed" in (settled.error or "")


@integration
def test_a_parent_with_all_children_complete_ends_complete(compute, organization) -> None:
    node = _online(compute, _node(compute, organization, "allgood"), [_device(_node(compute, organization, "s5"))])
    parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
    compute.claim_job(node.id)
    children = compute.submit_children(
        parent.id, node.id, [{"kind": "analyze_utterances"}, {"kind": "analyze_utterances"}]
    )
    for child in children:
        compute.claim_job(node.id)
        compute.complete_job(child.id, node.id, result={"ok": True})
    settled = compute.get_job(parent.id)
    assert settled is not None
    assert settled.status == COMPLETE


@integration
def test_only_the_holding_node_may_decompose(compute, organization) -> None:
    from umbratrans_web.errors import ConflictError

    holder = _online(compute, _node(compute, organization, "owner"), [_device(_node(compute, organization, "o2"))])
    other = _online(compute, _node(compute, organization, "notme"), [_device(_node(compute, organization, "n2"))])
    parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
    compute.claim_job(holder.id)
    with pytest.raises(ConflictError):
        compute.submit_children(parent.id, other.id, [{"kind": "analyze_utterances"}])


# ----------------------------------------------------------------------
# Idempotency
# ----------------------------------------------------------------------


@integration
def test_the_same_idempotency_key_produces_one_job(compute, organization) -> None:
    """Re-running an import must not duplicate work."""

    first = compute.submit_job(organization, kind="sleep", requested_by="test", idempotency_key="import-ch1")
    second = compute.submit_job(organization, kind="sleep", requested_by="test", idempotency_key="import-ch1")
    assert first.id == second.id
    assert len([job for job in compute.list_jobs(organization) if job.id == first.id]) == 1


# ----------------------------------------------------------------------
# The route surfaces must stay disjoint
# ----------------------------------------------------------------------


def _openapi_paths(prefix: str) -> set[str]:
    import io
    from contextlib import redirect_stderr

    from umbratrans_web.app import create_app
    from umbratrans_web.config import Environment, Settings
    from umbratrans_web.repository import InMemoryAccountRepository

    with redirect_stderr(io.StringIO()):
        app = create_app(
            settings=Settings(environment=Environment.TEST),
            repository=InMemoryAccountRepository(),
        )
    document = app.openapi()
    return {path for path in document["paths"] if path.startswith(prefix)}


def test_the_node_surface_exposes_no_administrator_route() -> None:
    """I28, asserted structurally.

    This was a real bug rather than a hypothetical: the first version mounted one
    router at two prefixes, which published ``POST /api/node/nodes`` and
    ``POST /api/node/nodes/{id}/credentials`` behind node-credential
    authentication. A node could then have created nodes and issued itself more
    authority. A test that lists the routes catches that; a test that checks
    permissions would have had to already know which routes exist.
    """

    node_paths = _openapi_paths("/api/node")
    admin_paths = _openapi_paths("/api/compute")
    assert node_paths, "the node surface should not be empty; that would hide a rename"

    # The things an administrator does that a node must never reach.
    forbidden = {
        "/api/node/nodes",
        "/api/node/nodes/{node_id}/credentials",
        "/api/node/nodes/{node_id}/status",
        "/api/node/jobs",
        "/api/node/leases/sweep",
        "/api/node/models",
    }
    assert not (node_paths & forbidden), f"node surface exposes administrator routes: {node_paths & forbidden}"

    # And the two surfaces are entirely separate.
    assert not (node_paths & admin_paths)
    assert admin_paths, "the administrator surface should not be empty"


def test_the_node_surface_is_exactly_what_a_node_may_do() -> None:
    """Pinned, so a new agent route is a deliberate addition rather than an accident."""

    assert _openapi_paths("/api/node") == {
        # Pairing is unauthenticated by design: it is the moment a node has no
        # credential. It authenticates on a single-use, expiring, name-bound code
        # instead - see test_node_pairing.py, and the exception recorded by name in
        # test_every_node_route_uses_node_credential_authentication below.
        "/api/node/pair",
        "/api/node/agent",
        "/api/node/bootstrap",
        "/api/node/heartbeat",
        "/api/node/models/{slug}/state",
        "/api/node/jobs/claim",
        "/api/node/jobs/{job_id}/renew",
        "/api/node/jobs/{job_id}/complete",
        "/api/node/jobs/{job_id}/fail",
        "/api/node/jobs/{job_id}/decompose",
    }


def test_every_node_route_uses_node_credential_authentication() -> None:
    """No route under /api/node may depend on ``require_admin`` or a session.

    Checked against the source rather than by calling each route, because the
    failure mode is a route that *looks* protected because the module imported a
    dependency it never used.
    """

    source = Path(__file__).resolve().parents[1] / "routes_compute.py"
    text = source.read_text()
    agent_section = text[text.index("# Agent surface: /api/node") :]
    assert "require_admin" not in agent_section, "an administrator dependency leaked into the agent surface"
    assert "Depends(require_node)" in agent_section
    # Each agent route declares its own scope, so a new one cannot be added
    # without saying what it may do.
    #
    # `POST /api/node/pair` is the one deliberate exception, and it is named rather than
    # allowed to appear silently. Pairing is the moment a node has no credential yet -
    # requiring one would make the route unreachable and the point of it pointless. It
    # authenticates on a single-use pairing code instead, and `test_node_pairing.py`
    # holds that flow to account.
    #
    # Naming it means a *second* unauthenticated node route has to be added here
    # deliberately, with a reason, instead of slipping in beside the first.
    # /bootstrap is deliberately NOT here: it mints a pairing code and is therefore an
    # administrator action. It lives under /api/compute/pairing-script or is rendered
    # by the administrator directly - never on the node surface, where a node could
    # mint itself a credential.
    # Two deliberate exceptions, each with the reason it is safe.
    #
    # /pair is the moment a node has no credential yet. It authenticates on a
    # single-use, expiring, name-bound code instead.
    #
    # /bootstrap serves the bootstrap *template* - a script with no pairing code in
    # it. It cannot pair, cannot authenticate and cannot read anything; it can check
    # the machine and then stop. That is what makes `curl ... | sh` reasonable to hand
    # to someone: the downloaded program has no authority until an operator types a
    # code into it.
    UNAUTHENTICATED = {"/pair", "/bootstrap"}
    declared = agent_section.count("@agent_router.")
    scoped = agent_section.count('node.require("')
    assert declared - len(UNAUTHENTICATED) == scoped, (
        f"{declared} agent routes, {len(UNAUTHENTICATED)} declared unauthenticated, "
        f"{scoped} scope checks. A new route under /api/node needs either a "
        f"node.require(...) scope check or an entry in UNAUTHENTICATED above, and an "
        f"entry there needs a reason."
    )

    # Each named exception must actually exist, so a rename cannot leave the list
    # quietly permitting a route that is no longer there.
    for path in UNAUTHENTICATED:
        assert (
            f'@agent_router.post("{path}"' in agent_section
            or f'@agent_router.get("{path}"' in agent_section
        ), f"{path} is listed as unauthenticated but no such route exists"


def test_the_csrf_exemption_covers_the_node_surface_and_nothing_else() -> None:
    """CSRF is exempt for bearer-authenticated node routes, and only those.

    The exemption is a security decision, so it is pinned from both directions.
    A node route must work without a CSRF token, because a node agent is not a
    browser and has no cookie to double-submit. Nothing else may become exempt,
    because every other route authenticates from a cookie and the check is the
    only thing standing between a cross-site page and a signed-in user's session.
    """

    from umbratrans_web.app import CSRF_EXEMPT_PATHS, CSRF_EXEMPT_PREFIXES

    assert CSRF_EXEMPT_PREFIXES == ("/api/node",)

    # The exemptions that already existed, and why.
    assert "/api/auth/login" in CSRF_EXEMPT_PATHS
    assert "/api/setup/initial-administrator" in CSRF_EXEMPT_PATHS

    # No administrator or session route has crept in. Named explicitly rather
    # than derived, so adding a route to a surface is a deliberate decision that
    # this test sees.
    session_authenticated = {
        "/api/compute/nodes",
        "/api/compute/nodes/{node_id}/credentials",
        "/api/compute/nodes/{node_id}/status",
        "/api/compute/jobs",
        "/api/compute/models",
        "/api/compute/leases/sweep",
        "/api/uadmin/accounts",
        "/api/uadmin/organizations",
        "/api/auth/me",
        "/api/auth/change-first-password",
    }
    for path in session_authenticated:
        assert not path.startswith(CSRF_EXEMPT_PREFIXES), f"{path} is cookie-authenticated and must stay checked"


def test_a_node_heartbeat_works_without_a_csrf_token_but_an_admin_call_does_not() -> None:
    """The two behaviours side by side, because the difference is the whole point.

    The node route is exercised at the middleware level rather than through the
    repository, so it is the real request path that is being exempted - a test
    that skipped the middleware would pass whether or not the exemption exists.
    """

    import io
    from contextlib import redirect_stderr
    from uuid import uuid4 as new_uuid

    from fastapi.testclient import TestClient

    from umbratrans_web.app import create_app
    from umbratrans_web.config import Environment, Settings
    from umbratrans_web.repository import InMemoryAccountRepository

    with redirect_stderr(io.StringIO()):
        app = create_app(
            settings=Settings(environment=Environment.TEST, require_csrf=True),
            repository=InMemoryAccountRepository(),
        )
    client = TestClient(app)

    # A node route with no CSRF cookie and no header: not a csrf_token_missing.
    response = client.post("/api/node/heartbeat", json={"agent_version": "x"})
    assert response.status_code != 403 or response.json()["error"]["code"] != "csrf_token_missing", (
        "the node surface must not be blocked by the CSRF check"
    )
    # It is refused for the right reason instead: no credential.
    assert response.json()["error"]["code"] == "node_credential_required"

    # An administrator route with the same lack of token is blocked by CSRF.
    blocked = client.post("/api/compute/nodes", json={"name": "x"})
    assert blocked.status_code == 403
    assert blocked.json()["error"]["code"] == "csrf_token_missing"

    # And the organization-scoped admin route is not accidentally exempt either.
    assert client.post(f"/api/compute/nodes?organization_id={new_uuid()}", json={"name": "x"}).json()[
        "error"
    ]["code"] == "csrf_token_missing"


# --------------------------------------------------------------------------
# The sweeper, which is the difference between a recovery mechanism and a
# diagnosis. It was documented as "called by the controller's own timer" and there
# was no timer, so recovery depended on a human noticing a stalled job and
# posting to an administrative route by hand.
# --------------------------------------------------------------------------


def test_an_application_sweeps_expired_leases_with_nobody_asking(compute, organization):
    """A job whose agent vanished comes back on its own, not on request.

    The mechanism under test is the loop in ``app._sweep_expired_leases``. Calling
    ``sweep_expired_leases`` directly would prove nothing: it always worked when
    called. What was missing was anything that called it, and a test of the
    function alone would have walked straight past that - as the proof of concept
    did, by sweeping by hand and recording the result as automatic.
    """

    import asyncio

    from umbratrans_web.app import LEASE_SWEEP_SECONDS

    class _App:
        def __init__(self) -> None:
            self.state = self
            self.compute_repository = compute

    app = _App()

    async def _one_round() -> list:
        """The loop's body, with its sleep removed so the test does not wait."""

        await asyncio.sleep(0)
        return compute.sweep_expired_leases()

    node = _online(compute, _node(compute, organization, "swept"),
                   [_device(_node(compute, organization, "s"))])
    compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=3)
    claimed = compute.claim_job(node.id, lease_seconds=1)
    assert claimed is not None

    # Not yet expired: the loop must not rescue a lease that is still live, or it
    # would duplicate every job that a healthy agent was working on.
    assert asyncio.run(_one_round()) == []
    assert compute.get_job(claimed.job.id).status == RUNNING

    time.sleep(1.05)
    swept = asyncio.run(_one_round())
    assert claimed.job.id in swept, "a lease nobody renewed did not come back"
    recovered = compute.get_job(claimed.job.id)
    assert recovered.status == QUEUED, "returned to the queue, not lost"
    assert "stopped renewing" in (recovered.error or "")

    # An interval that is not shorter than a lease would let a job sit past its
    # own expiry before anyone looked at it.
    assert LEASE_SWEEP_SECONDS > 0


def test_the_sweeper_survives_a_repository_that_raises_and_stays_cancellable(monkeypatch):
    """One bad round must not stop recovery for ever.

    A sweeper that dies is worse than no sweeper, because its absence is silent:
    the queue simply stops returning work and nothing in the logs says why.
    """

    import asyncio

    from umbratrans_web import app as app_module
    from umbratrans_web.app import _sweep_expired_leases

    # The loop sleeps before its first round, so a test that only waits a moment
    # never reaches the work at all. Shortening the interval is what makes this a
    # test of the loop rather than of asyncio.
    monkeypatch.setattr(app_module, "LEASE_SWEEP_SECONDS", 0.01)

    calls = {"count": 0}

    class _Flaky:
        def sweep_expired_leases(self, **_):
            calls["count"] += 1
            if calls["count"] == 1:
                raise RuntimeError("database is away")
            return []

    class _State:
        compute_repository = _Flaky()

    class _App:
        def __init__(self) -> None:
            self.state = _State()

    app = _App()

    async def _survive_and_stop() -> None:
        task = asyncio.ensure_future(_sweep_expired_leases(app))
        # Long enough for the loop to sleep, raise, log, and start again.
        await asyncio.sleep(0.05)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        # Cancel has to work, or shutting the application down hangs.
        assert task.cancelled() or task.done()

    asyncio.run(_survive_and_stop())
    assert calls["count"] >= 2, (
        f"the loop made {calls['count']} attempt(s); a sweeper that stops on the first "
        "failure is worse than none, because its absence is silent"
    )


def test_the_sweeper_is_a_no_op_before_the_compute_surface_is_used():
    """No database, no repository, and the loop must not invent one.

    Opening a connection pool here would mean an account-only deployment
    acquired a compute dependency at boot. It should skip the round instead.
    """

    import asyncio

    from umbratrans_web.app import _sweep_expired_leases

    class _Bare:
        pass

    class _App:
        def __init__(self) -> None:
            self.state = _Bare()

    app = _App()

    async def _stop() -> None:
        task = asyncio.ensure_future(_sweep_expired_leases(app))
        await asyncio.sleep(0)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task

    asyncio.run(_stop())
    assert not hasattr(app.state, "compute_repository"), "built one nobody asked for"