fix: make indexing review paths replay-safe

docs/ARCHITECTURE.md

@@ -83,7 +83,9 @@ Only `engine/` may import Recoll or invoke `recollindex`. Membership jobs first
 state in SQLite, atomically export the helper's `RCM1` flat map, then ask the engine to force-index
 affected canonical paths. Map records use length-prefixed absolute UTF-8 paths, so filenames can
 contain newlines. The C helper reads this private map without starting Python. Engine retries are
-therefore idempotent. Persistent directory rules reconcile at startup and on an explicit reindex,
+therefore idempotent. Jobs save paths needed for replay before changing rule membership, and collection
+deletion commits the cleanup job in the same SQLite transaction. Persistent directory rules reconcile
+at startup and on an explicit reindex, preserving prior membership for unreadable files or subtrees,
 not through live filesystem monitoring.
 
 ### Storage model
@@ -114,8 +116,9 @@ derives effective membership from SQLite and includes direct and rule provenance
 
 Search passes all requested collection IDs into one engine query, then verifies hit paths and effective
 membership in SQLite before returning them. Full-text and original-file routes resolve a live canonical
-path under the configured root. Original responses use range-capable `FileResponse`, with inline display
-limited to safe media types.
+path under the configured root. Original downloads open a contained regular file through no-follow
+directory descriptors and stream from the pinned descriptor, including single byte ranges; inline
+display is limited to safe media types.
 
 The image installs Recoll and its Python binding from Debian trixie, and syncs the application into a
 system-site-packages venv so the binding remains importable. A separate build stage compiles the C

docs/DECISIONS.md

@@ -3,6 +3,22 @@
 Append-only, newest first. Never edit or delete an entry — supersede it with a new one that links back.
 One entry per decision a future maintainer would otherwise have to reverse-engineer.
 
+## 2026-09-27 — Persist replay paths and pin original downloads
+**Status:** accepted
+**Context:** Review found that a job could mutate rule or collection state before its indexing step
+failed, leaving a retry without the old paths. `FileResponse` also reopened a path after containment
+checking, allowing a concurrent host-side symlink swap.
+**Decision:** persist affected paths in a running job before rule mutations; enqueue collection cleanup
+atomically with collection deletion. Open original files beneath the document root with no-follow
+directory descriptors and stream from that checked descriptor, including byte-range responses.
+**Alternatives:** recompute old paths on retry (the database may already have lost them); retain
+`FileResponse` after a path-only check (it reopens the path).
+**Consequences:** interrupted jobs retain their required reindex paths. File responses are tied to an
+opened regular file even if directory entries change later. The descriptor traversal relies on Linux
+`openat` semantics supplied by the current Docker deployment; desktop packaging must revisit it.
+
+---
+
 ## 2026-09-18 — Cross-platform desktop packaging is spec 005
 **Status:** accepted
 **Supersedes:** "The product is four specs, built in order" (below, same date), only by appending a fifth

docs/TODO.md

@@ -5,7 +5,7 @@ Small, specific, independently actionable. Remove only when actually done.
 
 ## Resume here
 
-Last touched 2026-09-26. Spec 001 implementation is in progress.
+Last touched 2026-09-27. Spec 001 implementation is in review closure.
 
 Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan -> TDD -> review):
 
@@ -19,9 +19,19 @@ Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan -
   identity, additive membership, durable jobs, the atomic membership map, the engine boundary,
   session authentication, the HTTP shell, membership/rule/job routes, scoped search and document
   retrieval, Docker packaging and acceptance coverage. Their checkboxes are current in the plan.
-  Final code review and branch handoff remain.
+  The first code review found replay, reconciliation, and download races. A fix checkpoint is on
+  `feat/001-document-discovery`; final verification and branch handoff remain.
 
-The next action is Task 14 Step 6, "Request review before integration." Do not repeat Tasks 1–14.
+The next action is Task 14 Step 6: run the full verification matrix, request a follow-up review of the
+fixes, then hand off the branch. Do not repeat Tasks 1–14. The review fixes include idempotent rule
+creation, persisted replay paths, atomic collection cleanup enqueue, unreadable-file preservation,
+departed-path reindexing, directory-rule pagination, recovery request validation, and descriptor-pinned
+downloads. The generated real-container lifecycle test covers these flows through Compose. Its 9-test
+contract/integration suite passed with `RECOLL_MODERN_REQUIRE_DOCKER=1` on 2026-09-27; the host suite
+passed with 157 passed and 4 skipped, and lint/type checking passed. The complete final matrix and
+follow-up review still need to run before completion. Docker Compose may be running on
+port 9124 with `RECOLL_MODERN_DOCUMENTS=/tmp/recoll-modern-corpus`; preserve its named volumes and
+document data. Use `docker compose down` without `-v` when done.
 
 Things a resuming agent will otherwise get wrong:
 
@@ -70,10 +80,6 @@ Things a resuming agent will otherwise get wrong:
       paths exist, and fail if `AGENTS.md` exceeds 150 lines.
 
 ## Known issues
-- Original file download checks containment before returning `FileResponse`; harden against a local
-  symlink swap between that check and the response's file open before shipping to untrusted local
-  writers. The Compose document mount is read-only to the service, but host writers may still replace
-  entries concurrently.
 - The API test suite reports deprecation warnings from the installed FastAPI/Starlette TestClient
   dependency. Replace its client integration when the dependency stack supports the successor API.
 - Directory sources currently require a named child of the document root; decide whether the root

src/recoll_modern/api/collections.py

@@ -78,9 +78,8 @@ def delete_collection(
     collection = request.app.state.collections.get(collection_id)
     if confirmation != collection.name:
         raise AppError("confirmation_required", "Collection name confirmation is required", 409)
-    paths = request.app.state.collections.delete(collection_id)
-    job = request.app.state.jobs.enqueue(
-        "collection_cleanup", {"version": 1, "paths": paths}, False
+    job = request.app.state.collection_repository.delete_with_cleanup(
+        collection_id, request.app.state.jobs
     )
     request.app.state.worker.notify()
     return {"job_id": job.id}

src/recoll_modern/api/documents.py

@@ -1,11 +1,16 @@
 """Complete extracted text and safe original-file responses."""
 
 import mimetypes
+import os
+import re
+import stat
+from collections.abc import Iterator
 from pathlib import Path
-from typing import cast
+from typing import BinaryIO, cast
+from urllib.parse import quote
 
 from fastapi import APIRouter, Request
-from fastapi.responses import FileResponse
+from fastapi.responses import Response, StreamingResponse
 
 from recoll_modern.api.dependencies import require_user
 from recoll_modern.domain.errors import AppError
@@ -16,6 +21,7 @@ SAFE_INLINE_TYPES = frozenset({
     "text/plain", "application/pdf", "image/png", "image/jpeg", "image/gif",
     "audio/mpeg", "audio/ogg", "video/mp4",
 })
+_RANGE = re.compile(r"bytes=(\d*)-(\d*)\Z")
 
 
 def _path(request: Request, document_hash: str) -> Path:
@@ -31,6 +37,67 @@ def _path(request: Request, document_hash: str) -> Path:
     return cast(Path, path)
 
 
+def open_contained_file(root: Path, relative_path: str) -> BinaryIO:
+    """Open beneath root before constructing a response; pin the checked inode."""
+    root = root.resolve()
+    resolved = (root / relative_path).resolve(strict=True)
+    if not resolved.is_relative_to(root):
+        raise ValueError("file is outside document root")
+    parts = resolved.relative_to(root).parts
+    if not parts:
+        raise ValueError("document path is not a file")
+    directory_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        for part in parts[:-1]:
+            next_fd = os.open(
+                part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd
+            )
+            os.close(directory_fd)
+            directory_fd = next_fd
+        file_fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd)
+        if not stat.S_ISREG(os.fstat(file_fd).st_mode):
+            os.close(file_fd)
+            raise ValueError("document path is not a regular file")
+        return os.fdopen(file_fd, "rb")
+    finally:
+        os.close(directory_fd)
+
+
+def _byte_range(header: str | None, size: int) -> tuple[int, int, bool] | None:
+    if header is None:
+        return (0, size - 1, False)
+    match = _RANGE.fullmatch(header.strip())
+    if match is None or size == 0:
+        return None
+    start_text, end_text = match.groups()
+    if not start_text and not end_text:
+        return None
+    if not start_text:
+        suffix = int(end_text)
+        if suffix <= 0:
+            return None
+        return (max(size - suffix, 0), size - 1, True)
+    start = int(start_text)
+    end = min(int(end_text), size - 1) if end_text else size - 1
+    if start >= size or end < start:
+        return None
+    return (start, end, True)
+
+
+def _chunks(stream: BinaryIO, start: int, length: int) -> Iterator[bytes]:
+    try:
+        stream.seek(start)
+        remaining = length
+        while remaining:
+            data = stream.read(min(64 * 1024, remaining))
+            if not data:
+                break
+            remaining -= len(data)
+            yield data
+    finally:
+        stream.close()
+
+
 @router.get("/documents/{document_hash}/text", name="document_text")
 def document_text(document_hash: str, request: Request) -> dict[str, str]:
     require_user(request)
@@ -42,12 +109,44 @@ def document_text(document_hash: str, request: Request) -> dict[str, str]:
 
 
 @router.get("/documents/{document_hash}/file", name="document_file")
-def document_file(document_hash: str, request: Request) -> FileResponse:
+def document_file(document_hash: str, request: Request) -> Response:
     require_user(request)
-    path = _path(request, document_hash)
-    filename = path.name.replace("\r", "_").replace("\n", "_").replace('"', "_")
+    document = request.app.state.documents.get(document_hash)
+    if document is None:
+        raise AppError("document_not_found", "Document was not found", 404)
+    if document.canonical_path is None:
+        raise AppError("document_missing", "Document is missing", 410)
+    try:
+        stream = open_contained_file(
+            request.app.state.path_policy.root, document.canonical_path
+        )
+    except (OSError, ValueError):
+        raise AppError("document_missing", "Document is missing", 410) from None
+    filename = Path(document.canonical_path).name
+    filename = filename.replace("\r", "_").replace("\n", "_").replace('"', "_")
     media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
-    return FileResponse(
-        path, media_type=media_type, filename=filename,
-        content_disposition_type="inline" if media_type in SAFE_INLINE_TYPES else "attachment",
+    disposition = "inline" if media_type in SAFE_INLINE_TYPES else "attachment"
+    quoted = quote(filename)
+    content_disposition = (
+        f'{disposition}; filename="{filename}"'
+        if quoted == filename else f"{disposition}; filename*=utf-8''{quoted}"
+    )
+    size = os.fstat(stream.fileno()).st_size
+    selected = _byte_range(request.headers.get("range"), size)
+    if selected is None:
+        stream.close()
+        return Response(status_code=416, headers={
+            "Accept-Ranges": "bytes", "Content-Range": f"bytes */{size}",
+        })
+    start, end, partial = selected
+    headers = {
+        "Accept-Ranges": "bytes",
+        "Content-Length": str(end - start + 1),
+        "Content-Disposition": content_disposition,
+    }
+    if partial:
+        headers["Content-Range"] = f"bytes {start}-{end}/{size}"
+    return StreamingResponse(
+        _chunks(stream, start, end - start + 1),
+        status_code=206 if partial else 200, media_type=media_type, headers=headers,
     )

src/recoll_modern/api/memberships.py

@@ -74,14 +74,18 @@ def remove_document(
 
 
 @router.get("/collections/{collection_id}/directory-rules")
-def list_rules(collection_id: str, request: Request) -> dict[str, object]:
+def list_rules(
+    collection_id: str, request: Request, limit: int = 50, offset: int = 0
+) -> dict[str, object]:
     require_user(request)
     request.app.state.collections.get(collection_id)
+    if not 1 <= limit <= 200 or offset < 0:
+        raise AppError("validation_failed", "Invalid pagination", 422)
     rules = [
         asdict(rule) for rule in request.app.state.memberships.list_rules()
         if rule.collection_id == collection_id
     ]
-    return {"items": rules, "total": len(rules)}
+    return {"items": rules[offset:offset + limit], "total": len(rules)}
 
 
 @router.patch("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202)

src/recoll_modern/app.py

@@ -66,6 +66,14 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
             if paths:
                 engine.force_reindex(tuple(str(policy.root / path) for path in sorted(set(paths))))
 
+        def current_rule_paths(rule_id: str) -> tuple[str, ...]:
+            paths = []
+            for content_hash in memberships.rule_documents(rule_id):
+                record = documents.get(content_hash)
+                if record is not None and record.canonical_path is not None:
+                    paths.append(record.canonical_path)
+            return tuple(paths)
+
         def cleanup(job: Job) -> None:
             export_membership_map(map_rows, settings.membership_map_path)
             paths = job.request.get("paths", [])
@@ -98,26 +106,39 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
                 raise ValueError("invalid rule patch job")
             if not all(isinstance(item, str) for item in exclusions):
                 raise ValueError("invalid rule patch job")
+            replay_paths = jobs.record_reindex_paths(job.id, current_rule_paths(rule_id))
             result = membership_service.plan_patch_rule(rule_id, tuple(exclusions))
-            reindex_changed(result.changed_paths)
+            reindex_changed(tuple(set(replay_paths) | set(result.changed_paths)))
 
         def delete_rule(job: Job) -> None:
             rule_id = job.request.get("rule_id")
             if not isinstance(rule_id, str):
                 raise ValueError("invalid rule deletion job")
-            reindex_changed(membership_service.plan_delete_rule(rule_id))
+            replay_paths = jobs.record_reindex_paths(job.id, current_rule_paths(rule_id))
+            changed = membership_service.plan_delete_rule(rule_id)
+            reindex_changed(tuple(set(replay_paths) | set(changed)))
 
         def reindex(job: Job) -> None:
             paths = job.request.get("paths", [])
             collections_requested = job.request.get("collections", [])
             if not isinstance(paths, list) or not isinstance(collections_requested, list):
                 raise ValueError("invalid reindex job")
-            for rule in memberships.list_rules():
-                reconcile_rule(rule.id, policy, documents, memberships)
+            rules = memberships.list_rules()
+            prior_paths = tuple(
+                path for rule in rules for path in current_rule_paths(rule.id)
+            )
+            replay_paths = jobs.record_reindex_paths(job.id, prior_paths)
+            changed: set[str] = set()
+            for rule in rules:
+                changed.update(
+                    reconcile_rule(rule.id, policy, documents, memberships).changed_paths
+                )
             export_membership_map(map_rows, settings.membership_map_path)
             selected = {
                 str(policy.root / path) for path in paths if isinstance(path, str)
             }
+            selected.update(str(policy.root / path) for path in changed)
+            selected.update(str(policy.root / path) for path in replay_paths)
             requested_ids = {item for item in collections_requested if isinstance(item, str)}
             for path, ids in memberships.membership_map_rows(policy.root):
                 if requested_ids.intersection(ids):
@@ -144,7 +165,9 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
         app.state.auth = auth_service
         app.state.engine = engine
         app.state.path_policy = policy
-        app.state.collections = CollectionService(SQLiteCollectionRepository(database))
+        collection_repository = SQLiteCollectionRepository(database)
+        app.state.collections = CollectionService(collection_repository)
+        app.state.collection_repository = collection_repository
         app.state.memberships = memberships
         app.state.documents = documents
         app.state.jobs = jobs

src/recoll_modern/domain/membership.py

@@ -57,7 +57,7 @@ class MembershipService:
         }
         for source, report in discovered:
             if source.type == "directory" and source.mode == "persistent":
-                rule = self._memberships.create_rule(
+                rule = self._memberships.ensure_rule(
                     collection_id,
                     self._policy.normalize(source.path),
                     tuple(self._policy.normalize(path) for path in source.excluded_directories),

src/recoll_modern/indexing/reconcile.py

@@ -28,18 +28,31 @@ def reconcile_rule(
         mode="persistent",
         excluded_directories=rule.excluded_directories,
     )])
-    skipped = (*rule.excluded_directories, *(error.relative_path for error in report.errors))
+    failed_paths = tuple(error.relative_path for error in report.errors)
+    skipped = (*rule.excluded_directories, *failed_paths)
+    existing_paths = documents.known_paths_under(rule.relative_path)
     known_paths = tuple(
-        path for path in documents.known_paths_under(rule.relative_path)
+        path for path in existing_paths
         if not any(path == prefix or path.startswith(prefix + "/") for prefix in skipped)
     )
+    retired_paths = set(known_paths) - set(report.relative_paths)
+    previous_hashes = memberships.rule_documents(rule_id)
+    protected_hashes = {
+        record.content_hash
+        for path in existing_paths
+        if any(path == failed or path.startswith(failed + "/") for failed in failed_paths)
+        if (record := documents.by_path(path)) is not None
+        and record.content_hash in previous_hashes
+    }
     observed = documents.observe_files(list(report.files), reconcile_paths=known_paths)
     changed = memberships.replace_rule_members(
-        rule_id, [file.content_hash for file in report.files]
+        rule_id, [file.content_hash for file in report.files] + sorted(protected_hashes)
     )
     canonical_changes = [
         document.canonical_path
         for document in observed.documents
         if document.canonical_path is not None
     ]
-    return ReconcileResult(report, tuple(sorted(set(changed + canonical_changes))))
+    return ReconcileResult(
+        report, tuple(sorted(set(changed + canonical_changes) | retired_paths))
+    )

src/recoll_modern/storage/collections.py

@@ -5,11 +5,14 @@ from __future__ import annotations
 import secrets
 import sqlite3
 from datetime import UTC, datetime
+from sqlite3 import Connection
 
 from recoll_modern.domain.collections import CollectionPage, CollectionPaths, normalize_name
 from recoll_modern.domain.errors import AppError
+from recoll_modern.domain.jobs import Job
 from recoll_modern.domain.models import Collection, CollectionCreate, CollectionPatch, _Unset
 from recoll_modern.storage.database import Database
+from recoll_modern.storage.jobs import JobRepository
 
 
 class SQLiteCollectionRepository:
@@ -113,32 +116,47 @@ class SQLiteCollectionRepository:
     def delete(self, collection_id: str) -> CollectionPaths:
         """Return effective canonical paths before SQLite cascades memberships away."""
         with self._database.transaction() as connection:
-            paths = [
-                str(row[0])
-                for row in connection.execute(
-                    "SELECT DISTINCT paths.relative_path "
-                    "FROM document_paths AS paths "
-                    "JOIN ("
-                    "  SELECT direct.document_hash "
-                    "  FROM direct_memberships AS direct "
-                    "  WHERE direct.collection_id = ? "
-                    "  UNION "
-                    "  SELECT rule_memberships.document_hash "
-                    "  FROM rule_memberships "
-                    "  JOIN directory_rules AS rules ON rules.id = rule_memberships.rule_id "
-                    "  WHERE rules.collection_id = ? "
-                    "    AND NOT EXISTS ("
-                    "      SELECT 1 FROM membership_exclusions AS exclusions "
-                    "      WHERE exclusions.collection_id = rules.collection_id "
-                    "        AND exclusions.document_hash = rule_memberships.document_hash"
-                    "    )"
-                    ") AS members ON members.document_hash = paths.document_hash "
-                    "WHERE paths.is_canonical = 1 "
-                    "ORDER BY paths.relative_path",
-                    (collection_id, collection_id),
-                )
-            ]
-            connection.execute("DELETE FROM collections WHERE id = ?", (collection_id,))
+            return self._delete_in_transaction(connection, collection_id)
+
+    def delete_with_cleanup(self, collection_id: str, jobs: JobRepository) -> Job:
+        """Delete metadata and queue its index cleanup in one durable transaction."""
+        with self._database.transaction() as connection:
+            paths = self._delete_in_transaction(connection, collection_id)
+            return jobs.enqueue_in_transaction(
+                connection, "collection_cleanup", {"version": 1, "paths": paths}, False
+            )
+
+    def _delete_in_transaction(
+        self, connection: Connection, collection_id: str
+    ) -> CollectionPaths:
+        paths = [
+            str(row[0])
+            for row in connection.execute(
+                "SELECT DISTINCT paths.relative_path "
+                "FROM document_paths AS paths "
+                "JOIN ("
+                "  SELECT direct.document_hash "
+                "  FROM direct_memberships AS direct "
+                "  WHERE direct.collection_id = ? "
+                "  UNION "
+                "  SELECT rule_memberships.document_hash "
+                "  FROM rule_memberships "
+                "  JOIN directory_rules AS rules ON rules.id = rule_memberships.rule_id "
+                "  WHERE rules.collection_id = ? "
+                "    AND NOT EXISTS ("
+                "      SELECT 1 FROM membership_exclusions AS exclusions "
+                "      WHERE exclusions.collection_id = rules.collection_id "
+                "        AND exclusions.document_hash = rule_memberships.document_hash"
+                "    )"
+                ") AS members ON members.document_hash = paths.document_hash "
+                "WHERE paths.is_canonical = 1 "
+                "ORDER BY paths.relative_path",
+                (collection_id, collection_id),
+            )
+        ]
+        deleted = connection.execute("DELETE FROM collections WHERE id = ?", (collection_id,))
+        if deleted.rowcount != 1:
+            raise _collection_not_found()
         return paths
 
 

src/recoll_modern/storage/jobs.py

@@ -7,6 +7,7 @@ import secrets
 import sqlite3
 from collections.abc import Mapping
 from datetime import UTC, datetime
+from sqlite3 import Connection
 
 from recoll_modern.domain.jobs import Job, JobFileError, JobState
 from recoll_modern.storage.database import Database
@@ -19,25 +20,32 @@ class JobRepository:
     def enqueue(
         self, kind: str, request: Mapping[str, object], fail_on_file_error: bool
     ) -> Job:
+        with self._database.transaction() as connection:
+            return self.enqueue_in_transaction(connection, kind, request, fail_on_file_error)
+
+    def enqueue_in_transaction(
+        self, connection: Connection, kind: str, request: Mapping[str, object],
+        fail_on_file_error: bool,
+    ) -> Job:
+        """Write a job in a caller-owned transaction when state and work must be atomic."""
         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),
-                    )
+                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
+            return Job(
+                job_id, kind, JobState.QUEUED, dict(request), fail_on_file_error, 0, 0
+            )
 
     def get(self, job_id: str) -> Job | None:
         with self._database.connect() as connection:
@@ -98,6 +106,26 @@ class JobRepository:
                 )
             ]
 
+    def record_reindex_paths(self, job_id: str, paths: tuple[str, ...]) -> tuple[str, ...]:
+        """Write paths ahead of a destructive membership mutation for replay."""
+        with self._database.transaction() as connection:
+            row = connection.execute(
+                "SELECT request_json FROM jobs WHERE id = ? AND state = 'running'", (job_id,)
+            ).fetchone()
+            if row is None:
+                raise ValueError("job is not running")
+            request = json.loads(str(row[0]))
+            prior = request.get("reindex_paths", [])
+            if not isinstance(prior, list) or not all(isinstance(item, str) for item in prior):
+                raise ValueError("invalid replay paths")
+            combined = tuple(sorted(set(prior) | set(paths)))
+            request["reindex_paths"] = list(combined)
+            connection.execute(
+                "UPDATE jobs SET request_json = ? WHERE id = ?",
+                (json.dumps(request, sort_keys=True), job_id),
+            )
+            return combined
+
     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:
@@ -156,11 +184,49 @@ def _valid_request(kind: str, raw: str) -> bool:
         return False
     if not isinstance(value, dict):
         return False
+    if value.get("version") != 1:
+        return False
+    replay_paths = value.get("reindex_paths", [])
+    if not _string_list(replay_paths):
+        return False
     if kind == "membership_add":
+        sources = value.get("sources")
+        if not isinstance(value.get("collection_id"), str) or not isinstance(sources, list):
+            return False
+        return all(_valid_source(source) for source in sources)
+    if kind == "membership_remove":
         return isinstance(value.get("collection_id"), str) and isinstance(
-            value.get("sources"), list
+            value.get("document_hash"), str
         )
-    return bool(kind)
+    if kind == "rule_patch":
+        return isinstance(value.get("rule_id"), str) and _string_list(
+            value.get("excluded_directories")
+        )
+    if kind == "rule_delete":
+        return isinstance(value.get("rule_id"), str)
+    if kind == "collection_cleanup":
+        return _string_list(value.get("paths"))
+    if kind == "reindex":
+        return _string_list(value.get("paths")) and _string_list(value.get("collections"))
+    if kind == "reconcile_rules":
+        return True
+    return False
+
+
+def _string_list(value: object) -> bool:
+    return isinstance(value, list) and all(isinstance(item, str) for item in value)
+
+
+def _valid_source(value: object) -> bool:
+    if not isinstance(value, dict) or not isinstance(value.get("path"), str):
+        return False
+    if value.get("type") == "file":
+        return True
+    if value.get("type") == "directory":
+        return value.get("mode") in ("snapshot", "persistent") and _string_list(
+            value.get("excluded_directories", [])
+        )
+    return False
 
 
 def _job_from_row(row: sqlite3.Row) -> Job:

src/recoll_modern/storage/membership.py

@@ -46,6 +46,26 @@ class MembershipRepository:
                 raise
             return DirectoryRule(rule_id, collection_id, relative_path, excluded_directories)
 
+    def ensure_rule(
+        self, collection_id: str, relative_path: str, excluded_directories: tuple[str, ...]
+    ) -> DirectoryRule:
+        """Replay a persistent source without creating a duplicate rule."""
+        with self._database.connect() as connection:
+            row = connection.execute(
+                "SELECT id FROM directory_rules WHERE collection_id = ? AND relative_path = ?",
+                (collection_id, relative_path),
+            ).fetchone()
+        if row is None:
+            return self.create_rule(collection_id, relative_path, excluded_directories)
+        rule_id = str(row[0])
+        rule = self.get_rule(rule_id)
+        assert rule is not None
+        if rule.excluded_directories != excluded_directories:
+            self.patch_rule(rule_id, excluded_directories)
+            rule = self.get_rule(rule_id)
+            assert rule is not None
+        return rule
+
     def get_rule(self, rule_id: str) -> DirectoryRule | None:
         with self._database.connect() as connection:
             row = connection.execute(

tests/api/test_collections.py

@@ -1,5 +1,6 @@
 from typing import cast
 
+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
@@ -45,4 +46,24 @@ def test_empty_create_and_rename_do_not_enqueue_jobs(
     client.patch(f"/collections/{collection_id}", json={"name": "Reading"}, headers=headers)
     with cast(FastAPI, client.app).state.database.connect() as connection:
         total = connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0]
-    assert total == 0
+        assert total == 0
+
+
+def test_delete_rolls_back_when_cleanup_job_cannot_be_written(
+    authenticated: tuple[TestClient, str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client, csrf = authenticated
+    collection_id = client.post(
+        "/collections", json={"name": "Books"}, headers={"X-CSRF-Token": csrf}
+    ).json()["id"]
+
+    def fail_id(_size: int) -> str:
+        raise RuntimeError("simulated job ID failure")
+
+    monkeypatch.setattr("recoll_modern.storage.jobs.secrets.token_hex", fail_id)
+    with pytest.raises(RuntimeError, match="simulated job ID failure"):
+        client.delete(
+            f"/collections/{collection_id}", params={"confirmation": "Books"},
+            headers={"X-CSRF-Token": csrf},
+        )
+    assert client.get(f"/collections/{collection_id}").status_code == 200

tests/api/test_documents.py

@@ -34,6 +34,8 @@ def test_text_and_byte_range(authenticated: tuple[TestClient, str], tmp_path: Pa
     assert client.get(
         f"/documents/{digest}/file", headers={"Range": "bytes=99-100"}
     ).status_code == 416
+    suffix = client.get(f"/documents/{digest}/file", headers={"Range": "bytes=-2"})
+    assert suffix.status_code == 206 and suffix.content == b"ef"
 
 
 def test_unknown_and_missing_documents(

tests/api/test_jobs.py

@@ -1,5 +1,8 @@
 import time
+from pathlib import Path
+from typing import cast
 
+from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 
@@ -22,6 +25,7 @@ def test_retry_refuses_completed_job(authenticated: tuple[TestClient, str]) -> N
         if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed":
             break
         time.sleep(0.01)
+        time.sleep(0.01)
     response = client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf})
     assert response.status_code == 409
     assert response.json()["error"]["code"] == "job_not_retryable"
@@ -32,3 +36,37 @@ def test_job_list_validates_pagination(authenticated: tuple[TestClient, str]) ->
     response = client.get("/jobs", params={"limit": 201})
     assert response.status_code == 422
     assert response.json()["error"]["code"] == "validation_failed"
+
+
+def test_collection_reindex_includes_departed_rule_path(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    books = tmp_path / "docs" / "books"
+    books.mkdir()
+    file = books / "gone.txt"
+    file.write_text("gone")
+    headers = {"X-CSRF-Token": csrf}
+    collection_id = client.post(
+        "/collections", headers=headers, json={"name": "Books"}
+    ).json()["id"]
+    created = client.post(
+        f"/collections/{collection_id}/memberships", headers=headers,
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]},
+    )
+    job_id = created.json()["job"]["id"]
+    for _ in range(100):
+        if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed":
+            break
+    engine = cast(FastAPI, client.app).state.engine
+    engine.forced_paths.clear()
+    file.unlink()
+    refreshed = client.post(
+        "/jobs/reindex", headers=headers, json={"collections": [collection_id]}
+    )
+    refresh_id = refreshed.json()["job"]["id"]
+    for _ in range(100):
+        if client.get(f"/jobs/{refresh_id}").json()["job"]["state"] == "completed":
+            break
+        time.sleep(0.01)
+    assert any(str(file) in batch for batch in engine.forced_paths)

tests/api/test_memberships.py

@@ -2,6 +2,8 @@ import time
 from pathlib import Path
 from typing import cast
 
+import pytest
+from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 
@@ -177,3 +179,90 @@ def test_remove_document_creates_override_then_direct_add_restores_it(
     assert len(members) == 1
     assert members[0]["direct"] is True
     assert members[0]["override"] is False
+
+
+def test_rule_deletion_retry_reindexes_removed_membership(
+    authenticated: tuple[TestClient, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client, csrf = authenticated
+    books = tmp_path / "docs" / "books"
+    books.mkdir()
+    file = books / "a.txt"
+    file.write_text("book")
+    collection_id = _collection(client, csrf)
+    created = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]},
+    )
+    assert _wait(client, created.json()["job"]["id"])["state"] == "completed"
+    rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"]
+    engine = cast(FastAPI, client.app).state.engine
+    original = engine.force_reindex
+    calls: list[tuple[str, ...]] = []
+
+    def fail_once(paths: tuple[str, ...]) -> None:
+        calls.append(paths)
+        if len(calls) == 1:
+            raise RuntimeError("simulated indexing failure")
+        original(paths)
+
+    monkeypatch.setattr(engine, "force_reindex", fail_once)
+    deleted = client.delete(
+        f"/collections/{collection_id}/directory-rules/{rule_id}",
+        headers={"X-CSRF-Token": csrf},
+    )
+    job_id = deleted.json()["job"]["id"]
+    assert _wait(client, job_id)["state"] == "failed"
+    retried = client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf})
+    assert retried.status_code == 202
+    assert _wait(client, job_id)["state"] == "completed"
+    assert calls == [(str(file),), (str(file),)]
+
+
+def test_rule_patch_retry_reindexes_excluded_member(
+    authenticated: tuple[TestClient, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client, csrf = authenticated
+    skipped = tmp_path / "docs" / "books" / "skip"
+    skipped.mkdir(parents=True)
+    file = skipped / "a.txt"
+    file.write_text("book")
+    collection_id = _collection(client, csrf)
+    created = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]},
+    )
+    assert _wait(client, created.json()["job"]["id"])["state"] == "completed"
+    rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"]
+    engine = cast(FastAPI, client.app).state.engine
+    original = engine.force_reindex
+    calls: list[tuple[str, ...]] = []
+
+    def fail_once(paths: tuple[str, ...]) -> None:
+        calls.append(paths)
+        if len(calls) == 1:
+            raise RuntimeError("simulated indexing failure")
+        original(paths)
+
+    monkeypatch.setattr(engine, "force_reindex", fail_once)
+    patched = client.patch(
+        f"/collections/{collection_id}/directory-rules/{rule_id}",
+        headers={"X-CSRF-Token": csrf}, json={"excluded_directories": ["books/skip"]},
+    )
+    job_id = patched.json()["job"]["id"]
+    assert _wait(client, job_id)["state"] == "failed"
+    client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf})
+    assert _wait(client, job_id)["state"] == "completed"
+    assert calls == [(str(file),), (str(file),)]
+
+
+def test_rule_list_validates_pagination(authenticated: tuple[TestClient, str]) -> None:
+    client, csrf = authenticated
+    collection_id = _collection(client, csrf)
+    response = client.get(
+        f"/collections/{collection_id}/directory-rules", params={"limit": 201}
+    )
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "validation_failed"

tests/contract/test_recoll_engine.py

@@ -74,13 +74,10 @@ class TestRecollEngine(EngineContract):
         if config is None:
             if os.environ.get("RECOLL_MODERN_REQUIRE_DOCKER") != "1":
                 pytest.skip("Recoll test container is not running")
-            corpus_file = os.environ.get("RECOLL_MODERN_TEST_CORPUS_FILE")
-            if corpus_file is None:
-                pytest.fail("RECOLL_MODERN_TEST_CORPUS_FILE is required for container contract")
             bootstrap = subprocess.run(
                 ["docker", "compose", "exec", "-T", "app", "recollindex", "-c",
                  "/var/lib/recoll-modern/recoll-config", "-i", "-Z",
-                 f"/documents/{corpus_file}"],
+                 "/documents"],
                 capture_output=True, text=True, check=False, timeout=30,
             )
             assert bootstrap.returncode == 0, bootstrap.stderr

tests/indexing/test_job_handlers.py

@@ -89,3 +89,36 @@ def test_handler_exports_map_before_forced_reindex(tmp_path: Path) -> None:
     handler.handle(claimed)
     assert calls == [(str(root / "good.txt"),)]
     assert jobs.get(job.id).state == JobState.COMPLETED  # type: ignore[union-attr]
+
+
+def test_persistent_source_replays_after_index_failure(tmp_path: Path) -> None:
+    root, jobs, membership, _, collection_id = _setup(tmp_path)
+    (root / "books").mkdir()
+    (root / "books" / "a.txt").write_text("book")
+    forced: list[tuple[str, ...]] = []
+
+    class FlakyIndexer:
+        def force_reindex(self, paths: tuple[str, ...]) -> None:
+            forced.append(paths)
+            if len(forced) == 1:
+                raise RuntimeError("simulated index failure")
+
+    handler = MembershipJobHandler(
+        PathPolicy(root), DocumentRepository(Database(tmp_path / "state.sqlite3")),
+        membership, jobs, tmp_path / "membership.map", FlakyIndexer(),
+    )
+    job = jobs.enqueue("membership_add", {
+        "version": 1, "collection_id": collection_id,
+        "sources": [{"type": "directory", "path": "books", "mode": "persistent",
+                     "excluded_directories": []}],
+    }, False)
+    claimed = jobs.claim_next()
+    assert claimed is not None
+    try:
+        handler.handle(claimed)
+    except RuntimeError:
+        pass
+    assert len(membership.list_rules()) == 1
+    handler.handle(claimed)
+    assert jobs.get(job.id).state == JobState.COMPLETED  # type: ignore[union-attr]
+    assert forced == [(str(root / "books" / "a.txt"),)] * 2

tests/indexing/test_reconcile.py

@@ -1,6 +1,9 @@
 from pathlib import Path
 
+import pytest
+
 from recoll_modern.domain.models import CollectionCreate
+from recoll_modern.indexing.discovery import DiscoveryReport, FileError
 from recoll_modern.indexing.paths import PathPolicy
 from recoll_modern.indexing.reconcile import reconcile_rule
 from recoll_modern.storage.collections import SQLiteCollectionRepository
@@ -52,3 +55,30 @@ def test_excluded_file_is_not_marked_missing(tmp_path: Path) -> None:
     record = documents.get(content_hash)
     assert record is not None and record.state == "present"
     assert membership.rule_documents(rule.id) == set()
+
+
+def test_unreadable_file_retains_rule_membership(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    root = tmp_path / "docs"
+    books = root / "books"
+    books.mkdir(parents=True)
+    (books / "a.txt").write_text("a")
+    database = Database(tmp_path / "state.sqlite3")
+    database.migrate()
+    collection_id = SQLiteCollectionRepository(database).create(CollectionCreate("Books")).id
+    membership = MembershipRepository(database)
+    documents = DocumentRepository(database)
+    rule = membership.create_rule(collection_id, "books", ())
+    first = reconcile_rule(rule.id, PathPolicy(root), documents, membership)
+    content_hash = first.discovery.files[0].content_hash
+    monkeypatch.setattr(
+        "recoll_modern.indexing.reconcile.discover_sources",
+        lambda _policy, _sources: DiscoveryReport(
+            (), (FileError("books/a.txt", "file_unreadable"),)
+        ),
+    )
+    second = reconcile_rule(rule.id, PathPolicy(root), documents, membership)
+    assert second.discovery.errors[0].code == "file_unreadable"
+    assert membership.rule_documents(rule.id) == {content_hash}
+    assert documents.get(content_hash).state == "present"  # type: ignore[union-attr]

tests/indexing/test_worker.py

@@ -48,10 +48,10 @@ 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)
+    job = jobs.enqueue("reindex", {"version": 1, "paths": [], "collections": []}, False)
     jobs.claim_next()
     handled: list[str] = []
-    worker = JobWorker(jobs, {"test": lambda claimed: handled.append(claimed.id)})
+    worker = JobWorker(jobs, {"reindex": lambda claimed: handled.append(claimed.id)})
     worker.start()
     for _ in range(100):
         current = jobs.get(job.id)

tests/integration/test_end_to_end.py

@@ -1,9 +1,14 @@
 """Exercise the complete HTTP lifecycle with durable state and a deterministic engine."""
 
+import os
+import subprocess
 import time
+import uuid
+from collections.abc import Callable
 from pathlib import Path
 from typing import cast
 
+import httpx
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
@@ -15,6 +20,24 @@ from recoll_modern.storage.database import Database
 from recoll_modern.storage.jobs import JobRepository
 
 
+def _live_wait(
+    client: httpx.Client, path: str, job_id: str, headers: dict[str, str]
+) -> dict[str, object]:
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        try:
+            response = client.get(f"{path}/jobs/{job_id}", headers=headers)
+        except httpx.TransportError:
+            time.sleep(0.1)
+            continue
+        if response.status_code == 200:
+            job = cast(dict[str, object], response.json()["job"])
+            if job["state"] not in ("queued", "running"):
+                return job
+        time.sleep(0.1)
+    raise AssertionError(f"container job {job_id} did not finish")
+
+
 def _wait(client: TestClient, job_id: str) -> dict[str, object]:
     deadline = time.monotonic() + 5
     while time.monotonic() < deadline:
@@ -118,3 +141,125 @@ def test_complete_collection_lifecycle(tmp_path: Path, monkeypatch: pytest.Monke
             "username": "admin", "password": "secret-password",
         })
         assert _wait(client, interrupted.id)["state"] == "completed"
+
+
+def test_real_container_collection_lifecycle(
+    compose_exec: Callable[[list[str]], subprocess.CompletedProcess[str]],
+) -> None:
+    document_root = os.environ.get("RECOLL_MODERN_DOCUMENTS")
+    if document_root is None:
+        pytest.fail("RECOLL_MODERN_DOCUMENTS is required for real-container acceptance")
+    relative = Path(f"recoll-modern-test-{uuid.uuid4().hex}")
+    corpus = Path(document_root) / relative
+    (corpus / "a" / ".calibre").mkdir(parents=True)
+    (corpus / "b").mkdir()
+    marker = uuid.uuid4().hex
+    content = {
+        "a/shared.txt": f"zebraplatypus duplicate {marker}",
+        "b/shared.txt": f"zebraplatypus duplicate {marker}",
+        "a/other.txt": f"zebraplatypus first distinct {marker}",
+        "b/other.txt": f"zebraplatypus second distinct {marker}",
+        "a/special.txt": f"zebraplatypus second only {marker}",
+        "a/.calibre/hidden.txt": f"zebraplatypus excluded {marker}",
+    }
+    for name, text in content.items():
+        file = corpus / name
+        file.write_text(text)
+    prefix = os.environ.get("RECOLL_MODERN_ROOT_PATH", "/").rstrip("/")
+    port = os.environ.get("RECOLL_MODERN_PUBLISH_PORT", "8000")
+    with httpx.Client(base_url=f"http://127.0.0.1:{port}", timeout=15) as client:
+        login = client.post(f"{prefix}/auth/login", json={
+            "username": os.environ["RECOLL_MODERN_ADMIN_USERNAME"],
+            "password": os.environ["RECOLL_MODERN_ADMIN_PASSWORD"],
+        })
+        assert login.status_code == 200, login.text
+        cookie = login.cookies.get("recoll_modern_session")
+        assert cookie
+        headers = {
+            "Cookie": f"recoll_modern_session={cookie}",
+            "X-CSRF-Token": login.json()["csrf_token"],
+        }
+        a = client.post(
+            f"{prefix}/collections", headers=headers,
+            json={"name": f"Real A {uuid.uuid4().hex}"},
+        ).json()["id"]
+        b = client.post(
+            f"{prefix}/collections", headers=headers,
+            json={"name": f"Real B {uuid.uuid4().hex}"},
+        ).json()["id"]
+        def add(collection_id: str, sources: list[dict[str, object]]) -> str:
+            response = client.post(
+                f"{prefix}/collections/{collection_id}/memberships", headers=headers,
+                json={"sources": sources},
+            )
+            assert response.status_code == 202, response.text
+            job_id = response.json()["job"]["id"]
+            assert _live_wait(client, prefix, job_id, headers)["state"] == "completed"
+            return cast(str, job_id)
+
+        rule_job = add(a, [{
+            "type": "directory", "path": (relative / "a").as_posix(),
+            "mode": "persistent",
+            "excluded_directories": [(relative / "a" / ".calibre").as_posix()],
+        }])
+        add(b, [{"type": "directory", "path": (relative / "b").as_posix(),
+                 "mode": "snapshot"}])
+        add(b, [{"type": "file", "path": (relative / "a" / "special.txt").as_posix()}])
+        docs_a = client.get(f"{prefix}/collections/{a}/documents", headers=headers).json()["items"]
+        docs_b = client.get(f"{prefix}/collections/{b}/documents", headers=headers).json()["items"]
+        assert len(docs_a) == 3 and len(docs_b) == 3
+        assert all(".calibre" not in item["path"] for item in docs_a)
+        shared = next(
+            item for item in docs_a
+            if item["document_id"] in {other["document_id"] for other in docs_b}
+            and item["path"].endswith("shared.txt")
+        )
+        assert shared["document_id"] in {item["document_id"] for item in docs_b}
+        combined = client.get(f"{prefix}/search", params=[
+            ("q", "zebraplatypus"), ("collections", a), ("collections", b),
+            ("fields", "text"), ("text_limit", "8"),
+        ], headers=headers)
+        assert combined.status_code == 200, combined.text
+        assert len(combined.json()["hits"]) == 4, combined.json()
+        assert all(hit["text"] == "zebrapla" for hit in combined.json()["hits"])
+        text_response = client.get(
+            f"{prefix}/documents/{shared['document_id']}/text", headers=headers
+        )
+        assert text_response.status_code == 200 and "duplicate" in text_response.json()["text"]
+        partial = client.get(
+            f"{prefix}/documents/{shared['document_id']}/file",
+            headers={**headers, "Range": "bytes=0-7"},
+        )
+        assert partial.status_code == 206 and partial.content == b"zebrapla"
+
+        (corpus / "a" / "shared.txt").unlink()
+        refreshed = client.post(f"{prefix}/jobs/reindex", headers=headers, json={})
+        assert _live_wait(client, prefix, refreshed.json()["job"]["id"], headers)[
+            "state"
+        ] == "completed"
+        promoted = client.get(f"{prefix}/collections/{b}/documents", headers=headers).json()[
+            "items"
+        ]
+        assert next(item for item in promoted if item["document_id"] == shared["document_id"])[
+            "path"
+        ] == (relative / "b" / "shared.txt").as_posix()
+        assert client.get(f"{prefix}/search", params={
+            "q": "zebraplatypus", "collections": b,
+        }, headers=headers).json()["hits"]
+
+        marked = compose_exec(["python", "-c", (
+            "from recoll_modern.storage.database import Database; "
+            "from pathlib import Path; "
+            "db=Database(Path('/var/lib/recoll-modern/recoll-modern.sqlite3')); "
+            "c=db.connect(); "
+            "c.execute(\"UPDATE jobs SET state='running' WHERE id=?\", "
+            "(__import__('sys').argv[1],)); "
+            "c.commit(); c.close()"
+        ), rule_job])
+        assert marked.returncode == 0, marked.stderr
+        restarted = subprocess.run(
+            ["docker", "compose", "restart", "app"], capture_output=True, text=True,
+            check=False, timeout=40,
+        )
+        assert restarted.returncode == 0, restarted.stderr
+        assert _live_wait(client, prefix, rule_job, headers)["state"] == "completed"

tests/integration/test_read_only_corpus.py

@@ -3,7 +3,9 @@
 import os
 import subprocess
 import time
+import uuid
 from collections.abc import Callable
+from pathlib import Path
 
 import httpx
 import pytest
@@ -21,7 +23,14 @@ def test_generated_corpus_indexes_and_searches(
 ) -> None:
     corpus_file = os.environ.get("RECOLL_MODERN_TEST_CORPUS_FILE")
     if corpus_file is None:
-        pytest.skip("Set RECOLL_MODERN_TEST_CORPUS_FILE for generated-corpus verification")
+        document_root = os.environ.get("RECOLL_MODERN_DOCUMENTS")
+        if document_root is None:
+            pytest.fail("RECOLL_MODERN_DOCUMENTS is required for generated-corpus verification")
+        relative = Path(f"recoll-modern-test-{uuid.uuid4().hex}") / "probe.txt"
+        file = Path(document_root) / relative
+        file.parent.mkdir(parents=True)
+        file.write_text("Unique zebraplatypus integration corpus.\n")
+        corpus_file = relative.as_posix()
     assert compose_exec(["test", "-f", f"/documents/{corpus_file}"]).returncode == 0
     port = os.environ.get("RECOLL_MODERN_PUBLISH_PORT", "8000")
     username = os.environ["RECOLL_MODERN_ADMIN_USERNAME"]

tests/security/test_downloads.py

@@ -6,6 +6,7 @@ from typing import cast
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
+from recoll_modern.api.documents import open_contained_file
 from recoll_modern.indexing.discovery import ObservedFile
 
 
@@ -23,3 +24,19 @@ def test_missing_document_is_gone(
     response = client.get(f"/documents/{digest}/file")
     assert response.status_code == 410
     assert response.json()["error"]["code"] == "document_missing"
+
+
+def test_open_descriptor_cannot_be_redirected_by_later_symlink_swap(tmp_path: Path) -> None:
+    root = tmp_path / "docs"
+    root.mkdir()
+    file = root / "safe.txt"
+    file.write_text("safe")
+    outside = tmp_path / "secret.txt"
+    outside.write_text("secret")
+    opened = open_contained_file(root, "safe.txt")
+    file.unlink()
+    file.symlink_to(outside)
+    try:
+        assert opened.read() == b"safe"
+    finally:
+        opened.close()

tests/storage/test_job_repository.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+import pytest
+
 from recoll_modern.domain.jobs import JobFileError, JobState
 from recoll_modern.storage.database import Database
 from recoll_modern.storage.jobs import JobRepository
@@ -11,10 +13,14 @@ def _repo(tmp_path: Path) -> JobRepository:
     return JobRepository(database)
 
 
+def _membership_request(sources: object = None) -> dict[str, object]:
+    return {"version": 1, "collection_id": "c1", "sources": [] if sources is None else sources}
+
+
 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)
+    first = repo.enqueue("membership_add", _membership_request(), False)
+    second = repo.enqueue("membership_add", _membership_request(), 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
@@ -22,7 +28,7 @@ def test_claim_next_is_durable_and_oldest_first(tmp_path: Path) -> 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)
+    job = repo.enqueue("membership_add", _membership_request(), False)
     repo.claim_next()
     repo.recover_running()
     assert repo.get(job.id).state == JobState.QUEUED  # type: ignore[union-attr]
@@ -32,7 +38,7 @@ 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)
+    job = repo.enqueue("membership_add", _membership_request(), False)
     repo.claim_next()
     with database.transaction() as connection:
         connection.execute("UPDATE jobs SET request_json = ? WHERE id = ?", ("{", job.id))
@@ -46,7 +52,7 @@ 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)
+    job = repo.enqueue("membership_add", _membership_request("not a list"), False)
     repo.claim_next()
     repo.recover_running()
     assert repo.get(job.id).state == JobState.FAILED  # type: ignore[union-attr]
@@ -54,7 +60,7 @@ def test_recover_running_fails_invalid_source_shape(tmp_path: Path) -> None:
 
 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)
+    job = repo.enqueue("membership_add", _membership_request(), False)
     repo.claim_next()
     repo.add_file_error(job.id, JobFileError("gone.txt", "file_vanished", "File vanished"))
     repo.recover_running()
@@ -63,8 +69,27 @@ def test_recover_running_clears_errors_from_interrupted_attempt(tmp_path: Path)
 
 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)
+    job = repo.enqueue("membership_add", _membership_request(), 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]
+
+
+@pytest.mark.parametrize("kind,payload", [
+    ("rule_delete", {"version": 1}),
+    ("rule_patch", {"version": 1, "rule_id": "r1", "excluded_directories": "bad"}),
+    ("reindex", {"version": 1, "paths": "bad", "collections": []}),
+    ("collection_cleanup", {"version": 2, "paths": []}),
+    ("unknown", {"version": 1}),
+])
+def test_recovery_rejects_invalid_kind_payload(
+    tmp_path: Path, kind: str, payload: dict[str, object]
+) -> None:
+    repo = _repo(tmp_path)
+    job = repo.enqueue(kind, payload, False)
+    repo.claim_next()
+    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"