"""Socle jobs (spec 2026-08-31 « socle jobs » §5) : claim, dédup, zombies, échecs.

Aucun type de job réel n'existe en T0 : ces tests enregistrent des types factices dans le
registre, exactement comme le fera un module métier. Le worker reste neutre.

`data_test` n'est pas vidée entre deux lancements de pytest : la fixture `_clean_test_jobs`
supprime les lignes des types factices avant chaque test (le worker prend le plus vieux job
`queued` *toutes organisations confondues* — un reliquat fausserait le test suivant).
"""

import asyncio
import uuid
from collections.abc import Mapping
from datetime import UTC, datetime, timedelta
from typing import Any

import pytest
import pytest_asyncio
from sqlalchemy import delete, select, update
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.core.errors import AppError
from app.db.engine import get_engine
from app.jobs import registry, service
from app.jobs import worker as worker_module
from app.jobs.models import Job
from app.jobs.registry import JobType
from app.jobs.worker import MAX_ATTEMPTS_ERROR, Worker, claim, finish

# --- types factices ------------------------------------------------------------------

STATE: dict[str, Any] = {}


class HardKill(BaseException):
    """Simule un kill -9 : le worker n'attrape que `Exception`, rien n'est finalisé."""


async def _ok(_session: AsyncSession, job: Job) -> Mapping[str, Any]:
    STATE.setdefault("ran", []).append(job.id)
    return {"ok": True}


async def _boom(_session: AsyncSession, job: Job) -> Mapping[str, Any]:
    STATE.setdefault("ran", []).append(job.id)
    raise RuntimeError("handler cassé")


async def _resume(session: AsyncSession, job: Job) -> Mapping[str, Any]:
    """Handler idempotent à checkpoint : reprend au curseur `progress.done`."""
    items: list[str] = job.payload["items"]
    start = (job.progress or {}).get("done", 0)
    for index in range(start, len(items)):
        STATE.setdefault("processed", []).append(items[index])
        if not await service.save_progress(session, job, {"done": index + 1, "total": len(items)}):
            # Usage de référence : dépossédé par le sweep, le handler s'arrête net.
            return {"dispossessed": True}
        if STATE.get("kill_at") == index + 1:
            STATE["kill_at"] = None
            raise HardKill
    return {"processed": len(items)}


async def _on_fail(_session: AsyncSession, job: Job) -> None:
    STATE.setdefault("failed", []).append((job.id, job.status, job.error))


TEST_TYPES = ("test_ok", "test_boom", "test_resume", "test_active", "test_unknown")

registry.register(JobType("test_ok", _ok, on_fail=_on_fail))
registry.register(JobType("test_boom", _boom, on_fail=_on_fail))
registry.register(JobType("test_resume", _resume, on_fail=_on_fail))
registry.register(JobType("test_active", _ok, dedup_scope="active", on_fail=_on_fail))


# --- outillage -----------------------------------------------------------------------


def new_org() -> str:
    return f"org_{uuid.uuid4().hex[:12]}"


def make_factory() -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(get_engine(), expire_on_commit=False)


@pytest.fixture
def session_factory() -> async_sessionmaker[AsyncSession]:
    return make_factory()


@pytest.fixture
def worker(session_factory: async_sessionmaker[AsyncSession]) -> Worker:
    return Worker(
        session_factory,
        poll_interval=0.01,
        sweep_interval=0.0,
        backoff_min=0.01,
        backoff_max=0.05,
    )


@pytest_asyncio.fixture(autouse=True)
async def _clean_test_jobs(session_factory: async_sessionmaker[AsyncSession]) -> None:
    async with session_factory() as session:
        await session.execute(delete(Job).where(Job.type.in_(TEST_TYPES)))
        await session.commit()
    STATE.clear()


async def enqueue(
    factory: async_sessionmaker[AsyncSession],
    org_id: str,
    type_: str,
    **kwargs: Any,
) -> Job:
    async with factory() as session:
        return await service.enqueue(session, org_id, type_, **kwargs)


async def reload(factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> Job:
    async with factory() as session:
        job = await session.scalar(select(Job).where(Job.id == job_id))
        assert job is not None
        return job


async def make_stale(
    factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, *, minutes: int
) -> None:
    """Vieillit `started_at` **et** le heartbeat : le sweep regarde le plus récent des deux."""
    stale = datetime.now(UTC) - timedelta(minutes=minutes)
    async with factory() as session:
        job = await session.scalar(select(Job).where(Job.id == job_id))
        assert job is not None
        values: dict[str, Any] = {"started_at": stale}
        if job.progress is not None:
            values["progress"] = {**job.progress, "updated_at": stale.isoformat()}
        await session.execute(update(Job).where(Job.id == job_id).values(**values))
        await session.commit()


async def age_started_at(
    factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, *, minutes: int
) -> None:
    """Vieillit `started_at` seul, en laissant le heartbeat tel quel."""
    stale = datetime.now(UTC) - timedelta(minutes=minutes)
    async with factory() as session:
        await session.execute(update(Job).where(Job.id == job_id).values(started_at=stale))
        await session.commit()


# --- claim : SKIP LOCKED -------------------------------------------------------------


async def test_two_workers_never_claim_the_same_job(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    ids = {(await enqueue(session_factory, org, "test_ok")).id for _ in range(2)}
    # Session A garde sa transaction ouverte : sa ligne est verrouillée quand B arrive.
    async with session_factory() as a, session_factory() as b:
        job_a = await claim(a)
        job_b = await claim(b)
        assert job_a is not None and job_b is not None
        assert {job_a.id, job_b.id} == ids  # deux jobs distincts, aucun pris deux fois
        await a.commit()
        await b.commit()


async def test_concurrent_claims_are_disjoint(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    for _ in range(4):
        await enqueue(session_factory, org, "test_ok")

    async def claim_one() -> uuid.UUID | None:
        async with session_factory() as session:
            job = await claim(session)
            await session.commit()
            return job.id if job else None

    claimed = await asyncio.gather(*(claim_one() for _ in range(4)))
    assert None not in claimed
    assert len(set(claimed)) == 4


async def test_claim_takes_the_oldest_first(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    first = await enqueue(session_factory, org, "test_ok")
    await enqueue(session_factory, org, "test_ok")
    async with session_factory() as session:
        job = await claim(session)
        await session.commit()
    assert job is not None and job.id == first.id
    assert job.status == "running" and job.attempts == 1 and job.started_at is not None


async def test_claim_returns_none_on_empty_queue(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    async with session_factory() as session:
        assert await claim(session) is None


# --- dédup ---------------------------------------------------------------------------


async def test_second_enqueue_returns_the_existing_job(
    session_factory: async_sessionmaker[AsyncSession],
) -> None:
    org = new_org()
    key = f"test_ok:{uuid.uuid4()}"
    first = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    second = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    assert second.id == first.id


async def test_queue_scope_releases_the_key_at_claim(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Anti lost-wakeup : un travail redevient enfilable dès que son run démarre."""
    org = new_org()
    key = f"test_ok:{uuid.uuid4()}"
    running = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    async with session_factory() as session:
        claimed = await claim(session)
        await session.commit()
    assert claimed is not None and claimed.id == running.id and claimed.dedup_key is None

    again = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    assert again.id != running.id  # ré-enfilé pendant le run : le réveil n'est pas perdu
    assert await worker.run_once() is True
    assert (await reload(session_factory, again.id)).status == "done"


async def test_active_scope_holds_the_key_until_done(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    key = f"test_active:{uuid.uuid4()}"
    first = await enqueue(session_factory, org, "test_active", dedup_key=key)
    async with session_factory() as session:
        claimed = await claim(session)
        await session.commit()
    assert claimed is not None and claimed.dedup_key == key  # tenue pendant le run

    assert (await enqueue(session_factory, org, "test_active", dedup_key=key)).id == first.id
    async with session_factory() as session:
        await finish(session, claimed.id, "done", attempts=claimed.attempts, report={"ok": True})
    assert (await reload(session_factory, first.id)).dedup_key is None
    assert (await enqueue(session_factory, org, "test_active", dedup_key=key)).id != first.id


async def test_enqueue_reinserts_when_key_vanishes_between_conflict_and_reread(
    session_factory: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
) -> None:
    """Course résiduelle (spec §1) : le gagnant a relâché sa clé, on ré-insère."""
    org = new_org()
    key = f"test_ok:{uuid.uuid4()}"
    first = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    original = service._find_by_dedup_key
    calls = {"n": 0}

    async def flaky(*args: Any, **kwargs: Any) -> Job | None:
        calls["n"] += 1
        if calls["n"] == 1:
            return None  # le gagnant a fini entre le conflit et la relecture
        # La clé est libérée pour de bon : la ré-insertion doit passer.
        async with session_factory() as session:
            await session.execute(update(Job).where(Job.id == first.id).values(dedup_key=None))
            await session.commit()
        return await original(*args, **kwargs)

    monkeypatch.setattr(service, "_find_by_dedup_key", flaky)
    second = await enqueue(session_factory, org, "test_ok", dedup_key=key)
    assert second.id != first.id
    assert calls["n"] == 2  # une relecture vide, une ré-insertion réussie


async def test_enqueue_gives_up_after_bounded_retries(
    session_factory: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
) -> None:
    org = new_org()
    key = f"test_ok:{uuid.uuid4()}"
    await enqueue(session_factory, org, "test_ok", dedup_key=key)

    async def never_found(*_args: Any, **_kwargs: Any) -> Job | None:
        return None

    monkeypatch.setattr(service, "_find_by_dedup_key", never_found)
    with pytest.raises(AppError) as exc:
        await enqueue(session_factory, org, "test_ok", dedup_key=key)
    assert exc.value.status == 409


async def test_enqueue_rejects_unknown_type(
    session_factory: async_sessionmaker[AsyncSession],
) -> None:
    with pytest.raises(ValueError, match="test_unknown"):
        await enqueue(session_factory, new_org(), "test_unknown")


# --- exécution et échecs -------------------------------------------------------------


async def test_run_once_marks_done_with_report(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    job = await enqueue(session_factory, new_org(), "test_ok", payload={"x": 1})
    assert await worker.run_once() is True
    row = await reload(session_factory, job.id)
    assert row.status == "done" and row.report == {"ok": True}
    assert row.finished_at is not None and row.error is None
    assert STATE["ran"] == [job.id]


async def test_run_once_on_empty_queue_does_nothing(worker: Worker) -> None:
    assert await worker.run_once() is False


async def test_handler_failure_is_definitive_and_calls_on_fail(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    broken = await enqueue(session_factory, org, "test_boom")
    healthy = await enqueue(session_factory, org, "test_ok")
    assert await worker.run_once() is True  # le worker survit à l'exception du handler
    row = await reload(session_factory, broken.id)
    assert row.status == "failed" and row.attempts == 1  # échec déterministe : pas de rejeu
    assert "handler cassé" in (row.error or "")
    assert STATE["failed"] == [(broken.id, "failed", row.error)]
    assert await worker.run_once() is True  # la file continue
    assert (await reload(session_factory, healthy.id)).status == "done"


async def test_unknown_type_fails_without_killing_the_worker(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    # Insertion directe : `enqueue` refuse un type inconnu — le cas réel est un job enfilé
    # par un déploiement plus récent que le worker.
    async with session_factory() as session:
        session.add(Job(org_id=new_org(), type="test_unknown", payload={}))
        await session.commit()
    assert await worker.run_once() is True
    async with session_factory() as session:
        row = await session.scalar(select(Job).where(Job.type == "test_unknown"))
    assert row is not None and row.status == "failed" and "inconnu" in (row.error or "")


# --- zombies -------------------------------------------------------------------------


async def test_fresh_heartbeat_protects_a_long_running_job(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Démarré il y a 30 min mais toujours vivant : c'est le heartbeat qui tranche."""
    job = await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        claimed = await claim(session)
        await session.commit()
        assert claimed is not None
        await service.save_progress(session, claimed, {"done": 1})
    await age_started_at(session_factory, job.id, minutes=30)  # heartbeat laissé frais
    assert await worker.sweep() == 0
    assert (await reload(session_factory, job.id)).status == "running"


@pytest.mark.parametrize(
    "progress",
    [
        {"updated_at": "2026-99-99T99:99:99"},  # date illisible
        {"updated_at": 42},  # pas une chaîne
        [1, 2],  # jsonb non-objet : `.get` lèverait un AttributeError
        "x",
        3,
    ],
    ids=["bad_date", "not_a_string", "array", "string", "number"],
)
async def test_unreadable_progress_never_breaks_the_sweep(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession], progress: Any
) -> None:
    """`progress` est du jsonb de forme libre : aucune valeur ne doit faire lever le sweep.

    Le verdict retombe sur `started_at` et le job est balayé normalement. Une exception
    ici gèlerait toute la file : le sweep est dans la boucle, qui ne meurt plus mais
    `continue` — process vivant, aucun job exécuté, à jamais.
    """
    job = await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        assert await claim(session) is not None
        await session.commit()
    async with session_factory() as session:
        await session.execute(update(Job).where(Job.id == job.id).values(progress=progress))
        await session.commit()
    await age_started_at(session_factory, job.id, minutes=30)
    assert await worker.sweep() == 1
    assert (await reload(session_factory, job.id)).status == "queued"


async def test_zombie_is_requeued_and_reclaimed_with_more_attempts(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    job = await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        assert await claim(session) is not None
        await session.commit()
    await make_stale(session_factory, job.id, minutes=30)
    assert await worker.sweep() == 1
    row = await reload(session_factory, job.id)
    assert row.status == "queued" and row.attempts == 1 and row.started_at is None
    assert await worker.run_once() is True
    row = await reload(session_factory, job.id)
    assert row.status == "done" and row.attempts == 2  # le claim ré-incrémente


async def test_zombie_at_max_attempts_fails_definitively_and_queue_continues(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    org = new_org()
    poison = await enqueue(session_factory, org, "test_ok", dedup_key=f"test_ok:{uuid.uuid4()}")
    healthy = await enqueue(session_factory, org, "test_ok")
    async with session_factory() as session:
        await session.execute(
            update(Job).where(Job.id == poison.id).values(status="running", attempts=3)
        )
        await session.commit()
    await make_stale(session_factory, poison.id, minutes=30)

    assert await worker.sweep() == 1
    row = await reload(session_factory, poison.id)
    assert row.status == "failed" and row.error == MAX_ATTEMPTS_ERROR
    assert row.finished_at is not None and row.dedup_key is None
    assert STATE["failed"] == [(poison.id, "failed", MAX_ATTEMPTS_ERROR)]
    # Un job empoisonné ne monopolise pas le worker : le suivant passe.
    assert await worker.run_once() is True
    assert (await reload(session_factory, healthy.id)).status == "done"


# --- fencing token : une finalisation périmée ne touche jamais la ligne ---------------


async def test_stale_finish_after_a_steal_is_a_noop(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Worker A vivant mais silencieux : le sweep lui vole son job, B le reprend.

    La finalisation tardive de A ne doit ni écraser le run de B, ni relâcher la
    `dedup_key` d'un type `'active'` — sinon deux instances actives deviennent possibles.
    """
    org = new_org()
    key = f"test_active:{uuid.uuid4()}"
    job = await enqueue(session_factory, org, "test_active", dedup_key=key)
    async with session_factory() as session:
        stolen = await claim(session)  # worker A
        await session.commit()
    assert stolen is not None and stolen.attempts == 1

    await make_stale(session_factory, job.id, minutes=30)
    assert await worker.sweep() == 1  # le sweep croit A mort
    async with session_factory() as session:
        retaken = await claim(session)  # worker B reprend
        await session.commit()
    assert retaken is not None and retaken.attempts == 2

    # A, toujours vivant, finalise avec le jeton du claim périmé : no-op.
    async with session_factory() as session:
        assert await finish(session, job.id, "done", attempts=1, report={"ok": True}) is None

    row = await reload(session_factory, job.id)
    assert row.status == "running" and row.attempts == 2  # le run de B est intact
    assert row.finished_at is None and row.report is None
    assert row.dedup_key == key  # la garantie « une seule instance active » tient


async def test_stale_save_progress_after_a_steal_is_a_noop(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Le jeton vaut aussi pour le curseur : A dépossédé n'écrit plus, heartbeat compris.

    Sans cette garde, A écrase le curseur de reprise de B *et* rafraîchit le heartbeat :
    si B meurt vraiment, le sweep croit le job vivant tant que A écrit — bloqué en
    `running` pour toujours, `dedup_key` `'active'` retenue avec lui.
    """
    org = new_org()
    key = f"test_active:{uuid.uuid4()}"
    job = await enqueue(session_factory, org, "test_active", dedup_key=key)
    async with session_factory() as session_a:
        stolen = await claim(session_a)  # worker A
        await session_a.commit()
        assert stolen is not None and stolen.attempts == 1
        assert await service.save_progress(session_a, stolen, {"done": 2}) is True

        await make_stale(session_factory, job.id, minutes=30)
        assert await worker.sweep() == 1  # le sweep vole le job à A
        async with session_factory() as session_b:
            retaken = await claim(session_b)  # worker B reprend
            await session_b.commit()
            assert retaken is not None and retaken.attempts == 2
            assert await service.save_progress(session_b, retaken, {"done": 500}) is True
        beat_of_b = (await reload(session_factory, job.id)).progress["updated_at"]

        # A, toujours vivant, tente d'avancer son curseur : dépossédé, il n'écrit rien.
        assert await service.save_progress(session_a, stolen, {"done": 3}) is False

    row = await reload(session_factory, job.id)
    assert row.progress["done"] == 500  # le curseur de reprise de B est intact
    assert row.progress["updated_at"] == beat_of_b  # A ne prolonge pas la vie du job
    assert row.dedup_key == key


async def test_dispossessed_lot_is_rolled_back_not_committed(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Le lot en vol d'un handler dépossédé est **jeté**, pas publié.

    `save_progress` est le seul point où le socle peut encore annuler : elle doit donc
    décider avant de committer. Sans quoi un auteur de handler qui lit « son travail ne
    sera pas enregistré » pourrait se permettre un effet de bord non idempotent dans ce
    lot — et le voir publié quand même.
    """
    org = new_org()
    job = await enqueue(session_factory, org, "test_ok")
    witness = await enqueue(session_factory, org, "test_ok")  # « travail métier » de A
    async with session_factory() as session_a:
        stolen = await claim(session_a)  # worker A prend le plus vieux
        await session_a.commit()
        assert stolen is not None and stolen.id == job.id

        await make_stale(session_factory, job.id, minutes=30)
        assert await worker.sweep() == 1
        async with session_factory() as session_b:
            retaken = await claim(session_b)  # worker B reprend le même job
            await session_b.commit()
        assert retaken is not None and retaken.id == job.id and retaken.attempts == 2

        # A écrit son lot métier, pas encore committé, puis avance son curseur.
        await session_a.execute(
            update(Job).where(Job.id == witness.id).values(error="témoin du lot de A")
        )
        assert await service.save_progress(session_a, stolen, {"done": 3}) is False

    assert (await reload(session_factory, witness.id)).error is None  # le lot a été jeté


async def test_save_progress_updates_the_instance_without_reopening_a_transaction(
    session_factory: async_sessionmaker[AsyncSession],
) -> None:
    """Contrat de `save_progress` : valeur à jour en mémoire, et aucune I/O après le commit.

    Une relecture rouvrirait une transaction qui resterait « idle in transaction » pendant
    tout le lot suivant du handler, à tenir le xmin du vacuum.
    """
    await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        claimed = await claim(session)
        await session.commit()
        assert claimed is not None
        assert await service.save_progress(session, claimed, {"done": 7}) is True
        assert claimed.progress["done"] == 7  # reposé sur l'instance, sans relecture
        assert session.in_transaction() is False  # rien n'a rouvert de transaction


async def test_finish_is_a_noop_on_an_already_terminal_job(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    job = await enqueue(session_factory, new_org(), "test_ok")
    assert await worker.run_once() is True
    async with session_factory() as session:
        assert await finish(session, job.id, "failed", attempts=1, error="trop tard") is None
    row = await reload(session_factory, job.id)
    assert row.status == "done" and row.error is None


# --- résilience de la boucle ---------------------------------------------------------


async def test_startup_sweep_recovers_zombies_of_the_previous_process(
    session_factory: async_sessionmaker[AsyncSession],
) -> None:
    """Le balayage de démarrage ne dépend pas de l'uptime de la machine.

    `sweep_interval` est laissé à sa valeur de production : seul le balayage *de démarrage*
    peut récupérer le zombie dans la fenêtre du test. Un worker lancé par systemd au boot
    doit le faire, alors que `time.monotonic()` y vaut quelques secondes.
    """
    job = await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        assert await claim(session) is not None  # le process précédent l'avait pris
        await session.commit()
    await make_stale(session_factory, job.id, minutes=30)  # puis il est mort dessus

    worker = Worker(session_factory, poll_interval=0.01, sweep_interval=60.0)
    task = asyncio.create_task(worker.run_forever())
    try:
        for _ in range(100):
            await asyncio.sleep(0.02)
            if (await reload(session_factory, job.id)).status == "done":
                break
    finally:
        worker.request_stop()
        await asyncio.wait_for(task, timeout=5)
    row = await reload(session_factory, job.id)
    assert row.status == "done" and row.attempts == 2


async def test_a_failing_sweep_never_starves_job_execution(
    worker: Worker,
    session_factory: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`run_once` passe avant `sweep` : un balayage cassé ne gèle pas la file.

    C'est la moitié « ordonnancement » de la protection contre le gel silencieux — sans
    elle, le `continue` du backoff empêcherait toute exécution de job, process vivant.
    """
    job = await enqueue(session_factory, new_org(), "test_ok")

    async def broken_sweep() -> int:
        raise RuntimeError("balayage cassé")

    monkeypatch.setattr(worker, "sweep", broken_sweep)
    task = asyncio.create_task(worker.run_forever())
    try:
        for _ in range(100):
            await asyncio.sleep(0.02)
            if (await reload(session_factory, job.id)).status == "done":
                break
    finally:
        worker.request_stop()
        await asyncio.wait_for(task, timeout=5)
    assert (await reload(session_factory, job.id)).status == "done"


async def test_sweep_re_evaluates_its_verdict_on_the_locked_row(
    worker: Worker,
    session_factory: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Un job qui reprend vie entre les deux lectures du sweep n'est pas balayé.

    La phase 1 est sans verrou : le verdict qu'elle rend est une présomption. Seule la
    phase 2, sur la ligne verrouillée et relue, décide.
    """
    job = await enqueue(session_factory, new_org(), "test_ok")
    async with session_factory() as session:
        assert await claim(session) is not None
        await session.commit()
    await age_started_at(session_factory, job.id, minutes=30)
    calls = {"n": 0}

    def blinking(_job: Job) -> datetime:
        calls["n"] += 1
        # Périmé pour la phase 1, frais pour la phase 2 : le worker a re-signalé sa vie.
        return datetime.now(UTC) - (timedelta(minutes=30) if calls["n"] == 1 else timedelta(0))

    monkeypatch.setattr(worker_module, "_heartbeat", blinking)
    assert await worker.sweep() == 0
    assert calls["n"] >= 2  # les deux phases ont bien tranché chacune leur tour
    assert (await reload(session_factory, job.id)).status == "running"


async def test_run_forever_survives_a_transient_infra_error(
    worker: Worker,
    session_factory: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Postgres qui redémarre ne doit pas arrêter la file en silence."""
    job = await enqueue(session_factory, new_org(), "test_ok")
    original = worker_module.claim
    calls = {"n": 0}

    async def flaky(session: AsyncSession) -> Job | None:
        calls["n"] += 1
        if calls["n"] == 1:
            raise OperationalError("SELECT 1", {}, Exception("connexion perdue"))
        return await original(session)

    monkeypatch.setattr(worker_module, "claim", flaky)
    task = asyncio.create_task(worker.run_forever())
    try:
        for _ in range(100):
            await asyncio.sleep(0.02)
            if (await reload(session_factory, job.id)).status == "done":
                break
    finally:
        worker.request_stop()
        await asyncio.wait_for(task, timeout=5)
    assert calls["n"] >= 2  # la boucle a bien retenté après la panne
    assert (await reload(session_factory, job.id)).status == "done"


# --- reprise à mi-course -------------------------------------------------------------


async def test_resume_midway_without_duplicate(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    """Worker tué à mi-course puis relancé : le checkpoint évite tout doublon."""
    items = ["a", "b", "c", "d", "e"]
    job = await enqueue(session_factory, new_org(), "test_resume", payload={"items": items})
    STATE["kill_at"] = 2
    with pytest.raises(HardKill):
        await worker.run_once()
    row = await reload(session_factory, job.id)
    assert row.status == "running" and row.progress["done"] == 2  # rien n'a été finalisé
    assert STATE["processed"] == ["a", "b"]

    assert await worker.sweep() == 0  # heartbeat encore frais
    await make_stale(session_factory, job.id, minutes=30)
    assert await worker.sweep() == 1
    assert (await reload(session_factory, job.id)).status == "queued"

    assert await worker.run_once() is True
    row = await reload(session_factory, job.id)
    assert row.status == "done" and row.report == {"processed": 5} and row.attempts == 2
    assert STATE["processed"] == items  # zéro doublon : c, d, e seulement


# --- arrêt propre --------------------------------------------------------------------


async def test_run_forever_stops_on_request(
    worker: Worker, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    job = await enqueue(session_factory, new_org(), "test_ok")
    task = asyncio.create_task(worker.run_forever())
    await asyncio.sleep(0.2)
    worker.request_stop()
    await asyncio.wait_for(task, timeout=5)
    assert (await reload(session_factory, job.id)).status == "done"