feat(documents-ingest): Paperless -> envelope adapter (module 5 phase 2 step 5)

Adds documents-ingest-paperless: paginated GET /api/documents/, maps each
doc to a source='paperless' envelope per plan §4.2-4.3, reusing
kb_mail.Envelope/insert_envelope unchanged (packages/kb-mail not touched).
Cross-source link (source_mail entity) is a deterministic join of
original_file_name against the phase-1 registry.json consume_name index —
no heuristics, no correspondent guessing (plan decision 4). Stats always
balance (fetched = already_in_db + inserted + errors) and main() now also
exits non-zero on imbalance, not just on errors>0, matching the exit-code
convention already established in gmail-bulk-import.

Verified live on PIHA (rsync to /tmp, ~/kb/venv, PIHA checkout untouched):
dry-run then --apply inserted 186/186 paperless envelopes (0 errors,
180 source_mail links), a second --apply reported inserted=0/already_in_db=186
(idempotent), gmail rows stayed at 225030 and document_chunk stayed empty.
Rotated the kb-ingest Paperless API token after it was accidentally
partially echoed during recon (old token now dead).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

jobs/documents-ingest/README.md

@@ -181,3 +181,123 @@ Covers: filename sanitization, consume-name collision handling, manifest
 filtering, MIME PDF-part extraction (including the RFC 2047 decoding
 mismatch), sha256 match/mismatch, duplicate detection, dry-run vs `--apply`
 behavior, and multi-attachment envelopes.
+
+---
+
+## Phase 2 — `documents-ingest-paperless` (Paperless -> envelope adapter)
+
+Module 5, phase 2 (`docs/kb/modules/05-faza2-plan.md`, §4.2-4.3, §6 step 5).
+Reads documents from the **Paperless REST API** (read-only — GET only, never
+writes to Paperless) and inserts them as `source='paperless'` rows into the
+`envelope` table on kb-postgres, reusing `kb_mail.envelope.Envelope` /
+`kb_mail.db.insert_envelope` from `packages/kb-mail` (untouched by this
+change — see plan §1.6). Existing `source='gmail'` rows and `document_chunk`
+are never touched; this job only ever `INSERT`s new `paperless` rows.
+
+### Cross-source link (`source_mail`)
+
+Per plan §1.9/§4.2, the deterministic join uses no heuristics: a document's
+`original_file_name` (from the Paperless API) is matched against
+`consume_name` in this job's **phase-1 registry**
+(`/opt/homelab/data/documents-ingest/registry.json`, produced by
+`extractor.py` — see above). A match appends a `source_mail` entity pointing
+back at the originating mail envelope; no match means the document was added
+outside the faktury-1 pipeline, and the entity is simply omitted — not an
+error.
+
+### Install
+
+```bash
+pip install -e packages/kb-mail/
+pip install -e jobs/documents-ingest/
+```
+
+### Usage
+
+```bash
+# Dry run (default) — fetch from Paperless, map, count; no DB writes:
+documents-ingest-paperless --dsn postgresql://kb:<pw>@localhost:5433/kb \
+    --paperless-token <token>
+
+# Real run — insert new envelope rows:
+documents-ingest-paperless --dsn ... --paperless-token ... --apply
+
+# Smoke-test slice:
+documents-ingest-paperless --dsn ... --paperless-token ... --limit 5
+```
+
+`--dsn` can come from `KB_DSN`, `--paperless-token` from `PAPERLESS_API_TOKEN`,
+`--paperless-url` from `PAPERLESS_URL` (defaults to Paperless' fixed LAN
+address, `http://192.168.31.5:8210`). No `--offset`: unlike the 225 030-row
+header backfill, a full re-scan of Paperless' ~186 documents is cheap and
+already idempotent, so there is no need for resumable partitioning — `--limit`
+exists only to cap a run for smoke-testing.
+
+### Mapping (plan §4.3)
+
+```
+id       = f"paperless:{document_id}"   -- prefixed: Paperless doc-ids are small
+                                         -- sequential ints that would otherwise
+                                         -- collide with any future source's ids
+ts       = documents_document.created   -- Paperless-detected date (content/filename),
+                                         -- not filesystem mtime
+geo      = NULL
+raw_ref  = str(document_id)             -- REFERENCE — Paperless is the source of truth,
+                                         -- no bytes are copied
+entities = content, correspondent, tag(s), filename, content_type,
+           and source_mail when the registry join hits (plan §4.2)
+```
+
+`correspondent`/`tag` are resolved from Paperless' `/api/correspondents/` and
+`/api/tags/` (fetched once, cached in memory for the run) and kept purely as
+informational metadata — nothing in this pipeline depends on them being
+non-null (plan decision 4). A document with empty OCR content (Paperless OCR
+sometimes produces none) still gets a normal envelope with `"text": ""` — not
+skipped, not an error, just counted (`empty_content`).
+
+### Idempotency
+
+A pre-fetched set of existing `source='paperless'` envelope ids (one query at
+the start of each run) skips documents already inserted; `insert_envelope`'s
+own `ON CONFLICT (id) DO NOTHING` is the second line of defense. Re-running
+`--apply` immediately after a successful run reports `inserted: 0` and
+`already_in_db` equal to the previous run's `inserted` count.
+
+### Stats must balance
+
+```
+fetched = already_in_db + inserted + errors
+```
+
+`source_mail_linked` and `empty_content` are informational subsets of
+`fetched`, not separate outcome buckets. A per-document mapping failure
+(e.g. an unparseable `created` date) is isolated, logged, and counted as
+`errors` — it never aborts the run. `main()` exits 1 on non-zero `errors`
+or if the balance invariant above doesn't hold (mirrors
+`gmail-bulk-import`'s exit-code convention) — a clean run always exits 0.
+
+### Tests
+
+```bash
+pip install -e packages/kb-mail/
+pip install -e jobs/documents-ingest/
+cd jobs/documents-ingest && pytest
+```
+
+Pure unit tests, no DB or real HTTP — `run()` is tested by monkeypatching
+`asyncpg.connect` (fake connection) and `aiohttp.ClientSession` (fake session
+serving canned JSON pages). Covers: mapping shape (content, correspondent,
+tag(s), filename, content_type, source_mail), the registry join (hit and
+miss), pagination (both the documents list and the correspondents/tags lookup
+tables), `--limit`, idempotency (pre-existing ids skipped, a second `--apply`
+run inserts nothing new), isolated per-document mapping errors, and the
+stats-balance invariant.
+
+### Definition of Done
+
+Per `CLAUDE.md`: smoke run is `documents-ingest-paperless --dsn ...
+--paperless-token ... --limit 5` (dry-run first) against kb-postgres@PIHA and
+the live Paperless API, over SSH — **not executed as part of this change**
+without operator confirmation (this job reads production Paperless data and
+writes production envelope rows on `--apply`). `pytest` passes locally before
+this commit.

jobs/documents-ingest/pyproject.toml

@@ -9,10 +9,13 @@ requires-python = ">=3.11"
 dependencies = [
     "asyncpg>=0.29",
     "structlog>=24.1",
+    "aiohttp>=3.9",
+    "kb-mail",
 ]
 
 [project.scripts]
 documents-ingest = "documents_ingest.extractor:main"
+documents-ingest-paperless = "documents_ingest.paperless_adapter:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]

jobs/documents-ingest/src/documents_ingest/paperless_adapter.py

@@ -0,0 +1,302 @@
+"""Paperless -> envelope adapter — module 5 phase 2 (docs/kb/modules/05-faza2-plan.md,
+§4.2-4.3, §6 step 5).
+
+Reads documents from the Paperless REST API (read-only — GET only, this job never
+writes to Paperless) and inserts them as `source='paperless'` rows into the `envelope`
+table on kb-postgres, reusing `kb_mail.envelope.Envelope` / `kb_mail.db.insert_envelope`
+(packages/kb-mail, untouched by this change — see plan §1.6). Existing `source='gmail'`
+rows and `document_chunk` are never touched.
+
+Cross-source proof (plan §1.9, §4.2): when a document's `original_file_name` matches a
+`consume_name` in the documents-ingest registry.json (produced by phase 1,
+jobs/documents-ingest/extractor.py), a `source_mail` entity links the document envelope
+back to the originating mail envelope. No match = no `source_mail` entity — a normal,
+expected outcome for documents added outside the faktury-1 pipeline.
+
+Runs on PIHA (needs a route to both the Paperless API and kb-postgres):
+
+Install (from repo root):
+    pip install -e packages/kb-mail/
+    pip install -e jobs/documents-ingest/
+
+Usage:
+    # Dry run (default) — fetch, map, count; no DB writes:
+    documents-ingest-paperless --dsn postgresql://kb:<pw>@localhost:5433/kb \\
+        --paperless-token <token>
+
+    # Real run:
+    documents-ingest-paperless --dsn ... --paperless-token ... --apply
+
+DSN can come from KB_DSN, token from PAPERLESS_API_TOKEN, URL from PAPERLESS_URL
+(defaults to Paperless' fixed LAN address per services/paperless/env.example).
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import AsyncIterator, Optional
+
+import aiohttp
+import asyncpg
+import structlog
+
+from kb_mail.db import insert_envelope
+from kb_mail.envelope import Envelope
+
+from .extractor import DEFAULT_REGISTRY, load_registry
+
+_log = structlog.get_logger(__name__)
+
+DEFAULT_PAPERLESS_URL = "http://192.168.31.5:8210"
+DEFAULT_PAGE_SIZE = 200
+
+
+def parse_document_ts(value: str) -> datetime:
+    """Parse Paperless `created` (ISO 8601, always carries an offset) to UTC-aware."""
+    dt = datetime.fromisoformat(value)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def build_registry_index(registry: dict[str, dict]) -> dict[str, dict]:
+    """Invert the sha256-keyed registry to consume_name -> {envelope_id, sha256, filename}.
+
+    consume_name (== Paperless `original_file_name` for faktury-1 uploads) is unique in
+    the registry (verified in plan §1.9), so this is a safe 1:1 index for the join.
+    """
+    return {
+        entry["consume_name"]: {
+            "envelope_id": entry["envelope_id"],
+            "sha256": sha256,
+            "filename": entry["filename"],
+        }
+        for sha256, entry in registry.items()
+    }
+
+
+def find_source_mail(original_file_name: Optional[str], registry_index: dict[str, dict]) -> Optional[dict]:
+    """Deterministic cross-source join: original_file_name == registry consume_name (plan §1.9).
+
+    Returns None when there is no match — a normal outcome for documents added outside
+    the faktury-1 pipeline (plan §4.2), never guessed via filename similarity or content.
+    """
+    if not original_file_name:
+        return None
+    return registry_index.get(original_file_name)
+
+
+def build_entities(
+    doc: dict,
+    correspondents: dict[int, Optional[str]],
+    tags: dict[int, Optional[str]],
+    source_mail: Optional[dict],
+) -> list[dict]:
+    """Build the `entities[]` list for one document, per plan §4.2."""
+    entities: list[dict] = [
+        {"type": "content", "text": doc.get("content") or ""},
+        {"type": "correspondent", "name": correspondents.get(doc.get("correspondent"))},
+    ]
+    for tag_id in doc.get("tags") or []:
+        entities.append({"type": "tag", "name": tags.get(tag_id)})
+    entities.append({"type": "filename", "value": doc.get("original_file_name")})
+    entities.append({"type": "content_type", "value": doc.get("mime_type")})
+    if source_mail is not None:
+        entities.append({
+            "type": "source_mail",
+            "envelope_id": source_mail["envelope_id"],
+            "attachment_sha256": source_mail["sha256"],
+            "attachment_filename": source_mail["filename"],
+        })
+    return entities
+
+
+def map_document(
+    doc: dict,
+    correspondents: dict[int, Optional[str]],
+    tags: dict[int, Optional[str]],
+    source_mail: Optional[dict],
+) -> Envelope:
+    """Map one Paperless API document record to a KB envelope (plan §4.2-4.3)."""
+    return Envelope(
+        id=f"paperless:{doc['id']}",
+        source="paperless",
+        ts=parse_document_ts(doc["created"]),
+        raw_ref=str(doc["id"]),
+        geo=None,
+        entities=build_entities(doc, correspondents, tags, source_mail),
+    )
+
+
+async def fetch_lookup(session: aiohttp.ClientSession, base_url: str, path: str) -> dict[int, Optional[str]]:
+    """Fetch a paginated Paperless id->name table (correspondents, tags) into a dict."""
+    result: dict[int, Optional[str]] = {}
+    url: Optional[str] = f"{base_url}{path}"
+    while url:
+        async with session.get(url) as resp:
+            resp.raise_for_status()
+            data = await resp.json()
+        for item in data.get("results", []):
+            result[item["id"]] = item.get("name")
+        url = data.get("next")
+    return result
+
+
+async def iter_documents(
+    session: aiohttp.ClientSession,
+    base_url: str,
+    page_size: int = DEFAULT_PAGE_SIZE,
+    limit: Optional[int] = None,
+) -> AsyncIterator[dict]:
+    """Yield every document from `GET /api/documents/`, following pagination.
+
+    `limit` caps the total number yielded (across pages) — for smoke-testing a slice
+    without pulling the whole collection. Full re-runs are cheap and idempotent (~186
+    docs today), so there is no --offset: unlike the 225k-row header backfill, this job
+    does not need resumable partitioning.
+    """
+    url: Optional[str] = f"{base_url}/api/documents/?page_size={page_size}&ordering=id"
+    count = 0
+    while url:
+        async with session.get(url) as resp:
+            resp.raise_for_status()
+            data = await resp.json()
+        for doc in data.get("results", []):
+            if limit is not None and count >= limit:
+                return
+            yield doc
+            count += 1
+        url = data.get("next")
+
+
+async def run(
+    dsn: str,
+    paperless_url: str,
+    paperless_token: str,
+    registry_path: Path = DEFAULT_REGISTRY,
+    limit: Optional[int] = None,
+    page_size: int = DEFAULT_PAGE_SIZE,
+    apply: bool = False,
+) -> dict[str, int]:
+    """Fetch Paperless documents, map to envelopes, and (if apply) insert new ones.
+
+    Returns stats that must always balance:
+
+        fetched = already_in_db + inserted + errors
+
+    `source_mail_linked` and `empty_content` are informational subsets of `fetched`,
+    not separate outcome buckets. dry-run (apply=False) never writes — `inserted`
+    reports what *would* be written, mirroring documents-ingest's extractor.py.
+    Idempotent via a pre-fetched set of existing `source='paperless'` envelope ids
+    (insert_envelope's own ON CONFLICT DO NOTHING is the second line of defense).
+    """
+    stats = {
+        "fetched": 0,
+        "already_in_db": 0,
+        "inserted": 0,
+        "source_mail_linked": 0,
+        "empty_content": 0,
+        "errors": 0,
+    }
+
+    registry_index = build_registry_index(load_registry(registry_path))
+
+    conn = await asyncpg.connect(dsn)
+    try:
+        existing_ids = {
+            r["id"] for r in await conn.fetch("SELECT id FROM envelope WHERE source = 'paperless'")
+        }
+
+        headers = {"Authorization": f"Token {paperless_token}"}
+        async with aiohttp.ClientSession(headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as session:
+            correspondents = await fetch_lookup(session, paperless_url, "/api/correspondents/")
+            tags = await fetch_lookup(session, paperless_url, "/api/tags/")
+
+            async for doc in iter_documents(session, paperless_url, page_size=page_size, limit=limit):
+                stats["fetched"] += 1
+
+                try:
+                    source_mail = find_source_mail(doc.get("original_file_name"), registry_index)
+                    env = map_document(doc, correspondents, tags, source_mail)
+                except Exception:
+                    _log.warning("skip.map_error", document_id=doc.get("id"), exc_info=True)
+                    stats["errors"] += 1
+                    continue
+
+                if not (doc.get("content") or "").strip():
+                    stats["empty_content"] += 1
+                if source_mail is not None:
+                    stats["source_mail_linked"] += 1
+
+                if env.id in existing_ids:
+                    stats["already_in_db"] += 1
+                    continue
+
+                if apply:
+                    await insert_envelope(conn, env)
+                    existing_ids.add(env.id)
+                stats["inserted"] += 1
+    finally:
+        await conn.close()
+
+    balance = stats["already_in_db"] + stats["inserted"] + stats["errors"]
+    if balance != stats["fetched"]:
+        _log.error("stats_mismatch", fetched=stats["fetched"], balance=balance, **stats)
+
+    _log.info("run_complete", apply=apply, **stats)
+    return stats
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Adapt Paperless documents into KB envelopes (module 5, phase 2 — plan §4.2-4.3)."
+    )
+    parser.add_argument("--dsn", default=os.environ.get("KB_DSN"),
+                        help="asyncpg DSN for kb-postgres (or set KB_DSN env var)")
+    parser.add_argument("--paperless-url", default=os.environ.get("PAPERLESS_URL", DEFAULT_PAPERLESS_URL),
+                        help=f"Paperless base URL (default: {DEFAULT_PAPERLESS_URL}, or set PAPERLESS_URL)")
+    parser.add_argument("--paperless-token", default=os.environ.get("PAPERLESS_API_TOKEN"),
+                        help="Paperless API token (or set PAPERLESS_API_TOKEN env var)")
+    parser.add_argument("--registry", type=Path, default=DEFAULT_REGISTRY,
+                        help=f"documents-ingest phase-1 registry JSON (default: {DEFAULT_REGISTRY})")
+    parser.add_argument("--limit", type=int, default=None,
+                        help="Max documents to process (default: all)")
+    parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE,
+                        help=f"Paperless API page size (default: {DEFAULT_PAGE_SIZE})")
+    parser.add_argument("--apply", action="store_true",
+                        help="Actually insert envelope rows. Default is dry-run.")
+    args = parser.parse_args()
+
+    if not args.dsn:
+        _log.error("missing_dsn", hint="pass --dsn or set KB_DSN")
+        sys.exit(1)
+    if not args.paperless_token:
+        _log.error("missing_token", hint="pass --paperless-token or set PAPERLESS_API_TOKEN")
+        sys.exit(1)
+
+    stats = asyncio.run(
+        run(
+            dsn=args.dsn,
+            paperless_url=args.paperless_url,
+            paperless_token=args.paperless_token,
+            registry_path=args.registry,
+            limit=args.limit,
+            page_size=args.page_size,
+            apply=args.apply,
+        )
+    )
+
+    mode = "APPLY" if args.apply else "DRY-RUN"
+    _log.info("summary", mode=mode, **stats)
+
+    balanced = stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"]
+    failed = stats["errors"] > 0 or not balanced
+    sys.exit(1 if failed else 0)
+
+
+if __name__ == "__main__":
+    main()

jobs/documents-ingest/tests/test_paperless_adapter.py

@@ -0,0 +1,424 @@
+"""Unit tests for the Paperless -> envelope adapter — no DB, no real HTTP."""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+
+import pytest
+
+from documents_ingest.paperless_adapter import (
+    build_entities,
+    build_registry_index,
+    find_source_mail,
+    fetch_lookup,
+    iter_documents,
+    map_document,
+    parse_document_ts,
+    run,
+)
+
+BASE_URL = "http://fake-paperless"
+
+
+def _doc(
+    doc_id=1,
+    created="2020-04-21T00:00:00+02:00",
+    content="some ocr text",
+    correspondent=None,
+    tags=None,
+    original_file_name="file.pdf",
+    mime_type="application/pdf",
+):
+    return {
+        "id": doc_id,
+        "correspondent": correspondent,
+        "tags": tags or [],
+        "content": content,
+        "created": created,
+        "original_file_name": original_file_name,
+        "mime_type": mime_type,
+    }
+
+
+def _registry_entry(sha256, envelope_id, filename, consume_name):
+    return {sha256: {"envelope_id": envelope_id, "filename": filename, "consume_name": consume_name,
+                      "size": 100, "ingested_at": "2026-07-13T17:50:49.281559+00:00"}}
+
+
+class TestParseDocumentTs:
+    def test_parses_offset_to_utc(self):
+        ts = parse_document_ts("2020-04-21T00:00:00+02:00")
+        assert ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc)
+
+    def test_result_is_utc_aware(self):
+        assert parse_document_ts("2026-06-09T12:00:00+00:00").tzinfo is not None
+
+    def test_parses_z_suffix(self):
+        ts = parse_document_ts("2026-06-09T12:00:00Z")
+        assert ts == datetime(2026, 6, 9, 12, 0, 0, tzinfo=timezone.utc)
+
+
+class TestBuildRegistryIndex:
+    def test_inverts_sha256_key_to_consume_name_key(self):
+        registry = {
+            "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"},
+        }
+        index = build_registry_index(registry)
+        assert index == {
+            "2026-06-09_a.pdf": {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"},
+        }
+
+
+class TestFindSourceMail:
+    def test_hit_on_matching_consume_name(self):
+        index = build_registry_index({
+            "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"},
+        })
+        match = find_source_mail("2026-06-09_a.pdf", index)
+        assert match == {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"}
+
+    def test_miss_returns_none(self):
+        index = build_registry_index({
+            "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"},
+        })
+        assert find_source_mail("polisa 920008969228.pdf", index) is None
+
+    def test_none_filename_returns_none(self):
+        assert find_source_mail(None, {}) is None
+
+
+class TestBuildEntities:
+    def test_shape_without_correspondent_tags_or_source_mail(self):
+        doc = _doc()
+        entities = build_entities(doc, correspondents={}, tags={}, source_mail=None)
+        assert entities == [
+            {"type": "content", "text": "some ocr text"},
+            {"type": "correspondent", "name": None},
+            {"type": "filename", "value": "file.pdf"},
+            {"type": "content_type", "value": "application/pdf"},
+        ]
+
+    def test_correspondent_name_resolved(self):
+        doc = _doc(correspondent=7)
+        entities = build_entities(doc, correspondents={7: "WARTA"}, tags={}, source_mail=None)
+        assert {"type": "correspondent", "name": "WARTA"} in entities
+
+    def test_multiple_tags_each_get_own_entity(self):
+        doc = _doc(tags=[1, 2])
+        entities = build_entities(doc, correspondents={}, tags={1: "faktury", 2: "2026"}, source_mail=None)
+        tag_entities = [e for e in entities if e["type"] == "tag"]
+        assert tag_entities == [{"type": "tag", "name": "faktury"}, {"type": "tag", "name": "2026"}]
+
+    def test_no_tags_means_no_tag_entities(self):
+        entities = build_entities(_doc(tags=[]), correspondents={}, tags={}, source_mail=None)
+        assert not [e for e in entities if e["type"] == "tag"]
+
+    def test_source_mail_entity_appended_when_present(self):
+        source_mail = {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"}
+        entities = build_entities(_doc(), correspondents={}, tags={}, source_mail=source_mail)
+        assert entities[-1] == {
+            "type": "source_mail",
+            "envelope_id": "env-a",
+            "attachment_sha256": "sha-a",
+            "attachment_filename": "a.pdf",
+        }
+
+    def test_empty_content_becomes_empty_string_not_none(self):
+        entities = build_entities(_doc(content=None), correspondents={}, tags={}, source_mail=None)
+        assert entities[0] == {"type": "content", "text": ""}
+
+
+class TestMapDocument:
+    def test_id_gets_paperless_prefix(self):
+        env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None)
+        assert env.id == "paperless:42"
+
+    def test_source_is_paperless(self):
+        env = map_document(_doc(), correspondents={}, tags={}, source_mail=None)
+        assert env.source == "paperless"
+
+    def test_raw_ref_is_document_id_string(self):
+        env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None)
+        assert env.raw_ref == "42"
+
+    def test_geo_is_none(self):
+        env = map_document(_doc(), correspondents={}, tags={}, source_mail=None)
+        assert env.geo is None
+
+    def test_ts_parsed_from_created(self):
+        env = map_document(_doc(created="2020-04-21T00:00:00+02:00"), correspondents={}, tags={}, source_mail=None)
+        assert env.ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc)
+
+
+class _FakeResponse:
+    def __init__(self, payload):
+        self._payload = payload
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def json(self):
+        return self._payload
+
+    def raise_for_status(self):
+        pass
+
+
+class _FakeSession:
+    def __init__(self, pages: dict[str, dict]):
+        self._pages = pages
+
+    def get(self, url):
+        return _FakeResponse(self._pages[url])
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+def _empty_page():
+    return {"count": 0, "next": None, "previous": None, "results": []}
+
+
+class TestFetchLookup:
+    async def test_single_page(self):
+        session = _FakeSession({
+            f"{BASE_URL}/api/tags/": {"count": 2, "next": None, "results": [
+                {"id": 1, "name": "faktury"}, {"id": 2, "name": "2026"},
+            ]},
+        })
+        result = await fetch_lookup(session, BASE_URL, "/api/tags/")
+        assert result == {1: "faktury", 2: "2026"}
+
+    async def test_follows_pagination(self):
+        session = _FakeSession({
+            f"{BASE_URL}/api/tags/": {
+                "count": 2, "next": f"{BASE_URL}/api/tags/?page=2",
+                "results": [{"id": 1, "name": "a"}],
+            },
+            f"{BASE_URL}/api/tags/?page=2": {
+                "count": 2, "next": None, "results": [{"id": 2, "name": "b"}],
+            },
+        })
+        result = await fetch_lookup(session, BASE_URL, "/api/tags/")
+        assert result == {1: "a", 2: "b"}
+
+    async def test_empty_results(self):
+        session = _FakeSession({f"{BASE_URL}/api/correspondents/": _empty_page()})
+        assert await fetch_lookup(session, BASE_URL, "/api/correspondents/") == {}
+
+
+class TestIterDocuments:
+    async def test_yields_all_docs_across_pages(self):
+        url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id"
+        url2 = f"{BASE_URL}/api/documents/?page=2"
+        session = _FakeSession({
+            url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]},
+            url2: {"count": 3, "next": None, "results": [_doc(3)]},
+        })
+        docs = [d async for d in iter_documents(session, BASE_URL, page_size=200)]
+        assert [d["id"] for d in docs] == [1, 2, 3]
+
+    async def test_limit_stops_early_across_pages(self):
+        url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id"
+        url2 = f"{BASE_URL}/api/documents/?page=2"
+        session = _FakeSession({
+            url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]},
+            url2: {"count": 3, "next": None, "results": [_doc(3)]},
+        })
+        docs = [d async for d in iter_documents(session, BASE_URL, page_size=200, limit=1)]
+        assert [d["id"] for d in docs] == [1]
+
+
+class _FakeConn:
+    def __init__(self, existing_ids=None):
+        self.existing_ids = list(existing_ids or [])
+        self.execute_calls: list[tuple] = []
+
+    async def fetch(self, query, *params):
+        return [{"id": i} for i in self.existing_ids]
+
+    async def execute(self, query, *params):
+        self.execute_calls.append(params)
+        return "INSERT 0 1"
+
+    async def close(self):
+        pass
+
+
+def _docs_page(docs, next_url=None):
+    return {"count": len(docs), "next": next_url, "results": docs}
+
+
+class TestRun:
+    def _patch(self, monkeypatch, conn, pages):
+        async def _fake_connect(dsn):
+            return conn
+        monkeypatch.setattr("documents_ingest.paperless_adapter.asyncpg.connect", _fake_connect)
+
+        def _fake_session_factory(*args, **kwargs):
+            return _FakeSession(pages)
+        monkeypatch.setattr("documents_ingest.paperless_adapter.aiohttp.ClientSession", _fake_session_factory)
+
+    def _base_pages(self, docs):
+        docs_url = f"{BASE_URL}/api/documents/?page_size=200&ordering=id"
+        return {
+            docs_url: _docs_page(docs),
+            f"{BASE_URL}/api/correspondents/": _empty_page(),
+            f"{BASE_URL}/api/tags/": _empty_page(),
+        }
+
+    async def test_dry_run_does_not_insert(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn()
+        pages = self._base_pages([_doc(1), _doc(2)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=False,
+        )
+
+        assert stats["fetched"] == 2
+        assert stats["inserted"] == 2
+        assert conn.execute_calls == []
+
+    async def test_apply_inserts_new_documents(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn()
+        pages = self._base_pages([_doc(1), _doc(2)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=True,
+        )
+
+        assert stats["inserted"] == 2
+        assert len(conn.execute_calls) == 2
+
+    async def test_idempotent_skips_existing_ids(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn(existing_ids=["paperless:1"])
+        pages = self._base_pages([_doc(1), _doc(2)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=True,
+        )
+
+        assert stats["already_in_db"] == 1
+        assert stats["inserted"] == 1
+        assert len(conn.execute_calls) == 1
+
+    async def test_rerun_after_apply_inserts_nothing_new(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        pages = self._base_pages([_doc(1), _doc(2)])
+
+        conn1 = _FakeConn()
+        self._patch(monkeypatch, conn1, pages)
+        first = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=True,
+        )
+        assert first["inserted"] == 2
+
+        # Second run sees both ids already in the DB (simulated via existing_ids).
+        conn2 = _FakeConn(existing_ids=["paperless:1", "paperless:2"])
+        self._patch(monkeypatch, conn2, pages)
+        second = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=True,
+        )
+        assert second["inserted"] == 0
+        assert second["already_in_db"] == 2
+        assert conn2.execute_calls == []
+
+    async def test_source_mail_linked_counted_on_registry_hit(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text(json.dumps(
+            _registry_entry("sha-a", "env-a", "invoice.pdf", "2026-06-09_invoice.pdf")
+        ))
+        conn = _FakeConn()
+        pages = self._base_pages([_doc(1, original_file_name="2026-06-09_invoice.pdf"), _doc(2)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=False,
+        )
+
+        assert stats["source_mail_linked"] == 1
+
+    async def test_empty_content_counted(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn()
+        pages = self._base_pages([_doc(1, content=""), _doc(2, content="text")])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=False,
+        )
+
+        assert stats["empty_content"] == 1
+
+    async def test_map_error_is_isolated_and_counted(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn()
+        bad_doc = _doc(1, created=None)  # parse_document_ts(None) raises
+        pages = self._base_pages([bad_doc, _doc(2)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=False,
+        )
+
+        assert stats["errors"] == 1
+        assert stats["inserted"] == 1
+        assert stats["fetched"] == 2
+
+    async def test_stats_balance_across_all_outcomes(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn(existing_ids=["paperless:1"])
+        pages = self._base_pages([_doc(1), _doc(2), _doc(3, created=None)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, apply=True,
+        )
+
+        assert stats["fetched"] == 3
+        assert stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"]
+        assert stats["already_in_db"] == 1
+        assert stats["inserted"] == 1
+        assert stats["errors"] == 1
+
+    async def test_limit_caps_documents_processed(self, monkeypatch, tmp_path):
+        registry_path = tmp_path / "registry.json"
+        registry_path.write_text("{}")
+        conn = _FakeConn()
+        pages = self._base_pages([_doc(1), _doc(2), _doc(3)])
+        self._patch(monkeypatch, conn, pages)
+
+        stats = await run(
+            dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok",
+            registry_path=registry_path, limit=2, apply=False,
+        )
+
+        assert stats["fetched"] == 2