feat: add durable indexing jobs

docs/ARCHITECTURE.md

@@ -4,9 +4,9 @@ How `recoll-modern` is built today. Update in the same commit as any structural
 
 ## Overview
 
-The Python service scaffold, SQLite schema, collection and membership repositories, and document
-discovery/identity modules are built. The API, job queue, and Recoll adapter remain intended structure.
-Recoll behaviour below was verified against a real container.
+The Python service scaffold, SQLite schema, collection and membership repositories, document
+discovery/identity modules, and durable job worker are built. The API and Recoll adapter remain
+intended structure. Recoll behaviour below was verified against a real container.
 
 The goal: a self-hosted service that wraps Recoll's indexing and full-text search behind (a) a modern
 web UI and (b) an MCP server, running in Docker, with several independent collections such as
@@ -24,6 +24,8 @@ web UI and (b) an MCP server, running in Docker, with several independent collec
 | `src/recoll_modern/storage/` | SQLite migrations, collection metadata and content-hash identity |
 | `src/recoll_modern/domain/membership.py` | Validates sources and applies direct additions and rule changes |
 | `src/recoll_modern/indexing/reconcile.py` | Refreshes persistent directory rules while retaining excluded or unreadable files as present |
+| `src/recoll_modern/indexing/worker.py` | Claims one durable job at a time and runs blocking handlers off the event loop |
+| `src/recoll_modern/indexing/handlers.py` | Scans membership sources before strict-mode mutations and records file errors |
 | `tests/` | Automated service tests |
 
 Discovery produces content hashes and root-relative paths without writing to the document volume.
@@ -80,7 +82,9 @@ is normalized: `direct_memberships` holds explicit files and snapshot expansions
 `membership_exclusions` holds per-file overrides against rule-derived membership. Effective membership
 is direct membership union non-excluded rule membership, so direct additions remain additive.
 
-`jobs` and `job_file_errors` make work and partial failures durable. `users` and hashed-token `sessions`
+`jobs` and `job_file_errors` make work and partial failures durable. At startup, the worker requeues
+well-formed interrupted jobs, clears errors from the prior attempt, and fails unreadable requests.
+`users` and hashed-token `sessions`
 provide authentication. `schema_migrations` is forward-only. The exact columns and constraints are the
 contract in `docs/specs/001-search-engine-api.md`.
 

docs/TODO.md

@@ -15,11 +15,12 @@ Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan -
   selectable search fields, subpath hosting and the packaging seam for deferred desktop spec 005.
 - Plan `docs/plans/001-search-engine-api.md`: **approved 2026-09-18.** Its listed Python runtime and
   development dependencies are approved. Execution follows the plan task by task.
-- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–5 add read-only discovery,
-  content-hash identity, additive membership and directory rules. Their checkboxes are current in the
-  plan. The next task is the durable job queue.
+- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–6 add read-only discovery,
+  content-hash identity, additive membership, directory rules and durable jobs. Their checkboxes are
+  current in the plan. The next task is membership-map export and the native helper.
 
-The next action is Task 6, "Add the durable single-worker job queue." Do not repeat Tasks 1–5.
+The next action is Task 7, "Export membership atomically and build the native helper." Do not repeat
+Tasks 1–6.
 
 Things a resuming agent will otherwise get wrong:
 

docs/plans/001-search-engine-api.md

@@ -478,7 +478,7 @@ git commit -m "feat: add membership provenance and rules"
 - Produces: `JobWorker.start()`, `JobWorker.stop()`, `JobWorker.run_once() -> bool`
 - Produces: `MembershipJobHandler.handle(job: Job) -> None`
 
-- [ ] **Step 1: Write failing transition, recovery, and policy tests**
+- [x] **Step 1: Write failing transition, recovery, and policy tests**
 
 ```python
 def test_recover_running_requeues_valid_jobs(repo: JobRepository) -> None:
@@ -498,17 +498,17 @@ def test_best_effort_applies_valid_files_and_records_errors(handler: MembershipJ
     assert jobs.current.state == JobState.COMPLETED_WITH_ERRORS
 ```
 
-- [ ] **Step 2: Run tests and observe failure**
+- [x] **Step 2: Run tests and observe failure**
 
 Run: `uv run pytest tests/storage/test_job_repository.py tests/indexing/test_worker.py tests/indexing/test_job_handlers.py -v`
 
-- [ ] **Step 3: Implement durable transitions and handlers**
+- [x] **Step 3: Implement durable transitions and handlers**
 
 Use opaque `j` plus 32 hex digits. `claim_next` runs under `BEGIN IMMEDIATE`, selects the oldest queued job, and changes it to running atomically. Valid transitions are queued→running, running→completed/completed_with_errors/failed, and failed→queued via retry. Startup requeues running jobs whose JSON decodes to the registered request type; malformed requests become failed with `invalid_job_request`.
 
 The worker owns one asyncio task, sleeps on an `asyncio.Event`, performs blocking discovery/database/engine work with `asyncio.to_thread`, and on shutdown stops claiming new work and leaves an interrupted running job recoverable. Membership handlers scan completely before opening the desired-state transaction in strict mode. Best-effort handlers persist every safe file error and continue.
 
-- [ ] **Step 4: Verify and commit**
+- [x] **Step 4: Verify and commit**
 
 Run: `uv run pytest tests/storage/test_job_repository.py tests/indexing/test_worker.py tests/indexing/test_job_handlers.py -v && uv run ruff check . && uv run mypy src tests`
 

src/recoll_modern/domain/jobs.py

@@ -0,0 +1,32 @@
+"""Durable indexing-job state shared by storage and workers."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class JobState(StrEnum):
+    QUEUED = "queued"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    COMPLETED_WITH_ERRORS = "completed_with_errors"
+    FAILED = "failed"
+
+
+@dataclass(frozen=True, slots=True)
+class Job:
+    id: str
+    kind: str
+    state: JobState
+    request: dict[str, object]
+    fail_on_file_error: bool
+    progress: int
+    total: int
+    error_code: str | None = None
+    error_message: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class JobFileError:
+    relative_path: str
+    code: str
+    message: str

src/recoll_modern/domain/membership.py

@@ -27,7 +27,16 @@ class MembershipService:
         self._memberships = memberships
 
     def plan_add(self, collection_id: str, sources: list[Source]) -> MembershipPlanResult:
-        for source in sources:
+        discovered = [(source, discover_sources(self._policy, [source])) for source in sources]
+        return self.apply_discovered(collection_id, discovered)
+
+    def apply_discovered(
+        self,
+        collection_id: str,
+        discovered: list[tuple[Source, DiscoveryReport]],
+    ) -> MembershipPlanResult:
+        """Apply already scanned sources so strict jobs never rescan after validation."""
+        for source, _ in discovered:
             if source.type == "directory":
                 parent = self._policy.normalize(source.path)
                 if any(
@@ -35,7 +44,6 @@ class MembershipService:
                     for path in source.excluded_directories
                 ):
                     raise ValueError("excluded directory must be a descendant")
-        discovered = [(source, discover_sources(self._policy, [source])) for source in sources]
         files = list({
             file.relative_path: file
             for _, report in discovered

src/recoll_modern/indexing/handlers.py

@@ -0,0 +1,71 @@
+"""Durable membership job application after complete source discovery."""
+
+from recoll_modern.domain.jobs import Job, JobFileError
+from recoll_modern.domain.membership import MembershipService
+from recoll_modern.indexing.discovery import Source, discover_sources
+from recoll_modern.indexing.paths import PathPolicy
+from recoll_modern.storage.documents import DocumentRepository
+from recoll_modern.storage.jobs import JobRepository
+from recoll_modern.storage.membership import MembershipRepository
+
+
+class MembershipJobHandler:
+    def __init__(
+        self,
+        policy: PathPolicy,
+        documents: DocumentRepository,
+        memberships: MembershipRepository,
+        jobs: JobRepository,
+    ) -> None:
+        self._policy = policy
+        self._service = MembershipService(policy, documents, memberships)
+        self._jobs = jobs
+
+    def handle(self, job: Job) -> None:
+        try:
+            collection_id = job.request["collection_id"]
+            source_values = job.request["sources"]
+            if not isinstance(collection_id, str) or not isinstance(source_values, list):
+                raise ValueError
+            sources = [_parse_source(value) for value in source_values]
+        except (KeyError, ValueError, TypeError):
+            self._jobs.fail(job.id, "invalid_job_request", "Job request is invalid")
+            return
+        discovered = [(source, discover_sources(self._policy, [source])) for source in sources]
+        errors = [error for _, report in discovered for error in report.errors]
+        for error in errors:
+            self._jobs.add_file_error(
+                job.id, JobFileError(error.relative_path, error.code, "File could not be read")
+            )
+        total = sum(len(report.files) + len(report.errors) for _, report in discovered)
+        self._jobs.progress(job.id, 0, total)
+        if errors and job.fail_on_file_error:
+            self._jobs.fail(job.id, "file_error", "One or more files could not be read")
+            return
+        try:
+            self._service.apply_discovered(collection_id, discovered)
+        except ValueError:
+            self._jobs.fail(job.id, "invalid_job_request", "Job request is invalid")
+            return
+        self._jobs.progress(job.id, total, total)
+        self._jobs.complete(job.id, with_errors=bool(errors))
+
+
+def _parse_source(value: object) -> Source:
+    if not isinstance(value, dict):
+        raise ValueError
+    kind = value.get("type")
+    path = value.get("path")
+    if not isinstance(path, str):
+        raise ValueError
+    if kind == "file":
+        return Source.file(path)
+    if kind == "directory":
+        mode = value.get("mode")
+        exclusions = value.get("excluded_directories", [])
+        if mode not in ("snapshot", "persistent") or not isinstance(exclusions, list):
+            raise ValueError
+        if not all(isinstance(item, str) for item in exclusions):
+            raise ValueError
+        return Source.directory(path, mode, tuple(exclusions))
+    raise ValueError

src/recoll_modern/indexing/worker.py

@@ -0,0 +1,68 @@
+"""Single asynchronous worker around blocking durable job handlers."""
+
+import asyncio
+from collections.abc import Callable, Mapping
+
+from recoll_modern.domain.jobs import Job, JobState
+from recoll_modern.storage.jobs import JobRepository
+
+
+class JobWorker:
+    def __init__(
+        self, jobs: JobRepository, handlers: Mapping[str, Callable[[Job], None]]
+    ) -> None:
+        self._jobs = jobs
+        self._handlers = handlers
+        self._wake = asyncio.Event()
+        self._stopping = False
+        self._task: asyncio.Task[None] | None = None
+
+    def start(self) -> None:
+        if self._task is not None:
+            return
+        self._jobs.recover_running()
+        self._stopping = False
+        self._task = asyncio.create_task(self._run())
+        self._wake.set()
+
+    async def stop(self) -> None:
+        self._stopping = True
+        self._wake.set()
+        if self._task is not None:
+            await self._task
+            self._task = None
+
+    def notify(self) -> None:
+        self._wake.set()
+
+    async def run_once(self) -> bool:
+        job = await asyncio.to_thread(self._jobs.claim_next)
+        if job is None:
+            return False
+        handler = self._handlers.get(job.kind)
+        if handler is None:
+            await asyncio.to_thread(
+                self._jobs.fail, job.id, "invalid_job_request", "Job kind is not supported"
+            )
+            return True
+        try:
+            await asyncio.to_thread(handler, job)
+        except Exception:
+            await asyncio.to_thread(
+                self._jobs.fail, job.id, "job_failed", "Job failed during processing"
+            )
+        else:
+            current = await asyncio.to_thread(self._jobs.get, job.id)
+            if current is not None and current.state is JobState.RUNNING:
+                await asyncio.to_thread(self._jobs.complete, job.id)
+        return True
+
+    async def _run(self) -> None:
+        while not self._stopping:
+            if await self.run_once():
+                continue
+            self._wake.clear()
+            try:
+                await asyncio.wait_for(self._wake.wait(), timeout=1.0)
+            except TimeoutError:
+                pass

src/recoll_modern/storage/jobs.py

@@ -0,0 +1,167 @@
+"""Atomic SQLite transitions for indexing jobs."""
+
+import json
+import secrets
+import sqlite3
+from collections.abc import Mapping
+from datetime import UTC, datetime
+
+from recoll_modern.domain.jobs import Job, JobFileError, JobState
+from recoll_modern.storage.database import Database
+
+
+class JobRepository:
+    def __init__(self, database: Database) -> None:
+        self._database = database
+
+    def enqueue(
+        self, kind: str, request: Mapping[str, object], fail_on_file_error: bool
+    ) -> Job:
+        serialized = json.dumps(dict(request), sort_keys=True)
+        now = datetime.now(UTC).isoformat()
+        while True:
+            job_id = f"j{secrets.token_hex(16)}"
+            try:
+                with self._database.transaction() as connection:
+                    connection.execute(
+                        "INSERT INTO jobs "
+                        "(id, kind, state, request_json, fail_on_file_error, created_at) "
+                        "VALUES (?, ?, 'queued', ?, ?, ?)",
+                        (job_id, kind, serialized, int(fail_on_file_error), now),
+                    )
+            except sqlite3.IntegrityError as error:
+                if "jobs.id" in str(error):
+                    continue
+                raise
+            job = self.get(job_id)
+            assert job is not None
+            return job
+
+    def get(self, job_id: str) -> Job | None:
+        with self._database.connect() as connection:
+            row = connection.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
+            return _job_from_row(row) if row is not None else None
+
+    def claim_next(self) -> Job | None:
+        with self._database.transaction() as connection:
+            row = connection.execute(
+                "SELECT * FROM jobs WHERE state = 'queued' ORDER BY created_at, rowid LIMIT 1"
+            ).fetchone()
+            if row is None:
+                return None
+            now = datetime.now(UTC).isoformat()
+            connection.execute(
+                "UPDATE jobs SET state = 'running', started_at = ?, finished_at = NULL "
+                "WHERE id = ? AND state = 'queued'",
+                (now, row["id"]),
+            )
+            updated = connection.execute(
+                "SELECT * FROM jobs WHERE id = ?", (row["id"],)
+            ).fetchone()
+            return _job_from_row(updated)
+
+    def progress(self, job_id: str, progress: int, total: int) -> None:
+        with self._database.transaction() as connection:
+            connection.execute(
+                "UPDATE jobs SET progress = ?, total = ? "
+                "WHERE id = ? AND state = 'running'",
+                (progress, total, job_id),
+            )
+
+    def add_file_error(self, job_id: str, error: JobFileError) -> None:
+        with self._database.transaction() as connection:
+            connection.execute(
+                "INSERT OR REPLACE INTO job_file_errors "
+                "(job_id, relative_path, code, message) VALUES (?, ?, ?, ?)",
+                (job_id, error.relative_path, error.code, error.message),
+            )
+
+    def file_errors(self, job_id: str) -> list[JobFileError]:
+        with self._database.connect() as connection:
+            return [
+                JobFileError(str(row[0]), str(row[1]), str(row[2]))
+                for row in connection.execute(
+                    "SELECT relative_path, code, message FROM job_file_errors "
+                    "WHERE job_id = ? ORDER BY relative_path",
+                    (job_id,),
+                )
+            ]
+
+    def complete(self, job_id: str, *, with_errors: bool = False) -> None:
+        state = "completed_with_errors" if with_errors else "completed"
+        with self._database.transaction() as connection:
+            connection.execute(
+                "UPDATE jobs SET state = ?, finished_at = ? "
+                "WHERE id = ? AND state = 'running'",
+                (state, datetime.now(UTC).isoformat(), job_id),
+            )
+
+    def fail(self, job_id: str, code: str, message: str) -> None:
+        with self._database.transaction() as connection:
+            connection.execute(
+                "UPDATE jobs SET state = 'failed', error_code = ?, error_message = ?, "
+                "finished_at = ? WHERE id = ? AND state = 'running'",
+                (code, message, datetime.now(UTC).isoformat(), job_id),
+            )
+
+    def recover_running(self) -> None:
+        with self._database.transaction() as connection:
+            for row in connection.execute(
+                "SELECT id, kind, request_json FROM jobs WHERE state = 'running'"
+            ).fetchall():
+                if _valid_request(str(row["kind"]), str(row["request_json"])):
+                    connection.execute(
+                        "DELETE FROM job_file_errors WHERE job_id = ?", (row["id"],)
+                    )
+                    connection.execute(
+                        "UPDATE jobs SET state = 'queued', started_at = NULL, "
+                        "progress = 0, total = 0 WHERE id = ?",
+                        (row["id"],),
+                    )
+                else:
+                    connection.execute(
+                        "UPDATE jobs SET state = 'failed', error_code = 'invalid_job_request', "
+                        "error_message = 'Job request is invalid', finished_at = ? WHERE id = ?",
+                        (datetime.now(UTC).isoformat(), row["id"]),
+                    )
+
+    def retry(self, job_id: str) -> None:
+        with self._database.transaction() as connection:
+            cursor = connection.execute(
+                "UPDATE jobs SET state = 'queued', error_code = NULL, error_message = NULL, "
+                "started_at = NULL, finished_at = NULL, progress = 0, total = 0 "
+                "WHERE id = ? AND state = 'failed'",
+                (job_id,),
+            )
+            if cursor.rowcount != 1:
+                raise ValueError("job is not retryable")
+            connection.execute("DELETE FROM job_file_errors WHERE job_id = ?", (job_id,))
+
+
+def _valid_request(kind: str, raw: str) -> bool:
+    try:
+        value = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return False
+    if not isinstance(value, dict):
+        return False
+    if kind == "membership_add":
+        return isinstance(value.get("collection_id"), str) and isinstance(
+            value.get("sources"), list
+        )
+    return bool(kind)
+
+
+def _job_from_row(row: sqlite3.Row) -> Job:
+    try:
+        value = json.loads(str(row["request_json"]))
+    except json.JSONDecodeError:
+        value = {}
+    request = value if isinstance(value, dict) else {}
+    return Job(
+        id=str(row["id"]), kind=str(row["kind"]), state=JobState(str(row["state"])),
+        request=request, fail_on_file_error=bool(row["fail_on_file_error"]),
+        progress=int(row["progress"]), total=int(row["total"]),
+        error_code=str(row["error_code"]) if row["error_code"] is not None else None,
+        error_message=str(row["error_message"]) if row["error_message"] is not None else None,
+    )

src/recoll_modern/storage/membership.py

@@ -138,6 +138,16 @@ class MembershipRepository:
                 (collection_id, content_hash),
             ).fetchone() is not None
 
+    def direct_documents(self, collection_id: str) -> set[str]:
+        with self._database.connect() as connection:
+            return {
+                str(row[0])
+                for row in connection.execute(
+                    "SELECT document_hash FROM direct_memberships WHERE collection_id = ?",
+                    (collection_id,),
+                )
+            }
+
     def exclude_document(self, collection_id: str, content_hash: str) -> None:
         with self._database.transaction() as connection:
             connection.execute(

tests/indexing/test_job_handlers.py

@@ -0,0 +1,58 @@
+from pathlib import Path
+
+from recoll_modern.domain.jobs import JobState
+from recoll_modern.domain.models import CollectionCreate
+from recoll_modern.indexing.handlers import MembershipJobHandler
+from recoll_modern.indexing.paths import PathPolicy
+from recoll_modern.storage.collections import SQLiteCollectionRepository
+from recoll_modern.storage.database import Database
+from recoll_modern.storage.documents import DocumentRepository
+from recoll_modern.storage.jobs import JobRepository
+from recoll_modern.storage.membership import MembershipRepository
+
+
+def _setup(
+    tmp_path: Path,
+) -> tuple[Path, JobRepository, MembershipRepository, MembershipJobHandler, str]:
+    root = tmp_path / "docs"
+    root.mkdir()
+    (root / "good.txt").write_text("good")
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    collection_id = SQLiteCollectionRepository(database).create(CollectionCreate("Books")).id
+    jobs = JobRepository(database)
+    membership = MembershipRepository(database)
+    handler = MembershipJobHandler(
+        PathPolicy(root), DocumentRepository(database), membership, jobs
+    )
+    return root, jobs, membership, handler, collection_id
+
+
+def _request(collection_id: str) -> dict[str, object]:
+    return {"collection_id": collection_id, "sources": [
+        {"type": "file", "path": "good.txt"},
+        {"type": "file", "path": "gone.txt"},
+    ]}
+
+
+def test_strict_discovery_error_applies_nothing(tmp_path: Path) -> None:
+    _, jobs, membership, handler, collection_id = _setup(tmp_path)
+    job = jobs.enqueue("membership_add", _request(collection_id), True)
+    claimed = jobs.claim_next()
+    assert claimed is not None
+    handler.handle(claimed)
+    assert jobs.get(job.id).state == JobState.FAILED  # type: ignore[union-attr]
+    assert membership.direct_documents(collection_id) == set()
+
+
+def test_best_effort_applies_valid_files_and_records_errors(tmp_path: Path) -> None:
+    _, jobs, membership, handler, collection_id = _setup(tmp_path)
+    job = jobs.enqueue("membership_add", _request(collection_id), False)
+    claimed = jobs.claim_next()
+    assert claimed is not None
+    handler.handle(claimed)
+    assert jobs.get(job.id).state == JobState.COMPLETED_WITH_ERRORS  # type: ignore[union-attr]
+    assert len(membership.direct_documents(collection_id)) == 1
+    assert [(error.relative_path, error.code) for error in jobs.file_errors(job.id)] == [
+        ("gone.txt", "file_vanished")
+    ]

tests/indexing/test_worker.py

@@ -0,0 +1,63 @@
+import asyncio
+from pathlib import Path
+
+import pytest
+
+from recoll_modern.domain.jobs import Job, JobState
+from recoll_modern.indexing.worker import JobWorker
+from recoll_modern.storage.database import Database
+from recoll_modern.storage.jobs import JobRepository
+
+
+@pytest.mark.asyncio
+async def test_worker_runs_one_job_and_records_completion(tmp_path: Path) -> None:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    jobs = JobRepository(database)
+    job = jobs.enqueue("test", {}, False)
+    handled: list[str] = []
+
+    def handle(claimed: Job) -> None:
+        handled.append(claimed.id)
+
+    worker = JobWorker(jobs, {"test": handle})
+    assert await worker.run_once()
+    assert handled == [job.id]
+    assert jobs.get(job.id).state == JobState.COMPLETED  # type: ignore[union-attr]
+
+
+@pytest.mark.asyncio
+async def test_worker_failure_is_durable(tmp_path: Path) -> None:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    jobs = JobRepository(database)
+    job = jobs.enqueue("test", {}, False)
+
+    def fail(_claimed: Job) -> None:
+        raise RuntimeError("private filesystem path")
+
+    worker = JobWorker(jobs, {"test": fail})
+    assert await worker.run_once()
+    failed = jobs.get(job.id)
+    assert failed is not None and failed.state == JobState.FAILED
+    assert "private" not in (failed.error_message or "")
+
+
+@pytest.mark.asyncio
+async def test_worker_start_recovers_and_drains_queue(tmp_path: Path) -> None:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    jobs = JobRepository(database)
+    job = jobs.enqueue("test", {}, False)
+    jobs.claim_next()
+    handled: list[str] = []
+    worker = JobWorker(jobs, {"test": lambda claimed: handled.append(claimed.id)})
+    worker.start()
+    for _ in range(100):
+        current = jobs.get(job.id)
+        if current is not None and current.state == JobState.COMPLETED:
+            break
+        await asyncio.sleep(0.01)
+    await worker.stop()
+    assert handled == [job.id]
+    assert jobs.get(job.id).state == JobState.COMPLETED  # type: ignore[union-attr]

tests/storage/test_job_repository.py

@@ -0,0 +1,70 @@
+from pathlib import Path
+
+from recoll_modern.domain.jobs import JobFileError, JobState
+from recoll_modern.storage.database import Database
+from recoll_modern.storage.jobs import JobRepository
+
+
+def _repo(tmp_path: Path) -> JobRepository:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    return JobRepository(database)
+
+
+def test_claim_next_is_durable_and_oldest_first(tmp_path: Path) -> None:
+    repo = _repo(tmp_path)
+    first = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    second = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    assert repo.claim_next().id == first.id  # type: ignore[union-attr]
+    assert repo.claim_next().id == second.id  # type: ignore[union-attr]
+    assert repo.claim_next() is None
+
+
+def test_recover_running_requeues_valid_jobs(tmp_path: Path) -> None:
+    repo = _repo(tmp_path)
+    job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    repo.claim_next()
+    repo.recover_running()
+    assert repo.get(job.id).state == JobState.QUEUED  # type: ignore[union-attr]
+
+
+def test_recover_running_fails_malformed_request(tmp_path: Path) -> None:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    repo = JobRepository(database)
+    job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    repo.claim_next()
+    with database.transaction() as connection:
+        connection.execute("UPDATE jobs SET request_json = ? WHERE id = ?", ("{", job.id))
+    repo.recover_running()
+    recovered = repo.get(job.id)
+    assert recovered is not None and recovered.state == JobState.FAILED
+    assert recovered.error_code == "invalid_job_request"
+
+
+def test_recover_running_fails_invalid_source_shape(tmp_path: Path) -> None:
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    repo = JobRepository(database)
+    job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": "not a list"}, False)
+    repo.claim_next()
+    repo.recover_running()
+    assert repo.get(job.id).state == JobState.FAILED  # type: ignore[union-attr]
+
+
+def test_recover_running_clears_errors_from_interrupted_attempt(tmp_path: Path) -> None:
+    repo = _repo(tmp_path)
+    job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    repo.claim_next()
+    repo.add_file_error(job.id, JobFileError("gone.txt", "file_vanished", "File vanished"))
+    repo.recover_running()
+    assert repo.file_errors(job.id) == []
+
+
+def test_failed_job_can_be_retried(tmp_path: Path) -> None:
+    repo = _repo(tmp_path)
+    job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False)
+    repo.claim_next()
+    repo.fail(job.id, "engine_unavailable", "Engine unavailable")
+    repo.retry(job.id)
+    assert repo.get(job.id).state == JobState.QUEUED  # type: ignore[union-attr]