diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 9642175b..f1a89b1b 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -1192,8 +1192,11 @@ def _semantic_sqlite_reinforce( threshold = _SEMANTIC_DEDUP_THRESHOLD if threshold > 0 and embedding is not None and self.items: # Type-filtered top-1 via the persistent matrix index — - # no per-item corpus rebuild. - hits = _vec_index_for(self).search( + # no per-item corpus rebuild. Bind the index once: calling + # _vec_index_for again after popping from self.items would + # see the size drift and force an O(n) rebuild. + idx = _vec_index_for(self) + hits = idx.search( embedding, k=1, memory_type=str(memory_type), ) if hits: @@ -1210,6 +1213,7 @@ def _semantic_sqlite_reinforce( extra = dict(matched.extra or {}) extra["reinforcement_count"] = extra.get("reinforcement_count", 1) + 1 extra["last_reinforced_at"] = now.isoformat() + row_written = False with self._sessions.session() as session: row = session.exec( _sel(self._memory_item_model).where( @@ -1221,10 +1225,26 @@ def _semantic_sqlite_reinforce( row.updated_at = now session.add(row) session.commit() - # Update in-memory cache - matched.extra = extra - matched.updated_at = now - return matched + row_written = True + if row_written: + # Update in-memory cache + matched.extra = extra + matched.updated_at = now + return matched + # The row is gone (deleted by another process), so + # the cache hit is stale, not authoritative. Evict it + # -- otherwise it stays a dedup magnet that silently + # drops every similar memorize -- and fall through to + # the real create path so this memory is stored. + logger.warning( + "Semantic dedup: cached %s item %s has no DB row " + "(deleted concurrently); evicting the stale entry " + "and creating instead", + memory_type, match_id, + ) + self.items.pop(match_id, None) + idx.remove(match_id) + idx.seen_items_len = len(self.items) return _original_sqlite_reinforce( self, diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 92037666..f80a983d 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1134,3 +1134,209 @@ async def test_transient_llm_error_still_raises_backend_unavailable(self, tmp_pa await bridge.memorize_file(str(target)) assert bridge._service.memorize.await_count == 1 + + +# --- Semantic-dedup reinforce: a stale cache hit must not report success ------ + + +def _semantic_reinforce_store(tmp_path, name, *, _models_cache={}): + """Build an isolated SQLiteStore with the real _patch_sqlite_bugs() applied. + + The scoped SQLA models are built once per process and shared: nerve's + _patched_get_models clears memu's model cache and rebuilds, and SQLAlchemy + refuses to reassign a Column object to a second Table, so a second bare + SQLiteStore(dsn=...) in one process raises "Column object 'url' already + assigned to Table 'memu_resources'". Each store still gets its own DB file. + """ + import memu.app # noqa: F401 -- FIRST, breaks a circular import in memu.database + import memu.database.sqlite.schema as schema_mod + from memu.database.sqlite.sqlite import SQLiteStore + from pydantic import BaseModel as _PydBaseModel + + if "models" not in _models_cache: + MemUBridge._patch_sqlite_bugs() + _models_cache["models"] = schema_mod.get_sqlite_sqlalchemy_models( + scope_model=_PydBaseModel, + ) + + db_path = tmp_path / f"{name}.sqlite" + store = SQLiteStore(dsn=f"sqlite:///{db_path}", sqla_models=_models_cache["models"]) + resource = store.resource_repo.create_resource( + url="mem://test", modality="text", local_path=str(tmp_path / "src.txt"), + caption=None, embedding=None, user_data={}, + ) + return store, resource, str(db_path) + + +def _item_row_count(db_path): + db = sqlite3.connect(db_path) + try: + return db.execute("SELECT COUNT(*) FROM memu_memory_items").fetchone()[0] + finally: + db.close() + + +def _item_extra_in_db(db_path, item_id): + """Read extra straight from the file -- proves persistence, not caching.""" + db = sqlite3.connect(db_path) + try: + row = db.execute( + "SELECT extra FROM memu_memory_items WHERE id = ?", (item_id,), + ).fetchone() + return json.loads(row[0]) if row and row[0] else None + finally: + db.close() + + +def _delete_row_externally(db_path, item_id): + """Delete the row over a SECOND connection -- what another process does.""" + db = sqlite3.connect(db_path) + try: + db.execute("DELETE FROM memu_memory_items WHERE id = ?", (item_id,)) + db.commit() + finally: + db.close() + + +class TestSemanticReinforceStaleCacheHit: + """A semantic-dedup hit on a row another process deleted must evict the stale + cache/index entry and actually persist the memory, not report a ghost. + """ + + # Cosine ~0.999 between the two, i.e. far above _SEMANTIC_DEDUP_THRESHOLD, + # so the second write is guaranteed to take the semantic-dedup branch. + EMB_SEED = [1.0, 0.0, 0.0, 0.0] + EMB_SIMILAR = [0.999, 0.0447, 0.0, 0.0] + + def _seed_then_delete_externally(self, tmp_path, name): + """Seed one item, warm the vector index, delete the row externally.""" + from nerve.memory.memu_bridge import _vec_index_for + + store, resource, db_path = _semantic_reinforce_store(tmp_path, name) + repo = store.memory_item_repo + + seeded = repo.create_item_reinforce( + resource_id=resource.id, memory_type="knowledge", + summary="the alpha fact about widgets", + embedding=self.EMB_SEED, user_data={}, + ) + assert _item_row_count(db_path) == 1, "seeding did not persist" + _vec_index_for(repo) # make the cached entry searchable + + _delete_row_externally(db_path, seeded.id) + assert _item_row_count(db_path) == 0, "external delete did not take" + # Precondition of the whole bug: the cache still serves the dead id. + assert seeded.id in repo.items + + return repo, resource, db_path, seeded.id + + def _reinforce_similar(self, repo, resource): + return repo.create_item_reinforce( + resource_id=resource.id, memory_type="knowledge", + summary="alpha fact regarding widgets", + embedding=self.EMB_SIMILAR, user_data={}, + ) + + def test_a_stale_cache_hit_creates_a_new_persisted_row(self, tmp_path): + repo, resource, db_path, dead_id = self._seed_then_delete_externally( + tmp_path, "stale-creates", + ) + + returned = self._reinforce_similar(repo, resource) + + assert returned.id != dead_id, ( + "returned the deleted id -- the ghost reinforce is back" + ) + assert (returned.extra or {}).get("reinforcement_count", 1) == 1, ( + "reinforcement_count > 1 makes memu/app/memorize.py:614 skip " + "creation and category linking, silently dropping the memory" + ) + assert _item_row_count(db_path) == 1, "the memory was not stored" + + def test_a_stale_cache_hit_evicts_the_cache_and_index_entry(self, tmp_path): + # _vec_index_note, NOT _vec_index_for: the latter REBUILDS whenever it + # sees the cache size drift, which silently repairs a missing + # idx.remove() and would make this assertion vacuous. + from nerve.memory.memu_bridge import _vec_index_note + + repo, resource, db_path, dead_id = self._seed_then_delete_externally( + tmp_path, "stale-evicts", + ) + + self._reinforce_similar(repo, resource) + + assert dead_id not in repo.items, "stale cache entry survived" + index = _vec_index_note(repo) + assert index is not None, "the vector index was never built" + assert dead_id not in index.id_to_row, ( + "stale vector-index entry survived -- still a dedup magnet" + ) + assert dead_id not in repo.list_items(), ( + "list_items() still serves the deleted id" + ) + + def test_after_a_stale_hit_the_new_item_is_visible_to_later_dedup(self, tmp_path): + """The eviction must leave the index in a REBUILDABLE state. + + ``_vec_index_for`` rebuilds only when ``seen_items_len`` differs from + ``len(items)``. Evicting without resyncing ``seen_items_len`` leaves + them equal (1 == 1) while the index itself is empty, so no rebuild ever + fires and the item created by the fall-through stays invisible to + semantic dedup for the rest of the process -- turning one silent drop + into permanently duplicated memories. + """ + repo, resource, db_path, _dead_id = self._seed_then_delete_externally( + tmp_path, "stale-then-visible", + ) + + created = self._reinforce_similar(repo, resource) + + # A third, still-similar memorize must dedup ONTO the new item. + third = repo.create_item_reinforce( + resource_id=resource.id, memory_type="knowledge", + summary="alpha facts on widgets now", + embedding=[0.998, 0.0632, 0.0, 0.0], user_data={}, + ) + + assert third.id == created.id, ( + "the item created after a stale hit is invisible to semantic dedup" + ) + assert _item_row_count(db_path) == 1, "dedup created a duplicate row" + + def test_the_returned_item_is_never_a_row_that_does_not_exist(self, tmp_path): + """The invariant, stated directly.""" + repo, resource, db_path, _dead_id = self._seed_then_delete_externally( + tmp_path, "stale-invariant", + ) + + returned = self._reinforce_similar(repo, resource) + + assert _item_extra_in_db(db_path, returned.id) is not None, ( + f"returned item {returned.id} has no row in the database" + ) + + def test_reinforce_still_dedups_when_the_row_exists(self, tmp_path): + """Control: the row-present path must be completely untouched.""" + from nerve.memory.memu_bridge import _vec_index_for + + store, resource, db_path = _semantic_reinforce_store(tmp_path, "row-present") + repo = store.memory_item_repo + + seeded = repo.create_item_reinforce( + resource_id=resource.id, memory_type="knowledge", + summary="the alpha fact about widgets", + embedding=self.EMB_SEED, user_data={}, + ) + _vec_index_for(repo) + + returned = self._reinforce_similar(repo, resource) + + assert returned.id == seeded.id, "semantic dedup stopped deduplicating" + assert (returned.extra or {}).get("reinforcement_count") == 2 + assert _item_row_count(db_path) == 1, "dedup created a duplicate row" + # The bump must be PERSISTED, not merely cached. + db_extra = _item_extra_in_db(db_path, seeded.id) + assert db_extra["reinforcement_count"] == 2, ( + "the reinforcement bump was not written to the database" + ) + assert "last_reinforced_at" in db_extra