diff --git a/src/documents/export/__init__.py b/src/documents/export/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/documents/export/sinks.py b/src/documents/export/sinks.py
new file mode 100644
index 000000000..43c7f073e
--- /dev/null
+++ b/src/documents/export/sinks.py
@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from django.core.serializers.json import DjangoJSONEncoder
+
+if TYPE_CHECKING:
+    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]")
diff --git a/src/documents/tests/export/__init__.py b/src/documents/tests/export/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/documents/tests/export/test_sinks.py b/src/documents/tests/export/test_sinks.py
new file mode 100644
index 000000000..296622f02
--- /dev/null
+++ b/src/documents/tests/export/test_sinks.py
@@ -0,0 +1,29 @@
+import io
+import json
+
+from documents.export.sinks import StreamingManifestWriter
+from documents.export.sinks import _dumps
+
+
+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()) == []