feat: arborist.embed — supported library-embedding surface

A stable façade so another Python app can use arborist as a
content-addressed / Merkle / audit-chained store without the CLI or a
wire protocol. Import from arborist.embed, not internal modules, so
refactors don't break embedders.

Surface: open_store(path), ingest_documents(conn, docs), search(conn, q),
plus re-exported Document/Edge/Source/Hit/IngestStats. Core only
(python+sqlite3) — no extras. _IterableSource adapts a plain doc iterable
into the Source contract.

This is the seam for using arborist as neopig's optional provenance
backend: neopig produces Documents from crawled pages, arborist gives
content-dedup (document_root) + FTS5 + an append-only audit chain
alongside neopig's existing md5/FileVault storage. Docs in
docs/embedding.md. 6 tests pin open/ingest/dedup/idempotence/edges/search.

CLAUDE.md

@@ -595,6 +595,9 @@ Architecture / ongoing work:
 - `docs/crawler.md` — web crawler: BFS discovery, robots/feed
   handling, polite vs `--fast`, and the content-addressed
   diagnostics (dedupe by `document_root`, orphan finding).
+- `docs/embedding.md` — embedding arborist as a library in another
+  Python app (`arborist.embed`): produce `Document`s → ingest →
+  dedup + FTS5 + audit chain. The neopig-backend seam.
 - `docs/benchmarks.md` — orientation: harnesses, fixtures,
   signal floor, make targets, bench-row schema, addenda index.
   Read first when running a bench.

arborist/embed.py

@@ -0,0 +1,100 @@
+"""arborist.embed — the supported surface for embedding arborist.
+
+Another Python app (e.g. neopig) that wants arborist as a content-
+addressed, Merkle-committed, audit-chained store imports **from here**,
+not from arborist's internal modules — so internal refactors don't break
+embedders. This module is the versioned seam.
+
+Contract, both sides being Python + SQLite:
+
+- The embedder *produces* :class:`Document` objects (a uri + normalized
+  text + optional outbound :class:`Edge` links) and hands them to
+  :func:`ingest_documents` (or implements a :class:`Source`).
+- arborist owns its own SQLite file. **Every write goes through this
+  API** — the embedder never touches arborist's tables directly (which
+  also keeps an ORM-based app's "no raw SQL" rule intact: there is no
+  SQL to write).
+- Ingest is idempotent: same content → same ``document_root`` → no-op.
+  This is the dedup-by-content property; the embedder's own
+  content-address key (md5, etc.) is orthogonal and untouched.
+
+Everything here is arborist *core* (python + sqlite3). No ``[crawler]``,
+``[nli]``, or ``[vec]`` extra is required to embed.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Iterator
+
+from arborist.document import Document, Edge
+from arborist.ingest import IngestStats, ingest_source
+from arborist.search import FTS5Backend, Hit
+from arborist.source import Source
+from arborist.store import connect
+
+__all__ = [
+    "Document",
+    "Edge",
+    "Source",
+    "Hit",
+    "IngestStats",
+    "open_store",
+    "ingest_documents",
+    "search",
+]
+
+
+def open_store(db_path: Path | str) -> sqlite3.Connection:
+    """Open (creating + migrating the schema if needed) an arborist store.
+
+    Returns a ``sqlite3.Connection`` the caller owns and must close. Safe
+    to call repeatedly against the same path; the schema is idempotent.
+    """
+    return connect(db_path)
+
+
+class _IterableSource(Source):
+    """Adapt a plain iterable of Documents into the Source contract.
+
+    Most embedders already have their documents in hand (one per crawled
+    page, say) and don't need a stateful corpus object — they just want
+    to push a batch through the pipeline. This wraps that case.
+    """
+
+    def __init__(self, documents: Iterable[Document], source_type: str = "embedded"):
+        self._documents = documents
+        self.source_type = source_type
+
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self._documents
+
+
+def ingest_documents(
+    conn: sqlite3.Connection,
+    documents: Iterable[Document],
+    *,
+    source_type: str = "embedded",
+    **ingest_kwargs,
+) -> IngestStats:
+    """Ingest an iterable of :class:`Document` objects into ``conn``.
+
+    Convenience wrapper over :func:`ingest_source` for the common case
+    where the embedder already holds the documents. Idempotent at the
+    content level (same content → same ``document_root`` → no-op insert).
+    Extra keyword args pass straight through to :func:`ingest_source`
+    (``chunker_name``, ``batch_size``, ``progress`` …). Returns
+    :class:`IngestStats`.
+    """
+    return ingest_source(conn, _IterableSource(documents, source_type), **ingest_kwargs)
+
+
+def search(conn: sqlite3.Connection, query: str, *, limit: int = 20) -> list[Hit]:
+    """FTS5 BM25 search over ingested chunk content. Returns ``list[Hit]``.
+
+    The lexical entry point — no LLM, no extras. For the full multi-route
+    retrieval / RAG path use ``arborist.qa.query`` directly.
+    """
+    return FTS5Backend(conn).search(query, limit=limit)

docs/embedding.md

@@ -0,0 +1,77 @@
+# Embedding arborist in another Python app
+
+arborist is usable as a library: another Python project can use it as a
+content-addressed, Merkle-committed, audit-chained store for documents it
+already has — without the CLI, a server, or any wire protocol. The
+supported surface is **`arborist.embed`**. Import from there, not from
+arborist's internal modules, so internal refactors don't break you.
+
+This is how a downstream archival/crawler app (e.g. neopig) can gain
+verifiable dedup, FTS5 search, and an append-only audit chain over its
+archived page text while keeping its own storage for everything else.
+
+## The contract
+
+1. **You produce `Document`s.** A `Document` is a `uri` + normalized text
+   `content` + a `source_type` tag (+ optional `title`, `edges`, `extra`).
+2. **You hand them to ingest.** arborist canonicalizes → chunks → Merkle-
+   roots → FTS5-indexes → writes one audit event. Idempotent: same content
+   → same `document_root` → no-op.
+3. **arborist owns its SQLite file.** Every write goes through this API;
+   you never touch arborist's tables. (That also means an ORM app with a
+   "no raw SQL" rule stays clean — there's no SQL for you to write.)
+4. **Your own content-address key is orthogonal.** If you already dedup by
+   md5 (or anything), keep it — arborist's `document_root` is an
+   *additional* hard-hash commitment, not a replacement. See the
+   soft-hash vs hard-hash rule in the project CLAUDE.md.
+
+`arborist.embed` is **core** (python + sqlite3); no `[crawler]`/`[nli]`/
+`[vec]` extra is needed to embed.
+
+## Minimal use
+
+```python
+from arborist.embed import open_store, ingest_documents, search, Document, Edge
+
+conn = open_store("data/arborist.db")          # creates + migrates schema
+
+ingest_documents(conn, [
+    Document(
+        uri="https://example.com/post",
+        content="the page's extracted prose",
+        source_type="neopig_html",
+        title="A Post",
+        edges=[Edge(edge_type="embeds_media", dst_uri="https://example.com/img.jpg")],
+        extra={"crawl_job_id": 7, "md5": "..."},   # your provenance, carried along
+    ),
+])
+
+for hit in search(conn, "extracted prose", limit=10):
+    print(hit.document_uri)
+
+conn.close()
+```
+
+For a stateful corpus, subclass `Source` (set `source_type`, implement
+`iter_documents()`) and call `arborist.ingest.ingest_source` — exactly how
+arborist's own `sources/` are written. For full multi-route retrieval /
+RAG, use `arborist.qa.query` directly.
+
+## What you get
+
+- **Content dedup with proofs** — identical content (across different URIs)
+  collapses to one `document_root`; per-chunk leaves expose partial
+  overlap. See `docs/crawler.md`.
+- **FTS5 search** over chunked content.
+- **Append-only audit chain** — every ingest is one hash-linked
+  `audit_events` row; tamper becomes detectable.
+- **Mesh-ready** — peers can re-derive and cross-verify roots.
+
+## Surface (`arborist.embed`)
+
+| Symbol | What |
+|---|---|
+| `open_store(db_path)` | open/create/migrate a store; returns a `sqlite3.Connection` you own |
+| `ingest_documents(conn, docs, *, source_type=…, **kw)` | ingest an iterable of `Document`s; returns `IngestStats` |
+| `search(conn, query, *, limit=20)` | FTS5 BM25 over chunk content; returns `list[Hit]` |
+| `Document`, `Edge`, `Source`, `Hit`, `IngestStats` | re-exported types |

tests/test_embed.py

@@ -0,0 +1,111 @@
+"""Tests for arborist.embed — the supported embedding surface.
+
+These pin the contract a downstream Python app (e.g. neopig) depends on:
+open a store, ingest Documents, get content-level dedup, search them.
+"""
+
+from __future__ import annotations
+
+from arborist.embed import (
+    Document,
+    Edge,
+    Source,
+    open_store,
+    ingest_documents,
+    search,
+)
+
+
+def _doc(uri, content, title=None, source_type="embedded"):
+    return Document(uri=uri, content=content, source_type=source_type, title=title)
+
+
+def test_open_store_creates_schema(tmp_path):
+    conn = open_store(tmp_path / "embed.db")
+    try:
+        # A fresh store has the documents table (schema migrated on open).
+        names = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+        assert "documents" in names
+        assert "audit_events" in names
+    finally:
+        conn.close()
+
+
+def test_ingest_documents_lands_and_is_searchable(tmp_path):
+    conn = open_store(tmp_path / "embed.db")
+    try:
+        stats = ingest_documents(
+            conn,
+            [
+                _doc("https://x.com/a", "the quick brown fox jumps", title="A"),
+                _doc("https://x.com/b", "lorem ipsum dolor sit amet", title="B"),
+            ],
+        )
+        assert stats.inserted == 2
+        n = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
+        assert n == 2
+
+        hits = search(conn, "brown fox", limit=10)
+        assert any("x.com/a" in h.document_uri for h in hits), hits
+    finally:
+        conn.close()
+
+
+def test_ingest_is_idempotent_on_content(tmp_path):
+    """Same content → same document_root → no-op re-insert (dedup)."""
+    conn = open_store(tmp_path / "embed.db")
+    try:
+        ingest_documents(conn, [_doc("https://x.com/a", "identical body text here")])
+        first = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
+        # Re-ingest the same content (same uri) — must not duplicate.
+        stats = ingest_documents(conn, [_doc("https://x.com/a", "identical body text here")])
+        second = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
+        assert first == second == 1
+        assert stats.inserted == 0
+    finally:
+        conn.close()
+
+
+def test_dedup_is_content_not_uri(tmp_path):
+    """Two different URIs with identical content share one document_root —
+    the property an embedder leans on for cross-source dedup."""
+    conn = open_store(tmp_path / "embed.db")
+    try:
+        ingest_documents(
+            conn,
+            [
+                _doc("https://x.com/a", "byte identical content"),
+                _doc("https://y.com/mirror", "byte identical content"),
+            ],
+        )
+        roots = [r[0] for r in conn.execute("SELECT DISTINCT document_root FROM documents")]
+        assert len(roots) == 1, "identical content must collapse to one root"
+    finally:
+        conn.close()
+
+
+def test_edges_survive_ingest(tmp_path):
+    conn = open_store(tmp_path / "embed.db")
+    try:
+        page = Document(
+            uri="https://x.com/page",
+            content="a page that links to an image",
+            source_type="neopig_html",
+            title="Page",
+            edges=[Edge(edge_type="embeds_media", dst_uri="https://x.com/img.jpg")],
+        )
+        ingest_documents(conn, [page])
+        n_edges = conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0]
+        assert n_edges >= 1
+    finally:
+        conn.close()
+
+
+def test_iterable_source_is_a_source_subclass():
+    """The embed wrapper must satisfy the Source contract so it flows
+    through the same ingest pipeline as any first-class corpus."""
+    from arborist.embed import _IterableSource
+
+    src = _IterableSource([], source_type="embedded")
+    assert isinstance(src, Source)
+    assert src.source_type == "embedded"