Fix: de-duplicate source_file fixture across sink test classes

TestDirectoryExportSink and TestZipExportSink each defined an identical source_file fixture; hoist it to module scope.
Feature: add ZipExportSink with atomic finalize and manifest spooling
2026-07-23 04:14:55 +00:00 · 2026-07-22 15:38:55 -07:00 · 2026-07-22 15:36:50 -07:00 · 2026-07-22 15:07:24 -07:00 · 2026-07-22 14:57:48 -07:00 · 2026-07-22 14:34:55 -07:00
8 changed files with 3297 additions and 0 deletions
@@ -0,0 +1,839 @@
+# Export Zip Compression Control Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add `--zip-compression {stored,deflated,bzip2,lzma,zstd}` and `--zip-compression-level N` flags to `document_exporter`, threaded into `ZipExportSink`, with import-side safety for codecs the running Python can't read.
+
+**Architecture:** A new pure-data module `documents/export/compression.py` owns the method↔constant map, per-method level bounds, the runtime availability probe, and a compress-type readability check. `ZipExportSink` gains `compression`/`compresslevel` constructor params. The command validates flags up front (fail-fast `CommandError`) and constructs the sink; the importer pre-checks entry compress types before extracting.
+
+**Tech Stack:** Python ≥3.11 (zstd only on 3.14+), `zipfile`, `compression.zstd` (PEP 784), pytest + pytest-mock + factory-boy. Backend tests run on the Linux VM (Python 3.11 — zstd positive tests are `skipif`-guarded); `ruff` runs locally.
+
+**Spec:** `docs/superpowers/specs/2026-06-16-export-zip-compression-design.md`
+
+**PREREQUISITE:** The base refactor `docs/superpowers/plans/2026-06-16-export-sink-architecture.md` MUST be merged first. This plan assumes `src/documents/export/sinks.py` exists with `ZipExportSink(target, zip_name, *, delete=False)` opening its `ZipFile` in `_open()`.
+
+---
+
+## Verified facts (CPython 3.14.3, via `uv run --python 3.14 --no-project`)
+
+- Constants: `ZIP_STORED=0`, `ZIP_DEFLATED=8`, `ZIP_BZIP2=12`, `ZIP_LZMA=14`, `ZIP_ZSTANDARD=93` (zstd added 3.14; absent on < 3.14).
+- `ZipFile(file, "w", compression=…, compresslevel=…)` applies both as the default for every `write`/`writestr` — no per-entry args needed (verified).
+- Level bounds: `deflated` 0–9, `bzip2` 1–9, `lzma`/`stored` ignore level, `zstd` -131072…22 (`compression.zstd.CompressionParameter.compression_level.bounds() == (-131072, 22)`).
+- An invalid level fails at the **first write** (`ValueError: Invalid initialization option` / `compresslevel must be between 1 and 9`), plus GC-time `AttributeError` noise on close — hence up-front validation.
+- zstd is backed by `compression.zstd`; `zipfile` raises `RuntimeError` if it's unavailable.
+
+## Conventions for every task
+
+- **Run backend tests on the VM:** `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "<targets>"` (never locally).
+- **Lint locally:** `ruff check <paths> && ruff format <paths>` (global ruff, not `uv run`).
+- **Tests are pytest-style:** classes, `@pytest.mark.django_db` on the class only where DB is needed (the `compression.py` and sink tests need no DB), factory-boy, `mocker`, `parametrize`, full type annotations.
+- The VM runs Python 3.11, so **zstd positive tests must be `@pytest.mark.skipif(...)`-guarded**; they will simply not run there. zstd _rejection_ tests (the < 3.14 path) DO run on the VM.
+
+## File structure
+
+- **Create** `src/documents/export/compression.py` — method map, CLI choices, level bounds, `compression_available()`, `level_error()`, `compress_type_readable()`, `unreadable_method_names()`. Pure, no Django.
+- **Create** `src/documents/tests/export/test_compression.py` — unit tests for the above.
+- **Modify** `src/documents/export/sinks.py` — `ZipExportSink.__init__` gains `compression`/`compresslevel`; `_open()` passes them to `ZipFile`.
+- **Modify** `src/documents/tests/export/test_sinks.py` — assert the chosen `compress_type` is applied.
+- **Modify** `src/documents/management/commands/document_exporter.py` — add the two CLI flags, up-front validation, and pass resolved values to `ZipExportSink`.
+- **Modify** `src/documents/tests/test_management_exporter.py` — flag validation + default-unchanged tests.
+- **Modify** `src/documents/management/commands/document_importer.py` — pre-extract compress-type check.
+- **Modify** `src/documents/tests/test_management_importer.py` — unsupported-codec → `CommandError`.
+- **Modify** `docs/administration.md` — document both flags + zstd portability caveat.
+
+---
+
+## Task 1: `documents/export/compression.py` (pure compression policy)
+
+**Files:**
+
+- Create: `src/documents/export/compression.py`
+- Test: `src/documents/tests/export/test_compression.py`
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `src/documents/tests/export/test_compression.py`:
+
+```python
+import sys
+import zipfile
+
+import pytest
+
+from documents.export import compression
+
+
+class TestCompressionMethods:
+    def test_choices_always_include_zstd(self) -> None:
+        # zstd is offered regardless of runtime; availability is checked separately
+        assert compression.COMPRESSION_CHOICES == (
+            "stored",
+            "deflated",
+            "bzip2",
+            "lzma",
+            "zstd",
+        )
+
+    @pytest.mark.parametrize(
+        ("name", "constant"),
+        [
+            ("stored", zipfile.ZIP_STORED),
+            ("deflated", zipfile.ZIP_DEFLATED),
+            ("bzip2", zipfile.ZIP_BZIP2),
+            ("lzma", zipfile.ZIP_LZMA),
+        ],
+    )
+    def test_method_maps_to_zipfile_constant(self, name: str, constant: int) -> None:
+        assert compression.COMPRESSION_METHODS[name] == constant
+
+    def test_stored_and_deflated_always_available(self) -> None:
+        assert compression.compression_available("stored")
+        assert compression.compression_available("deflated")
+
+    def test_zstd_availability_tracks_runtime(self) -> None:
+        expected: bool = sys.version_info >= (3, 14)
+        assert compression.compression_available("zstd") == expected
+
+
+class TestLevelError:
+    @pytest.mark.parametrize(
+        ("method", "level"),
+        [
+            ("deflated", 0),
+            ("deflated", 9),
+            ("bzip2", 1),
+            ("bzip2", 9),
+            ("deflated", None),
+            ("stored", None),
+        ],
+    )
+    def test_valid_levels_return_none(self, method: str, level: int | None) -> None:
+        assert compression.level_error(method, level) is None
+
+    @pytest.mark.parametrize(
+        ("method", "level"),
+        [
+            ("deflated", 10),
+            ("deflated", -1),
+            ("bzip2", 0),
+            ("bzip2", 10),
+        ],
+    )
+    def test_out_of_range_levels_return_message(
+        self,
+        method: str,
+        level: int,
+    ) -> None:
+        msg: str | None = compression.level_error(method, level)
+        assert msg is not None
+        assert "between" in msg
+
+    @pytest.mark.parametrize("method", ["stored", "lzma"])
+    def test_level_on_levelless_method_is_rejected(self, method: str) -> None:
+        msg: str | None = compression.level_error(method, 5)
+        assert msg is not None
+        assert "no effect" in msg
+
+
+class TestCompressTypeReadable:
+    @pytest.mark.parametrize("ct", [zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED])
+    def test_stored_and_deflated_always_readable(self, ct: int) -> None:
+        assert compression.compress_type_readable(ct)
+
+    def test_zstd_compress_type_readability_tracks_runtime(self) -> None:
+        # 93 = ZIP_ZSTANDARD; 20 = legacy zstd method id (read-only)
+        expected: bool = sys.version_info >= (3, 14)
+        assert compression.compress_type_readable(93) == expected
+        assert compression.compress_type_readable(20) == expected
+
+    def test_unknown_compress_type_is_unreadable(self) -> None:
+        assert not compression.compress_type_readable(9999)
+
+    def test_unreadable_method_names_lists_methods(self) -> None:
+        # An unknown method id maps to no name and is reported generically.
+        names: set[str] = compression.unreadable_method_names({9999})
+        assert names == {"method 9999"}
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/export/test_compression.py -v"`
+Expected: FAIL with `ModuleNotFoundError: No module named 'documents.export.compression'`.
+
+- [ ] **Step 3: Implement `compression.py`**
+
+Create `src/documents/export/compression.py`:
+
+```python
+from __future__ import annotations
+
+import importlib
+import zipfile
+
+# ZIP_ZSTANDARD exists only on Python 3.14+ (PEP 784). None elsewhere.
+ZSTD: int | None = getattr(zipfile, "ZIP_ZSTANDARD", None)
+
+# CLI choices are fixed across runtimes so argparse never hides zstd; runtime
+# availability is enforced separately in compression_available().
+COMPRESSION_CHOICES: tuple[str, ...] = (
+    "stored",
+    "deflated",
+    "bzip2",
+    "lzma",
+    "zstd",
+)
+
+# Method name -> zipfile compression constant (zstd only when supported).
+COMPRESSION_METHODS: dict[str, int] = {
+    "stored": zipfile.ZIP_STORED,
+    "deflated": zipfile.ZIP_DEFLATED,
+    "bzip2": zipfile.ZIP_BZIP2,
+    "lzma": zipfile.ZIP_LZMA,
+}
+if ZSTD is not None:
+    COMPRESSION_METHODS["zstd"] = ZSTD
+
+# Inclusive (min, max) level bounds per method; None => level not applicable.
+# Verified on CPython 3.14.3.
+LEVEL_BOUNDS: dict[str, tuple[int, int] | None] = {
+    "stored": None,
+    "deflated": (0, 9),
+    "bzip2": (1, 9),
+    "lzma": None,
+    "zstd": (-131072, 22),
+}
+
+# zipfile compress_type id -> method name. 93 = current zstd id, 20 = legacy
+# zstd id that zipfile can still read.
+_COMPRESS_TYPE_TO_METHOD: dict[int, str] = {
+    zipfile.ZIP_STORED: "stored",
+    zipfile.ZIP_DEFLATED: "deflated",
+    zipfile.ZIP_BZIP2: "bzip2",
+    zipfile.ZIP_LZMA: "lzma",
+    93: "zstd",
+    20: "zstd",
+}
+
+
+def compression_available(method: str) -> bool:
+    """Whether the running interpreter can actually use the given method."""
+    if method in ("stored", "deflated"):
+        # zlib is a hard CPython dependency; stored needs nothing.
+        return True
+    if method == "bzip2":
+        return _module_importable("bz2")
+    if method == "lzma":
+        return _module_importable("lzma")
+    if method == "zstd":
+        return ZSTD is not None and _module_importable("compression.zstd")
+    return False
+
+
+def _module_importable(name: str) -> bool:
+    try:
+        importlib.import_module(name)
+    except ImportError:
+        return False
+    return True
+
+
+def level_error(method: str, level: int | None) -> str | None:
+    """Return a human message if (method, level) is invalid, else None."""
+    if level is None:
+        return None
+    bounds = LEVEL_BOUNDS[method]
+    if bounds is None:
+        return f"--zip-compression-level has no effect for '{method}'"
+    low, high = bounds
+    if not (low <= level <= high):
+        return (
+            f"--zip-compression-level for '{method}' must be between "
+            f"{low} and {high}"
+        )
+    return None
+
+
+def compress_type_readable(compress_type: int) -> bool:
+    """Whether this interpreter can decompress an entry of the given type."""
+    method = _COMPRESS_TYPE_TO_METHOD.get(compress_type)
+    if method is None:
+        return False
+    return compression_available(method)
+
+
+def unreadable_method_names(compress_types: set[int]) -> set[str]:
+    """Map a set of compress_type ids to human method names for error messages."""
+    names: set[str] = set()
+    for ct in compress_types:
+        names.add(_COMPRESS_TYPE_TO_METHOD.get(ct, f"method {ct}"))
+    return names
+```
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/export/test_compression.py -v"`
+Expected: PASS (on the 3.11 VM, `test_zstd_availability_tracks_runtime` and `test_zstd_compress_type_readability_tracks_runtime` assert `False`).
+
+- [ ] **Step 5: Lint**
+
+Run: `ruff check src/documents/export/compression.py src/documents/tests/export/test_compression.py && ruff format src/documents/export/compression.py src/documents/tests/export/test_compression.py`
+Expected: no errors.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/documents/export/compression.py src/documents/tests/export/test_compression.py
+git commit -m "Feature: add export compression policy module"
+```
+
+---
+
+## Task 2: `ZipExportSink` accepts compression method + level
+
+**Files:**
+
+- Modify: `src/documents/export/sinks.py`
+- Test: `src/documents/tests/export/test_sinks.py`
+
+- [ ] **Step 1: Write the failing test**
+
+Append to `src/documents/tests/export/test_sinks.py` (the top-of-file block already imports `zipfile`, `Path`, `pytest`, `ZipExportSink`, `StreamingManifestWriter` from the base-refactor plan):
+
+```python
+class TestZipExportSinkCompression:
+    @pytest.fixture()
+    def source_file(self, tmp_path: Path) -> Path:
+        src: Path = tmp_path / "src" / "doc.pdf"
+        src.parent.mkdir(parents=True)
+        src.write_bytes(b"PDF-CONTENT" * 100)
+        return src
+
+    @pytest.mark.parametrize(
+        ("method", "constant"),
+        [
+            ("stored", zipfile.ZIP_STORED),
+            ("deflated", zipfile.ZIP_DEFLATED),
+            ("bzip2", zipfile.ZIP_BZIP2),
+            ("lzma", zipfile.ZIP_LZMA),
+        ],
+    )
+    def test_compression_method_is_applied_to_file_entries(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+        method: str,
+        constant: int,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with ZipExportSink(
+            target,
+            "export",
+            delete=False,
+            compression=constant,
+        ) as sink:
+            sink.add_file(source_file, "doc.pdf")
+        with zipfile.ZipFile(target / "export.zip") as zf:
+            info = zf.getinfo("doc.pdf")
+            assert info.compress_type == constant
+
+    def test_compressing_method_beats_stored(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        # Robust size invariant: a compressing method must be <= stored on
+        # compressible content (avoids flaky level-9-vs-level-1 comparisons).
+        sizes: dict[str, int] = {}
+        for name, constant in (("stored", zipfile.ZIP_STORED), ("deflated", zipfile.ZIP_DEFLATED)):
+            target: Path = tmp_path / name
+            target.mkdir()
+            with ZipExportSink(target, "export", delete=False, compression=constant) as sink:
+                sink.add_file(source_file, "doc.pdf")
+            sizes[name] = (target / "export.zip").stat().st_size
+        assert sizes["deflated"] <= sizes["stored"]
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/export/test_sinks.py::TestZipExportSinkCompression -v"`
+Expected: FAIL with `TypeError: __init__() got an unexpected keyword argument 'compression'`.
+
+- [ ] **Step 3: Add the params to `ZipExportSink`**
+
+In `src/documents/export/sinks.py`, change `ZipExportSink.__init__` to accept the new keyword-only params and store them, and pass them in `_open()`:
+
+```python
+    def __init__(
+        self,
+        target: Path,
+        zip_name: str,
+        *,
+        delete: bool = False,
+        compression: int = zipfile.ZIP_DEFLATED,
+        compresslevel: int | None = None,
+    ) -> None:
+        self._target = target.resolve()
+        self._zip_path = (self._target / zip_name).with_suffix(".zip")
+        self._tmp_path = self._zip_path.with_name(self._zip_path.name + ".tmp")
+        self._delete = delete
+        self._compression = compression
+        self._compresslevel = compresslevel
+        self._zip: zipfile.ZipFile | None = None
+        self._dirs: set[str] = set()
+        self._pending_manifest: tuple[Path, str] | None = None
+        self._stream_open = False
+```
+
+And in `_open()`:
+
+```python
+    def _open(self) -> None:
+        settings.SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
+        self._zip = zipfile.ZipFile(
+            self._tmp_path,
+            "w",
+            compression=self._compression,
+            compresslevel=self._compresslevel,
+            allowZip64=True,
+        )
+```
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/export/test_sinks.py -v"`
+Expected: PASS (all sink tests, including the four method params and the size invariant). `bzip2`/`lzma` are present on the VM's CPython, so those params pass.
+
+- [ ] **Step 5: Lint**
+
+Run: `ruff check src/documents/export/sinks.py && ruff format src/documents/export/sinks.py`
+Expected: no errors.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/documents/export/sinks.py src/documents/tests/export/test_sinks.py
+git commit -m "Feature: ZipExportSink accepts compression method and level"
+```
+
+---
+
+## Task 3: Wire CLI flags + validation into `document_exporter`
+
+**Files:**
+
+- Modify: `src/documents/management/commands/document_exporter.py`
+- Test: `src/documents/tests/test_management_exporter.py`
+
+- [ ] **Step 1: Add the argparse flags**
+
+In `document_exporter.py`, add the import near the other `documents.export` import:
+
+```python
+from documents.export.compression import COMPRESSION_CHOICES
+from documents.export.compression import COMPRESSION_METHODS
+from documents.export.compression import compression_available
+from documents.export.compression import level_error
+from documents.export.compression import ZSTD
+```
+
+In `add_arguments`, after the `--zip-name` argument, add:
+
+```python
+        parser.add_argument(
+            "--zip-compression",
+            choices=COMPRESSION_CHOICES,
+            default=None,
+            help=(
+                "Compression method for the export zip (requires --zip). "
+                "Default: deflated. 'zstd' requires Python 3.14+ on both the "
+                "exporting and importing machine."
+            ),
+        )
+
+        parser.add_argument(
+            "--zip-compression-level",
+            type=int,
+            default=None,
+            help=(
+                "Compression level for the export zip (requires --zip). "
+                "deflated: 0-9, bzip2: 1-9, zstd: -131072..22; ignored for "
+                "stored/lzma."
+            ),
+        )
+```
+
+- [ ] **Step 2: Read + validate the flags in `handle()`**
+
+In `handle()`, after the existing `--compare-*` + `--zip` guard, add the compression flag handling. Insert before the sink construction:
+
+```python
+        zip_compression: str | None = options["zip_compression"]
+        zip_compression_level: int | None = options["zip_compression_level"]
+
+        if not self.zip_export and (
+            zip_compression is not None or zip_compression_level is not None
+        ):
+            raise CommandError(
+                "--zip-compression and --zip-compression-level require --zip",
+            )
+
+        compression_method = zip_compression or "deflated"
+        if self.zip_export:
+            if not compression_available(compression_method):
+                if compression_method == "zstd" and ZSTD is None:
+                    raise CommandError(
+                        "zstd compression requires Python 3.14 or newer",
+                    )
+                raise CommandError(
+                    f"Compression method '{compression_method}' is not "
+                    f"available on this Python runtime",
+                )
+            level_msg = level_error(compression_method, zip_compression_level)
+            if level_msg is not None:
+                raise CommandError(level_msg)
+```
+
+- [ ] **Step 3: Pass the resolved values into `ZipExportSink`**
+
+Change the `ZipExportSink(...)` construction in `handle()` to:
+
+```python
+        if self.zip_export:
+            sink = ZipExportSink(
+                self.target,
+                options["zip_name"],
+                delete=self.delete,
+                compression=COMPRESSION_METHODS[compression_method],
+                compresslevel=zip_compression_level,
+            )
+        else:
+            sink = DirectoryExportSink(
+                self.target,
+                compare_checksums=self.compare_checksums,
+                compare_json=self.compare_json,
+                delete=self.delete,
+            )
+```
+
+- [ ] **Step 4: Write the command-level tests**
+
+Add to the `TestExportImport` class in `src/documents/tests/test_management_exporter.py` (imports `call_command`, `CommandError`, `ZipFile`, `timezone` already present):
+
+```python
+    def test_compression_flags_require_zip(self) -> None:
+        for args in (
+            ["--zip-compression", "lzma"],
+            ["--zip-compression-level", "5"],
+        ):
+            with self.assertRaises(CommandError):
+                call_command(
+                    "document_exporter",
+                    self.target,
+                    *args,
+                    skip_checks=True,
+                )
+
+    def test_zip_compression_level_out_of_range_raises(self) -> None:
+        with self.assertRaises(CommandError):
+            call_command(
+                "document_exporter",
+                self.target,
+                "--zip",
+                "--zip-compression",
+                "deflated",
+                "--zip-compression-level",
+                "99",
+                skip_checks=True,
+            )
+
+    def test_zip_compression_level_rejected_for_stored(self) -> None:
+        with self.assertRaises(CommandError):
+            call_command(
+                "document_exporter",
+                self.target,
+                "--zip",
+                "--zip-compression",
+                "stored",
+                "--zip-compression-level",
+                "5",
+                skip_checks=True,
+            )
+
+    def test_zip_lzma_compression_round_trips(self) -> None:
+        call_command(
+            "document_exporter",
+            self.target,
+            "--zip",
+            "--zip-compression",
+            "lzma",
+            skip_checks=True,
+        )
+        expected = str(
+            self.target / f"export-{timezone.localdate().isoformat()}.zip",
+        )
+        self.assertIsFile(expected)
+        with ZipFile(expected) as zip_file:
+            info = zip_file.getinfo("manifest.json")
+            # manifest.json carries the chosen method; deflated is the default
+            self.assertEqual(info.compress_type, 14)  # ZIP_LZMA
+
+    def test_default_zip_uses_deflate(self) -> None:
+        call_command(
+            "document_exporter",
+            self.target,
+            "--zip",
+            skip_checks=True,
+        )
+        expected = str(
+            self.target / f"export-{timezone.localdate().isoformat()}.zip",
+        )
+        with ZipFile(expected) as zip_file:
+            info = zip_file.getinfo("manifest.json")
+            self.assertEqual(info.compress_type, 8)  # ZIP_DEFLATED
+```
+
+- [ ] **Step 5: Run the tests**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/test_management_exporter.py -v"`
+Expected: PASS — the new tests plus all existing exporter tests stay green.
+
+- [ ] **Step 6: Lint**
+
+Run: `ruff check src/documents/management/commands/document_exporter.py src/documents/tests/test_management_exporter.py && ruff format src/documents/management/commands/document_exporter.py src/documents/tests/test_management_exporter.py`
+Expected: no errors.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add src/documents/management/commands/document_exporter.py src/documents/tests/test_management_exporter.py
+git commit -m "Feature: add --zip-compression and --zip-compression-level flags"
+```
+
+---
+
+## Task 4: Importer pre-check for unreadable codecs
+
+**Files:**
+
+- Modify: `src/documents/management/commands/document_importer.py`
+- Test: `src/documents/tests/test_management_importer.py`
+
+- [ ] **Step 1: Write the failing test**
+
+The importer test file `src/documents/tests/test_management_importer.py` is
+`TestCase`-style (`class TestCommandImport(... TestCase)`, `self.assertRaises`,
+`DirectoriesMixin` gives `self.dirs.scratch_dir`). Match that style. Add this
+method to `TestCommandImport`. It builds a valid zip and patches the readability
+probe so the check fires deterministically on any runtime:
+
+```python
+    def test_import_rejects_unreadable_compression(self) -> None:
+        """
+        GIVEN:
+            - A zip archive with an entry whose compression this Python can't read
+        WHEN:
+            - Import is attempted
+        THEN:
+            - A CommandError naming the issue is raised, before extraction
+        """
+        import zipfile
+        from unittest import mock
+
+        archive = Path(self.dirs.scratch_dir) / "export.zip"
+        with zipfile.ZipFile(archive, "w") as zf:
+            zf.writestr("manifest.json", "[]")
+
+        with mock.patch(
+            "documents.management.commands.document_importer.compress_type_readable",
+            return_value=False,
+        ):
+            with self.assertRaises(CommandError) as e:
+                call_command(
+                    "document_importer",
+                    str(archive),
+                    "--no-progress-bar",
+                    skip_checks=True,
+                )
+        self.assertIn("compression", str(e.exception))
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/test_management_importer.py -k unreadable_compression -v"`
+Expected: FAIL — no pre-check exists yet, so the import proceeds (or fails with a different error).
+
+- [ ] **Step 3: Implement the pre-check**
+
+In `document_importer.py`, add the import:
+
+```python
+from documents.export.compression import compress_type_readable
+from documents.export.compression import unreadable_method_names
+```
+
+Find the zip-handling block (around `document_importer.py:453`):
+
+```python
+                with ZipFile(self.source) as zf:
+                    zf.extractall(tmp_dir)
+```
+
+Replace it with a pre-check before extraction:
+
+```python
+                with ZipFile(self.source) as zf:
+                    unsupported = {
+                        info.compress_type
+                        for info in zf.infolist()
+                        if not compress_type_readable(info.compress_type)
+                    }
+                    if unsupported:
+                        names = ", ".join(sorted(unreadable_method_names(unsupported)))
+                        raise CommandError(
+                            f"This archive uses compression this Python cannot "
+                            f"read ({names}). zstd archives require Python 3.14+.",
+                        )
+                    zf.extractall(tmp_dir)
+```
+
+Confirm `CommandError` is imported in `document_importer.py` (it is used elsewhere; if not, add `from django.core.management.base import CommandError`).
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/test_management_importer.py -v"`
+Expected: PASS — the new test plus all existing importer tests (normal deflated/stored archives still import).
+
+- [ ] **Step 5: Lint**
+
+Run: `ruff check src/documents/management/commands/document_importer.py src/documents/tests/test_management_importer.py && ruff format src/documents/management/commands/document_importer.py src/documents/tests/test_management_importer.py`
+Expected: no errors.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/documents/management/commands/document_importer.py src/documents/tests/test_management_importer.py
+git commit -m "Feature: importer rejects archives with unreadable compression"
+```
+
+---
+
+## Task 5: Document the flags
+
+**Files:**
+
+- Modify: `docs/administration.md`
+
+- [ ] **Step 1: Add the flags to the option list**
+
+In `docs/administration.md`, update the usage block (around line 257) to include the new flags:
+
+```
+document_exporter target [-c] [-d] [-f] [-na] [-nt] [-p] [-sm] [-z]
+
+optional arguments:
+-c,  --compare-checksums
+-cj, --compare-json
+-d,  --delete
+-f,  --use-filename-format
+-na, --no-archive
+-nt, --no-thumbnail
+-p,  --use-folder-prefix
+-sm, --split-manifest
+-z,  --zip
+-zn, --zip-name
+--zip-compression
+--zip-compression-level
+--data-only
+--no-progress-bar
+--passphrase
+```
+
+- [ ] **Step 2: Add the prose**
+
+After the `-z`/`--zip` paragraph (around line 330), add:
+
+```markdown
+The compression method for the zip can be set with `--zip-compression`
+(`stored`, `deflated` (default), `bzip2`, `lzma`, or `zstd`) and tuned with
+`--zip-compression-level` (deflated: 0–9, bzip2: 1–9, zstd: -131072–22; ignored
+for `stored` and `lzma`). Both options require `--zip`.
+
+!!! warning
+
+    `zstd` compression requires Python 3.14 or newer on **both** the machine
+    creating the export and any machine importing it. An archive compressed with
+    `zstd` (or `lzma`/`bzip2` where those modules are unavailable) cannot be
+    imported on a runtime that lacks the codec; the importer will refuse it with
+    a clear error. The default `deflated` is universally readable.
+```
+
+- [ ] **Step 3: Verify the docs build is not broken (lint markdown)**
+
+Run: `ruff check docs/ 2>/dev/null; echo "docs are markdown; rely on prettier pre-commit"`
+(No code to test. The prettier pre-commit hook will reformat on commit.)
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add docs/administration.md
+git commit -m "Docs: document --zip-compression and --zip-compression-level"
+```
+
+---
+
+## Task 6: Final verification
+
+**Files:** none (verification only).
+
+- [ ] **Step 1: Full backend suites on the VM**
+
+Run: `bash /c/Users/tholmes/Documents/Coding/paperless/vmtest.sh "src/documents/tests/export/ src/documents/tests/test_management_exporter.py src/documents/tests/test_management_importer.py -v"`
+Expected: PASS, no failures.
+
+- [ ] **Step 2: Spot-check the zstd happy path on Python 3.14 (cannot run under Django on the 3.11 VM)**
+
+The zstd positive round-trip can't run in the 3.11 test env. Confirm the policy module behaves on a real 3.14 interpreter with a standalone check (no Django needed):
+
+Run:
+
+```bash
+uv run --python 3.14 --no-project python -c "import sys; sys.path.insert(0,'src'); import django; print('skip')" 2>/dev/null || \
+uv run --python 3.14 --no-project python -c "
+import zipfile, io
+from compression.zstd import CompressionParameter as CP
+print('zstd const', zipfile.ZIP_ZSTANDARD, 'bounds', CP.compression_level.bounds())
+buf = io.BytesIO()
+with zipfile.ZipFile(buf,'w',compression=zipfile.ZIP_ZSTANDARD,compresslevel=19) as zf:
+    zf.writestr('a.txt','x'*1000)
+with zipfile.ZipFile(buf) as zf:
+    assert zf.getinfo('a.txt').compress_type == zipfile.ZIP_ZSTANDARD
+    assert zf.read('a.txt') == b'x'*1000
+print('zstd round-trip OK')
+"
+```
+
+Expected: prints `zstd const 93 bounds (-131072, 22)` and `zstd round-trip OK`. This validates the constant, bounds, and that a zstd archive round-trips — the parts the 3.11 CI cannot exercise.
+
+- [ ] **Step 3: Type-check on the VM (pyrefly)**
+
+```bash
+tar czf - src pyproject.toml uv.lock .pyrefly-baseline.json | ssh -o BatchMode=yes -p 2244 trenton@localhost 'tar xzf - -C ~/projects/paperless-ngx'
+ssh -o BatchMode=yes -p 2244 trenton@localhost 'bash -lc "cd ~/projects/paperless-ngx && uv run pyrefly check"'
+```
+
+Expected: no new type errors beyond the baseline. (Note: `import compression.zstd` is guarded behind `importlib.import_module`, so it is never statically resolved on the 3.11 baseline.)
+
+- [ ] **Step 4: Final lint**
+
+Run: `ruff check src/documents/export/ src/documents/management/commands/document_exporter.py src/documents/management/commands/document_importer.py && ruff format --check src/documents/export/ src/documents/management/commands/document_exporter.py src/documents/management/commands/document_importer.py`
+Expected: clean.
+
+---
+
+## Notes for the implementer
+
+- **Default behavior is unchanged:** with no flags, the sink is constructed with `compression=ZIP_DEFLATED, compresslevel=None` — byte-method-identical to today (`shutil.make_archive` used `ZIP_DEFLATED` with no level). `test_default_zip_uses_deflate` pins this.
+- **zstd availability is gated three ways and never imported statically:** the constant via `getattr`, the codec via `importlib.import_module("compression.zstd")`, and the CLI value rejected with a friendly message on < 3.14. The choices list always contains `zstd` so argparse doesn't hide it.
+- **The importer pre-check is the safety net** for portability foot-guns — without it an unreadable entry raises a bare `NotImplementedError` mid-`extractall`. The check runs on `infolist()` (metadata only) before any extraction.
+- **Why `--zip-compression` defaults to `None`, not `"deflated"`:** so `handle()` can detect "user passed it without `--zip`" and fail fast. The effective default is resolved as `zip_compression or "deflated"`.
@@ -0,0 +1,337 @@
+# Export Sink Architecture — Design
+
+**Date:** 2026-06-16
+**Branch base:** `dev`
+**Status:** Approved design, pending implementation plan
+
+## Problem
+
+The `document_exporter` management command can export to a folder or to a zip
+file, but the zip support is bolted on rather than designed in:
+
+- **Zip mode is a temp-dir detour.** `handle()` redirects `self.target` to a
+  `tempfile.TemporaryDirectory` in `SCRATCH_DIR`, runs the entire export against
+  that directory, then calls `shutil.make_archive` to zip the whole tree and
+  cleans the temp dir up (`document_exporter.py:322-358`). The export is written
+  to disk twice (loose files, then the zip).
+
+- **An attempted "direct to zip" refactor leaks the destination everywhere.**
+  The prior work on `feature-direct-zip-export` threads `if self.zip_export:`
+  branches through `check_and_copy`, `check_and_write_json`,
+  `_write_split_manifest`, `dump`, `handle`, and `StreamingManifestWriter`. Each
+  write site grew a second code path plus a `.resolve().relative_to(self.target)`
+  arcname dance. The destination became a cross-cutting concern smeared across
+  the command.
+
+- **The command owns logic that isn't about the export contents.** Incremental
+  sync — the `files_in_export_dir` snapshot, the `--compare-checksums` /
+  `--compare-json` skip-if-unchanged checks, and the `--delete` stale-file prune —
+  is interleaved with the logic that decides _what_ to export. These behaviors
+  only make sense for a folder destination, yet they live in the command body.
+
+- **Atomicity is informal.** A backup must never look complete when it isn't.
+  The temp-dir approach happens to be safe (the zip is built last), but there is
+  no explicit "produce the archive only if the whole run succeeded" contract, and
+  the direct-to-zip branch had to hand-manage a `.tmp` file inline.
+
+## Goal
+
+Separate **what** is exported (the command's job) from **where/how** it lands
+(the destination's job), behind a small `ExportSink` abstraction. The command
+declares files, JSON blobs, and a streamed manifest; the sink decides whether and
+how to persist each one. Folder and zip become two interchangeable sinks, and a
+future `S3ExportSink` is a third implementation rather than a fourth set of
+branches. The zip is produced **only** if the entire export succeeds.
+
+## Scope
+
+In scope:
+
+- New `documents/export/` package with the `ExportSink` ABC and two concrete
+  sinks (`DirectoryExportSink`, `ZipExportSink`).
+- Move all incremental-sync machinery (snapshot, compare, prune) out of the
+  command and into `DirectoryExportSink`.
+- Rewrite `document_exporter.handle()` / `dump()` to be destination-agnostic.
+- Simplify `StreamingManifestWriter` to write to a sink-provided handle.
+- Unit tests for each sink; keep existing command-level tests green.
+
+Out of scope:
+
+- `bulk_download.py` / `BulkArchiveStrategy` and share-link bundle zipping. Those
+  select _which document files_ go in and stream to an HTTP response with no
+  atomic-finalize requirement — a different axis from the backup sink. Untouched.
+- Actually implementing an S3 (or any cloud) sink. The interface is designed to
+  _allow_ one; we do not build one (YAGNI).
+- Changing the export's on-disk/in-zip layout, manifest schema, crypto, or any
+  CLI flag's meaning. Behavior is preserved; only the destination plumbing moves.
+- Zip compression control (method / level). The `ZipExportSink` keeps today's
+  fixed `ZIP_DEFLATED` here; making compression configurable is a follow-up —
+  see `2026-06-16-export-zip-compression-design.md`, which depends on this
+  refactor landing first. The sink is the single seam that makes it a small,
+  isolated change.
+
+## Decisions
+
+These were settled during brainstorming:
+
+1. **Scope is the `document_exporter` command only.** Design the interface so an
+   S3 sink could be added later; do not refactor `bulk_download` or share bundles.
+2. **`--compare-*` are folder-only (hard error with `--zip`); `--delete` is kept
+   for both.** `--compare-checksums` / `--compare-json` are genuine no-ops in zip
+   mode today (the temp dir is always empty, so the compare always copies), so
+   combining either with `--zip` raises a `CommandError` up front. **`--delete`,
+   however, is an existing tested feature in zip mode** — it wipes the destination
+   directory of pre-existing files/dirs before the archive lands
+   (`test_export_zipped_with_delete`). Its meaning differs by destination: folder
+   `--delete` prunes stale exported files; zip `--delete` clears the target dir.
+   Both are preserved — `--delete` is a parameter of _both_ sinks, not an error.
+3. **The zip manifest spools to a temp file, not memory.** The sink exposes a
+   streaming-write handle. The zip sink streams the manifest to a single temp
+   file in `SCRATCH_DIR` and adds it as the manifest entry at finalize, keeping
+   peak memory flat regardless of library size. The only "temp" artifact is one
+   manifest file, not a whole export tree.
+
+## Architecture
+
+### The `ExportSink` interface
+
+New module `documents/export/sinks.py`:
+
+```python
+class ExportSink(AbstractContextManager):
+    """Destination for a document export.
+
+    The command declares export contents via the three verbs below; the sink
+    decides whether and how to persist each item. arcname is always a relative
+    POSIX path (e.g. "manifest.json", "originals/foo.pdf").
+    """
+
+    def add_file(
+        self,
+        source: Path,
+        arcname: str,
+        *,
+        checksum: str | None = None,
+    ) -> None:
+        """Persist an existing file at the relative arcname."""
+
+    def add_json(self, content: list | dict, arcname: str) -> None:
+        """Persist JSON-serializable content at the relative arcname."""
+
+    def stream(self, arcname: str) -> ContextManager[TextIO]:
+        """Yield a writable text handle for incrementally produced content.
+
+        Reserved for the bulk manifest. At most one stream may be open at a
+        time; add_file/add_json may be called freely while it is open.
+        """
+
+    # __enter__ opens the sink and returns self.
+    # __exit__ calls finalize() on success, abort() on exception.
+```
+
+**Contract / invariants** (the checklist a future sink author honors):
+
+- `arcname` is relative and **POSIX-style (forward slashes)**; the sink maps it to
+  its own namespace (folder: joined under the target; zip: the entry name). The
+  command must build arcnames with `Path(...).as_posix()` — `str(Path(...))`
+  yields backslashes on Windows, which corrupts zip entry names and makes the
+  manifest's stored paths non-portable. The same string is used both as the sink
+  key and as the value stored in the manifest (`EXPORTER_FILE_NAME` etc.), so it
+  must be POSIX at the point of construction. (The share-link bundle path already
+  uses `.as_posix()`; the document targets currently do not and must be fixed.)
+- At most one `stream()` is open at a time. It is the manifest. `add_file` /
+  `add_json` may be interleaved with an open stream — implementations that can't
+  interleave a real stream (zip, S3) must spool the stream to a side buffer and
+  emit it at `finalize()`.
+- The sink is a context manager. Normal exit finalizes; an exception aborts.
+  **No partial or failed run may leave a "complete-looking" artifact.**
+
+### `DirectoryExportSink(target, *, compare_checksums, compare_json, delete)`
+
+Owns everything the command currently does for folder mode:
+
+- On open: snapshot existing files under `target` (today's `files_in_export_dir`).
+- `add_file`: the `check_and_copy` skip logic (mtime/size, or checksum when
+  `compare_checksums`), then copy with stat preservation. Records the arcname as
+  "seen this run".
+- `add_json`: the `check_and_write_json` blake2b compare-or-write (honoring
+  `compare_json`). Records the arcname as seen.
+- `stream`: yields a handle writing to `<arcname>.tmp`; on context close, applies
+  the `compare_json` blake2b compare and renames-or-discards (today's
+  `StreamingManifestWriter` finalize). Records the arcname as seen.
+- `finalize()` (success only): if `delete`, prune every snapshot file not seen
+  this run and clean up emptied directories (today's stale-delete pass).
+- `abort()` (on exception): discard any in-flight `.tmp`; leave existing files
+  intact; do **not** run the prune.
+
+The folder sink is inherently in-place/incremental, not atomic — that is its
+nature and is unchanged. Its safety is the per-file `.tmp`+rename it already does.
+
+### `ZipExportSink(target, zip_name, *, delete)`
+
+- On open: ensure `SCRATCH_DIR` exists (`mkdir(parents=True, exist_ok=True)` —
+  today's `handle()` does this before using it; the sink must do it now), then
+  open a `zipfile.ZipFile` at `<target>/<zip_name>.zip.tmp` (`ZIP_DEFLATED`,
+  `allowZip64=True`). The `.zip.tmp` lives in the same directory as the final
+  `.zip` so the finalize rename is atomic (same filesystem).
+- `add_file` / `add_json`: write the entry directly, first emitting directory
+  marker entries for parent paths so every zip viewer shows the folder structure
+  (today's `_ensure_zip_dirs`). A _flat_ export (no `--use-folder-prefix`, no
+  nested arcnames) has no parent dirs, so it emits **zero** markers — matching
+  today's `make_archive` output for flat trees (keeps the `namelist()` count
+  assertions in `test_export_zipped` valid). Nested/prefixed exports gain marker
+  entries; any count assertion on those must be audited.
+- `stream`: yields a handle writing to a single temp file in `SCRATCH_DIR`.
+- `finalize()` (success only): add the spooled manifest temp file as its entry,
+  close the zip, then **if `delete`, wipe the destination directory** of every
+  pre-existing file/dir except the in-progress `.zip.tmp` and any prior `.zip`
+  (today's zip `--delete` behavior), then atomically rename `.zip.tmp` → `.zip`.
+- `abort()` (on exception): close the zip, unlink the `.zip.tmp`, delete the
+  manifest temp file. **A `.zip` therefore exists only after a fully successful
+  run**, and on abort the destination is never wiped.
+- Rejects `compare_*` (the command guards this before constructing the sink). It
+  does **not** reject `delete` — that is a supported zip behavior (see above).
+
+### Command changes (`document_exporter.py`)
+
+- **`handle()`**: validate the target, then _up front_ raise `CommandError` if
+  `--compare-checksums` or `--compare-json` is combined with `--zip` (those are
+  no-ops in zip mode). `--delete` is **not** rejected — it is passed to whichever
+  sink is built. Construct the appropriate sink (`delete=` passed to both). Run
+  the export as `with sink: self.dump(sink)`. Delete the temp-dir /
+  `shutil.make_archive` block entirely.
+- **`--data-only`**: unchanged in meaning — it simply skips every `sink.add_file`
+  call (no document/thumbnail/archive/bundle files) while the manifest stream and
+  `metadata.json` are still written. Works identically for both sinks; no sink
+  code is data-only-aware. (`test_export_data_only` and its zip equivalent stay
+  green.)
+- **`dump(sink)`**: destination-agnostic. Builds relative arcnames and calls
+  `sink.add_file(...)`, `sink.add_json(...)`, and `sink.stream("manifest.json")`.
+  `self.files_in_export_dir`, `check_and_copy`, `check_and_write_json`, and the
+  stale-delete pass are removed (their logic now lives in the folder sink).
+- **`generate_document_targets`**: returns relative arcnames
+  (`originals/<name>`, `<name>-thumbnail.webp`, `archive/<name>-archive.pdf`)
+  instead of absolute `self.target / ...` paths. It already writes the relative
+  name into `document_dict[EXPORTER_FILE_NAME]` etc.; we just drop the absolute
+  half.
+- **`StreamingManifestWriter`**: simplified to write JSON-array records to the
+  text handle returned by `sink.stream("manifest.json")`. It no longer knows
+  folder vs zip, owns no `.tmp` logic, and has no compare/zip parameters — that
+  behavior moved into each sink's `stream()`.
+- **Crypto / passphrase** handling stays in the command: it transforms record
+  _contents_ before they reach the sink, which is independent of destination.
+- **Progress tracking stays in the command — the sinks know nothing about it.**
+  `PaperlessCommand.track()` wraps the _document iterable_ in `dump()` and ticks
+  the Rich bar once per document. That loop stays in the command; each iteration
+  calls `sink.add_file(...)`, so the per-document progress is preserved
+  unchanged. The sinks deliberately do **not** depend on `PaperlessCommand`,
+  `track()`, or Rich — coupling the destination abstraction to the command
+  framework would defeat the isolation goal and make the sinks impossible to unit
+  -test without a full command. (A sink is a plain context-managed I/O object; it
+  is constructed by `handle()` and exercised directly in `test_sinks.py`.) If
+  finer-grained progress is ever wanted for a single very large file, that is a
+  future enhancement layered via an optional callback — not a `PaperlessCommand`
+  dependency, and out of scope here.
+
+### How `--split-manifest` fits (no sink special-casing)
+
+`--split-manifest` is purely a command-level choice and touches no sink code:
+
+- The single bulk `manifest.json` is always the one and only `sink.stream(...)`
+  handle. In split mode it simply carries fewer record types (document records,
+  notes, and custom-field-instances are redirected out).
+- Per-document `<base>-manifest.json` files are small _complete_ JSON blobs — they
+  were never streamed. `_write_split_manifest` collapses to building the content
+  list and one `sink.add_json(content, "<base>-manifest.json")` call, exactly
+  like `metadata.json`.
+
+Because the manifest stream is backed by its own handle (a `.tmp` file in the
+folder sink, a `SCRATCH_DIR` temp file in the zip sink) and never an open zip
+entry, the per-document `add_json` / `add_file` calls made _while the bulk
+manifest stream is open_ never collide with it.
+
+## Data flow
+
+```
+handle(options)
+  ├─ validate target; reject --compare-* + --zip  → CommandError  (--delete allowed)
+  ├─ sink = DirectoryExportSink(..., delete=…) | ZipExportSink(..., delete=…)
+  └─ with FileLock(MEDIA_LOCK), sink:
+       dump(sink)
+         ├─ with sink.stream("manifest.json") as mh:
+         │    writer = StreamingManifestWriter(mh)
+         │    ├─ global querysets        → writer.write_batch(...)   (encrypted inline)
+         │    ├─ per document:
+         │    │    ├─ sink.add_file(source, "originals/…", checksum=…)
+         │    │    ├─ sink.add_file(thumb,  "…-thumbnail.webp")
+         │    │    ├─ sink.add_file(archive,"archive/…-archive.pdf", checksum=…)
+         │    │    └─ split? sink.add_json(doc_bundle, "…-manifest.json")
+         │    │            : writer.write_record(doc_record)
+         │    └─ per share-link bundle: sink.add_file(...) + writer.write_record(...)
+         └─ sink.add_json(metadata, "metadata.json")
+  (success → sink.finalize();  exception → sink.abort())
+```
+
+## Error handling & atomicity
+
+- Any exception in `dump()` propagates through `with sink:` → `__exit__` →
+  `abort()`. Zip: the `.zip.tmp` and the manifest temp file are deleted, and the
+  destination is **not** wiped; **no `.zip` is produced.** Folder: in-flight
+  `.tmp` files are discarded, existing files are left intact, and the stale-prune
+  does not run.
+- `finalize()` runs only on clean exit, after all contents are written. For the
+  zip: optionally wipe the destination (`--delete`), then the single `.zip.tmp` →
+  `.zip` rename (atomic on the same filesystem). For the folder: the optional
+  stale-delete prune.
+- **Honest limits of the atomicity guarantee.** The guarantee is "no
+  _complete-looking_ `.zip` after a failed run," not "no leftovers." If the
+  process is `SIGKILL`ed or the rename itself fails _after_ the zip is closed, a
+  `.zip.tmp` may be orphaned — that is the safe direction (no false-complete
+  `.zip`), but stale `.zip.tmp` files are **not** auto-cleaned on a later run
+  (matching the prior branch). `KeyboardInterrupt` is a `BaseException` but
+  `__exit__` still runs, so `abort()` fires normally. The rename being atomic and
+  these runs not racing each other both rely on `FileLock(settings.MEDIA_LOCK)`,
+  which serializes exports; concurrent same-`--zip-name` runs are out of scope.
+- The `FileLock(settings.MEDIA_LOCK)` wrapping is unchanged.
+
+## Testing
+
+New `documents/export/tests/test_sinks.py`, unit-testing each sink in isolation
+(pytest classes, factory-boy factories, the `mocker` fixture, `parametrize`, full
+type annotations; run on the Linux VM):
+
+- **Round-trip** (both sinks, parametrized): `add_file` + `add_json` + a streamed
+  manifest produce the expected files/entries with correct relative arcnames.
+- **Folder incremental**: unchanged file is skipped under `compare_checksums` and
+  under `compare_json`; `delete` prunes a snapshot file not written this run and
+  removes emptied directories; without `delete`, stale files remain.
+- **Zip atomicity**: injecting an exception mid-export (via `mocker`) leaves no
+  `.zip` and no leftover `.zip.tmp`, and does not wipe the destination even with
+  `--delete`; a clean run yields exactly the `.zip`. A nested/prefixed export has
+  directory marker entries; a flat export has none.
+- **Zip `--delete`**: a clean `--zip --delete` run wipes pre-existing
+  files/dirs in the destination and produces the `.zip` (preserves
+  `test_export_zipped_with_delete`).
+- **POSIX arcnames**: nested arcnames are stored with forward slashes in both the
+  zip entry names and the manifest values, regardless of host OS (guards the
+  Windows backslash bug).
+- **`--data-only`**: both sinks produce only `manifest.json` + `metadata.json`,
+  no document files.
+- **Stream contract**: opening a second concurrent `stream()` is rejected;
+  `add_file`/`add_json` while a stream is open succeed.
+- **Command guard**: `--zip` with `--compare-checksums` or `--compare-json`
+  raises `CommandError`; `--zip --delete` does **not** error.
+
+Existing `test_management_exporter.py` and `test_management_importer.py` stay
+green unchanged — the export's external behavior (layout, manifest, round-trip
+import, `--zip --delete`, `--data-only`) is preserved.
+
+## Risks
+
+- **Behavior drift in the folder path.** The incremental logic is subtle
+  (mtime/size vs checksum, blake2b json compare, empty-dir cleanup). Mitigation:
+  move it verbatim into the sink and lean on the unchanged command-level tests
+  plus new focused sink tests.
+- **Manifest interleaving in zip mode.** Relies on the spool-to-temp-file
+  decision; the stream contract makes this explicit and the stream-contract test
+  guards it.
@@ -0,0 +1,236 @@
+# Export Zip Compression Control — Design
+
+**Date:** 2026-06-16
+**Branch base:** `dev`
+**Status:** Design complete (zstd facts verified on CPython 3.14.3) — **depends on**
+`2026-06-16-export-sink-architecture-design.md` being implemented first.
+
+## Prerequisite
+
+This builds directly on the export sink refactor. It assumes `ZipExportSink`
+already exists and is the single place that owns `zipfile.ZipFile` creation and
+entry writes. Do not start this until that refactor has landed; without it, the
+change would have to touch the command's zip branches again.
+
+## Problem
+
+Zip export is hardwired to `ZIP_DEFLATED` at the library default level. Users
+have no way to trade speed against archive size — a fast `ZIP_STORED` pass for a
+quick local copy, or a maximal `ZIP_LZMA` pass for the smallest off-site backup.
+The sink refactor turns "which compression" into a single constructor argument,
+so exposing it is now a small, isolated change.
+
+## Goal
+
+Let the operator choose the zip compression method and level from the CLI, with
+behavior identical to today when the flags are omitted. All knowledge of
+compression stays inside `ZipExportSink`; the command only parses flags and maps
+them to sink arguments.
+
+## Scope
+
+In scope:
+
+- `ZipExportSink` gains `compression: int` and `compresslevel: int | None`
+  constructor parameters (default `ZIP_DEFLATED`, `None` → library default),
+  passed straight to `zipfile.ZipFile(...)`.
+- New `document_exporter` flags: `--zip-compression` and
+  `--zip-compression-level`, valid only with `--zip`.
+- Validation: method availability, level range per method, and the
+  requires-`--zip` guard.
+- Import-side: a pre-extract support check in `document_importer` that turns an
+  unsupported codec into a clear `CommandError` (the importer otherwise decompresses
+  transparently via `ZipFile.extractall`).
+- Docs: add both flags and the zstd-portability caveat to `docs/administration.md`
+  (the `document_exporter` option list, lines ~257-270 and the `-z`/`-zn` section,
+  lines ~328-330). New flags are long-form only (`--zip-compression`,
+  `--zip-compression-level`) — no short aliases, to avoid `-zc`/`-zl` collisions
+  with the existing `-z`/`-zn`.
+
+Out of scope:
+
+- Compression for any non-zip sink (folder has none; a future S3 sink would
+  handle its own object storage compression separately).
+- Changing the default. Omitting the flags must produce a byte-compatible-method
+  archive to today's (`ZIP_DEFLATED`, default level).
+
+## Design
+
+### `ZipExportSink` changes
+
+The base sink's signature is `ZipExportSink(target, zip_name, *, delete)`; this
+adds two keyword-only params after `delete`:
+
+```python
+def __init__(
+    self,
+    target: Path,
+    zip_name: str,
+    *,
+    delete: bool = False,
+    compression: int = zipfile.ZIP_DEFLATED,
+    compresslevel: int | None = None,
+) -> None:
+    ...
+    # opened in __enter__:
+    self._zip = zipfile.ZipFile(
+        self._tmp_path,
+        "w",
+        compression=compression,
+        compresslevel=compresslevel,
+        allowZip64=True,
+    )
+```
+
+`ZipFile` applies `compression`/`compresslevel` as the default for every
+`write`/`writestr` (verified: a `ZipFile(..., compression=ZIP_BZIP2)` yields
+entries with `compress_type == ZIP_BZIP2` without per-call args), so `add_file` /
+`add_json` / the manifest entry need no changes. Directory marker entries are
+empty so their compressed payload is zero, but they are still _tagged_ with the
+chosen `compress_type` — harmless, but tests that read `infolist()` should filter
+or account for marker entries (see Testing).
+
+### CLI flags (`document_exporter`)
+
+- `--zip-compression {stored,deflated,bzip2,lzma}` — and `zstd` **when the
+  runtime supports it** (see below). Maps to the matching `zipfile.ZIP_*`
+  constant. Default `deflated`.
+- `--zip-compression-level N` — integer. Per-method accepted ranges (verified
+  against the [3.14 `zipfile` docs](https://docs.python.org/3.14/library/zipfile.html#zipfile.ZipFile)):
+  - `deflated`: **0–9** (`zlib` also accepts `-1` = "default", identical to
+    omitting the flag / `compresslevel=None`).
+  - `bzip2`: **1–9** (`0` is invalid for bzip2).
+  - `lzma`, `stored`: level has **no effect** — passing `--zip-compression-level`
+    with either is a `CommandError`, not a silent accept (consistent with the
+    base refactor's fail-fast posture).
+  - `zstd`: **-131072 … 22** (the documented commonly-accepted range; the
+    authoritative bounds are
+    `compression.zstd.CompressionParameter.compression_level.bounds()`).
+
+  Default: unset → library default (`compresslevel=None`).
+
+Both flags require `--zip`; passing either without `--zip` raises a
+`CommandError`, matching the incremental-flag rule from the base refactor.
+
+**Why validate up front (not let `zipfile` raise) — verified on 3.14.3:** an
+invalid level does _not_ fail at `ZipFile(...)` construction — it fails at the
+**first `write`/`writestr` call**, with an opaque message
+(`ValueError: Invalid initialization option` for deflated > 9, or
+`ValueError: compresslevel must be between 1 and 9` for bzip2). Worse, on context
+exit the half-initialized write handle emits a secondary
+`AttributeError: '_ZipWriteFile' object has no attribute '_compressor'` during GC
+finalization, so the user sees stack-trace noise unrelated to the real cause.
+Up-front validation turns all of that into a single clean `CommandError`.
+
+### Validation (in `handle()`, before constructing the sink)
+
+1. **Requires `--zip`.** Either flag without `--zip` → `CommandError`.
+2. **Method availability — via a named, patchable seam.** Expose a module-level
+   helper `compression_available(method: str) -> bool` that does
+   `try: import bz2 / import lzma / from compression import zstd except ImportError:
+return False` — **not** `importlib.util.find_spec`, which can report a stdlib
+   C-extension as present when importing it actually fails. `stored`/`deflated`
+   are always available (`zlib` is a hard CPython dependency). For `zstd` the probe
+   must import `compression.zstd` (3.14+), not merely check that
+   `zipfile.ZIP_ZSTANDARD` exists. Making this a named function is also what lets
+   the test patch "method unavailable" with `mocker`. If the chosen method is
+   unavailable, raise a `CommandError` naming the missing capability — `zipfile`
+   itself would otherwise raise a bare `RuntimeError`
+   ("Compression requires the (missing) … module").
+3. **Level range.** Reject an out-of-range `--zip-compression-level` for the
+   chosen method with a clear `CommandError`; reject the flag entirely for
+   `stored`/`lzma` (see above).
+
+### zstd (Python 3.14+)
+
+**Verified empirically on CPython 3.14.3** (via `uv run --python 3.14 --no-project`)
+and against [PEP 784](https://peps.python.org/pep-0784/) +
+[the 3.14 `zipfile` docs](https://docs.python.org/3.14/library/zipfile.html):
+
+- The compression-method constant is **`zipfile.ZIP_ZSTANDARD`** (added 3.14; its
+  numeric value is `93`). It does **not** exist on < 3.14.
+- It is backed by the new **`compression.zstd`** stdlib module (PEP 784 added a
+  `compression` namespace package; legacy `bz2`/`lzma`/`zlib` imports are
+  unchanged). `zipfile` raises `RuntimeError` if `compression.zstd` is
+  unavailable when zstd is requested.
+- Accepted `compresslevel` is **`-131072 … 22`**, confirmed at runtime via
+  `compression.zstd.CompressionParameter.compression_level.bounds() == (-131072, 22)`.
+
+Gate everything zstd-related at runtime so nothing is imported or referenced on
+< 3.14 (the project targets Python ≥ 3.11):
+
+```python
+_ZSTD: int | None = getattr(zipfile, "ZIP_ZSTANDARD", None)  # None before 3.14
+```
+
+Presence of the _constant_ does not guarantee the _codec_ is usable, so the
+availability probe (validation step 2) imports `compression.zstd`, not merely
+checks the constant.
+
+Keep `zstd` in the `--zip-compression` `choices` **always** (even on < 3.14), and
+reject it in validation with a friendly "zstd requires Python 3.14+" message. If
+it were dropped from `choices` on older runtimes, argparse would emit a generic
+"invalid choice" that reads as though the option never existed — worse UX.
+
+### Import-side compatibility
+
+`document_importer` reads zips with `ZipFile(self.source).extractall(...)`
+(`document_importer.py:453`), which decompresses each entry transparently using
+whatever method it was stored with — **provided the matching module exists on the
+importing machine.**
+
+The failure mode when it doesn't is unfriendly and must be handled: a zstd (or
+otherwise unsupported) entry raises a bare `NotImplementedError` **per-entry,
+during `extractall`** — _not_ at `ZipFile(self.source)` open, and `is_zipfile()`
+still returns true (a zstd archive is a valid zip container). So the importer
+enters the zip branch, creates its temp dir, may partially extract other entries,
+then blows up mid-extract with no context. **Mitigation (in scope here):** before
+extracting, inspect `ZipFile(self.source).infolist()` compress types and, if any
+is unsupported on this runtime, raise a `CommandError` naming the method and the
+requirement (e.g. "this archive uses zstd, which needs Python 3.14+") instead of
+letting `NotImplementedError` escape.
+
+Per-method summary (document in help text + `administration.md`):
+
+- `deflated`/`stored`: universally importable.
+- `bzip2`/`lzma`: importable wherever the `bz2`/`lzma` modules are present
+  (essentially always).
+- `zstd`: importable only on Python 3.14+. An archive compressed with `zstd` is
+  **not** importable on older runtimes.
+
+## Testing
+
+New cases in the sink tests and an export→import round-trip
+(pytest classes, factory-boy, `mocker`, `parametrize`, typed; run on the VM):
+
+- **Round-trip per method.** Parametrize over the available methods (skip `zstd`
+  below 3.14, skip `bzip2`/`lzma` if the module is somehow absent): export a
+  small library, import it back, assert documents/manifest match.
+- **Method is applied.** Assert each written _file_ entry's `compress_type`
+  equals the requested method (read back via `ZipFile.infolist()`), filtering out
+  directory marker entries (which are tagged but empty).
+- **Level affects size — robustly.** Do **not** compare deflate level 9 vs 1
+  (on small or incompressible fixtures level 9 can equal or slightly exceed level
+  1, causing flaky CI). Instead assert that a compressing method on a
+  moderately-compressible fixture yields a total smaller than `stored`
+  (`ZIP_STORED`), which is a stable invariant.
+- **Validation.** Each flag without `--zip` → `CommandError`; out-of-range level
+  (`--zip-compression-level 99`) → a clean `CommandError` from validation
+  (asserting we never reach the `writestr` that would raise the masked
+  `ValueError`); `--zip-compression-level` with `stored`/`lzma` → `CommandError`;
+  unavailable method (patch the named availability seam with `mocker`) →
+  `CommandError`; on < 3.14, `--zip-compression zstd` → the friendly
+  "requires 3.14+" `CommandError`.
+- **Import pre-check.** An archive containing an unsupported compress type
+  produces a `CommandError` from the importer naming the method, not a raw
+  `NotImplementedError` (simulate by patching the importer's support probe).
+- **Default unchanged.** Omitting both flags yields file entries with
+  `compress_type == ZIP_DEFLATED`, identical to pre-feature behavior.
+
+## Risks
+
+- **Foot-gun archives.** A user could produce a `zstd`/`lzma` archive their
+  import target can't read. Mitigation: explicit help text and the import-side
+  notes above; the default stays the universally-readable `deflated`.
+- **Optional-module assumptions.** Don't assume `bz2`/`lzma` are always compiled
+  in; probe and error clearly. Mitigation: the availability validation step.
@@ -0,0 +1,335 @@
+from __future__ import annotations
+
+import abc
+import hashlib
+import json
+import os
+import shutil
+import tempfile
+import zipfile
+from contextlib import AbstractContextManager
+from contextlib import contextmanager
+from pathlib import Path
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING
+
+from django.conf import settings
+from django.core.serializers.json import DjangoJSONEncoder
+
+from documents.file_handling import delete_empty_directories
+from documents.utils import compute_checksum
+from documents.utils import copy_file_with_basic_stats
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from typing import TextIO
+
+
+def _dumps(content: list | dict) -> str:
+    """Serialize export JSON consistently across all sinks."""
+    return json.dumps(content, cls=DjangoJSONEncoder, indent=2, ensure_ascii=False)
+
+
+class StreamingManifestWriter:
+    """Incrementally writes a JSON array to a text handle, one record at a time.
+
+    Knows nothing about folders or zips: it writes the array framing and records
+    to whatever handle the sink's ``stream()`` yields. The sink owns the handle's
+    lifecycle (atomic rename, compare, spooling).
+    """
+
+    def __init__(self, handle: TextIO) -> None:
+        self._file = handle
+        self._first = True
+        self._file.write("[")
+
+    def write_record(self, record: dict) -> None:
+        if not self._first:
+            self._file.write(",\n")
+        else:
+            self._first = False
+        self._file.write(_dumps(record))
+
+    def write_batch(self, records: list[dict]) -> None:
+        for record in records:
+            self.write_record(record)
+
+    def close(self) -> None:
+        """Write the closing bracket. Does NOT close the handle (the sink owns it)."""
+        self._file.write("\n]")
+
+
+class ExportSink(AbstractContextManager, abc.ABC):
+    """Destination for a document export.
+
+    The command declares export contents via three verbs; the sink decides how to
+    persist each. ``arcname`` is always a relative POSIX path
+    (e.g. ``"manifest.json"``, ``"originals/foo.pdf"``).
+
+    Contract:
+      * At most one ``stream()`` open at a time (it is the manifest);
+        ``add_file``/``add_json`` may be called while it is open.
+      * Context-manager: normal exit finalizes, an exception aborts. No partial or
+        failed run leaves a complete-looking artifact.
+    """
+
+    @abc.abstractmethod
+    def add_file(
+        self,
+        source: Path,
+        arcname: str,
+        *,
+        checksum: str | None = None,
+    ) -> None: ...
+
+    @abc.abstractmethod
+    def add_json(self, content: list | dict, arcname: str) -> None: ...
+
+    @abc.abstractmethod
+    def stream(self, arcname: str):  # -> contextmanager yielding TextIO
+        ...
+
+    def _open(self) -> None:
+        """Hook called on context entry. Override as needed."""
+
+    @abc.abstractmethod
+    def _finalize(self) -> None:
+        """Commit on clean exit."""
+
+    @abc.abstractmethod
+    def _abort(self) -> None:
+        """Roll back on exception."""
+
+    def __enter__(self) -> ExportSink:
+        self._open()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        if exc_type is not None:
+            self._abort()
+        else:
+            self._finalize()
+
+
+class DirectoryExportSink(ExportSink):
+    """Writes loose files into a target directory, with incremental sync.
+
+    Owns the snapshot/skip/compare/prune machinery that used to live in the
+    command (``files_in_export_dir``, ``check_and_copy``, ``check_and_write_json``,
+    and the ``--delete`` pass).
+    """
+
+    def __init__(
+        self,
+        target: Path,
+        *,
+        compare_checksums: bool,
+        compare_json: bool,
+        delete: bool,
+    ) -> None:
+        self._target = target.resolve()
+        self._compare_checksums = compare_checksums
+        self._compare_json = compare_json
+        self._delete = delete
+        self._snapshot: set[Path] = set()
+        self._stream_open = False
+
+    def _open(self) -> None:
+        for x in self._target.glob("**/*"):
+            if x.is_file():
+                self._snapshot.add(x.resolve())
+
+    def add_file(
+        self,
+        source: Path,
+        arcname: str,
+        *,
+        checksum: str | None = None,
+    ) -> None:
+        target = (self._target / arcname).resolve()
+        self._snapshot.discard(target)
+        perform_copy = False
+        if target.exists():
+            source_stat = source.stat()
+            target_stat = target.stat()
+            if self._compare_checksums and checksum:
+                perform_copy = compute_checksum(target) != checksum
+            elif (
+                source_stat.st_mtime != target_stat.st_mtime
+                or source_stat.st_size != target_stat.st_size
+            ):
+                perform_copy = True
+        else:
+            perform_copy = True
+        if perform_copy:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            copy_file_with_basic_stats(source, target)
+
+    def add_json(self, content: list | dict, arcname: str) -> None:
+        target = (self._target / arcname).resolve()
+        json_str = _dumps(content)
+        perform_write = True
+        if target in self._snapshot:
+            self._snapshot.discard(target)
+            if self._compare_json:
+                target_checksum = hashlib.blake2b(target.read_bytes()).hexdigest()
+                src_checksum = hashlib.blake2b(json_str.encode("utf-8")).hexdigest()
+                if src_checksum == target_checksum:
+                    perform_write = False
+        if perform_write:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(json_str, encoding="utf-8")
+
+    @contextmanager
+    def stream(self, arcname: str) -> Iterator[TextIO]:
+        if self._stream_open:
+            raise RuntimeError("A stream is already open on this sink")
+        target = (self._target / arcname).resolve()
+        tmp = target.with_suffix(target.suffix + ".tmp")
+        target.parent.mkdir(parents=True, exist_ok=True)
+        handle = tmp.open("w", encoding="utf-8")
+        self._stream_open = True
+        try:
+            yield handle
+        except BaseException:
+            handle.close()
+            tmp.unlink(missing_ok=True)
+            raise
+        else:
+            handle.close()
+            self._commit_streamed_file(target, tmp)
+        finally:
+            self._stream_open = False
+
+    def _commit_streamed_file(self, target: Path, tmp: Path) -> None:
+        if target in self._snapshot:
+            self._snapshot.discard(target)
+            if self._compare_json:
+                existing = hashlib.blake2b(target.read_bytes()).hexdigest()
+                new = hashlib.blake2b(tmp.read_bytes()).hexdigest()
+                if existing == new:
+                    tmp.unlink()
+                    return
+        tmp.rename(target)
+
+    def _finalize(self) -> None:
+        if self._delete:
+            for f in self._snapshot:
+                f.unlink()
+                delete_empty_directories(f.parent, self._target)
+
+    def _abort(self) -> None:
+        # Folder mode is in-place/incremental: streamed .tmp files are already
+        # cleaned in stream(); leave everything else intact and skip the prune.
+        return None
+
+
+class ZipExportSink(ExportSink):
+    """Writes a single zip archive, produced atomically only on success.
+
+    Builds into ``<target>/<zip_name>.zip.tmp`` and renames to ``.zip`` on clean
+    finalize. The manifest stream is spooled to a temp file in SCRATCH_DIR and
+    added as an entry at finalize (a zip entry cannot be interleaved with others).
+    """
+
+    def __init__(self, target: Path, zip_name: str, *, delete: bool = False) -> None:
+        self._target = target.resolve()
+        self._zip_path = (self._target / zip_name).with_suffix(".zip")
+        self._tmp_path = self._zip_path.with_name(self._zip_path.name + ".tmp")
+        self._delete = delete
+        self._zip: zipfile.ZipFile | None = None
+        self._dirs: set[str] = set()
+        self._pending_manifest: tuple[Path, str] | None = None
+        self._stream_open = False
+
+    def _open(self) -> None:
+        settings.SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
+        self._zip = zipfile.ZipFile(
+            self._tmp_path,
+            "w",
+            compression=zipfile.ZIP_DEFLATED,
+            allowZip64=True,
+        )
+
+    def _ensure_dirs(self, arcname: str) -> None:
+        parts = PurePosixPath(arcname).parts[:-1]
+        for i in range(len(parts)):
+            dir_arc = "/".join(parts[: i + 1]) + "/"
+            if dir_arc not in self._dirs:
+                self._dirs.add(dir_arc)
+                assert self._zip is not None
+                self._zip.mkdir(dir_arc)
+
+    def add_file(
+        self,
+        source: Path,
+        arcname: str,
+        *,
+        checksum: str | None = None,
+    ) -> None:
+        assert self._zip is not None
+        self._ensure_dirs(arcname)
+        self._zip.write(source, arcname=arcname)
+
+    def add_json(self, content: list | dict, arcname: str) -> None:
+        assert self._zip is not None
+        self._ensure_dirs(arcname)
+        self._zip.writestr(arcname, _dumps(content))
+
+    @contextmanager
+    def stream(self, arcname: str) -> Iterator[TextIO]:
+        if self._stream_open:
+            raise RuntimeError("A stream is already open on this sink")
+        settings.SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
+        fd, tmp_name = tempfile.mkstemp(
+            dir=settings.SCRATCH_DIR,
+            prefix="export-manifest-",
+            suffix=".json",
+        )
+        tmp = Path(tmp_name)
+        handle = os.fdopen(fd, "w", encoding="utf-8")
+        self._stream_open = True
+        try:
+            yield handle
+        except BaseException:
+            handle.close()
+            tmp.unlink(missing_ok=True)
+            raise
+        else:
+            handle.close()
+            self._pending_manifest = (tmp, arcname)
+        finally:
+            self._stream_open = False
+
+    def _finalize(self) -> None:
+        assert self._zip is not None
+        if self._pending_manifest is not None:
+            tmp, arcname = self._pending_manifest
+            self._ensure_dirs(arcname)
+            self._zip.write(tmp, arcname=arcname)
+            tmp.unlink(missing_ok=True)
+            self._pending_manifest = None
+        self._zip.close()
+        self._zip = None
+        if self._delete:
+            self._wipe_destination()
+        self._tmp_path.rename(self._zip_path)
+
+    def _wipe_destination(self) -> None:
+        skip = {self._zip_path.resolve(), self._tmp_path.resolve()}
+        for item in self._target.glob("*"):
+            if item.resolve() in skip:
+                continue
+            if item.is_dir():
+                shutil.rmtree(item)
+            else:
+                item.unlink()
+
+    def _abort(self) -> None:
+        if self._zip is not None:
+            self._zip.close()
+            self._zip = None
+        self._tmp_path.unlink(missing_ok=True)
+        if self._pending_manifest is not None:
+            self._pending_manifest[0].unlink(missing_ok=True)
+            self._pending_manifest = None
@@ -0,0 +1,289 @@
+import io
+import json
+import os
+import zipfile
+from pathlib import Path
+
+import pytest
+
+from documents.export.sinks import DirectoryExportSink
+from documents.export.sinks import ExportSink
+from documents.export.sinks import StreamingManifestWriter
+from documents.export.sinks import ZipExportSink
+from documents.export.sinks import _dumps
+
+
+@pytest.fixture()
+def source_file(tmp_path: Path) -> Path:
+    src: Path = tmp_path / "src" / "doc.pdf"
+    src.parent.mkdir(parents=True)
+    src.write_bytes(b"PDF-CONTENT")
+    return src
+
+
+class TestDumps:
+    def test_dumps_is_indented_unicode_json(self) -> None:
+        result: str = _dumps({"a": "é", "b": 1})
+        assert '"é"' in result  # ensure_ascii=False keeps unicode literal
+        assert "\n" in result  # indent=2 produces newlines
+        assert json.loads(result) == {"a": "é", "b": 1}
+
+
+class TestStreamingManifestWriter:
+    def test_writes_json_array_of_records(self) -> None:
+        handle: io.StringIO = io.StringIO()
+        writer: StreamingManifestWriter = StreamingManifestWriter(handle)
+        writer.write_batch([{"pk": 1}, {"pk": 2}])
+        writer.write_record({"pk": 3})
+        writer.close()
+        assert json.loads(handle.getvalue()) == [{"pk": 1}, {"pk": 2}, {"pk": 3}]
+
+    def test_empty_manifest_is_valid_empty_array(self) -> None:
+        handle: io.StringIO = io.StringIO()
+        writer: StreamingManifestWriter = StreamingManifestWriter(handle)
+        writer.close()
+        assert json.loads(handle.getvalue()) == []
+
+
+class TestDirectoryExportSink:
+    def test_add_file_copies_to_relative_arcname(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            sink.add_file(source_file, "originals/doc.pdf")
+        assert (target / "originals" / "doc.pdf").read_bytes() == b"PDF-CONTENT"
+
+    def test_add_json_writes_file(self, tmp_path: Path) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            sink.add_json({"version": "x"}, "metadata.json")
+        assert json.loads((target / "metadata.json").read_text()) == {"version": "x"}
+
+    def test_stream_writes_manifest(self, tmp_path: Path) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            with sink.stream("manifest.json") as handle:
+                writer: StreamingManifestWriter = StreamingManifestWriter(handle)
+                writer.write_record({"pk": 1})
+                writer.close()
+        assert json.loads((target / "manifest.json").read_text()) == [{"pk": 1}]
+
+    def test_add_file_skips_when_size_and_mtime_match(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        # Pre-existing target with identical size+mtime but DIFFERENT content:
+        # if add_file skips (no compare-checksums), the old content survives.
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        existing: Path = target / "originals" / "doc.pdf"
+        existing.parent.mkdir(parents=True)
+        # Same byte length as the source but different content + matching mtime,
+        # so a size/mtime comparison treats it as unchanged and skips the copy.
+        existing.write_bytes(b"X" * len(b"PDF-CONTENT"))
+        stat = source_file.stat()
+        os.utime(existing, (stat.st_atime, stat.st_mtime))
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            sink.add_file(source_file, "originals/doc.pdf", checksum="abc")
+        assert existing.read_bytes() == b"X" * len(b"PDF-CONTENT")  # skipped
+
+    def test_add_file_recopies_when_compare_checksums_differ(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        existing: Path = target / "originals" / "doc.pdf"
+        existing.parent.mkdir(parents=True)
+        existing.write_bytes(b"X" * len(b"PDF-CONTENT"))
+        stat = source_file.stat()
+        os.utime(existing, (stat.st_atime, stat.st_mtime))
+        with DirectoryExportSink(
+            target,
+            compare_checksums=True,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            # wrong checksum forces recopy despite matching size/mtime
+            sink.add_file(source_file, "originals/doc.pdf", checksum="not-the-real-sum")
+        assert existing.read_bytes() == b"PDF-CONTENT"  # recopied
+
+    def test_delete_prunes_unwritten_snapshot_files(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        stale: Path = target / "stale.pdf"
+        stale.write_bytes(b"STALE")
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=True,
+        ) as sink:
+            sink.add_file(source_file, "originals/doc.pdf")
+        assert not stale.exists()
+        assert (target / "originals" / "doc.pdf").exists()
+
+    def test_no_delete_keeps_unwritten_files(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        stale: Path = target / "stale.pdf"
+        stale.write_bytes(b"STALE")
+        with DirectoryExportSink(
+            target,
+            compare_checksums=False,
+            compare_json=False,
+            delete=False,
+        ) as sink:
+            sink.add_file(source_file, "originals/doc.pdf")
+        assert stale.exists()
+
+
+class TestZipExportSink:
+    def test_round_trip_files_json_and_stream(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with ZipExportSink(target, "export", delete=False) as sink:
+            sink.add_file(source_file, "originals/doc.pdf")
+            sink.add_json({"version": "x"}, "metadata.json")
+            with sink.stream("manifest.json") as handle:
+                writer = StreamingManifestWriter(handle)
+                writer.write_record({"pk": 1})
+                writer.close()
+        zip_path: Path = target / "export.zip"
+        assert zip_path.exists()
+        assert not (target / "export.zip.tmp").exists()
+        with zipfile.ZipFile(zip_path) as zf:
+            names = set(zf.namelist())
+            assert {"originals/doc.pdf", "metadata.json", "manifest.json"} <= names
+            assert zf.read("originals/doc.pdf") == b"PDF-CONTENT"
+            assert json.loads(zf.read("manifest.json")) == [{"pk": 1}]
+
+    def test_nested_arcname_emits_directory_marker(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with ZipExportSink(target, "export", delete=False) as sink:
+            sink.add_file(source_file, "originals/doc.pdf")
+        with zipfile.ZipFile(target / "export.zip") as zf:
+            assert "originals/" in zf.namelist()
+
+    def test_flat_arcname_has_no_directory_markers(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with ZipExportSink(target, "export", delete=False) as sink:
+            sink.add_file(source_file, "doc.pdf")
+        with zipfile.ZipFile(target / "export.zip") as zf:
+            assert all(not n.endswith("/") for n in zf.namelist())
+
+    def test_exception_leaves_no_zip_and_no_tmp(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        with pytest.raises(RuntimeError):
+            with ZipExportSink(target, "export", delete=False) as sink:
+                sink.add_file(source_file, "doc.pdf")
+                raise RuntimeError("boom")
+        assert not (target / "export.zip").exists()
+        assert not (target / "export.zip.tmp").exists()
+
+    def test_delete_wipes_destination_on_success(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        (target / "preexisting.txt").write_text("old")
+        (target / "olddir").mkdir()
+        with ZipExportSink(target, "export", delete=True) as sink:
+            sink.add_file(source_file, "doc.pdf")
+        assert (target / "export.zip").exists()
+        assert not (target / "preexisting.txt").exists()
+        assert not (target / "olddir").exists()
+
+    def test_abort_with_delete_does_not_wipe_destination(
+        self,
+        tmp_path: Path,
+        source_file: Path,
+    ) -> None:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        (target / "preexisting.txt").write_text("old")
+        with pytest.raises(RuntimeError):
+            with ZipExportSink(target, "export", delete=True) as sink:
+                sink.add_file(source_file, "doc.pdf")
+                raise RuntimeError("boom")
+        assert (target / "preexisting.txt").exists()
+        assert not (target / "export.zip").exists()
+
+
+class TestStreamContract:
+    @pytest.fixture(params=["dir", "zip"])
+    def sink(self, request: pytest.FixtureRequest, tmp_path: Path) -> ExportSink:
+        target: Path = tmp_path / "out"
+        target.mkdir()
+        if request.param == "dir":
+            return DirectoryExportSink(
+                target,
+                compare_checksums=False,
+                compare_json=False,
+                delete=False,
+            )
+        return ZipExportSink(target, "export", delete=False)
+
+    def test_second_concurrent_stream_is_rejected(self, sink: ExportSink) -> None:
+        with sink:
+            with sink.stream("manifest.json"):
+                with pytest.raises(RuntimeError, match="already open"):
+                    with sink.stream("other.json"):
+                        pass
Author	SHA1	Message	Date
stumpylog	02b97fbaaf	Fix: de-duplicate source_file fixture across sink test classes TestDirectoryExportSink and TestZipExportSink each defined an identical source_file fixture; hoist it to module scope.	2026-07-22 15:38:55 -07:00
stumpylog	0e538acdef	Feature: add ZipExportSink with atomic finalize and manifest spooling	2026-07-22 15:36:50 -07:00
stumpylog	7f58d16758	Fix: make ExportSink a real ABC per design spec The implementation plan for Task 2 diverged from the design spec (export-sink-architecture-design.md), leaving ExportSink as a plain class with NotImplementedError bodies instead of the specified AbstractContextManager subclass. Use abc.ABC + @abstractmethod so a concrete sink missing a required method fails at instantiation rather than at first call.	2026-07-22 15:07:24 -07:00
stumpylog	344414bb47	Feature: add ExportSink ABC and DirectoryExportSink	2026-07-22 14:57:48 -07:00
stumpylog	8263f75643	Feature: add export package with StreamingManifestWriter and _dumps	2026-07-22 14:34:55 -07:00
stumpylog	72d031a7e2	Docs: remove specs/plans unrelated to direct-to-zip export This branch is scoped to the export sink architecture + zip compression control work; drop the search, bulk-edit, ingestion-staging, vector-store, and other unrelated design docs and scratch profiling scripts that had accumulated on it.	2026-07-22 14:26:08 -07:00
stumpylogandClaude Opus 4.8	1a253d0f26	Docs: ingestion staging & enqueue unification implementation plan Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	dc4cb829a6	Docs: revise ingestion staging spec per critical review Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	cad2d2d151	Docs: ingestion staging & enqueue unification design spec Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	8db09d7c92	Docs: bulk-edit operation registry spec + implementation plan Adds the extensibility design that hoists the bulk-edit operation definition (today smeared across serialisers.py, views.py, bulk_edit.py and keyed three different ways) behind a BulkEditOperation registry + PermissionRequirements value object, with per-operation OpenAPI examples. Contract-preserving refactor; both docs reviewed across multiple passes (permission matrix verified, both view call sites accounted for). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	b02a151e45	Docs: export zip compression implementation plan Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	0083443e77	Docs: verify zstd/3.14 facts in compression spec (empirical, uv --python 3.14) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	2cd9b94c43	Docs: export sink architecture implementation plan Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	b76bd9a7bb	Docs: revise export specs per critical review - Preserve --zip --delete (real tested feature), only --compare-* hard-error - Mandate .as_posix() arcnames; --data-only, SCRATCH_DIR, atomicity-honesty notes - Progress tracking stays in command; sinks decoupled from PaperlessCommand - Compression spec: gate/verify zstd, fix import failure mode, patchable probe seam Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	87a9fcadc8	Docs: export sink architecture + zip compression design specs Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylogandClaude Opus 4.8	1c4f6b6144	Docs(beta): flesh out search error-shapes spec and implementation plan Replace the search-error-shapes stub with a full design spec and a TDD implementation plan for friendlier advanced-search error messages. Empirically validated against a live Tantivy index: three error families (UnknownFieldError, InvalidFieldValueError, MalformedQueryError), proactive numeric validation plus a parse_query backstop, comparison operators confirmed working, and a parse-based field drift guard. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-22 14:17:28 -07:00
stumpylog	2d985961b0	Fture search error ideas	2026-07-22 14:17:28 -07:00
Trenton Holmes	08a2186de1	Tracks the tanvity update blocker	2026-07-22 14:17:28 -07:00
Trenton Holmes	a89fdfb033	Tracks the tanvity update blocker	2026-07-22 14:17:28 -07:00
Trenton Holmes	c0fc6c0475	More done work	2026-07-22 14:17:27 -07:00
Trenton Holmes	28f291c503	Store more profiling files	2026-07-22 14:17:27 -07:00
stumpylog	8263460b7b	Removes these for starting implementation	2026-07-22 14:17:27 -07:00
stumpylog	f4e997e524	Marks some things as done	2026-07-22 14:17:27 -07:00
stumpylog	1944208de4	Storing more stuff for stuff	2026-07-22 14:17:27 -07:00
Trenton Holmes	db14bd6b61	Storing more ideas/plans	2026-07-22 14:17:27 -07:00
stumpylog	f41181f992	Additional ideas and plans	2026-07-22 14:17:27 -07:00
stumpylog	b4729ec5d2	Minor plan updates	2026-07-22 14:17:26 -07:00
stumpylog	3cd772d8e9	Just for later ideas, store some brainstorming sessions with Claude	2026-07-22 14:17:26 -07:00