fix(memory): sweep legacy orphaned message vectors

Repair vector rows left behind by older databases at store startup. Keep the existing single-statement delete path from main and avoid reintroducing the redundant batched helper.

synapse/memory/store.py

@@ -251,6 +251,8 @@ class PersistentMemoryStore:
             "DELETE FROM settings WHERE key IN ('anthropic_api_key', 'escalation_model')"
         )
 
+        self._sweep_orphan_msg_vectors(conn)
+
         conn.commit()
         conn.close()
 
@@ -788,6 +790,29 @@ class PersistentMemoryStore:
         except Exception:
             pass
 
+    def _sweep_orphan_msg_vectors(self, conn) -> None:
+        """One-time repair for databases written before delete_conversation
+        cleaned up after itself: drop vectors whose message is already gone."""
+        try:
+            ids = [r["message_id"] for r in conn.execute(
+                "SELECT v.message_id FROM message_vectors v "
+                "LEFT JOIN messages m ON m.id = v.message_id WHERE m.id IS NULL"
+            ).fetchall()]
+            if ids:
+                conn.execute(
+                    "DELETE FROM message_vectors WHERE message_id NOT IN "
+                    "(SELECT id FROM messages)"
+                )
+                if self.vec_enabled and conn.execute(
+                    "SELECT 1 FROM sqlite_master WHERE name = 'vec_messages'"
+                ).fetchone():
+                    for message_id in ids:
+                        conn.execute(
+                            "DELETE FROM vec_messages WHERE rowid = ?", (message_id,)
+                        )
+        except Exception:
+            pass
+
     def _backfill_vec_msgs(self, conn, dim: int) -> None:
         """Index any message_vectors rows missing from vec_messages."""
         try:
@@ -1215,4 +1240,4 @@ class PersistentMemoryStore:
 from ..nexus_config import MEMORY_DB
 
 DB_PATH = MEMORY_DB
-store = PersistentMemoryStore(DB_PATH)
+store = PersistentMemoryStore(DB_PATH)

tests/test_documents.py

@@ -136,6 +136,27 @@ def test_conversation_recall_uses_vec_and_matches_brute_force():
     asyncio.run(run())
 
 
+def test_startup_sweeps_pre_existing_orphan_vectors():
+    """Databases written before delete_conversation cleaned up after itself are
+    repaired the next time the store opens them."""
+    import json
+    path = Path(tempfile.mkdtemp()) / "t.db"
+    s = PersistentMemoryStore(path)
+    s.create_conversation("c1")
+    mid = s.add_message("c1", "user", "lego star wars")
+    conn = s._connect()
+    conn.execute("INSERT INTO message_vectors (message_id, embedding) VALUES (?, ?)",
+                 (mid, json.dumps([1.0, 0.0])))
+    conn.execute("DELETE FROM messages WHERE id = ?", (mid,))  # the old leaky delete
+    conn.commit()
+    conn.close()
+
+    reopened = PersistentMemoryStore(path)
+    conn = reopened._connect()
+    assert conn.execute("SELECT COUNT(*) FROM message_vectors").fetchone()[0] == 0
+    conn.close()
+
+
 def test_projects_scope_documents_and_survive_delete():
     s = _store()