feat: expose membership and job APIs

docs/ARCHITECTURE.md

@@ -6,7 +6,8 @@ How `recoll-modern` is built today. Update in the same commit as any structural
 
 The Python service scaffold, SQLite schema, collection and membership repositories, document
 discovery/identity modules, durable job worker, engine adapter, authentication service, and initial
-HTTP shell are built. Membership, search and job resource routes remain to be added. Recoll behaviour
+HTTP shell, membership routes, and job routes are built. Search and document retrieval remain to be
+added. 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
@@ -31,7 +32,7 @@ web UI and (b) an MCP server, running in Docker, with several independent collec
 | `helper/` | C reader run by Recoll's `metadatacmds` for each indexed file |
 | `src/recoll_modern/engine/` | Shared Protocol, deterministic fake, and isolated Recoll Python/indexer adapter |
 | `src/recoll_modern/domain/auth.py`, `storage/auth.py` | Argon2 password checks, seeded administrator, hashed server-side sessions and CSRF verification |
-| `src/recoll_modern/api/` | HTTP auth, health, file browsing, collection metadata, prefix/proxy handling and JSON errors |
+| `src/recoll_modern/api/` | HTTP auth, health, file browsing, collection/membership/rule/job resources, prefix/proxy handling and JSON errors |
 | `tests/` | Automated service tests |
 
 Discovery produces content hashes and root-relative paths without writing to the document volume.
@@ -104,6 +105,10 @@ Core routes use a server-side session cookie and CSRF token; `GET /auth/session`
 the database only stores hashes. Forwarded scheme, host and prefix are applied only for configured
 proxy networks.
 
+Membership, rule, and reindex writes create versioned, durable jobs. The worker updates SQLite
+provenance, exports the membership map, and forces affected paths through the engine. Document listing
+derives effective membership from SQLite and includes direct and rule provenance.
+
 The resource groups are auth, health, read-only file browsing, collections and membership sources,
 directory rules, scoped search, document text/original retrieval, and indexing jobs. Search callers
 select returned fields; only `document_id` is unconditional, and inline full text is bounded. Every

docs/TODO.md

@@ -15,12 +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–10 add discovery, content-hash
+- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–11 add discovery, content-hash
   identity, additive membership, durable jobs, the atomic membership map, the engine boundary,
-  session authentication and the HTTP shell. Their checkboxes are current in the plan. The next task
-  is membership, rule and job routes.
+  session authentication, the HTTP shell, and membership/rule/job routes. Their checkboxes are current
+  in the plan. The next task is scoped search and document retrieval.
 
-The next action is Task 11, "Expose membership, rule, and job APIs." Do not repeat Tasks 1–10.
+The next action is Task 12, "Expose scoped search and document retrieval." Do not repeat Tasks 1–11.
 
 Things a resuming agent will otherwise get wrong:
 
@@ -49,7 +49,7 @@ Things a resuming agent will otherwise get wrong:
       then record the verified ones in `AGENTS.md`.
 - [ ] Dockerfile + compose: Recoll in the image, document volume mounted `:ro`, writable volumes for
       the index and SQLite.
-- [ ] Write the `metadatacmds` membership helper as a fast-starting non-Python program, fed by a flat
+- [x] Write the `metadatacmds` membership helper as a fast-starting non-Python program, fed by a flat
       map exported from SQLite. A `python3` helper made a 400-file index 9x slower; Recoll spawns it
       once per file per index run.
 - [ ] Re-measure indexing overhead against real documents (PDF, EPUB) rather than small text files —

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

@@ -787,7 +787,7 @@ git commit -m "feat: add authenticated HTTP service"
 - Consumes: Task 5 membership plans and Task 6 job repository
 - Produces: spec endpoints under `/collections/{id}/memberships`, `/directory-rules`, and `/jobs`
 
-- [ ] **Step 1: Write failing request/response tests**
+- [x] **Step 1: Write failing request/response tests**
 
 ```python
 def test_persistent_directory_request_enqueues_replayable_job(client: TestClient) -> None:
@@ -808,11 +808,11 @@ def test_removing_rule_preserves_overlapping_membership(client: TestClient) -> N
 
 Cover invalid source combinations, exclusions outside source, default best-effort, strict atomicity, effective provenance, per-file overrides, patching rule exclusions, non-retryable jobs, and 50/200 pagination.
 
-- [ ] **Step 2: Run tests and observe failure**
+- [x] **Step 2: Run tests and observe failure**
 
 Run: `uv run pytest tests/api/test_memberships.py tests/api/test_jobs.py -v`
 
-- [ ] **Step 3: Implement validated schemas and routes**
+- [x] **Step 3: Implement validated schemas and routes**
 
 Use discriminated Pydantic models `FileSource(type="file", path)` and
 `DirectorySource(type="directory", path, mode, excluded_directories)`. Reject mode/exclusions on files
@@ -823,7 +823,7 @@ effective membership plus `direct`, contributing rule IDs, and override state. I
 collection IDs and uses the whole document root when neither is present. Retry accepts failed or
 completed-with-errors jobs only.
 
-- [ ] **Step 4: Verify and commit**
+- [x] **Step 4: Verify and commit**
 
 Run: `uv run pytest tests/api/test_memberships.py tests/api/test_jobs.py -v && uv run ruff check . && uv run mypy src tests`
 

src/recoll_modern/api/collections.py

@@ -7,7 +7,7 @@ from fastapi import APIRouter, Request, Response
 from fastapi.encoders import jsonable_encoder
 
 from recoll_modern.api.dependencies import require_user
-from recoll_modern.api.schemas import CollectionCreateBody, CollectionPatchBody
+from recoll_modern.api.schemas import CollectionCreateBody, CollectionPatchBody, validated_sources
 from recoll_modern.domain.errors import AppError
 from recoll_modern.domain.models import UNSET, CollectionCreate, CollectionPatch
 
@@ -37,14 +37,18 @@ def create_collection(
     body: CollectionCreateBody, request: Request, response: Response
 ) -> dict[str, object]:
     require_user(request)
+    sources = (
+        validated_sources(request.app.state.path_policy, body.sources)
+        if body.sources else []
+    )
     collection = request.app.state.collections.create(
         CollectionCreate(body.name, body.description)
     )
     payload = jsonable_encoder(asdict(collection))
-    if body.sources:
+    if sources:
         job = request.app.state.jobs.enqueue(
             "membership_add",
-            {"collection_id": collection.id, "sources": body.sources},
+            {"version": 1, "collection_id": collection.id, "sources": sources},
             body.fail_on_file_error,
         )
         request.app.state.worker.notify()
@@ -76,7 +80,7 @@ def delete_collection(
         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", {"paths": paths}, False
+        "collection_cleanup", {"version": 1, "paths": paths}, False
     )
     request.app.state.worker.notify()
     return {"job_id": job.id}

src/recoll_modern/api/jobs.py

@@ -0,0 +1,78 @@
+"""Durable indexing-job HTTP resources."""
+
+from dataclasses import asdict
+from typing import cast
+
+from fastapi import APIRouter, Request
+from fastapi.encoders import jsonable_encoder
+
+from recoll_modern.api.dependencies import require_user
+from recoll_modern.api.schemas import ReindexBody
+from recoll_modern.domain.errors import AppError
+from recoll_modern.domain.jobs import Job
+from recoll_modern.indexing.paths import PathPolicyError
+
+router = APIRouter()
+
+
+def _job_payload(job: Job) -> dict[str, object]:
+    return cast(dict[str, object], jsonable_encoder(asdict(job)))
+
+
+@router.get("/jobs")
+def list_jobs(request: Request, limit: int = 50, offset: int = 0) -> dict[str, object]:
+    require_user(request)
+    if not 1 <= limit <= 200 or offset < 0:
+        raise AppError("validation_failed", "Invalid pagination", 422)
+    jobs, total = request.app.state.jobs.list_jobs(limit, offset)
+    return {"items": [_job_payload(job) for job in jobs], "total": total}
+
+
+@router.post("/jobs/reindex", status_code=202)
+def reindex(body: ReindexBody, request: Request) -> dict[str, object]:
+    require_user(request)
+    policy = request.app.state.path_policy
+    paths: list[str] = []
+    for path in body.paths or []:
+        try:
+            normalized = policy.normalize(path)
+            policy.resolve_file(normalized)
+        except PathPolicyError:
+            raise AppError("invalid_source", "Reindex path is invalid", 422) from None
+        paths.append(normalized)
+    for collection_id in body.collections or []:
+        request.app.state.collections.get(collection_id)
+    job = request.app.state.jobs.enqueue(
+        "reindex",
+        {"version": 1, "paths": sorted(set(paths)),
+         "collections": sorted(set(body.collections or []))},
+        body.fail_on_file_error,
+    )
+    request.app.state.worker.notify()
+    return {"job": _job_payload(job)}
+
+
+@router.get("/jobs/{job_id}")
+def get_job(job_id: str, request: Request) -> dict[str, object]:
+    require_user(request)
+    job = request.app.state.jobs.get(job_id)
+    if job is None:
+        raise AppError("job_not_found", "Job was not found", 404)
+    errors = request.app.state.jobs.file_errors(job_id)
+    return {"job": _job_payload(job), "file_errors": jsonable_encoder(errors)}
+
+
+@router.post("/jobs/{job_id}/retry", status_code=202)
+def retry_job(job_id: str, request: Request) -> dict[str, object]:
+    require_user(request)
+    job = request.app.state.jobs.get(job_id)
+    if job is None:
+        raise AppError("job_not_found", "Job was not found", 404)
+    try:
+        request.app.state.jobs.retry(job_id)
+    except ValueError:
+        raise AppError("job_not_retryable", "Job cannot be retried", 409) from None
+    request.app.state.worker.notify()
+    retried = request.app.state.jobs.get(job_id)
+    assert retried is not None
+    return {"job": _job_payload(retried)}

src/recoll_modern/api/memberships.py

@@ -0,0 +1,115 @@
+"""Collection membership, document provenance, and directory-rule routes."""
+
+from dataclasses import asdict
+from typing import cast
+
+from fastapi import APIRouter, Request
+from fastapi.encoders import jsonable_encoder
+
+from recoll_modern.api.dependencies import require_user
+from recoll_modern.api.schemas import (
+    DirectorySource,
+    MembershipBody,
+    RulePatchBody,
+    validated_sources,
+)
+from recoll_modern.domain.errors import AppError
+from recoll_modern.storage.membership import DirectoryRule
+
+router = APIRouter()
+
+
+def _rule(request: Request, collection_id: str, rule_id: str) -> DirectoryRule:
+    rule = cast(DirectoryRule | None, request.app.state.memberships.get_rule(rule_id))
+    if rule is None or rule.collection_id != collection_id:
+        raise AppError("directory_rule_not_found", "Directory rule was not found", 404)
+    return rule
+
+
+@router.post("/collections/{collection_id}/memberships", status_code=202)
+def add_memberships(
+    collection_id: str, body: MembershipBody, request: Request
+) -> dict[str, object]:
+    require_user(request)
+    request.app.state.collections.get(collection_id)
+    if not body.sources:
+        raise AppError("invalid_source", "At least one source is required", 422)
+    sources = validated_sources(request.app.state.path_policy, body.sources)
+    job = request.app.state.jobs.enqueue(
+        "membership_add",
+        {"version": 1, "collection_id": collection_id, "sources": sources},
+        body.fail_on_file_error,
+    )
+    request.app.state.worker.notify()
+    return {"job": jsonable_encoder(asdict(job))}
+
+
+@router.get("/collections/{collection_id}/documents")
+def list_documents(
+    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)
+    items, total = request.app.state.memberships.list_documents(collection_id, limit, offset)
+    return {"items": items, "total": total}
+
+
+@router.delete("/collections/{collection_id}/documents/{document_hash}", status_code=202)
+def remove_document(
+    collection_id: str, document_hash: str, request: Request
+) -> dict[str, object]:
+    require_user(request)
+    request.app.state.collections.get(collection_id)
+    if request.app.state.documents.get(document_hash) is None:
+        raise AppError("document_not_found", "Document was not found", 404)
+    job = request.app.state.jobs.enqueue(
+        "membership_remove",
+        {"version": 1, "collection_id": collection_id, "document_hash": document_hash},
+        False,
+    )
+    request.app.state.worker.notify()
+    return {"job": jsonable_encoder(asdict(job))}
+
+
+@router.get("/collections/{collection_id}/directory-rules")
+def list_rules(collection_id: str, request: Request) -> dict[str, object]:
+    require_user(request)
+    request.app.state.collections.get(collection_id)
+    rules = [
+        asdict(rule) for rule in request.app.state.memberships.list_rules()
+        if rule.collection_id == collection_id
+    ]
+    return {"items": rules, "total": len(rules)}
+
+
+@router.patch("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202)
+def patch_rule(
+    collection_id: str, rule_id: str, body: RulePatchBody, request: Request
+) -> dict[str, object]:
+    require_user(request)
+    rule = _rule(request, collection_id, rule_id)
+    validated = validated_sources(request.app.state.path_policy, [DirectorySource(
+        type="directory", path=rule.relative_path, mode="persistent",
+        excluded_directories=body.excluded_directories,
+    )])
+    job = request.app.state.jobs.enqueue(
+        "rule_patch",
+        {"version": 1, "rule_id": rule_id,
+         "excluded_directories": validated[0]["excluded_directories"]},
+        False,
+    )
+    request.app.state.worker.notify()
+    return {"job": jsonable_encoder(asdict(job))}
+
+
+@router.delete("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202)
+def delete_rule(collection_id: str, rule_id: str, request: Request) -> dict[str, object]:
+    require_user(request)
+    _rule(request, collection_id, rule_id)
+    job = request.app.state.jobs.enqueue(
+        "rule_delete", {"version": 1, "rule_id": rule_id}, False
+    )
+    request.app.state.worker.notify()
+    return {"job": jsonable_encoder(asdict(job))}

src/recoll_modern/api/schemas.py

@@ -1,15 +1,83 @@
 """Small HTTP request bodies for collection metadata."""
 
-from pydantic import BaseModel
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from recoll_modern.domain.errors import AppError
+from recoll_modern.indexing.paths import PathPolicy, PathPolicyError
+
+
+class FileSource(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    type: Literal["file"]
+    path: str
+
+
+class DirectorySource(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    type: Literal["directory"]
+    path: str
+    mode: Literal["snapshot", "persistent"]
+    excluded_directories: list[str] = Field(default_factory=list)
+
+
+type SourceBody = Annotated[FileSource | DirectorySource, Field(discriminator="type")]
+
+
+class MembershipBody(BaseModel):
+    sources: list[SourceBody]
+    fail_on_file_error: bool = False
+
+
+class RulePatchBody(BaseModel):
+    excluded_directories: list[str]
+
+
+class ReindexBody(BaseModel):
+    paths: list[str] | None = None
+    collections: list[str] | None = None
+    fail_on_file_error: bool = False
 
 
 class CollectionCreateBody(BaseModel):
     name: str
     description: str | None = None
-    sources: list[dict[str, object]] | None = None
+    sources: list[SourceBody] | None = None
     fail_on_file_error: bool = False
 
 
 class CollectionPatchBody(BaseModel):
     name: str | None = None
     description: str | None = None
+
+
+def validated_sources(
+    policy: PathPolicy, sources: list[FileSource | DirectorySource]
+) -> list[dict[str, object]]:
+    """Normalize sources and reject invalid or repeated root-relative selections."""
+    result: list[dict[str, object]] = []
+    seen: set[tuple[str, str]] = set()
+    for source in sources:
+        try:
+            path = policy.normalize(source.path)
+            if source.type == "file":
+                policy.resolve_file(path)
+                item: dict[str, object] = {"type": "file", "path": path}
+            else:
+                policy.resolve_directory(path)
+                excluded = [policy.normalize(value) for value in source.excluded_directories]
+                if any(not value.startswith(path + "/") for value in excluded):
+                    raise PathPolicyError("invalid_source")
+                item = {
+                    "type": "directory", "path": path, "mode": source.mode,
+                    "excluded_directories": excluded,
+                }
+        except PathPolicyError:
+            raise AppError("invalid_source", "Membership source is invalid", 422) from None
+        key = (source.type, path)
+        if key in seen:
+            raise AppError("invalid_source", "Membership source is duplicated", 422)
+        seen.add(key)
+        result.append(item)
+    return result

src/recoll_modern/app.py

@@ -8,12 +8,15 @@ from datetime import timedelta
 from fastapi import FastAPI
 
 from recoll_modern.api import auth, collections, files, health
+from recoll_modern.api import jobs as job_routes
+from recoll_modern.api import memberships as membership_routes
 from recoll_modern.api.errors import install_error_handlers
 from recoll_modern.api.middleware import PrefixMiddleware
 from recoll_modern.config import Settings
 from recoll_modern.domain.auth import AuthService
 from recoll_modern.domain.collections import CollectionService
 from recoll_modern.domain.jobs import Job
+from recoll_modern.domain.membership import MembershipService
 from recoll_modern.engine.protocol import Engine
 from recoll_modern.engine.recoll import RecollEngine
 from recoll_modern.indexing.handlers import MembershipJobHandler
@@ -53,6 +56,12 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
         jobs = JobRepository(database)
         map_rows = MembershipMapRows(memberships, policy.root)
         export_membership_map(map_rows, settings.membership_map_path)
+        membership_service = MembershipService(policy, documents, memberships)
+
+        def reindex_changed(paths: tuple[str, ...]) -> None:
+            export_membership_map(map_rows, settings.membership_map_path)
+            if paths:
+                engine.force_reindex(tuple(str(policy.root / path) for path in sorted(set(paths))))
 
         def cleanup(job: Job) -> None:
             export_membership_map(map_rows, settings.membership_map_path)
@@ -71,6 +80,50 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
             if changed:
                 engine.force_reindex(tuple(str(policy.root / path) for path in sorted(changed)))
 
+        def remove_membership(job: Job) -> None:
+            collection_id = job.request.get("collection_id")
+            content_hash = job.request.get("document_hash")
+            if not isinstance(collection_id, str) or not isinstance(content_hash, str):
+                raise ValueError("invalid membership removal job")
+            changed = membership_service.plan_remove_document(collection_id, content_hash)
+            reindex_changed(changed)
+
+        def patch_rule(job: Job) -> None:
+            rule_id = job.request.get("rule_id")
+            exclusions = job.request.get("excluded_directories")
+            if not isinstance(rule_id, str) or not isinstance(exclusions, list):
+                raise ValueError("invalid rule patch job")
+            if not all(isinstance(item, str) for item in exclusions):
+                raise ValueError("invalid rule patch job")
+            result = membership_service.plan_patch_rule(rule_id, tuple(exclusions))
+            reindex_changed(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))
+
+        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)
+            export_membership_map(map_rows, settings.membership_map_path)
+            selected = {
+                str(policy.root / path) for path in paths if isinstance(path, str)
+            }
+            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):
+                    selected.add(path)
+            if not selected and not paths and not collections_requested:
+                selected.add(str(policy.root))
+            if selected:
+                engine.force_reindex(tuple(sorted(selected)))
+
         membership_handler = MembershipJobHandler(
             policy, documents, memberships, jobs, settings.membership_map_path, engine
         )
@@ -78,6 +131,10 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
             "membership_add": membership_handler.handle,
             "collection_cleanup": cleanup,
             "reconcile_rules": reconcile_all,
+            "membership_remove": remove_membership,
+            "rule_patch": patch_rule,
+            "rule_delete": delete_rule,
+            "reindex": reindex,
         })
         app.state.settings = settings
         app.state.database = database
@@ -90,7 +147,7 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
         app.state.jobs = jobs
         app.state.worker = worker
         if memberships.list_rules():
-            jobs.enqueue("reconcile_rules", {}, False)
+            jobs.enqueue("reconcile_rules", {"version": 1}, False)
         worker.start()
         try:
             yield
@@ -104,4 +161,6 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) -
     app.include_router(health.router)
     app.include_router(files.router)
     app.include_router(collections.router)
+    app.include_router(membership_routes.router)
+    app.include_router(job_routes.router)
     return app

src/recoll_modern/storage/jobs.py

@@ -1,5 +1,7 @@
 """Atomic SQLite transitions for indexing jobs."""
 
+from __future__ import annotations
+
 import json
 import secrets
 import sqlite3
@@ -42,6 +44,15 @@ class JobRepository:
             row = connection.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
             return _job_from_row(row) if row is not None else None
 
+    def list_jobs(self, limit: int, offset: int) -> tuple[list[Job], int]:
+        with self._database.connect() as connection:
+            total = int(connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0])
+            rows = connection.execute(
+                "SELECT * FROM jobs ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?",
+                (limit, offset),
+            ).fetchall()
+            return [_job_from_row(row) for row in rows], total
+
     def claim_next(self) -> Job | None:
         with self._database.transaction() as connection:
             row = connection.execute(
@@ -130,7 +141,7 @@ class JobRepository:
             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'",
+                "WHERE id = ? AND state IN ('failed', 'completed_with_errors')",
                 (job_id,),
             )
             if cursor.rowcount != 1:

src/recoll_modern/storage/membership.py

@@ -195,6 +195,58 @@ class MembershipRepository:
                 if (collections := _effective(connection, str(row[1])))
             ]
 
+    def list_documents(
+        self, collection_id: str, limit: int, offset: int
+    ) -> tuple[list[dict[str, object]], int]:
+        with self._database.connect() as connection:
+            hashes = [
+                str(row[0])
+                for row in connection.execute(
+                    "SELECT content_hash FROM documents ORDER BY content_hash"
+                )
+                if collection_id in _effective(connection, str(row[0]))
+            ]
+            items: list[dict[str, object]] = []
+            for content_hash in hashes[offset:offset + limit]:
+                doc = connection.execute(
+                    "SELECT byte_size, state FROM documents WHERE content_hash = ?",
+                    (content_hash,),
+                ).fetchone()
+                canonical = connection.execute(
+                    "SELECT relative_path FROM document_paths "
+                    "WHERE document_hash = ? AND is_canonical = 1",
+                    (content_hash,),
+                ).fetchone()
+                direct = connection.execute(
+                    "SELECT 1 FROM direct_memberships "
+                    "WHERE collection_id = ? AND document_hash = ?",
+                    (collection_id, content_hash),
+                ).fetchone() is not None
+                rules = [
+                    str(row[0])
+                    for row in connection.execute(
+                        "SELECT r.id FROM rule_memberships AS rm "
+                        "JOIN directory_rules AS r ON r.id = rm.rule_id "
+                        "WHERE r.collection_id = ? AND rm.document_hash = ? ORDER BY r.id",
+                        (collection_id, content_hash),
+                    )
+                ]
+                override = connection.execute(
+                    "SELECT 1 FROM membership_exclusions "
+                    "WHERE collection_id = ? AND document_hash = ?",
+                    (collection_id, content_hash),
+                ).fetchone() is not None
+                items.append({
+                    "document_id": content_hash,
+                    "path": str(canonical[0]) if canonical is not None else None,
+                    "size": int(doc[0]),
+                    "state": str(doc[1]),
+                    "direct": direct,
+                    "contributing_rules": rules,
+                    "override": override,
+                })
+            return items, len(hashes)
+
 
 def _effective(connection: sqlite3.Connection, content_hash: str) -> list[str]:
     return [

tests/api/test_jobs.py

@@ -0,0 +1,34 @@
+import time
+
+from fastapi.testclient import TestClient
+
+
+def test_job_reindex_enqueues_versioned_request(authenticated: tuple[TestClient, str]) -> None:
+    client, csrf = authenticated
+    response = client.post("/jobs/reindex", json={}, headers={"X-CSRF-Token": csrf})
+    assert response.status_code == 202
+    job = response.json()["job"]
+    assert job["request"]["version"] == 1
+    assert job["fail_on_file_error"] is False
+    listed = client.get("/jobs").json()
+    assert listed["total"] == 1
+
+
+def test_retry_refuses_completed_job(authenticated: tuple[TestClient, str]) -> None:
+    client, csrf = authenticated
+    response = client.post("/jobs/reindex", json={}, headers={"X-CSRF-Token": csrf})
+    job_id = response.json()["job"]["id"]
+    for _ in range(100):
+        if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed":
+            break
+        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"
+
+
+def test_job_list_validates_pagination(authenticated: tuple[TestClient, str]) -> None:
+    client, _ = authenticated
+    response = client.get("/jobs", params={"limit": 201})
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "validation_failed"

tests/api/test_memberships.py

@@ -0,0 +1,179 @@
+import time
+from pathlib import Path
+from typing import cast
+
+from fastapi.testclient import TestClient
+
+
+def _collection(client: TestClient, csrf: str) -> str:
+    response = client.post(
+        "/collections", json={"name": "Books"}, headers={"X-CSRF-Token": csrf}
+    )
+    return cast(str, response.json()["id"])
+
+
+def _wait(client: TestClient, job_id: str) -> dict[str, object]:
+    for _ in range(100):
+        job = client.get(f"/jobs/{job_id}").json()["job"]
+        if job["state"] not in ("queued", "running"):
+            return cast(dict[str, object], job)
+        time.sleep(0.01)
+    raise AssertionError("job did not finish")
+
+
+def test_persistent_directory_request_enqueues_replayable_job(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    books = tmp_path / "docs" / "books"
+    books.mkdir()
+    (books / "a.txt").write_text("a")
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent",
+                           "excluded_directories": []}]},
+    )
+    assert response.status_code == 202
+    assert response.json()["job"]["state"] == "queued"
+    job_id = response.json()["job"]["id"]
+    assert _wait(client, job_id)["state"] == "completed"
+    rules = client.get(f"/collections/{collection_id}/directory-rules").json()["items"]
+    assert len(rules) == 1
+    members = client.get(f"/collections/{collection_id}/documents").json()["items"]
+    assert len(members) == 1
+    assert members[0]["contributing_rules"] == [rules[0]["id"]]
+
+
+def test_invalid_file_source_combination_is_rejected(
+    authenticated: tuple[TestClient, str]
+) -> None:
+    client, csrf = authenticated
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "file", "path": "a.txt", "mode": "persistent"}]},
+    )
+    assert response.status_code == 422
+
+
+def test_duplicate_normalized_sources_are_rejected(
+    authenticated: tuple[TestClient, str]
+) -> None:
+    client, csrf = authenticated
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [
+            {"type": "file", "path": "books/a.txt"},
+            {"type": "file", "path": "books//a.txt"},
+        ]},
+    )
+    assert response.status_code == 422
+
+
+def test_strict_job_keeps_membership_empty_on_file_error(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    (tmp_path / "docs" / "good.txt").write_text("good")
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [
+            {"type": "file", "path": "good.txt"},
+            {"type": "file", "path": "gone.txt"},
+        ], "fail_on_file_error": True},
+    )
+    assert _wait(client, response.json()["job"]["id"])["state"] == "failed"
+    assert client.get(f"/collections/{collection_id}/documents").json()["items"] == []
+
+
+def test_removing_one_rule_preserves_overlapping_membership(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    root = tmp_path / "docs"
+    for name in ("books", "other"):
+        directory = root / name
+        directory.mkdir()
+        (directory / "a.txt").write_text("same")
+    collection_id = _collection(client, csrf)
+    for name in ("books", "other"):
+        response = client.post(
+            f"/collections/{collection_id}/memberships",
+            headers={"X-CSRF-Token": csrf},
+            json={"sources": [{"type": "directory", "path": name, "mode": "persistent"}]},
+        )
+        assert _wait(client, response.json()["job"]["id"])["state"] == "completed"
+    rules = client.get(f"/collections/{collection_id}/directory-rules").json()["items"]
+    response = client.delete(
+        f"/collections/{collection_id}/directory-rules/{rules[0]['id']}",
+        headers={"X-CSRF-Token": csrf},
+    )
+    assert _wait(client, response.json()["job"]["id"])["state"] == "completed"
+    members = client.get(f"/collections/{collection_id}/documents").json()["items"]
+    assert len(members) == 1
+    assert members[0]["contributing_rules"] == [rules[1]["id"]]
+
+
+def test_rule_exclusion_patch_removes_rule_membership(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    skipped = tmp_path / "docs" / "books" / "skip"
+    skipped.mkdir(parents=True)
+    (skipped / "a.txt").write_text("a")
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]},
+    )
+    assert _wait(client, response.json()["job"]["id"])["state"] == "completed"
+    rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"]
+    patch = client.patch(
+        f"/collections/{collection_id}/directory-rules/{rule_id}",
+        headers={"X-CSRF-Token": csrf},
+        json={"excluded_directories": ["books/skip"]},
+    )
+    assert _wait(client, patch.json()["job"]["id"])["state"] == "completed"
+    assert client.get(f"/collections/{collection_id}/documents").json()["items"] == []
+
+
+def test_remove_document_creates_override_then_direct_add_restores_it(
+    authenticated: tuple[TestClient, str], tmp_path: Path
+) -> None:
+    client, csrf = authenticated
+    books = tmp_path / "docs" / "books"
+    books.mkdir()
+    (books / "a.txt").write_text("a")
+    collection_id = _collection(client, csrf)
+    response = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]},
+    )
+    assert _wait(client, response.json()["job"]["id"])["state"] == "completed"
+    documents = client.get(f"/collections/{collection_id}/documents").json()["items"]
+    document_id = documents[0]["document_id"]
+    removed = client.delete(
+        f"/collections/{collection_id}/documents/{document_id}",
+        headers={"X-CSRF-Token": csrf},
+    )
+    assert _wait(client, removed.json()["job"]["id"])["state"] == "completed"
+    assert client.get(f"/collections/{collection_id}/documents").json()["items"] == []
+    added = client.post(
+        f"/collections/{collection_id}/memberships",
+        headers={"X-CSRF-Token": csrf},
+        json={"sources": [{"type": "file", "path": "books/a.txt"}]},
+    )
+    assert _wait(client, added.json()["job"]["id"])["state"] == "completed"
+    members = client.get(f"/collections/{collection_id}/documents").json()["items"]
+    assert len(members) == 1
+    assert members[0]["direct"] is True
+    assert members[0]["override"] is False