feat(plugin-embeddings): batch embed() calls through a SQLite vector store

docs/plugins.md

@@ -93,9 +93,10 @@ scope.
 ### Vector cache
 
 Computed vectors are cached at `~/.cache/dirsql/embeddings/` (respecting
-`XDG_CACHE_HOME`), keyed on the SHA-256 of the value bytes plus the model
-identifier. Changing either recomputes — switching models never serves stale
-vectors — and re-running a query over unchanged files is cache hits all the
-way. There is no eviction: the directory is **safe to wipe at any time**; the
-only cost is re-embedding. The cache never lives inside a queried tree — the
-worker writes nothing into the directories you query.
+`XDG_CACHE_HOME`): one SQLite database per model identifier, each row keyed
+on the SHA-256 of the value bytes. Changing the value or the model recomputes
+— switching models never serves stale vectors — and re-running a query over
+unchanged files is cache hits all the way, a whole batch answered by one
+query. There is no eviction: the directory is **safe to wipe at any time**;
+the only cost is re-embedding. The cache never lives inside a queried tree —
+the worker writes nothing into the directories you query.

plugins/dirsql-plugin-embeddings/README.md

@@ -53,9 +53,10 @@ Results print one `path<TAB>distance` line per match, closest first.
 `embed()` is inert until a query calls it: no worker process is spawned and
 no model is loaded for queries that never use it. On the first call, dirsql
 spawns the plugin's worker process (`dirsql-plugin-embeddings worker`), which
-serves every call of the invocation over stdin/stdout. Only the values the
-query actually selects are embedded — the worker receives values, not paths,
-and never opens files itself.
+serves every call of the invocation over stdin/stdout. A statement's values
+reach the worker in batches of up to 4096, each encoded in one model call.
+Only the values the query actually selects are embedded — the worker receives
+values, not paths, and never opens files itself.
 
 ## Model
 
@@ -79,12 +80,14 @@ The one-liner's `--model` flag templates the same second argument.
 ## Vector cache
 
 Computed vectors are cached at `~/.cache/dirsql/embeddings/` (or
-`$XDG_CACHE_HOME/dirsql/embeddings/` when `XDG_CACHE_HOME` is set), keyed by
-the SHA-256 of the value bytes plus the model identifier — changing either
-recomputes; switching models never serves stale vectors. There is no
-eviction: **the directory is safe to wipe at any time**; the only cost is
-re-embedding. The cache never lives inside a queried tree — the worker
-receives values, not paths, and writes nothing anywhere else.
+`$XDG_CACHE_HOME/dirsql/embeddings/` when `XDG_CACHE_HOME` is set): one
+SQLite database per model identifier, each row keyed by the SHA-256 of the
+value bytes. Changing the value or the model recomputes; switching models
+never serves stale vectors. A batch is looked up in one query and its misses
+written in one transaction. There is no eviction: **the directory is safe to
+wipe at any time**; the only cost is re-embedding. The cache never lives
+inside a queried tree — the worker receives values, not paths, and writes
+nothing anywhere else.
 
 ## Docs
 

plugins/dirsql-plugin-embeddings/changelog.d/2026-10-02-batched-embed-sqlite-cache.md

@@ -0,0 +1,4 @@
+**Changed**
+
+- **`embed()` now encodes a statement's values in batches.** The packaged fragment sets `batch = 4096`, so dirsql hands the worker up to 4096 values per request and the worker answers them with one `model.encode()` call per model instead of one per value. Embedding every abstract in a corpus, startup excluded, went from 4.5x the time of a bare batch encode (32768 abstracts) to 0.83x of it (65536 abstracts). The worker still answers single `{"call": [...]}` requests. (#1204)
+- **The vector cache is one SQLite database per model.** `~/.cache/dirsql/embeddings/` now holds `<sha256 of model identifier>.db` files, each a `vectors(digest, vector)` table; a batch is looked up in one query and its misses stored in one transaction. The `cachetta` dependency is gone. Vectors written by earlier releases (one file per value) are ignored, never read; wipe the directory to reclaim their space. (#1204)

plugins/dirsql-plugin-embeddings/e2e-attestations/claude-1204-embed-plugin-batch.json

@@ -0,0 +1,7 @@
+{
+  "command": "uv run --with-editable ../../packages/python python -m pytest tests/e2e/ -x -q",
+  "ran_at": 1790960575,
+  "exit_code": 0,
+  "commit": "64524926303ffd11d9f107fd843b6c2f21648bbd",
+  "branch": "claude/1204-embed-plugin-batch"
+}

plugins/dirsql-plugin-embeddings/migrations.d/2026-10-02-batched-embed-sqlite-cache.md

@@ -0,0 +1,44 @@
+### dirsql-plugin-embeddings: vector cache moves to SQLite, worker requests are batched
+
+#### Summary
+
+The on-disk vector cache changes format: `~/.cache/dirsql/embeddings/` (or
+`$XDG_CACHE_HOME/dirsql/embeddings/`) now holds one SQLite database per model
+identifier instead of one `cachetta` file per cached value. Nothing reads the
+old files. The first query after upgrading re-embeds everything it touches;
+later runs hit the new store. The worker protocol gains the batched
+`{"calls": [...]}` request alongside the single `{"call": [...]}` request,
+and the packaged fragment opts in with `batch = 4096`, which needs dirsql
+0.4.60 or later.
+
+#### Required changes
+
+| Surface | Before | After |
+| ------- | ------ | ----- |
+| Cache directory contents | One `cachetta` file per cached value | One `<sha256 of model identifier>.db` SQLite database per model; old files are dead weight — wipe the directory to reclaim their space |
+| `dirsql` version with the packaged fragment | Any release with `[[dirsql.function]]` | 0.4.60 or later (`batch` is rejected as an unknown field by older releases) |
+| Worker protocol | `{"call": [value, model_id?]}` only | Also `{"calls": [[value, model_id?], ...]}` answered by `{"results": [...]}` in request order |
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+- The first query after upgrading reports every `embed()` call as computed
+  rather than cached, even over files embedded before, because the old
+  per-value files are never read.
+- A statement's `embed()` values reach the worker in batches of up to 4096;
+  a failing model fails every call in its batch group with the same error
+  instead of one call at a time.
+
+#### Verification
+
+```bash
+cd "$(mktemp -d)"
+printf 'hello world\n' > note.md
+uvx --with dirsql-plugin-embeddings dirsql query "SELECT path, vec_length(embed(content)) AS dims FROM '*.md'"
+# expected: [{"path":"note.md","dims":512}]
+ls "${XDG_CACHE_HOME:-$HOME/.cache}/dirsql/embeddings/"
+# expected: exactly one <64 hex chars>.db file for the default model
+```

plugins/dirsql-plugin-embeddings/pyproject.toml

@@ -17,7 +17,8 @@ requires-python = ">=3.10"
 #
 # model2vec (NOT sentence-transformers/torch) is the inference stack: static
 # embeddings whose runtime deps are numpy + tokenizers, viable in uvx
-# ephemeral envs. cachetta is the vector cache; tqdm the stderr progress UI.
+# ephemeral envs. The vector cache is stdlib sqlite3; tqdm the stderr
+# progress UI.
 #
 # dirsql is the SDK the one-liner search command runs its generated SQL
 # through. Functionally it needs >= 0.4.17 (the first release with the core
@@ -30,7 +31,6 @@ dependencies = [
     "dirsql",
     "sqlite-vec>=0.1",
     "model2vec>=0.6",
-    "cachetta>=0.7",
     "tqdm>=4",
     "click>=8",
 ]

plugins/dirsql-plugin-embeddings/src/dirsql_plugin_embeddings/dirsql.toml

@@ -7,10 +7,12 @@ entrypoint = "sqlite3_vec_init"
 
 # The embed() SQL scalar function, served by the plugin's worker process over
 # newline-delimited JSON on stdin/stdout. Inert until a query calls it. The
-# generous timeout absorbs the first call's model download.
+# generous timeout absorbs the first call's model download. A statement's
+# values reach the worker in batches, each encoded in one model call.
 [[dirsql.function]]
 name = "embed"
 args = [1, 2]
 command = "dirsql-plugin-embeddings worker"
 deterministic = true
 timeout = "600s"
+batch = 4096

plugins/dirsql-plugin-embeddings/src/dirsql_plugin_embeddings/embedding/cache.py

@@ -1,19 +1,8 @@
 import os
-from datetime import timedelta
 from pathlib import Path
 
-from cachetta import Cachetta
-
-# No eviction: entries stay until the user wipes the directory (documented as
-# safe -- the only cost is re-embedding).
-NO_EVICTION = timedelta(days=365000)
-
 
 def cache_dir():
     xdg = os.environ.get("XDG_CACHE_HOME", "")
     base = Path(xdg) if xdg else Path.home() / ".cache"
     return base / "dirsql" / "embeddings"
-
-
-def make_cache():
-    return Cachetta(path=cache_dir(), hashed=True, duration=NO_EVICTION)

plugins/dirsql-plugin-embeddings/src/dirsql_plugin_embeddings/embedding/cache_test.py

@@ -1,4 +1,3 @@
-from datetime import timedelta
 from pathlib import Path
 from unittest.mock import patch
 
@@ -19,20 +18,3 @@ def it_treats_an_empty_xdg_cache_home_as_unset():
         with patch.dict(cache.os.environ, {"XDG_CACHE_HOME": ""}):
             with patch.object(cache.Path, "home", return_value=Path("/home/u")):
                 assert cache.cache_dir() == Path("/home/u/.cache/dirsql/embeddings")
-
-
-def describe_make_cache():
-    def it_builds_a_hashed_cachetta_at_the_cache_dir_with_no_eviction():
-        with patch.object(cache, "Cachetta") as cachetta:
-            with patch.object(
-                cache, "cache_dir", return_value=Path("/somewhere")
-            ) as directory:
-                built = cache.make_cache()
-        directory.assert_called_once_with()
-        cachetta.assert_called_once_with(
-            path=Path("/somewhere"), hashed=True, duration=cache.NO_EVICTION
-        )
-        assert built is cachetta.return_value
-
-    def it_pins_no_eviction_to_a_thousand_years():
-        assert cache.NO_EVICTION == timedelta(days=365000)

plugins/dirsql-plugin-embeddings/src/dirsql_plugin_embeddings/embedding/store.py

@@ -0,0 +1,53 @@
+import sqlite3
+from array import array
+from hashlib import sha256
+from json import dumps
+
+SCHEMA = (
+    "CREATE TABLE IF NOT EXISTS vectors"
+    " (digest TEXT PRIMARY KEY, vector BLOB NOT NULL)"
+)
+LOOKUP = (
+    "SELECT digest, vector FROM vectors"
+    " WHERE digest IN (SELECT value FROM json_each(?))"
+)
+INSERT = "INSERT OR REPLACE INTO vectors (digest, vector) VALUES (?, ?)"
+
+
+def pack(vector):
+    return array("d", vector).tobytes()
+
+
+def unpack(blob):
+    doubles = array("d")
+    doubles.frombytes(blob)
+    return doubles.tolist()
+
+
+class VectorStore:
+    """One SQLite database per model under ``directory``, keyed by text digest."""
+
+    def __init__(self, directory):
+        self._directory = directory
+        self._connections = {}
+
+    def _connection(self, identifier):
+        if identifier not in self._connections:
+            self._directory.mkdir(parents=True, exist_ok=True)
+            name = sha256(identifier.encode("utf-8")).hexdigest() + ".db"
+            connection = sqlite3.connect(self._directory / name)
+            connection.execute(SCHEMA)
+            self._connections[identifier] = connection
+        return self._connections[identifier]
+
+    def lookup(self, identifier, digests):
+        """The stored vectors among ``digests``, keyed by digest."""
+        rows = self._connection(identifier).execute(LOOKUP, (dumps(list(digests)),))
+        return {digest: unpack(blob) for digest, blob in rows}
+
+    def insert(self, identifier, vectors):
+        """Store ``vectors`` (digest to vector) in one transaction."""
+        connection = self._connection(identifier)
+        rows = [(digest, pack(vector)) for digest, vector in vectors.items()]
+        with connection:
+            connection.executemany(INSERT, rows)

plugins/dirsql-plugin-embeddings/src/dirsql_plugin_embeddings/embedding/store_test.py

@@ -0,0 +1,123 @@
+from hashlib import sha256
+from unittest.mock import MagicMock, call, patch
+
+from . import store
+
+
+def make_store():
+    directory = MagicMock()
+    return store.VectorStore(directory), directory
+
+
+def describe_databases():
+    def it_keeps_one_database_per_model_named_by_the_identifier_digest():
+        built, directory = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            built.lookup("m1", ["d"])
+            built.lookup("m2", ["d"])
+        assert directory.__truediv__.call_args_list == [
+            call(sha256(b"m1").hexdigest() + ".db"),
+            call(sha256(b"m2").hexdigest() + ".db"),
+        ]
+        assert connect.call_args_list == [call(directory / "x")] * 2
+
+    def it_creates_the_cache_directory_before_opening():
+        built, directory = make_store()
+        with patch.object(store.sqlite3, "connect"):
+            built.lookup("m1", [])
+        directory.mkdir.assert_called_once_with(parents=True, exist_ok=True)
+
+    def it_creates_the_table_on_open():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            built.lookup("m1", [])
+        assert connect.return_value.execute.call_args_list[0] == call(store.SCHEMA)
+
+    def it_opens_each_model_once():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            built.lookup("m1", ["a"])
+            built.insert("m1", {"a": [1.0]})
+            built.lookup("m1", ["a"])
+        assert connect.call_count == 1
+
+
+def describe_lookup():
+    def it_finds_nothing_on_a_cold_lookup():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            connect.return_value.execute.return_value = []
+            assert built.lookup("m1", ["a", "b"]) == {}
+
+    def it_unpacks_every_stored_row_keyed_by_digest():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            connect.return_value.execute.return_value = [
+                ("a", store.pack([1.0, -0.5])),
+                ("c", store.pack([3.0])),
+            ]
+            found = built.lookup("m1", ["a", "b", "c"])
+        assert found == {"a": [1.0, -0.5], "c": [3.0]}
+
+    def it_asks_for_the_whole_batch_in_one_query():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            connect.return_value.execute.return_value = []
+            built.lookup("m1", ["a", "b", "c"])
+        selects = [
+            args
+            for args, _ in connect.return_value.execute.call_args_list
+            if args[0] is store.LOOKUP
+        ]
+        assert selects == [(store.LOOKUP, ('["a", "b", "c"]',))]
+
+    def it_accepts_any_iterable_of_digests():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            connect.return_value.execute.return_value = []
+            built.lookup("m1", iter(["a", "b"]))
+        connect.return_value.execute.assert_called_with(
+            store.LOOKUP, ('["a", "b"]',)
+        )
+
+
+def describe_insert():
+    def it_writes_the_packed_batch_in_one_transaction():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            connection = connect.return_value
+            built.insert("m1", {"a": [1.0], "b": [2.0, 3.0]})
+        connection.__enter__.assert_called_once()
+        connection.__exit__.assert_called_once()
+        connection.executemany.assert_called_once_with(
+            store.INSERT,
+            [("a", store.pack([1.0])), ("b", store.pack([2.0, 3.0]))],
+        )
+
+    def it_writes_inside_the_transaction_not_before_it():
+        built, _ = make_store()
+        events = []
+        with patch.object(store.sqlite3, "connect") as connect:
+            connection = connect.return_value
+            connection.__enter__.side_effect = lambda: events.append("enter")
+            connection.executemany.side_effect = lambda *_: events.append("write")
+            built.insert("m1", {"a": [1.0]})
+        assert events == ["enter", "write"]
+
+    def it_accepts_an_empty_batch():
+        built, _ = make_store()
+        with patch.object(store.sqlite3, "connect") as connect:
+            built.insert("m1", {})
+        connect.return_value.executemany.assert_called_once_with(store.INSERT, [])
+
+
+def describe_pack():
+    def it_encodes_doubles_and_unpack_restores_them_exactly():
+        vector = [0.1, -2.5e-8, 1e300, 0.0, 3.0]
+        assert store.unpack(store.pack(vector)) == vector
+
+    def it_encodes_eight_bytes_per_component():
+        assert len(store.pack([1.0, 2.0, 3.0])) == 24
+
+    def it_unpacks_an_empty_blob_to_an_empty_vector():
+        assert store.unpack(b"") == []