feat: ndjson source adapter — ingest a JSON-Lines spool of pre-extracted records

A generic first-party source adapter (RFC 002 §3.2): each NDJSON line is one
pre-extracted record {content, source_file, metadata, chunk_index} filed as one
verbatim drawer. Byte-preserving (no declared transformations), per-source_file
chunk-index so records sharing a source_file never collide on the deterministic
drawer id, optional --wing routing fallback. Registered first-party in
sources/__init__.py (not the third-party entry-point group). 11 conformance tests.

Carries zero caller-specific vocabulary — any pipeline that pre-extracts records
can mine --source ndjson <spool>.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: joshuafontany

mempalace/sources/__init__.py

@@ -35,6 +35,7 @@
     TransformationViolationError,
 )
 from .context import PalaceContext, ProgressHook
+from .ndjson import NdjsonSourceAdapter
 from .registry import (
     available_adapters,
     get_adapter,
@@ -45,6 +46,11 @@
     unregister,
 )
 
+# First-party in-tree adapters register on import (RFC 002 §3.2 — explicit
+# registration, not the third-party entry-point group). This makes
+# ``mine --source ndjson`` resolve without an install step.
+register(NdjsonSourceAdapter.name, NdjsonSourceAdapter)
+
 __all__ = [
     "AdapterClosedError",
     "AdapterSchema",
@@ -54,6 +60,7 @@
     "FieldSpec",
     "IngestMode",
     "IngestResult",
+    "NdjsonSourceAdapter",
     "PalaceContext",
     "ProgressHook",
     "RouteHint",

mempalace/sources/ndjson.py

@@ -0,0 +1,118 @@
+"""NDJSON source adapter (RFC 002) — ingest a JSON-Lines spool of pre-extracted records.
+
+For any pipeline that has already extracted and chunked content elsewhere and wants
+to hand MemPalace ready-to-file records: each line of the spool is one record, filed
+as one drawer. The adapter is byte-preserving — it stores ``content`` verbatim and
+declares no transformations (the foundational promise; see ``CLAUDE.md`` "Verbatim
+always"). It carries no opinion about *where* the records came from.
+
+One NDJSON line is one record::
+
+    {"content": "...", "source_file": "...", "metadata": {...}, "chunk_index": 0}
+
+``content`` and ``source_file`` are required strings; ``metadata`` (flat scalars) and
+``chunk_index`` (int) are optional. ``metadata`` passes through verbatim — producers
+may attach any flat-scalar fields, opaque to this adapter. When ``chunk_index`` is
+absent the adapter assigns a running per-``source_file`` ordinal so two records that
+share a ``source_file`` never collide on the deterministic drawer id
+(``sha256(source_file)_chunk``).
+
+``SourceRef.local_path`` points at the spool file. The optional ``--wing`` routing
+flag (``SourceRef.options['wing']``) fills the ``wing`` metadata key only where a
+record left it unset (RFC 002 §2.5 — the record's own routing wins).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Iterator
+
+from .base import (
+    AdapterSchema,
+    BaseSourceAdapter,
+    DrawerRecord,
+    IngestResult,
+    SourceAdapterError,
+    SourceNotFoundError,
+    SourceRef,
+)
+
+
+class NdjsonSourceAdapter(BaseSourceAdapter):
+    """File a newline-delimited JSON spool of pre-extracted records into verbatim drawers.
+
+    First-party, in-tree — registered manually (RFC 002 §3.2), not via the
+    third-party entry-point group.
+    """
+
+    name = "ndjson"
+    adapter_version = "0.1.0"
+    capabilities = frozenset({"byte_preserving"})
+    supported_modes = frozenset({"whole_record"})
+    declared_transformations = frozenset()  # verbatim — the producer pre-chunked
+    default_privacy_class = "pii_potential"
+
+    def ingest(
+        self,
+        *,
+        source: SourceRef,
+        palace: "object",  # PalaceContext — broad to avoid the import cycle
+    ) -> Iterator[IngestResult]:
+        path = source.local_path
+        if not path or not os.path.isfile(path):
+            raise SourceNotFoundError(
+                f"ndjson: spool file not found: {path!r} (expected a newline-delimited JSON file)"
+            )
+
+        # Honor the --wing routing precedence (§2.5): a record's own wing wins;
+        # the flag fills in only where the producer left routing unannotated.
+        fallback_wing = source.options.get("wing") if source.options else None
+
+        # Per-source_file ordinal so absent chunk_index values never collide on
+        # the deterministic drawer id. Provided chunk_index values pass through.
+        ordinals: dict[str, int] = {}
+
+        with open(path, encoding="utf-8") as fh:
+            for line_no, raw in enumerate(fh, start=1):
+                line = raw.strip()
+                if not line:
+                    continue
+                try:
+                    record = json.loads(line)
+                except json.JSONDecodeError as exc:
+                    raise SourceAdapterError(
+                        f"ndjson: malformed JSON at {path}:{line_no}: {exc}"
+                    ) from exc
+                if not isinstance(record, dict):
+                    raise SourceAdapterError(f"ndjson: line {path}:{line_no} is not a JSON object")
+
+                content = record.get("content")
+                source_file = record.get("source_file")
+                if not isinstance(content, str) or not isinstance(source_file, str):
+                    raise SourceAdapterError(
+                        f"ndjson: line {path}:{line_no} lacks string 'content' and 'source_file'"
+                    )
+
+                metadata = dict(record.get("metadata") or {})
+                if fallback_wing and "wing" not in metadata:
+                    metadata["wing"] = fallback_wing
+
+                provided = record.get("chunk_index")
+                if isinstance(provided, int):
+                    chunk_index = provided
+                else:
+                    chunk_index = ordinals.get(source_file, 0)
+                ordinals[source_file] = chunk_index + 1
+
+                yield DrawerRecord(
+                    content=content,
+                    source_file=source_file,
+                    chunk_index=chunk_index,
+                    metadata=metadata,
+                )
+
+    def describe_schema(self) -> AdapterSchema:
+        # No fixed structured schema: metadata is producer-defined and flows
+        # through verbatim (upsert_drawer does not reject undeclared fields).
+        return AdapterSchema(version="1.0", fields={})

tests/test_ndjson_source_adapter.py

@@ -0,0 +1,169 @@
+"""Tests for the first-party ``ndjson`` source adapter (RFC 002).
+
+A pipeline that pre-extracts records writes a JSON-Lines spool and runs
+``mempalace mine --source ndjson <spool>``; this adapter reads it back into
+verbatim drawers.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from argparse import Namespace
+
+import pytest
+
+from mempalace.sources import available_adapters
+from mempalace.sources.base import SourceAdapterError, SourceNotFoundError, SourceRef
+from mempalace.sources.ndjson import NdjsonSourceAdapter
+
+
+def _write_spool(tmp_dir, records) -> str:
+    path = os.path.join(tmp_dir, "batch-0.ndjson")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write("\n".join(json.dumps(r) for r in records) + "\n")
+    return path
+
+
+def test_ndjson_is_registered_first_party():
+    # Registered on import of mempalace.sources — no install step needed.
+    assert "ndjson" in available_adapters()
+
+
+def test_ingest_reads_spool_verbatim(tmp_dir):
+    spool = _write_spool(
+        tmp_dir,
+        [
+            {
+                "content": "the record stored exactly as written",
+                "source_file": "feed://abc/item/1",
+                "metadata": {"opaque_field": "kept", "wing": "wing_a"},
+            }
+        ],
+    )
+    adapter = NdjsonSourceAdapter()
+    out = list(adapter.ingest(source=SourceRef(local_path=spool), palace=None))
+
+    assert len(out) == 1
+    drawer = out[0]
+    assert drawer.content == "the record stored exactly as written"  # verbatim
+    assert drawer.source_file == "feed://abc/item/1"
+    assert drawer.chunk_index == 0
+    assert drawer.metadata["opaque_field"] == "kept"  # producer metadata flows through
+    assert drawer.metadata["wing"] == "wing_a"
+
+
+def test_absent_chunk_index_gets_per_source_ordinal(tmp_dir):
+    # Two records sharing a source_file with no chunk_index must NOT collide on
+    # the deterministic drawer id (sha256(source_file)_chunk).
+    spool = _write_spool(
+        tmp_dir,
+        [
+            {"content": "first", "source_file": "feed://x"},
+            {"content": "second", "source_file": "feed://x"},
+            {"content": "other", "source_file": "feed://y"},
+        ],
+    )
+    out = list(NdjsonSourceAdapter().ingest(source=SourceRef(local_path=spool), palace=None))
+
+    triples = [(d.source_file, d.chunk_index, d.content) for d in out]
+    assert triples == [
+        ("feed://x", 0, "first"),
+        ("feed://x", 1, "second"),
+        ("feed://y", 0, "other"),
+    ]
+
+
+def test_provided_chunk_index_passes_through(tmp_dir):
+    spool = _write_spool(
+        tmp_dir,
+        [{"content": "c", "source_file": "feed://x", "chunk_index": 7}],
+    )
+    out = list(NdjsonSourceAdapter().ingest(source=SourceRef(local_path=spool), palace=None))
+    assert out[0].chunk_index == 7
+
+
+def test_wing_option_fills_only_when_absent(tmp_dir):
+    spool = _write_spool(
+        tmp_dir,
+        [
+            {"content": "a", "source_file": "s://1"},
+            {"content": "b", "source_file": "s://2", "metadata": {"wing": "wing_own"}},
+        ],
+    )
+    out = list(
+        NdjsonSourceAdapter().ingest(
+            source=SourceRef(local_path=spool, options={"wing": "wing_flag"}), palace=None
+        )
+    )
+    assert out[0].metadata["wing"] == "wing_flag"  # filled from the flag
+    assert out[1].metadata["wing"] == "wing_own"  # the record's own wing wins
+
+
+def test_blank_lines_skipped(tmp_dir):
+    path = os.path.join(tmp_dir, "batch.ndjson")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write('\n{"content": "c", "source_file": "s"}\n\n')
+    out = list(NdjsonSourceAdapter().ingest(source=SourceRef(local_path=path), palace=None))
+    assert len(out) == 1
+
+
+def test_malformed_json_raises(tmp_dir):
+    path = os.path.join(tmp_dir, "bad.ndjson")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write("{not json}\n")
+    with pytest.raises(SourceAdapterError):
+        list(NdjsonSourceAdapter().ingest(source=SourceRef(local_path=path), palace=None))
+
+
+def test_missing_required_fields_raises(tmp_dir):
+    spool = _write_spool(tmp_dir, [{"content": "no source_file"}])
+    with pytest.raises(SourceAdapterError):
+        list(NdjsonSourceAdapter().ingest(source=SourceRef(local_path=spool), palace=None))
+
+
+def test_missing_spool_raises_not_found(tmp_dir):
+    with pytest.raises(SourceNotFoundError):
+        list(
+            NdjsonSourceAdapter().ingest(
+                source=SourceRef(local_path=os.path.join(tmp_dir, "nope.ndjson")), palace=None
+            )
+        )
+
+
+def test_end_to_end_mine_source_ndjson_files_drawers(tmp_dir, palace_path):
+    from mempalace.cli import _mine_via_source_adapter
+    from mempalace.palace import get_collection
+
+    spool = _write_spool(
+        tmp_dir,
+        [
+            {
+                "content": "pre-extracted record line",
+                "source_file": "feed://e2e/1",
+                "metadata": {"opaque_field": "kept"},
+            }
+        ],
+    )
+    args = Namespace(source="ndjson", dir=spool, wing=None, dry_run=False)
+    _mine_via_source_adapter(args, palace_path)
+
+    col = get_collection(palace_path, create=True)
+    got = col.get(where={"source_file": "feed://e2e/1"})
+    assert got["documents"] == ["pre-extracted record line"]
+    meta = got["metadatas"][0]
+    assert meta["opaque_field"] == "kept"
+    assert meta["adapter_name"] == "ndjson"  # stamped by PalaceContext, not by the adapter
+    assert meta["adapter_version"] == "0.1.0"
+
+
+def test_end_to_end_dry_run_files_nothing(tmp_dir, palace_path):
+    from mempalace.cli import _mine_via_source_adapter
+    from mempalace.palace import get_collection
+
+    spool = _write_spool(tmp_dir, [{"content": "c", "source_file": "feed://dry/1"}])
+    args = Namespace(source="ndjson", dir=spool, wing=None, dry_run=True)
+    _mine_via_source_adapter(args, palace_path)
+
+    col = get_collection(palace_path, create=True)
+    assert col.get(where={"source_file": "feed://dry/1"})["ids"] == []