feat(p2): embedding cache in the Embedder base, fused hybrid search, rebuild paths

Caching/batching/prefixing moved into Embedder so FakeEmbedder exercises the
same cache the Bifrost client uses. Rebuilding an index with an unchanged model
embeds nothing; a dim or model change drops the vectors and re-embeds. A
token-free query short-circuits before the embedding call.

src/mdmem/embed.py

@@ -88,11 +88,39 @@ class EmbeddingCache:
 
 
 class Embedder(ABC):
+    """Prefixing, batching and caching live here so every backend shares them."""
+
     model: str
     dim: int
+    cache: EmbeddingCache | None = None
+
+    async def embed(self, texts: list[str], *, is_query: bool = False) -> np.ndarray:
+        if not texts:
+            return np.zeros((0, self.dim), dtype=np.float32)
+        prepared = [QUERY_INSTRUCT + t for t in texts] if is_query else list(texts)
+
+        out: list[np.ndarray | None] = [None] * len(prepared)
+        todo: list[int] = []
+        for i, t in enumerate(prepared):
+            cached = self.cache.get(self.model, self.dim, t) if self.cache else None
+            if cached is not None:
+                out[i] = cached
+            else:
+                todo.append(i)
+
+        for start in range(0, len(todo), BATCH):
+            idxs = todo[start : start + BATCH]
+            vecs = await self._embed_batch([prepared[i] for i in idxs])
+            for i, v in zip(idxs, vecs, strict=True):
+                out[i] = v
+                if self.cache:
+                    self.cache.put(self.model, self.dim, prepared[i], v)
+
+        return np.vstack([v for v in out if v is not None]).astype(np.float32)
 
     @abstractmethod
-    async def embed(self, texts: list[str], *, is_query: bool = False) -> np.ndarray: ...
+    async def _embed_batch(self, prepared: list[str]) -> np.ndarray:
+        """Embed one batch of at most BATCH already-prefixed texts."""
 
     async def selftest(self) -> dict[str, object]:
         started = time.monotonic()
@@ -129,31 +157,7 @@ class BifrostEmbedder(Embedder):
     async def aclose(self) -> None:
         await self._client.aclose()
 
-    async def embed(self, texts: list[str], *, is_query: bool = False) -> np.ndarray:
-        if not texts:
-            return np.zeros((0, self.dim), dtype=np.float32)
-        prepared = [QUERY_INSTRUCT + t for t in texts] if is_query else list(texts)
-
-        out: list[np.ndarray | None] = [None] * len(prepared)
-        todo: list[int] = []
-        for i, t in enumerate(prepared):
-            cached = self.cache.get(self.model, self.dim, t) if self.cache else None
-            if cached is not None:
-                out[i] = cached
-            else:
-                todo.append(i)
-
-        for start in range(0, len(todo), BATCH):
-            idxs = todo[start : start + BATCH]
-            vecs = await self._call([prepared[i] for i in idxs])
-            for i, v in zip(idxs, vecs, strict=True):
-                out[i] = v
-                if self.cache:
-                    self.cache.put(self.model, self.dim, prepared[i], v)
-
-        return np.vstack([v for v in out if v is not None]).astype(np.float32)
-
-    async def _call(self, batch: list[str]) -> np.ndarray:
+    async def _embed_batch(self, batch: list[str]) -> np.ndarray:
         delay = 0.5
         last: Exception | None = None
         for attempt in range(1, RETRIES + 1):
@@ -198,16 +202,20 @@ class BifrostEmbedder(Embedder):
 class FakeEmbedder(Embedder):
     """Deterministic seeded vectors — tests run with no network and no GPU (§12a.1)."""
 
-    def __init__(self, dim: int = 16, model: str = "fake") -> None:
+    def __init__(self, dim: int = 16, model: str = "fake", cache: EmbeddingCache | None = None) -> None:
         self.dim = dim
         self.model = model
+        self.cache = cache
         self.calls = 0
+        self.texts_embedded = 0
+        self.last_latency_ms: float | None = 0.0
+        self.errors = 0
 
-    async def embed(self, texts: list[str], *, is_query: bool = False) -> np.ndarray:
+    async def _embed_batch(self, batch: list[str]) -> np.ndarray:
         self.calls += 1
+        self.texts_embedded += len(batch)
         rows = []
-        for t in texts:
-            base = QUERY_INSTRUCT + t if is_query else t
+        for base in batch:
             # Bag-of-words hashing: texts sharing words end up close, which makes
             # ranking tests meaningful without a model.
             v = np.zeros(self.dim, dtype=np.float32)

src/mdmem/search.py

@@ -103,6 +103,11 @@ def search(
     inbox_weight: float | None = None,
     debug: bool = False,
 ) -> SearchResponse:
+    if not to_match(query):
+        # No word tokens at all: neither channel can say anything useful, and a
+        # dense-only search on punctuation would just return the whole vault.
+        return SearchResponse(fusion=(fusion or settings.fusion).lower())
+
     conn = db.read()
     include_inbox = settings.include_inbox_default if include_inbox is None else include_inbox
     max_chars = settings.max_chars_default if max_chars is None else max_chars

src/mdmem/service.py

@@ -211,7 +211,9 @@ class MemoryService:
     async def search(self, query: str, **kw: Any) -> search_mod.SearchResponse:
         started = time.monotonic()
         vec: bytes | None = None
-        if self.embedder is not None:
+        from mdmem.ftsquery import to_match
+
+        if self.embedder is not None and to_match(query):
             try:
                 v = await self.embedder.embed([query], is_query=True)
                 vec = to_blob(v[0])

tests/test_embed.py

@@ -0,0 +1,135 @@
+import httpx
+import numpy as np
+import pytest
+
+from mdmem.embed import QUERY_INSTRUCT, BifrostEmbedder, EmbeddingCache, EmbeddingError, FakeEmbedder
+
+
+def _transport(handler):
+    return httpx.MockTransport(handler)
+
+
+def _embedder(handler, dim=4, cache=None, model="llamaswap/qwen3-embedding-4b"):
+    e = BifrostEmbedder("http://bifrost", "vk", model, dim, cache=cache)
+    e._client = httpx.AsyncClient(transport=_transport(handler))
+    return e
+
+
+def _reply(vectors):
+    return httpx.Response(200, json={"data": [{"embedding": v} for v in vectors]})
+
+
+async def test_documents_get_no_instruction_prefix_but_queries_do():
+    seen = []
+
+    def handler(request):
+        body = request.read().decode()
+        seen.append(body)
+        import json
+
+        n = len(json.loads(body)["input"])
+        return _reply([[1.0, 0, 0, 0]] * n)
+
+    e = _embedder(handler)
+    await e.embed(["doc text"], is_query=False)
+    assert QUERY_INSTRUCT.split("\n")[0] not in seen[0]
+    await e.embed(["query text"], is_query=True)
+    assert "Instruct:" in seen[1]
+
+
+async def test_vectors_are_truncated_and_normalised():
+    def handler(request):
+        return _reply([[3.0, 4.0, 99.0, 99.0, 99.0]])
+
+    e = _embedder(handler, dim=2)
+    v = await e.embed(["x"])
+    assert v.shape == (1, 2)
+    assert np.allclose(v[0], [0.6, 0.8])
+
+
+async def test_dim_too_small_fails_loudly_with_the_model_name():
+    def handler(request):
+        return _reply([[1.0, 2.0]])
+
+    e = _embedder(handler, dim=2560)
+    with pytest.raises(EmbeddingError) as exc:
+        await e.embed(["x"])
+    assert "llamaswap/qwen3-embedding-4b" in str(exc.value)
+    assert "provider prefix" in str(exc.value)
+
+
+async def test_http_error_body_is_surfaced():
+    def handler(request):
+        return httpx.Response(404, text="model not found")
+
+    e = _embedder(handler)
+    with pytest.raises(EmbeddingError, match="404"):
+        await e.embed(["x"])
+
+
+async def test_transient_failure_is_retried(monkeypatch):
+    calls = {"n": 0}
+
+    def handler(request):
+        calls["n"] += 1
+        if calls["n"] < 3:
+            return httpx.Response(503, text="upstream busy")
+        return _reply([[1.0, 0, 0, 0]])
+
+    monkeypatch.setattr("mdmem.embed.asyncio.sleep", lambda *_: _noop())
+    e = _embedder(handler)
+    v = await e.embed(["x"])
+    assert calls["n"] == 3 and v.shape == (1, 4)
+
+
+async def _noop():
+    return None
+
+
+async def test_batches_are_capped_at_16():
+    sizes = []
+
+    def handler(request):
+        import json
+
+        n = len(json.loads(request.read())["input"])
+        sizes.append(n)
+        return _reply([[1.0, 0, 0, 0]] * n)
+
+    e = _embedder(handler)
+    v = await e.embed([f"t{i}" for i in range(35)])
+    assert sizes == [16, 16, 3] and v.shape == (35, 4)
+
+
+async def test_cache_avoids_a_second_call(tmp_path):
+    calls = {"n": 0}
+
+    def handler(request):
+        calls["n"] += 1
+        import json
+
+        n = len(json.loads(request.read())["input"])
+        return _reply([[1.0, 0, 0, 0]] * n)
+
+    cache = EmbeddingCache(tmp_path / "cache.db")
+    e = _embedder(handler, cache=cache)
+    await e.embed(["same text"])
+    await e.embed(["same text"])
+    assert calls["n"] == 1
+    assert cache.hits == 1 and cache.misses == 1
+
+
+def test_cache_key_includes_model_and_dim(tmp_path):
+    c = EmbeddingCache(tmp_path / "c.db")
+    a = c.key("m1", 8, "t")
+    assert a != c.key("m2", 8, "t")
+    assert a != c.key("m1", 16, "t")
+    assert a == c.key("m1", 8, "t")
+
+
+async def test_fake_embedder_is_deterministic_and_normalised():
+    e = FakeEmbedder(dim=32)
+    a = await e.embed(["llama swap"])
+    b = await e.embed(["llama swap"])
+    assert np.allclose(a, b)
+    assert np.allclose(np.linalg.norm(a, axis=1), 1.0)

tests/test_rebuild.py

@@ -0,0 +1,115 @@
+"""Index rebuilds must be free when the model is unchanged, and automatic when it is not."""
+
+import pytest
+
+from mdmem.embed import EmbeddingCache, FakeEmbedder
+from mdmem.service import MemoryService
+
+
+async def _seed(svc, n=5):
+    ids = []
+    for i in range(n):
+        r = await svc.write_note(
+            title=f"Notiz {i}",
+            body=f"Inhalt Nummer {i} über Bifrost",
+            type="fact",
+            summary=f"s{i}",
+            keywords=["bifrost"],
+        )
+        ids.append(r["id"])
+    return ids
+
+
+@pytest.fixture()
+def cache(settings):
+    c = EmbeddingCache(settings.cache_db)
+    yield c
+    c.close()
+
+
+async def test_rebuild_with_unchanged_model_is_a_100_percent_cache_hit(settings, cache):
+    emb = FakeEmbedder(settings.embed_dim, cache=cache)
+    svc = MemoryService.build(settings, embedder=emb)
+    await svc.start(watch=False)
+    await _seed(svc, 5)
+    embedded_first = emb.texts_embedded
+    assert embedded_first >= 5
+    await svc.stop()
+
+    settings.index_db.unlink()
+    cache.hits = cache.misses = 0
+    emb2 = FakeEmbedder(settings.embed_dim, cache=cache)
+    svc2 = MemoryService.build(settings, embedder=emb2)
+    await svc2.start(watch=False)
+    try:
+        assert svc2.db.read().execute("SELECT count(*) FROM chunks_vec").fetchone()[0] == 5
+        assert svc2.db.read().execute("SELECT count(*) FROM chunks WHERE vec_pending=1").fetchone()[0] == 0
+        assert emb2.texts_embedded == 0, "rebuild re-embedded text the cache already had"
+        assert cache.misses == 0 and cache.hits >= 5
+    finally:
+        await svc2.stop()
+
+
+async def test_dim_change_drops_vectors_and_re_embeds(settings, monkeypatch, cache):
+    svc = MemoryService.build(settings, embedder=FakeEmbedder(settings.embed_dim, cache=cache))
+    await svc.start(watch=False)
+    ids = await _seed(svc, 3)
+    await svc.stop()
+
+    from dataclasses import replace
+
+    smaller = replace(settings, embed_dim=8)
+    emb2 = FakeEmbedder(8, cache=cache)
+    svc2 = MemoryService.build(smaller, embedder=emb2)
+    await svc2.start(watch=False)
+    try:
+        assert svc2.db.meta_get("embed_dim") == "8"
+        assert svc2.db.read().execute("SELECT count(*) FROM chunks WHERE vec_pending=1").fetchone()[0] == 0
+        # 3 chunks + the "ping" self-test, which a new dim also misses in the cache
+        assert emb2.texts_embedded == 4, "a dim change must re-embed, not reuse the old vectors"
+        r = await svc2.search("Bifrost", k=5)
+        assert {x.id for x in r.results} >= set(ids[:1])
+    finally:
+        await svc2.stop()
+
+
+async def test_model_change_re_embeds(settings, cache):
+    svc = MemoryService.build(settings, embedder=FakeEmbedder(settings.embed_dim, cache=cache))
+    await svc.start(watch=False)
+    await _seed(svc, 2)
+    await svc.stop()
+
+    from dataclasses import replace
+
+    other = replace(settings, embed_model="other/model")
+    emb2 = FakeEmbedder(settings.embed_dim, model="other/model", cache=cache)
+    svc2 = MemoryService.build(other, embedder=emb2)
+    await svc2.start(watch=False)
+    try:
+        assert emb2.texts_embedded == 3  # 2 chunks + the "ping" self-test
+        assert svc2.db.meta_get("embed_model") == "other/model"
+    finally:
+        await svc2.stop()
+
+
+async def test_embeddings_down_leaves_chunks_pending_and_retries_later(settings):
+    emb = FakeEmbedder(settings.embed_dim)
+
+    async def boom(*_a, **_kw):
+        raise RuntimeError("bifrost down")
+
+    real = emb._embed_batch
+    emb._embed_batch = boom
+    svc = MemoryService.build(settings, embedder=emb)
+    await svc.start(watch=False)
+    try:
+        res = await svc.write_note(title="Trotzdem", body="Bifrost ist weg", type="fact")
+        assert svc.db.read().execute("SELECT count(*) FROM chunks WHERE vec_pending=1").fetchone()[0] == 1
+        assert (await svc.search("Bifrost")).results[0].id == res["id"]  # lexical still works
+
+        emb._embed_batch = real
+        n = await svc.embed_pending()
+        assert n == 1
+        assert svc.db.read().execute("SELECT count(*) FROM chunks WHERE vec_pending=1").fetchone()[0] == 0
+    finally:
+        await svc.stop()

tests/test_search.py

@@ -0,0 +1,156 @@
+import pytest
+
+from mdmem.embed import FakeEmbedder
+from mdmem.service import MemoryService
+
+
+@pytest.fixture()
+async def svc(settings):
+    s = MemoryService.build(settings, embedder=FakeEmbedder(settings.embed_dim))
+    await s.start(watch=False)
+    yield s
+    await s.stop()
+
+
+async def _add(svc, *, state="active", **kw):
+    res = await svc.write_note(**kw)
+    if state != "inbox":
+        await svc.set_state(res["id"], state)
+    return res["id"]
+
+
+async def test_lexical_and_dense_results_are_fused(svc):
+    a = await _add(
+        svc,
+        title="llama-swap routet Modelle",
+        body="llama-swap fronts local models",
+        type="fact",
+        keywords=["router"],
+    )
+    await _add(svc, title="Kaffeemaschine", body="entkalken alle 3 Monate", type="procedure")
+
+    r = await svc.search("llama-swap", k=5, debug=True)
+    assert r.results[0].id == a
+    dbg = r.results[0].debug
+    assert dbg["via"] in ("both", "lex", "sem")
+    assert dbg["fused"] > 0
+
+
+async def test_dense_finds_a_note_with_no_lexical_overlap(svc):
+    target = await _add(
+        svc,
+        title="Wartung",
+        body="entkalken alle drei Monate mit Essig",
+        type="procedure",
+        keywords=["kaffee"],
+    )
+    r = await svc.search("entkalken alle drei Monate mit Essig", k=5, debug=True)
+    assert r.results and r.results[0].id == target
+
+
+async def test_search_still_works_when_embeddings_are_down(svc, monkeypatch):
+    a = await _add(svc, title="Bifrost", body="Gateway für Modelle", type="fact")
+
+    async def boom(*_a, **_kw):
+        raise RuntimeError("bifrost down")
+
+    monkeypatch.setattr(svc.embedder, "embed", boom)
+    r = await svc.search("Bifrost", k=5, debug=True)
+    assert [x.id for x in r.results] == [a]
+    assert r.results[0].debug["sem_rank"] is None
+    assert svc.embed_ok is False
+
+
+async def test_inbox_results_are_penalised_and_unverified(svc):
+    active = await _add(svc, title="Gleicher Text hier", body="Bifrost Gateway", type="fact")
+    inbox = await _add(svc, title="Gleicher Text hier", body="Bifrost Gateway", type="fact", state="inbox")
+    r = await svc.search("Bifrost Gateway", k=5)
+    ids = [x.id for x in r.results]
+    assert ids.index(active) < ids.index(inbox)
+    assert next(x for x in r.results if x.id == inbox).verified is False
+    assert next(x for x in r.results if x.id == active).verified is True
+
+
+async def test_include_inbox_false_hides_unreviewed(svc):
+    await _add(svc, title="Nur im Posteingang", body="Bifrost", type="fact", state="inbox")
+    r = await svc.search("Bifrost", include_inbox=False)
+    assert r.results == []
+
+
+async def test_archive_is_excluded(svc):
+    nid = await _add(svc, title="Altlast", body="Bifrost", type="fact")
+    await svc.set_state(nid, "archive")
+    assert (await svc.search("Bifrost")).results == []
+
+
+async def test_superseded_notes_are_filtered_unless_asked(svc):
+    old = await _add(svc, title="Altes Faktum", body="Der Router heißt A", type="fact")
+    res = await svc.supersede(old, title="Neues Faktum", body="Der Router heißt A", type="fact")
+    await svc.approve(res["id"])
+
+    ids = [x.id for x in (await svc.search("Router heißt")).results]
+    assert old not in ids and res["id"] in ids
+    ids2 = [x.id for x in (await svc.search("Router heißt", include_superseded=True)).results]
+    assert old in ids2
+
+
+async def test_type_and_tag_filters(svc):
+    f = await _add(svc, title="Ein Fakt", body="Thema Alpha", type="fact", tags=["homelab"])
+    await _add(svc, title="Eine Prozedur", body="Thema Alpha", type="procedure", tags=["küche"])
+    r = await svc.search("Thema Alpha", types=["fact"])
+    assert [x.id for x in r.results] == [f]
+    r = await svc.search("Thema Alpha", tags=["homelab"])
+    assert [x.id for x in r.results] == [f]
+    assert (await svc.search("Thema Alpha", tags=["gibtsnicht"])).results == []
+
+
+async def test_valid_at_filter(svc):
+    nid = await _add(
+        svc,
+        title="Zeitlich begrenzt",
+        body="Gilt nur 2025",
+        type="fact",
+        valid_from="2025-01-01T00:00:00Z",
+        valid_until="2026-01-01T00:00:00Z",
+    )
+    assert [x.id for x in (await svc.search("Gilt nur", valid_at="2025-06-01T00:00:00Z")).results] == [nid]
+    assert (await svc.search("Gilt nur", valid_at="2026-06-01T00:00:00Z")).results == []
+
+
+async def test_since_until_filter_on_updated(svc):
+    nid = await _add(svc, title="Neulich", body="Ereignis", type="event")
+    assert (await svc.search("Ereignis", since="2100-01-01T00:00:00Z")).results == []
+    assert [x.id for x in (await svc.search("Ereignis", until="2100-01-01T00:00:00Z")).results] == [nid]
+
+
+async def test_budget_truncates_and_reports(svc):
+    for i in range(6):
+        await _add(svc, title=f"Notiz {i}", body="Bifrost " + ("Text " * 200), type="note")
+    r = await svc.search("Bifrost", k=6, max_chars=1000)
+    assert r.truncated is True
+    assert r.chars <= 1000 + 800  # the first result is always emitted
+    assert len(r.results) < 6
+
+
+async def test_min_score_abstains(svc):
+    await _add(svc, title="Irgendwas", body="völlig anderes Thema", type="note")
+    r = await svc.search("Bifrost llama swap", min_score=5.0)
+    assert r.results == []
+
+
+async def test_cc_fusion_returns_results_too(svc):
+    a = await _add(svc, title="llama-swap", body="Modelle routen", type="fact")
+    r = await svc.search("llama-swap Modelle", fusion="cc", cc_alpha=0.5, debug=True)
+    assert r.fusion == "cc" and r.results[0].id == a
+    assert 0.0 <= r.results[0].score <= 1.0
+
+
+async def test_empty_query_returns_nothing(svc):
+    await _add(svc, title="Etwas", body="Inhalt", type="note")
+    assert (await svc.search("   ")).results == []
+
+
+async def test_snippet_is_bounded(svc):
+    await _add(svc, title="Lang", body="Bifrost " + ("wort " * 1500), type="note")
+    r = await svc.search("Bifrost")
+    assert len(r.results[0].snippet) <= 801