Report cache hits from the embeddings worker (plugin)

plugins/dirsql-plugin-embeddings/changelog.d/2026-09-04-report-cache-hits.md

@@ -0,0 +1,3 @@
+**Added**
+
+- **The worker now tells dirsql which embeddings cost nothing.** Every `ok` response carries `"meta": {"cached": <bool>}`, so the progress line can split its worker-call count — `dirsql: ran 41231 worker calls in 2m41s (38104 cached)` — instead of reporting one undifferentiated total for a run that was mostly disk reads. `cachetta`'s wrapper returns a hit and a miss identically, so the flag is recovered from whether the wrapped compute actually ran. (#1034)

plugins/dirsql-plugin-embeddings/e2e-attestations/claude-tackle-1034-plugin-flag.json

@@ -0,0 +1,7 @@
+{
+  "command": "uv run --with-editable ../../packages/python python -m pytest tests/e2e/ -x -q",
+  "ran_at": 1788558384,
+  "exit_code": 0,
+  "commit": "fc3ee82adb022a9b6da85687695ca752dd178f03",
+  "branch": "claude/tackle-1034-plugin-flag"
+}

plugins/dirsql-plugin-embeddings/migrations.d/2026-09-04-report-cache-hits.md

@@ -0,0 +1,39 @@
+### dirsql-plugin-embeddings: `ok` responses now carry cache metadata
+
+#### Summary
+
+The worker's success responses gain an optional `"meta"` object:
+`{"ok": [floats...], "meta": {"cached": true|false}}`. Nothing is removed and
+no signature a user calls changes — the plugin exports no Python API, and its
+CLI arguments, exit codes and vectors are all untouched. The one observable
+difference is the extra key on the wire, and the progress line dirsql draws
+from it.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+- A worker `ok` line previously read `{"ok": [1.0, 0.0]}` and now reads
+  `{"ok": [1.0, 0.0], "meta": {"cached": false}}`. dirsql ignores unknown keys,
+  so this is additive for it; a third party parsing worker output with an
+  exact-equality check on the whole object will see a mismatch and should
+  compare `response["ok"]` instead.
+- `{"ok": null}` for a NULL value is unchanged — no compute is attempted, so
+  there is no cache state to report.
+- With the plugin installed, `dirsql` runs that hit the on-disk cache now show
+  the split in the progress line (`ran 41231 worker calls in 2m41s
+  (38104 cached)`). A run with no hits looks exactly as it did before.
+
+#### Verification
+
+```bash
+printf '{"call": ["hello"]}\n{"call": ["hello"]}\n' \
+  | uvx --from dirsql-plugin-embeddings dirsql-plugin-embeddings worker
+# expected: first line "cached":false, second line "cached":true
+```

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

@@ -18,13 +18,18 @@ def __init__(self):
         self._cache = make_cache()
         self._compute_cached = self._cache.wrap(self._compute)
         self._pending = None
+        self._computed = False
 
     def _model(self, model_id):
         if model_id not in self._models:
             self._models[model_id] = model.load_model(model_id)
         return self._models[model_id]
 
     def _compute(self, digest, identifier):
+        # cachetta's wrapper returns a hit and a miss identically, so the only
+        # place a miss is observable is here: _compute runs on a miss and never
+        # on a hit. embed() clears the flag and reads it back afterwards.
+        self._computed = True
         text, loaded = self._pending
         # No progress bar: the protocol embeds one value per round trip, so a
         # per-call bar can only ever say 1/1. The model *download* bar, which
@@ -33,11 +38,21 @@ def _compute(self, digest, identifier):
         return [float(component) for component in vector]
 
     def embed(self, text, model_id):
+        """Return the vector for ``text``, and whether the cache served it."""
         loaded = self._model(model_id)
         identifier = model.model_identifier(model_id, loaded)
         digest = sha256(text.encode("utf-8")).hexdigest()
         self._pending = (text, loaded)
-        return self._compute_cached(digest, identifier)
+        self._computed = False
+        vector = self._compute_cached(digest, identifier)
+        return vector, not self._computed
+
+    def _embed_response(self, text, model_id):
+        try:
+            vector, cached = self.embed(text, model_id)
+        except Exception as error:
+            return {"err": f"embed({model_id!r}) failed: {error}"}
+        return {"ok": vector, "meta": {"cached": cached}}
 
     def handle(self, line):
         try:
@@ -59,10 +74,7 @@ def handle(self, line):
         (model_id,) = rest or [model.DEFAULT_MODEL_ID]
         if not isinstance(model_id, str):
             return {"err": MALFORMED_MODEL_ID}
-        try:
-            return {"ok": self.embed(text, model_id)}
-        except Exception as error:
-            return {"err": f"embed({model_id!r}) failed: {error}"}
+        return self._embed_response(text, model_id)
 
     def serve(self, stdin, stdout):
         for line in stdin:

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

@@ -23,6 +23,7 @@ def it_starts_with_no_loaded_models_and_nothing_pending():
         built, _ = make_worker()
         assert built._models == {}
         assert built._pending is None
+        assert built._computed is False
 
 
 def describe_model_loading():
@@ -43,6 +44,17 @@ def it_loads_each_distinct_model_id():
 
 
 def describe_compute():
+    def it_records_that_it_ran_so_a_miss_is_distinguishable_from_a_hit():
+        # _compute runs only on a cache miss, so it is the one place a miss is
+        # observable -- cachetta's wrapper returns both cases identically.
+        built, _ = make_worker()
+        loaded = MagicMock()
+        loaded.encode.return_value = [[1.0]]
+        built._pending = ("hello", loaded)
+        built._computed = False
+        built._compute(HELLO_DIGEST, "m1")
+        assert built._computed is True
+
     def it_encodes_the_pending_text_and_returns_plain_floats():
         built, _ = make_worker()
         loaded = MagicMock()
@@ -77,7 +89,36 @@ def it_keys_the_cache_on_the_sha256_of_the_text_and_the_identifier():
             "m1", model.load_model.return_value
         )
         built._compute_cached.assert_called_once_with(HELLO_DIGEST, "m1@9.9")
-        assert result == [1.0]
+        assert result == ([1.0], True)
+
+    def it_reports_a_hit_when_compute_never_ran():
+        built, _ = make_worker()
+        built._compute_cached = MagicMock(return_value=[1.0])
+        with patch.object(worker, "model"):
+            _, cached = built.embed("hello", "m1")
+        assert cached is True
+
+    def it_reports_a_miss_when_compute_ran():
+        built, _ = make_worker()
+
+        def compute(digest, identifier):
+            built._computed = True
+            return [1.0]
+
+        built._compute_cached = MagicMock(side_effect=compute)
+        with patch.object(worker, "model"):
+            _, cached = built.embed("hello", "m1")
+        assert cached is False
+
+    def it_clears_a_previous_calls_miss_before_consulting_the_cache():
+        # Without the reset, one miss would make every later hit read as a
+        # miss for the rest of the process.
+        built, _ = make_worker()
+        built._computed = True
+        built._compute_cached = MagicMock(return_value=[1.0])
+        with patch.object(worker, "model"):
+            _, cached = built.embed("hello", "m1")
+        assert cached is True
 
     def it_stages_the_text_and_model_for_compute():
         built, _ = make_worker()
@@ -99,19 +140,27 @@ def it_hashes_the_utf8_bytes_of_the_text():
 def describe_handle():
     def it_answers_ok_with_the_embedding():
         built, _ = make_worker()
-        with patch.object(built, "embed", return_value=[1.0, 0.0]) as embed:
+        with patch.object(
+            built, "embed", return_value=([1.0, 0.0], False)
+        ) as embed:
             response = built.handle('{"call": ["hello", "m1"]}')
         embed.assert_called_once_with("hello", "m1")
-        assert response == {"ok": [1.0, 0.0]}
+        assert response == {"ok": [1.0, 0.0], "meta": {"cached": False}}
+
+    def it_reports_a_cache_hit_in_the_response_metadata():
+        built, _ = make_worker()
+        with patch.object(built, "embed", return_value=([1.0], True)):
+            response = built.handle('{"call": ["hello", "m1"]}')
+        assert response == {"ok": [1.0], "meta": {"cached": True}}
 
     def it_uses_the_default_model_for_a_single_argument_call():
         built, _ = make_worker()
-        with patch.object(built, "embed", return_value=[1.0]) as embed:
+        with patch.object(built, "embed", return_value=([1.0], False)) as embed:
             with patch.object(worker, "model") as model:
                 model.DEFAULT_MODEL_ID = "default/model"
                 response = built.handle('{"call": ["hello"]}')
         embed.assert_called_once_with("hello", "default/model")
-        assert response == {"ok": [1.0]}
+        assert response == {"ok": [1.0], "meta": {"cached": False}}
 
     def it_answers_err_on_invalid_json():
         built, _ = make_worker()

plugins/dirsql-plugin-embeddings/tests/e2e/embed_function_test.py

@@ -34,10 +34,12 @@
 _FRAGMENT = str(resources.files("dirsql_plugin_embeddings").joinpath("dirsql.toml"))
 
 
-def _run(query, cwd, cache_home, path=None):
+def _run(query, cwd, cache_home, path=None, progress=None):
     env = {**os.environ, "XDG_CACHE_HOME": str(cache_home)}
     if path is not None:
         env["PATH"] = path
+    if progress is not None:
+        env["DIRSQL_PROGRESS"] = progress
     return subprocess.run(
         [
             sys.executable,
@@ -175,3 +177,47 @@ def it_observes_a_spawn_when_embed_is_called_positive_control(
             f"stdout={result.stdout!r} stderr={result.stderr!r}"
         )
         assert spawn_log.read_text(encoding="utf-8") == "spawned\n"
+
+
+def describe_cache_reporting():
+    """The progress line's cache split, end to end (#1034).
+
+    The disk cache is what makes a re-run cheap, and a bare round-trip count
+    cannot say so: 2 worker calls reads identically whether both ran a model or
+    both were dictionary lookups. A cold run and the warm run after it must
+    read differently.
+    """
+
+    @pytest.fixture
+    def tree(tmp_path):
+        notes = tmp_path / "notes"
+        notes.mkdir()
+        (notes / "a.txt").write_text("hello", encoding="utf-8")
+        (notes / "b.txt").write_text("world", encoding="utf-8")
+        return tmp_path
+
+    def _query(model):
+        return (
+            "SELECT embed(content, '" + model + "') AS emb FROM './notes/*.txt'"
+        )
+
+    def it_reports_no_cache_hits_on_a_cold_run(tree, tiny_model, cache_home):
+        result = _run(
+            _query(tiny_model), tree, cache_home, progress="always"
+        )
+        assert result.returncode == 0, (
+            f"stdout={result.stdout!r} stderr={result.stderr!r}"
+        )
+        assert "ran 2 worker calls" in result.stderr
+        assert "cached" not in result.stderr
+
+    def it_reports_the_warm_run_as_entirely_cache_served(
+        tree, tiny_model, cache_home
+    ):
+        _run(_query(tiny_model), tree, cache_home, progress="always")
+        warm = _run(_query(tiny_model), tree, cache_home, progress="always")
+        assert warm.returncode == 0, (
+            f"stdout={warm.stdout!r} stderr={warm.stderr!r}"
+        )
+        assert "ran 2 worker calls" in warm.stderr
+        assert "(2 cached)" in warm.stderr

plugins/dirsql-plugin-embeddings/tests/e2e/worker_cli_test.py

@@ -28,15 +28,18 @@ def worker_argv():
 def describe_worker_cli():
     def it_serves_an_embed_request_over_real_pipes(spawn_worker, tiny_model):
         worker = spawn_worker(argv=worker_argv())
-        assert worker.request("hello", tiny_model) == {"ok": [1.0, 0.0]}
+        assert worker.request("hello", tiny_model)["ok"] == [1.0, 0.0]
 
     def it_survives_a_malformed_request_and_keeps_serving(
         spawn_worker, tiny_model
     ):
         worker = spawn_worker(argv=worker_argv())
         response = worker.send_line("{definitely not json")
         assert "err" in response
-        assert worker.request("hello world", tiny_model) == {"ok": [0.5, 0.5]}
+        assert worker.request("hello world", tiny_model)["ok"] == [
+            0.5,
+            0.5,
+        ]
 
     def it_stays_silent_on_stderr_when_stderr_is_not_a_tty(
         spawn_worker, tiny_model

plugins/dirsql-plugin-embeddings/tests/integration/cache_behavior_test.py

@@ -57,7 +57,7 @@ def it_serves_a_repeat_of_the_same_content_and_model_from_one_entry(
         worker = spawn_worker()
         first = worker.request("hello", tiny_model)
         second = worker.request("hello", tiny_model)
-        assert first == second == {"ok": [1.0, 0.0]}
+        assert first["ok"] == second["ok"] == [1.0, 0.0]
         assert entries(cache_home) == 1
 
     def it_survives_the_process_a_fresh_worker_hits_the_same_entry(
@@ -68,7 +68,7 @@ def it_survives_the_process_a_fresh_worker_hits_the_same_entry(
         first_worker.close()
         second_worker = spawn_worker()
         second = second_worker.request("hello", tiny_model)
-        assert first == second == {"ok": [1.0, 0.0]}
+        assert first["ok"] == second["ok"] == [1.0, 0.0]
         assert entries(cache_home) == 1
 
     def it_treats_text_and_blob_of_the_same_bytes_as_one_value(
@@ -94,8 +94,8 @@ def it_misses_when_the_model_changes(
         spawn_worker, cache_home, tiny_model, other_model
     ):
         worker = spawn_worker()
-        assert worker.request("hello", tiny_model) == {"ok": [1.0, 0.0]}
-        assert worker.request("hello", other_model) == {"ok": [0.0, 2.0]}
+        assert worker.request("hello", tiny_model)["ok"] == [1.0, 0.0]
+        assert worker.request("hello", other_model)["ok"] == [0.0, 2.0]
         assert entries(cache_home) == 2
 
 
@@ -107,5 +107,49 @@ def it_shares_one_entry_between_default_and_explicit_same_id(
         worker = spawn_worker(argv=argv)
         by_default = worker.send_line('{"call": ["hello"]}')
         explicit = worker.request("hello", tiny_model)
-        assert by_default == explicit == {"ok": [1.0, 0.0]}
+        assert by_default["ok"] == explicit["ok"] == [1.0, 0.0]
         assert entries(cache_home) == 1
+
+
+def describe_cache_reporting():
+    """The worker tells dirsql which answers cost nothing (#1034).
+
+    ``cachetta``'s wrapper returns a hit and a miss identically, so the worker
+    has to recover the signal itself; these pin that it recovers the right one.
+    """
+
+    def it_flags_a_recomputed_value_as_not_cached(spawn_worker, tiny_model):
+        worker = spawn_worker()
+        response = worker.request("hello", tiny_model)
+        assert response["meta"] == {"cached": False}
+
+    def it_flags_a_repeat_as_served_from_cache(spawn_worker, tiny_model):
+        worker = spawn_worker()
+        worker.request("hello", tiny_model)
+        response = worker.request("hello", tiny_model)
+        assert response["meta"] == {"cached": True}
+
+    def it_flags_a_hit_from_a_fresh_process(spawn_worker, tiny_model):
+        first = spawn_worker()
+        first.request("hello", tiny_model)
+        first.close()
+        response = spawn_worker().request("hello", tiny_model)
+        assert response["meta"] == {"cached": True}
+
+    def it_flags_each_distinct_value_as_computed(spawn_worker, tiny_model):
+        worker = spawn_worker()
+        first = worker.request("hello", tiny_model)
+        second = worker.request("world", tiny_model)
+        assert first["meta"] == second["meta"] == {"cached": False}
+
+    def it_flags_a_new_model_for_a_seen_value_as_computed(
+        spawn_worker, tiny_model, other_model
+    ):
+        worker = spawn_worker()
+        worker.request("hello", tiny_model)
+        response = worker.request("hello", other_model)
+        assert response["meta"] == {"cached": False}
+
+    def it_reports_no_cache_state_for_a_null_value(spawn_worker, tiny_model):
+        worker = spawn_worker()
+        assert worker.request(None, tiny_model) == {"ok": None}