feat(retrieval): Wilson prior, exploration slot, embeddings + three-leg fusion

better_memory/cli/backfill_embeddings.py

@@ -0,0 +1,118 @@
+"""One-shot embedding backfill for reflections + semantic memories.
+
+Run at deploy after migration 0014:
+
+    python -m better_memory.cli.backfill_embeddings
+
+Idempotent: only rows missing a vector are embedded. The lazy self-heal in
+memory.retrieve covers stragglers afterwards; this exists so the historical
+corpus doesn't wait to be retrieved before becoming searchable.
+
+One event loop and one embedder for the whole job (the embedder's
+httpx.AsyncClient is loop-bound); batches of 50 per HTTP request.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+import sqlite_vec
+
+from better_memory.db.connection import connect
+from better_memory.db.schema import apply_migrations
+from better_memory.services.reflection import _embedding_source_text
+
+_BATCH = 50
+
+
+def backfill(conn, embedder) -> dict[str, int]:
+    stats = {"reflections": 0, "semantics": 0, "skipped": 0}
+
+    refl = conn.execute(
+        """SELECT r.id, r.title, r.use_cases, r.hints FROM reflections r
+           WHERE r.status IN ('pending_review', 'confirmed')
+             AND r.id NOT IN (SELECT reflection_id FROM reflection_embeddings)"""
+    ).fetchall()
+    sems = conn.execute(
+        """SELECT id, content FROM semantic_memories
+           WHERE id NOT IN (SELECT memory_id FROM semantic_embeddings)"""
+    ).fetchall()
+
+    jobs = [
+        ("reflections", "reflection_embeddings", "reflection_id", r["id"],
+         _embedding_source_text(r["title"], r["use_cases"],
+                                json.loads(r["hints"])))
+        for r in refl
+    ] + [
+        ("semantics", "semantic_embeddings", "memory_id", s["id"], s["content"])
+        for s in sems
+    ]
+
+    async def _embed_all() -> list[list[list[float]] | None]:
+        out = []
+        for i in range(0, len(jobs), _BATCH):
+            texts = [j[4] for j in jobs[i:i + _BATCH]]
+            try:
+                out.append(await embedder.embed_batch(texts))
+            except Exception:
+                out.append(None)
+        return out
+
+    batches = asyncio.run(_embed_all()) if jobs else []
+
+    for bi, vectors in enumerate(batches):
+        chunk = jobs[bi * _BATCH:(bi + 1) * _BATCH]
+        if vectors is None:
+            stats["skipped"] += len(chunk)
+            continue
+        for (kind, table, col, row_id, _), vec in zip(chunk, vectors):
+            conn.execute(
+                f"INSERT INTO {table} ({col}, embedding) VALUES (?, ?)",
+                (row_id, sqlite_vec.serialize_float32(vec)))
+            stats[kind] += 1
+    conn.commit()
+    return stats
+
+
+def main(argv: list[str] | None = None) -> None:
+    from better_memory.config import get_config
+    from better_memory.embeddings.ollama import OllamaEmbedder
+
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--home", default=None,
+                    help="BETTER_MEMORY_HOME override (default: config)")
+    args = ap.parse_args(argv)
+
+    config = get_config()
+    db = Path(args.home) / "memory.db" if args.home else config.memory_db
+    conn = connect(db)
+    try:
+        apply_migrations(conn)
+
+        if config.embeddings_backend != "ollama":
+            print("embeddings backend is not ollama; nothing to backfill")
+            return
+
+        embedder = OllamaEmbedder()
+        try:
+            stats = backfill(conn, embedder)
+        finally:
+            try:
+                asyncio.run(embedder.aclose())
+            except Exception:
+                pass
+        print(f"backfilled reflections={stats['reflections']} "
+              f"semantics={stats['semantics']} skipped={stats['skipped']}")
+        if stats["skipped"]:
+            print("warning: some rows skipped (Ollama unreachable?); "
+                  "re-run later or let retrieve self-heal them", file=sys.stderr)
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

better_memory/db/migrations/0014_semantic_embeddings.sql

@@ -0,0 +1,12 @@
+-- Migration 0014: vector index for semantic memories.
+--
+-- reflection_embeddings has existed since 0002; semantic memories never got
+-- the parallel table, so they have no semantic-search substrate. Written by
+-- SemanticMemoryService on create/update, healed lazily on retrieve, and
+-- backfilled once by cli.backfill_embeddings. 768 dims = nomic-embed-text,
+-- matching observation_embeddings and reflection_embeddings.
+
+CREATE VIRTUAL TABLE semantic_embeddings USING vec0(
+    memory_id TEXT PRIMARY KEY,
+    embedding FLOAT[768]
+);

better_memory/embeddings/sync_embed.py

@@ -0,0 +1,75 @@
+"""Sync facade over the async embedder, for thread-bound sync services.
+
+Two hazards this class exists to contain:
+
+1. ``httpx.AsyncClient`` is bound to the event loop that first uses it,
+   and ``run_async_in_worker`` runs a fresh loop per call — so the
+   embedder must be CONSTRUCTED inside the worker coroutine and closed
+   there. The factory is invoked per call; ``OllamaEmbedder`` construction
+   is cheap and contacts nothing.
+2. Retrieval must never hang on a dead Ollama. Any failure opens a
+   circuit breaker: for ``cooldown`` seconds every call returns ``None``
+   immediately. Worst case is one bounded stall per cooldown window.
+
+Returns ``None`` on every failure path — callers treat a missing vector
+as "no vec leg", never as an error.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from typing import Any
+
+from better_memory.async_bridge import run_async_in_worker
+
+#: Bridge-level hard stop. The wiring uses OllamaEmbedder(timeout=5.0,
+#: max_retries=1), so a healthy-but-slow call ends well inside this; the
+#: bridge timeout is the backstop against pathological hangs.
+_WORKER_TIMEOUT = 15.0
+_DEFAULT_COOLDOWN = 60.0
+
+
+class SyncEmbedder:
+    def __init__(
+        self,
+        factory: Callable[[], Any] | None,
+        *,
+        clock: Callable[[], float] = time.monotonic,
+        cooldown: float = _DEFAULT_COOLDOWN,
+        timeout: float = _WORKER_TIMEOUT,
+    ) -> None:
+        self._factory = factory
+        self._clock = clock
+        self._cooldown = cooldown
+        self._timeout = timeout
+        self._down_until = 0.0
+
+    def embed_text(self, text: str) -> list[float] | None:
+        return self._run(lambda emb: emb.embed(text))
+
+    def embed_batch(self, texts: list[str]) -> list[list[float]] | None:
+        return self._run(lambda emb: emb.embed_batch(texts))
+
+    def _run(self, op: Callable[[Any], Any]):
+        if self._factory is None:
+            return None
+        if self._clock() < self._down_until:
+            return None
+
+        factory = self._factory
+
+        async def _go():
+            emb = factory()
+            try:
+                return await op(emb)
+            finally:
+                aclose = getattr(emb, "aclose", None)
+                if aclose is not None:
+                    await aclose()
+
+        try:
+            return run_async_in_worker(_go, timeout=self._timeout)
+        except Exception:
+            self._down_until = self._clock() + self._cooldown
+            return None

better_memory/mcp/server.py

@@ -62,6 +62,7 @@
 from better_memory.db.connection import connect
 from better_memory.db.schema import apply_migrations
 from better_memory.embeddings.ollama import OllamaEmbedder
+from better_memory.embeddings.sync_embed import SyncEmbedder
 from better_memory.mcp._util import resolve_session_id as _resolve_session_id
 from better_memory.mcp.handlers import (
     EpisodeToolHandlers,
@@ -180,6 +181,14 @@ def create_server() -> tuple[
         # memory.retrieve will succeed on their next call without a restart.
         _probe_ollama(config.ollama_host)
 
+    # Fresh short-timeout embedder per bridge call (loop-bound client);
+    # shared instance so the breaker state is process-wide.
+    sync_embedder: SyncEmbedder | None = None
+    if embedder is not None:
+        sync_embedder = SyncEmbedder(
+            lambda: OllamaEmbedder(timeout=5.0, max_retries=1)
+        )
+
     # Concurrency invariant: the memory-side services below
     # (EpisodeService, ObservationService, ReflectionSynthesisService,
     # RetentionService, SpoolService, SemanticMemoryService,
@@ -212,17 +221,18 @@ def create_server() -> tuple[
         config=config,
         memory_conn=memory_conn,
         embedder=embedder,
+        sync_embedder=sync_embedder,
         session_id=startup_session_id,
         project=startup_project,
     )
 
     # Reflection synthesis is driven by the IDE-LLM via two MCP tools
     # (memory.synthesize_next_get_context / _apply). The service no
     # longer holds a chat client.
-    reflections = ReflectionSynthesisService(memory_conn)
+    reflections = ReflectionSynthesisService(memory_conn, sync_embedder=sync_embedder)
     retention = RetentionService(conn=memory_conn)
     memory_rating = MemoryRatingService(memory_conn)
-    semantic = SemanticMemoryService(memory_conn)
+    semantic = SemanticMemoryService(memory_conn, sync_embedder=sync_embedder)
     session_bootstrap = SessionBootstrapService(memory_conn)
 
     knowledge = KnowledgeService(

better_memory/mcp/tools.py

@@ -189,9 +189,9 @@ def tool_definitions(
                 # `query` is deliberately NOT required. Making it mandatory
                 # would break every existing caller (and the start_episode /
                 # bootstrap internal paths) for a param whose absence degrades
-                # gracefully to the popularity order. The description urges it
+                # gracefully to the Wilson-prior. The description urges it
                 # instead; the real usefulness gains come from the shortlist
-                # default and the ignored-history demotion, not from coercing
+                # default and the Wilson-score confidence boost, not from coercing
                 # a query on every call.
             },
         ),

better_memory/services/memory_rating.py

@@ -61,23 +61,6 @@ class ApplySessionRatingsResult(TypedDict):
 _VALID_CLASSES: set[str] = {"cited", "shaped", "ignored", "misled", "overlooked"}
 _CREDIT_CLASSES: set[str] = {"cited", "shaped", "misled", "overlooked"}
 
-# Retrieval-ranking weight for the `overlooked` class. One `overlooked`
-# rating contributes this many points to a memory's rank score — the same
-# unit as one `useful_count`. Per issue #60, an overlooked memory should
-# rank harder than a cited one. Imported by reflection.py and semantic.py.
-OVERLOOKED_RANKING_WEIGHT = 3
-
-# Demotion for memories that keep being served and keep not mattering.
-# A memory is only demoted once it has been ignored in at least
-# IGNORED_DEMOTION_FLOOR distinct sessions AND has never been useful — one or
-# two ignores say nothing (the task simply wasn't relevant that day), but a
-# double-digit run of them with no hits is the memory telling you it does not
-# belong in the default set. Memories with any useful history are never
-# demoted, however often they are also ignored: the best-performing memory on
-# the live DB is ignored in 125 of the 192 sessions it appears in.
-IGNORED_DEMOTION_FLOOR = 10
-IGNORED_DEMOTION_WEIGHT = 1
-
 
 class MemoryRatingService:
     """Writes useful_count / times_misled on reflections + semantic memories,