diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 9642175b..4af42837 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -888,6 +888,23 @@ def _patch_sqlite_bugs(): Fix: use stored category embeddings. 5. list_items/list_categories bypass the vector cache. Fix: return cache when populated and unfiltered. + 7. Semantic dedup in create_item_reinforce (nerve addition, not a memU + bug fix). Its writeback seeds `extra` from the ROW inside the write + transaction, because read paths build MemoryItem without extra= and + writing a cache-derived extra back would drop content_hash and any + key another writer added. + 8. A content-only memory_update unlinks EVERY category: memU maps a + missing `categories` to [], so its diff removes the item's whole + current set. Fix: neutralise membership for that one call through a + relation-repo proxy passed down in `state` (nothing process-wide is + mutated), then rebuild the (old, new) pairs from the surviving links. + 9. update_item never refreshes extra.content_hash, which + create_item_reinforce dedups on, so an updated item keeps its old + text's hash. Fix: recompute it from the DB row when summary or + memory_type changes (only when a hash already exists). + 10. delete_item leaves the item's memu_category_items rows behind (no FK + / cascade). Fix: delete relations and the item in ONE transaction, + installed before Fix 3 so its index hook still wraps it. """ try: from pydantic import BaseModel @@ -985,14 +1002,230 @@ def _safe_create_tables(self): SQLiteStore._create_tables = _safe_create_tables + from memu.database.sqlite.repositories.memory_item_repo import SQLiteMemoryItemRepo + from memu.database.inmemory.vector import cosine_topk, cosine_topk_salience + + # Fix 10: delete_item deletes the item row and leaves its + # memu_category_items rows behind (there is no FK / ON DELETE + # CASCADE), so every memory_delete orphans that item's category + # relations. Replace the BASE implementation with one that deletes + # relations and the item in ONE transaction, so a failure can never + # leave a surviving item stripped of its links. + # + # Installed BEFORE Fix 3 so Fix 3's _indexed_delete_item wraps this + # and the vector-index hook still fires. Needs no idempotency guard: + # this is a full replacement that calls no captured original, so a + # repeated _patch_sqlite_bugs() cannot stack it (unlike the + # wrapper-style fixes). A marker attribute is kept for tests. + from sqlmodel import delete as _del, select as _sel_del + + # Stash memU's own implementation the first time, symmetrically with + # Fix 9: it documents what was replaced and gives tests a pristine + # reference to compare against. + if not hasattr(SQLiteMemoryItemRepo, "_nerve_memu_delete_item"): + SQLiteMemoryItemRepo._nerve_memu_delete_item = SQLiteMemoryItemRepo.delete_item + + def _cascade_delete_item(self, item_id): + rel_model = self._sqla_models.CategoryItem + with self._sessions.session() as session: + row = session.exec( + _sel_del(self._memory_item_model).where( + self._memory_item_model.id == item_id + ) + ).first() + # Relations are deleted on BOTH paths: when the item row is + # already gone (another process removed it) its + # memu_category_items rows are exactly the dangling rows + # this fix exists to prevent, so skipping them there would + # leave the changelog contract half-kept. Deleting the + # item is still conditional -- an unknown id stays a + # no-op, since the relation DELETE matches nothing. + session.exec(_del(rel_model).where(rel_model.item_id == item_id)) + if row is not None: + session.delete(row) + session.commit() + + # Evict on BOTH paths, like memU's own delete_item. When the row + # is already gone (another process removed it) returning early + # would leave the id in self.items, and Fix 5 serves that cache + # unfiltered, so list_items() would keep returning a deleted item. + self.items.pop(item_id, None) + relations = self._state.relations + relations[:] = [r for r in relations if r.item_id != item_id] + + _cascade_delete_item._nerve_cascade_delete = True # type: ignore[attr-defined] + SQLiteMemoryItemRepo.delete_item = _cascade_delete_item + + # Fix 9: update_item rewrites summary/memory_type but never refreshes + # extra.content_hash, which create_item_reinforce dedups on. An + # updated item therefore keeps its OLD text's hash forever, so + # re-memorizing the old wording reinforces the corrected row instead + # of being recognised as different content. + # + # The hash is written from the row memU's write LEFT BEHIND, never + # from a pre-read snapshot: a pre-read closes its session before that + # write, so any writer outside the memU loop thread (the date sweep + # runs on _blocking_pool with its own connection; `nerve memory` is a + # second process) could land in between and leave a hash of text the + # row does not hold. Reading the row afterwards also removes the + # need to guess the effective summary/type - the row IS the answer. + # + # The writeback is a single conditional UPDATE, so it is a no-op + # unless summary and memory_type still hold what was just written: + # if another writer moved them, we leave THEIR value alone rather + # than stamping a hash for text that is already gone. json_set + # merges into the row's current extra, so a concurrent writer's keys + # survive. Never from get_item(): read paths build MemoryItem + # without extra=, so a cached item has extra == {} and a cache-based + # version would silently refresh nothing in any long-lived process. + # + # Installed BEFORE Fix 3, like Fix 10. memU's own update_item is + # stashed on the class the first time, so a repeated + # _patch_sqlite_bugs() re-derives the SAME one-deep chain instead of + # stacking. A marker on the outermost function cannot achieve that: + # Fix 3 re-wraps whatever it finds, hiding our marker. + from memu.database.models import compute_content_hash as _content_hash + from sqlalchemy import text as _sql_text + from sqlmodel import select as _sel_upd + + _memu_update_item = getattr( + SQLiteMemoryItemRepo, "_nerve_memu_update_item", None) + if _memu_update_item is None: + _memu_update_item = SQLiteMemoryItemRepo.update_item + SQLiteMemoryItemRepo._nerve_memu_update_item = _memu_update_item + + def _hash_refreshing_update_item( + self, *, item_id, memory_type=None, summary=None, + embedding=None, extra=None, tool_record=None, + ): + # Delegate first, unchanged, and return what it returns: callers + # depend on the MemoryItem, and the caller's own `extra` must + # flow through this call (memorize() passes extra={"ref_id": ...} + # with no summary/type, which skips the refresh entirely). + result = _memu_update_item( + self, item_id=item_id, memory_type=memory_type, + summary=summary, embedding=embedding, extra=extra, + tool_record=tool_record, + ) + + # Everything below is BEST-EFFORT. It runs after memU's content + # write has already committed, so a failure here (a lock, say) + # must not turn a landed update into a raising call: that would + # leave the content written but the categories undiffed + # (_patch_update_memory_item diffs AFTER update_item), the vector + # index on the old embedding, and bridge.update_item reporting + # False for an update that partly applied. Degrading to a stale + # hash is exactly base's unconditional behaviour, so it is the + # strictly safer failure. The delegation itself stays OUTSIDE + # this guard: a genuine update failure must still propagate. + try: + if memory_type is None and summary is None: + return result + + with self._sessions.session() as session: + row = session.exec( + _sel_upd(self._memory_item_model).where( + self._memory_item_model.id == item_id + ) + ).first() + if row is None: + return result + current = dict(row.extra or {}) + # Only refresh a hash that already exists: an item + # created without one must not be newly enrolled into + # hash-dedup. + if not current.get("content_hash"): + return result + want = _content_hash(row.summary, str(row.memory_type)) + # The only interpolation is the model's own __tablename__ + # (Fix 6 renames memu's tables, so it cannot be a + # literal); every value is a bound parameter. RETURNING + # hands back the row's own merged extra, so the cache is + # never rebuilt from the pre-write snapshot. + returned = session.execute( + _sql_text( + f"UPDATE {self._memory_item_model.__tablename__} " + "SET extra = json_set(" + "coalesce(extra, '{}'), '$.content_hash', :want) " + "WHERE id = :item_id " + "AND json_extract(extra, '$.content_hash') IS NOT NULL " + "AND summary = :summary AND memory_type = :memory_type " + "RETURNING extra" + ), + { + "want": want, + "item_id": item_id, + "summary": row.summary, + "memory_type": str(row.memory_type), + }, + ).fetchall() + session.commit() + # rowcount is not meaningful with RETURNING; the row + # count is. Empty means the CAS declined: another writer + # moved summary/type, or removed the hash, between the + # read above and this statement. Nothing was written, so + # claiming a refreshed hash in the cache would make it + # disagree with the row. + if not returned: + return result + # json_set merges into the row's CURRENT extra while the + # CAS binds only summary/memory_type, so a writer that + # changed any OTHER key in the read -> write window + # committed and is present here. Reconstructing + # {**current, ...} would silently revert that key in the + # cache and the returned item - including + # reinforcement_count, which Fix 3's salience ranking + # reads. The JSON column may hand back either a string + # or an already-decoded dict. + # + # The CAS has already COMMITTED here, so a decode failure + # must not skip the cache assignments below: that would + # leave the cache and the returned item on the OLD hash + # while the row holds the new one. Only the decode is + # guarded, and the fallback IS the row's content on this + # path, because the CAS bound summary/memory_type and the + # only key it set is content_hash. + written = returned[0][0] + try: + if isinstance(written, str): + written = json.loads(written) + written = dict(written or {}) + except Exception as decode_exc: # noqa: BLE001 + logger.warning( + "content_hash refresh committed but its RETURNING " + "extra could not be decoded for memU item %s: %s; " + "rebuilding the cache from the pre-write snapshot", + item_id, decode_exc, + ) + written = {**current, "content_hash": want} + + # Keep the cache consistent with what was written, like the + # Fix 7 writeback does. + cached = self.items.get(item_id) + if cached is not None: + cached.extra = written + if getattr(result, "id", None) == item_id: + result.extra = written + return result + except Exception as exc: # noqa: BLE001 + # The row keeps its old hash, i.e. base's behaviour. Logged + # so a persistently failing refresh is visible rather than + # silent. + logger.warning( + "content_hash refresh skipped for memU item %s: %s", + item_id, exc, + ) + return result + + _hash_refreshing_update_item._nerve_hash_refresh = True # type: ignore[attr-defined] + SQLiteMemoryItemRepo.update_item = _hash_refreshing_update_item + # Fix 3: vector_search_items calls list_items() on every query, # re-reading and JSON-parsing all embeddings from SQLite (~2s for 3K items). # Worse, cosine_topk re-stacks every embedding into a brand-new # (n, dim) float32 matrix per call (~130 MB with 20K items). # Use the persistent incremental _VectorIndex instead: one # mat-vec per query, rows appended as items are created. - from memu.database.sqlite.repositories.memory_item_repo import SQLiteMemoryItemRepo - from memu.database.inmemory.vector import cosine_topk, cosine_topk_salience _original_vector_search = SQLiteMemoryItemRepo.vector_search_items @@ -1204,24 +1437,33 @@ def _semantic_sqlite_reinforce( "Semantic dedup: reinforcing %s item %s (%.3f) instead of creating '%s'", memory_type, match_id, score, summary[:80], ) - # Update DB row + # Update DB row. Seed `extra` from the ROW, not from + # the cache: read paths build MemoryItem without + # extra=, so a cached extra is {} in any process that + # did not itself write the item, and writing that back + # would drop content_hash and every key another writer + # added (e.g. mentioned_at, ref_id). from sqlmodel import select as _sel now = self._now() - extra = dict(matched.extra or {}) - extra["reinforcement_count"] = extra.get("reinforcement_count", 1) + 1 - extra["last_reinforced_at"] = now.isoformat() + extra: dict[str, Any] = {} with self._sessions.session() as session: row = session.exec( _sel(self._memory_item_model).where( self._memory_item_model.id == match_id ) ).first() - if row: + if row is not None: + extra = dict(row.extra or {}) + elif matched.extra: + extra = dict(matched.extra) + extra["reinforcement_count"] = extra.get("reinforcement_count", 1) + 1 + extra["last_reinforced_at"] = now.isoformat() + if row is not None: row.extra = extra row.updated_at = now session.add(row) session.commit() - # Update in-memory cache + # Keep the cache consistent with what was written matched.extra = extra matched.updated_at = now return matched @@ -1281,6 +1523,116 @@ def _semantic_inmemory_reinforce( _SEMANTIC_DEDUP_THRESHOLD, ) + # Fix 8: a content-only memory_update destroys every category link. + # memU has one sentinel for two meanings: _patch_update_memory_item + # maps a missing `categories` to [] (crud.py _map_category_names_to_ids + # returns [] for a falsy list), so cats_to_remove becomes the item's + # ENTIRE current set and each link is unlinked. + # + # An omitted `categories` therefore performs no membership mutation: + # the delegated call sees a relation repo whose link/unlink are + # no-ops, so no link can be removed by a diff it never should have + # computed. The shared repo is never modified -- the no-ops live on + # a proxy built per call and reachable only through the `store` this + # call passes down, so a concurrent coroutine on the same memU loop + # keeps the real mutators. The (old, new) pairs the LLM + # category-summary step consumes are rebuilt here from the links + # that actually survive. + from memu.app.crud import CRUDMixin as _CRUDMixin + + if not getattr(_CRUDMixin._patch_update_memory_item, "_nerve_keeps_categories", False): + _pre_keep_update_handler = _CRUDMixin._patch_update_memory_item + _CRUDMixin._nerve_memu_update_handler = _pre_keep_update_handler + + class _KeepRelRepoProxy: + """Per-call relation repo whose membership mutators are no-ops. + + Every other attribute is forwarded to the real repo, so the + delegated handler's reads (get_item_categories) see live + data. Instantiated inside the call, so nothing shared is + written and there is no restore to get wrong. + """ + + __slots__ = ("_nerve_real",) + + def __init__(self, real): + object.__setattr__(self, "_nerve_real", real) + + def unlink_item_category(self, *_args, **_kwargs): + return None + + def link_item_category(self, *_args, **_kwargs): + return None + + def __getattr__(self, name): + return getattr(object.__getattribute__(self, "_nerve_real"), name) + + class _KeepStoreProxy: + """Per-call store serving the relation proxy, nothing else changed.""" + + __slots__ = ("_nerve_real", "_nerve_rel") + + def __init__(self, real, rel_proxy): + object.__setattr__(self, "_nerve_real", real) + object.__setattr__(self, "_nerve_rel", rel_proxy) + + @property + def category_item_repo(self): + return object.__getattribute__(self, "_nerve_rel") + + def __getattr__(self, name): + return getattr(object.__getattribute__(self, "_nerve_real"), name) + + async def _category_preserving_update(self, state, step_context): + payload = state.get("memory_payload") or {} + # An explicit list still replaces; an explicit [] still clears. + if payload.get("categories") is not None: + return await _pre_keep_update_handler(self, state, step_context) + + store = state["store"] + memory_id = state["memory_id"] + rel_repo = store.category_item_repo + item_before = store.memory_item_repo.get_item(memory_id) + old_content = getattr(item_before, "summary", None) + + # Neutralise membership for THIS call only. The delegated + # handler reads store.memory_item_repo and + # store.category_item_repo; the proxies serve both. + store_proxy = _KeepStoreProxy(store, _KeepRelRepoProxy(rel_repo)) + out = await _pre_keep_update_handler( + self, {**state, "store": store_proxy}, step_context, + ) + + # run_steps threads the returned mapping into persist_index + # and build_response, so the proxy must not escape its + # window: hand the REAL store back. + out = dict(out) + out["store"] = store + + # memU diffed against a no-op, so its pairs are wrong: a live + # link marked (old, None) renders as "content is discarded". + if payload.get("content"): + new_content = getattr(out.get("memory_item"), "summary", None) + # Keep only ids the response step can resolve: + # _patch_build_response subscripts + # memory_category_repo.categories UNGUARDED, so a + # dangling relation row would raise KeyError after the + # content write has already committed. + known = store.memory_category_repo.categories + out["category_updates"] = { + rel.category_id: (old_content, new_content) + for rel in rel_repo.get_item_categories(memory_id) + if rel.category_id in known + } + else: + out["category_updates"] = {} + return out + + _category_preserving_update._nerve_keeps_categories = True # type: ignore[attr-defined] + _CRUDMixin._patch_update_memory_item = _category_preserving_update + + logger.info("Patched memU write paths (category preservation, hash refresh, cascade delete)") + except Exception as e: logger.error( "memU monkey-patching failed (expected memu-py==%s): %s. " diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 92037666..73573f04 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -2,11 +2,14 @@ import asyncio import json +import logging import sqlite3 +import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch +import numpy as np import pytest import pytest_asyncio @@ -975,6 +978,295 @@ def test_missing_db_does_not_raise(self, tmp_path): # --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- +# memU write-path integrity (Fixes 7-correction, 8, 9, 10) +# --------------------------------------------------------------------------- + + +_MEMU_MODELS_CACHE = {} + + +def _memu_models(): + """Build the SQLAlchemy models ONCE per process. + + nerve's Fix 6 (_patched_get_models) clears memu's _MODEL_CACHE on every + call, so a second SQLiteStore(dsn=...) that rebuilds them raises + ArgumentError("Column object 'url' already assigned"). Sharing one build + across stores is what lets these tests use several isolated stores. + """ + if "models" in _MEMU_MODELS_CACHE: + return _MEMU_MODELS_CACHE["models"] + + import memu.app.service # noqa: F401 - initialize the package graph + import memu.database.sqlite.schema as schema_mod + from memu.app.crud import CRUDMixin + from memu.database.sqlite.repositories.memory_item_repo import ( + SQLiteMemoryItemRepo as Repo, + ) + + # Building the models needs the patches applied (Fix 6 renames memu's own + # "sqlite_*" tables, which SQLite reserves), but this helper must leave + # global state exactly as it found it: leaking a patched Repo out of a + # module-level helper is how one test silently breaks its neighbours. + saved_repo = {n: Repo.__dict__.get(n) for n in _PATCHED_ITEM_REPO_ATTRS} + saved_handler = CRUDMixin.__dict__.get("_patch_update_memory_item") + saved_stash = CRUDMixin.__dict__.get("_nerve_memu_update_handler") + try: + MemUBridge._patch_sqlite_bugs() + # Resolve through the MODULE, never a from-import: Fix 6 replaces this + # attribute. + models = schema_mod.get_sqlite_sqlalchemy_models(scope_model=None) + finally: + _restore_attrs(Repo, saved_repo) + _restore_attrs(CRUDMixin, { + "_patch_update_memory_item": saved_handler, + "_nerve_memu_update_handler": saved_stash, + }) + + _MEMU_MODELS_CACHE["models"] = models + return models + + +def _restore_attrs(cls, saved): + """Put class attributes back, deleting those that did not exist before.""" + for name, value in saved.items(): + if value is None: + if name in cls.__dict__: + delattr(cls, name) + else: + setattr(cls, name, value) + + +def _content_hash(summary, memory_type): + """memu's compute_content_hash, imported safely. + + ``from memu.database.models import ...`` FIRST triggers a circular import + inside memu itself (database/__init__ -> factory -> app/__init__ -> service + -> factory), so memu.app.service must be imported before it. Going through + this helper is what lets a single test in these classes run alone. + """ + import memu.app.service # noqa: F401 + from memu.database.models import compute_content_hash + + return compute_content_hash(summary, memory_type) + + +def _race_extra_in_the_write_window(fx, repo, item_id, mutate_extra): + """Land ``mutate_extra`` on the row's extra between the hash read and write. + + Hooks ``session.execute`` and fires once on the conditional hash write, so + the concurrent commit lands strictly inside the read -> write window; memU's + own update_item goes through exec/add/commit and is not intercepted. Returns + ``(result, raced)`` so the caller can assert the race really happened. + """ + sessions = repo._sessions + original_session = sessions.session + raced = [] + + def racing_session(): + session = original_session() + real_execute = session.execute + + def _execute(stmt, *args, **kwargs): + if "json_set" in str(stmt) and not raced: + raced.append(True) + conn = sqlite3.connect(fx.path) + row = conn.execute( + "SELECT extra FROM memu_memory_items WHERE id = ?", (item_id,), + ).fetchone() + extra = json.loads(row[0]) if row and row[0] else {} + mutate_extra(extra) + conn.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = ?", + (json.dumps(extra), item_id), + ) + conn.commit() + conn.close() + return real_execute(stmt, *args, **kwargs) + + session.execute = _execute + return session + + sessions.session = racing_session + try: + return repo.update_item(item_id=item_id, summary="my new text"), raced + finally: + sessions.session = original_session + + +# Every memu-py class attribute the patches reassign, so each test can restore +# global state and never leak into another test (or another suite: a module that +# mutates shared globals at import time breaks its neighbours invisibly, so all +# patching happens INSIDE tests). +class _NoRowSessions: + """Session manager whose queries return no row. + + Lets a bare ``object.__new__(Repo)`` stub satisfy the hash-refresh wrapper's + row lookup: with no row there is nothing to recompute a hash from, so it + delegates straight through - which is what the forwarding test measures. + """ + + class _Session: + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def exec(self, *args, **kwargs): + return self + + def first(self): + return None + + def session(self): + return self._Session() + + +_PATCHED_ITEM_REPO_ATTRS = ( + "update_item", "delete_item", "clear_items", "list_items", + "create_item", "create_item_reinforce", "vector_search_items", + "_nerve_memu_update_item", "_nerve_memu_delete_item", +) + + +class _MemuPatchFixture: + """Isolated memU stores over a temp file with _patch_sqlite_bugs() applied.""" + + def __init__(self, tmp_path): + self.models = _memu_models() + self.path = str(tmp_path / "memu.sqlite") + self._stores = [] + self.saved_item_repo = {} + self.saved_handler = None + + def __enter__(self): + import memu.app.service # noqa: F401 + from memu.app.crud import CRUDMixin + from memu.database.sqlite.repositories.memory_item_repo import ( + SQLiteMemoryItemRepo as Repo, + ) + + self.Repo = Repo + self.CRUDMixin = CRUDMixin + self.saved_item_repo = { + n: Repo.__dict__.get(n) for n in _PATCHED_ITEM_REPO_ATTRS + } + self.saved_handler = CRUDMixin.__dict__.get("_patch_update_memory_item") + self.saved_handler_stash = CRUDMixin.__dict__.get("_nerve_memu_update_handler") + MemUBridge._patch_sqlite_bugs() + return self + + def __exit__(self, *exc): + for store in self._stores: + try: + store.close() + except Exception: + pass + _restore_attrs(self.Repo, self.saved_item_repo) + _restore_attrs(self.CRUDMixin, { + "_patch_update_memory_item": self.saved_handler, + "_nerve_memu_update_handler": self.saved_handler_stash, + }) + return False + + def store(self): + """A fresh store over the same file (a simulated process restart).""" + from memu.database.sqlite.sqlite import SQLiteStore + + store = SQLiteStore(dsn=f"sqlite:///{self.path}", sqla_models=self.models) + self._stores.append(store) + return store + + def setup(self, n_categories=2): + store = self.store() + cats = {} + for name in ("procedures", "patterns", "decisions")[:n_categories]: + cat = store.memory_category_repo.get_or_create_category( + name=name, description="d", embedding=None, user_data={}, + ) + cats[name] = cat.id + return store, cats + + def add_item(self, store, summary="a fact", memory_type="knowledge", embedding=None): + return store.memory_item_repo.create_item_reinforce( + resource_id=None, memory_type=memory_type, summary=summary, + embedding=embedding, user_data={}, + ) + + def link_all(self, store, item_id, cats): + for cid in cats.values(): + store.category_item_repo.link_item_category(item_id, cid, user_data={}) + + def raw_extra(self, item_id): + row = sqlite3.connect(self.path).execute( + "SELECT extra FROM memu_memory_items WHERE id = ?", (item_id,), + ).fetchone() + return json.loads(row[0]) if row and row[0] else {} + + def raw_count(self, sql, *params): + return sqlite3.connect(self.path).execute(sql, params).fetchone()[0] + + def run_update(self, store, cats, item_id, *, content=None, memory_type=None, + categories=None, ctx=None): + """Drive the REAL memU update workflow handler (no mock of it).""" + class _Ctx: + def __init__(self, mapping): + self.category_name_to_id = dict(mapping) + self.category_ids = list(mapping.values()) + + class _Embed: + async def embed(self, payload): + return [None] + + svc = object.__new__(self.CRUDMixin) + svc._get_step_embedding_client = lambda step_ctx: _Embed() + state = { + "memory_id": item_id, + "memory_payload": {"content": content, "type": memory_type, + "categories": categories}, + "ctx": ctx if ctx is not None else _Ctx(cats), + "store": store, + "user": {}, + } + return asyncio.run( + self.CRUDMixin._patch_update_memory_item(svc, state, None), + ) + + def update_coro(self, store, cats, item_id, *, content=None, memory_type=None, + categories=None, ctx=None, embed=None): + """Same call as run_update, but returns the coroutine unawaited. + + ``embed`` lets a test suspend the delegated handler mid-flight (its only + await is the embedding call), which is what makes two updates genuinely + interleave rather than run back to back. + """ + class _Embed: + async def embed(self, payload): + return [None] + + svc = object.__new__(self.CRUDMixin) + client = embed if embed is not None else _Embed() + svc._get_step_embedding_client = lambda step_ctx: client + state = { + "memory_id": item_id, + "memory_payload": {"content": content, "type": memory_type, + "categories": categories}, + "ctx": ctx if ctx is not None else self.ctx(cats), + "store": store, + "user": {}, + } + return self.CRUDMixin._patch_update_memory_item(svc, state, None) + + def ctx(self, mapping): + class _Ctx: + def __init__(self, m): + self.category_name_to_id = dict(m) + self.category_ids = list(m.values()) + + return _Ctx(mapping) + + class TestIndexedUpdateItemForwarding: """Regression: the _indexed_update_item monkeypatch must forward item_id by keyword. @@ -1013,13 +1305,20 @@ def spy_update(self, *, item_id, memory_type=None, summary=None, calls.append(item_id) return "spy-result" - # Snapshot the item-repo methods _patch_sqlite_bugs() reassigns so the - # test restores global state and does not leak into other tests. - names = ( - "update_item", "delete_item", "clear_items", "list_items", - "create_item", "create_item_reinforce", "vector_search_items", - ) + # Snapshot every item-repo attribute _patch_sqlite_bugs() reassigns so + # the test restores global state and does not leak into other tests. + # The _nerve_memu_* stashes matter as much as the methods: leaving one + # behind hands the NEXT test this spy as "memU's own implementation". + names = _PATCHED_ITEM_REPO_ATTRS saved = {n: Repo.__dict__.get(n) for n in names} + # Fix 8 installs on CRUDMixin, not on Repo, so those two attributes leak + # out of this class unless they are snapshotted here as well. + from memu.app.crud import CRUDMixin + + saved_crud = { + n: CRUDMixin.__dict__.get(n) + for n in ("_patch_update_memory_item", "_nerve_memu_update_handler") + } Repo.update_item = spy_update try: @@ -1029,6 +1328,11 @@ def spy_update(self, *, item_id, memory_type=None, summary=None, assert Repo.update_item is not spy_update stub = object.__new__(Repo) # no _nerve_vec_index → index hook skipped + # The hash-refresh wrapper reads the item's current row, so the stub + # needs the two attributes that read uses. Returning no row makes it + # skip the refresh and delegate, which is what this test measures. + stub._memory_item_model = _memu_models().MemoryItem + stub._sessions = _NoRowSessions() # Exactly how memu's crud layer calls it (all keyword) — used to raise. result = Repo.update_item( stub, item_id="mem-123", memory_type=None, @@ -1043,6 +1347,7 @@ def spy_update(self, *, item_id, memory_type=None, summary=None, delattr(Repo, name) else: setattr(Repo, name, fn) + _restore_attrs(CRUDMixin, saved_crud) class TestSqliteLockedClassifier: @@ -1134,3 +1439,1716 @@ 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 + + +class TestUpdatePreservesCategories: + """Fix 8: a content-only memory_update must not unlink every category. + + memU has ONE sentinel for two meanings: _patch_update_memory_item maps a + missing ``categories`` to ``[]`` (_map_category_names_to_ids returns [] for + a falsy list), so ``cats_to_remove`` becomes the item's ENTIRE current set. + Measured on the live store before the fix: 154 of 154 items ever updated + without a categories argument held ZERO category links. + """ + + def test_content_only_update_keeps_links_and_rows(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + before = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + + out = fx.run_update(store, cats, item.id, content="a corrected fact") + + after = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + assert len(after) == 2 + # Same relation row ids: preserved, not deleted and re-created. + assert after == before + # The summary-patch step must see "updated", never "discarded": + # (old, None) renders as "This memory content is discarded" and + # would make the LLM drop an item whose link still exists. + assert sorted(out["category_updates"].values()) == [ + ("a fact", "a corrected fact"), + ] * 2 + + def test_unpatched_handler_loses_every_link(self, tmp_path): + """The control arm: without Fix 8 the same call unlinks everything.""" + with _MemuPatchFixture(tmp_path) as fx: + # Use memU's own handler from the stash, not fx.saved_handler: + # a previous test in this process may already have patched the class, + # so the snapshot is not guaranteed to be the pristine original. + fx.CRUDMixin._patch_update_memory_item = ( + fx.CRUDMixin._nerve_memu_update_handler + ) + + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + + out = fx.run_update(store, cats, item.id, content="a corrected fact") + + assert store.category_item_repo.get_item_categories(item.id) == [] + assert sorted(out["category_updates"].values()) == [("a fact", None)] * 2 + + def test_explicit_list_still_replaces(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + + fx.run_update(store, cats, item.id, content="y", categories=["patterns"]) + + links = [r.category_id for r in + store.category_item_repo.get_item_categories(item.id)] + assert links == [cats["patterns"]] + + def test_explicit_empty_list_still_clears(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + + fx.run_update(store, cats, item.id, content="y", categories=[]) + + assert store.category_item_repo.get_item_categories(item.id) == [] + + def test_three_links_preserved(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(3) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + before = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + + out = fx.run_update(store, cats, item.id, content="revised") + + after = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + assert len(after) == 3 + assert after == before + assert set(out["category_updates"].values()) == {("a fact", "revised")} + + def test_incomplete_map_preserves_every_link(self, tmp_path): + """An incomplete ctx map must be harmless, not merely detected. + + The earlier design round-tripped the links through + ctx.category_name_to_id and raised when a link did not map back, because + an unmapped link would have been silently dropped. Membership is no + longer mutated at all, so the map is not consulted and the STRONGER + property holds: every link survives even when the map cannot name it. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(3) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + before = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + partial = {k: v for k, v in cats.items() if k != "decisions"} + ctx = fx.ctx(partial) + snapshot = dict(ctx.category_name_to_id) + + out = fx.run_update(store, cats, item.id, content="revised", ctx=ctx) + + after = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + # Same relation rows, including the one `partial` cannot name. + assert after == before + assert len(after) == 3 + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 3 + # Every surviving link is reported as updated, unmapped one included. + assert set(out["category_updates"]) == set(cats.values()) + assert set(out["category_updates"].values()) == {("a fact", "revised")} + # The update really happened, or the assertions above are free. + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ? AND summary = ?", + item.id, "revised", + ) == 1 + # Service-lifetime state (shared with memorize) must not be mutated. + assert ctx.category_name_to_id == snapshot + + def test_no_category_name_lookup_is_performed(self, tmp_path): + """Preservation must not depend on the name->id map at all. + + A ctx whose category_name_to_id raises on access proves the block reads + it nowhere; that is what makes an incomplete map structurally harmless + rather than harmless-by-luck. + """ + class _ExplodingCtx: + category_ids: list = [] + + @property + def category_name_to_id(self): + raise AssertionError("category_name_to_id must not be read") + + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + + fx.run_update(store, cats, item.id, content="revised", + ctx=_ExplodingCtx()) + + assert len(store.category_item_repo.get_item_categories(item.id)) == 2 + + def test_item_with_no_links_is_a_noop(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + + out = fx.run_update(store, cats, item.id, content="y") + + assert store.category_item_repo.get_item_categories(item.id) == [] + assert out["category_updates"] == {} + + def test_type_only_update_keeps_links(self, tmp_path): + """Preservation must not depend on `content` being supplied. + + bridge.update_item allows a type-only change (memory_type set, content + None) from both the tool handler and the web route, and that shape + reaches memU's diff exactly like a content-only one. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + before = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + assert len(before) == 2 + + out = fx.run_update(store, cats, item.id, memory_type="profile") + + after = {r.id for r in store.category_item_repo.get_item_categories(item.id)} + assert after == before + # No content changed, so there is nothing for the summary step to + # patch: reporting (old, old) pairs would re-summarise every category + # of every type-only edit. + assert out["category_updates"] == {} + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + # The update must not have been a no-op, or the assertion above is free. + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ? AND memory_type = ?", + item.id, "profile", + ) == 1 + + def test_a_link_added_between_the_two_reads_is_preserved(self, tmp_path): + """The former residual window is CLOSED, and pinned so it stays closed. + + The earlier design synthesized a categories payload from the links it + saw, which memU then re-read to diff: a link inserted between those two + reads was absent from the payload, so the diff removed it. Membership is + no longer mutated at all, so a racing insert survives. + + The race MUST be inserted by raw SQL. Going through + rel.link_item_category would be swallowed by the no-op stub this fix + installs for the delegated call, and the test would pass vacuously. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + repo = store.category_item_repo + repo.link_item_category(item.id, cats["procedures"], user_data={}) + + real_get = repo.get_item_categories + calls = [] + landed = [] + + def counting_get(iid): + calls.append(iid) + out = real_get(iid) + if len(calls) == 1: + con = sqlite3.connect(fx.path) + con.execute( + "INSERT INTO memu_category_items" + " (id, item_id, category_id, created_at, updated_at)" + " VALUES (?, ?, ?, datetime('now'), datetime('now'))", + (str(uuid.uuid4()), item.id, cats["patterns"]), + ) + con.commit() + con.close() + # The race is real only if the row is in the DB already. + landed.append(fx.raw_count( + "SELECT count(*) FROM memu_category_items" + " WHERE item_id = ? AND category_id = ?", + item.id, cats["patterns"], + )) + return out + + repo.get_item_categories = counting_get + try: + out = fx.run_update(store, cats, item.id, content="revised") + finally: + repo.get_item_categories = real_get + + # Both reads happened, or the window was never opened. + assert len(calls) == 2 + # The raw insert landed before the second read, or there was no race. + assert landed == [1] + links = { + r[0] for r in sqlite3.connect(fx.path).execute( + "SELECT category_id FROM memu_category_items WHERE item_id = ?", + (item.id,), + ).fetchall() + } + assert cats["procedures"] in links + # The link added inside the window survives: the window is closed. + assert cats["patterns"] in links + # ... and it is reported to the summary step, not silently dropped. + assert out["category_updates"][cats["patterns"]] == ("a fact", "revised") + + def test_the_stubs_never_reach_the_shared_repo_under_overlap(self, tmp_path): + """The load-bearing item-1 test: two omitted-``categories`` updates that + genuinely interleave must not neutralise each other's repo. + + The neutralisation lives on a per-call proxy, so it is reachable only + through the ``store`` mapping one call passes down. An instance-level + ``setattr`` on ``store.category_item_repo`` would instead be in force for + every coroutine on the memU loop for the whole duration of the delegated + ``await`` (crud.py's embed call), and its restore is not reentrant: with A + entering, B entering, then A finishing first, B's ``finally`` puts A's + stub back permanently. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + first = fx.add_item(store, summary="first fact") + second = fx.add_item(store, summary="second fact") + fx.link_all(store, first.id, cats) + fx.link_all(store, second.id, cats) + repo = store.category_item_repo + names = ("unlink_item_category", "link_item_category") + snapshots = [] + + gate = None + + class _GatedEmbed: + """First caller parks inside the delegated handler until released.""" + + def __init__(self): + self.entered = asyncio.Event() + + async def embed(self, payload): + # Observe the SHARED repo from inside the window: this is + # where an instance stub would be visible. + snapshots.append( + {n: n in repo.__dict__ for n in names}, + ) + if not self.entered.is_set(): + self.entered.set() + await gate.wait() + return [None] + + embed = _GatedEmbed() + + async def drive(): + nonlocal gate + gate = asyncio.Event() + a = asyncio.ensure_future( + fx.update_coro(store, cats, first.id, + content="first revised", embed=embed), + ) + # B must enter A's window, or the arms never overlap. BOUNDED: + # a handler that raises before reaching embed never sets this, + # and an unbounded wait would hang the whole suite instead of + # failing -- which is exactly how a broken mutant stalls a + # mutation matrix rather than being reported. + try: + await asyncio.wait_for(embed.entered.wait(), timeout=10) + except TimeoutError: # pragma: no cover - diagnostic path + gate.set() + exc = a.exception() if a.done() else None + msg = ("the first update never reached its embed step, so the " + f"overlap window never opened (task exception: {exc!r})") + raise AssertionError(msg) from exc + b = asyncio.ensure_future( + fx.update_coro(store, cats, second.id, + content="second revised", embed=embed), + ) + # Let B reach its own embed before A is allowed to finish, so + # both are inside the window at the same time. + for _ in range(50): + if len(snapshots) >= 2: + break + await asyncio.sleep(0) + gate.set() + return await asyncio.wait_for(asyncio.gather(a, b), timeout=30) + + out_a, out_b = asyncio.run(drive()) + + # Both arms really were inside the window together. + assert len(snapshots) >= 2, snapshots + # (b) the shared repo never carried either stub, at any point. + assert all(s == {n: False for n in names} for s in snapshots), snapshots + assert not any(n in repo.__dict__ for n in names) + + # (a) both items keep all their links. + for item in (first, second): + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", + item.id, + ) == 2 + assert set(out_a["category_updates"]) == set(cats.values()) + assert set(out_b["category_updates"]) == set(cats.values()) + + # (c) the real mutator still works afterwards. + repo.unlink_item_category(first.id, cats["procedures"]) + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", first.id, + ) == 1 + + def test_a_concurrent_link_inside_the_window_still_lands(self, tmp_path): + """The cross-pipeline victim: another caller's membership write must work. + + An instance-level stub silently swallows every ``link_item_category`` on + the shared repo while one content-only update is parked in its + ``await`` - including an explicit-list update from another coroutine, + which then reports success having changed nothing. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store, summary="a fact") + victim = fx.add_item(store, summary="another fact") + fx.link_all(store, item.id, cats) + repo = store.category_item_repo + target = cats["patterns"] + + class _GatedEmbed: + def __init__(self): + self.entered = asyncio.Event() + self.release = asyncio.Event() + + async def embed(self, payload): + self.entered.set() + await self.release.wait() + return [None] + + embed = _GatedEmbed() + + async def drive(): + task = asyncio.ensure_future( + fx.update_coro(store, cats, item.id, + content="revised", embed=embed), + ) + # BOUNDED, like the overlap test above: a handler that raises + # before reaching embed must FAIL this test, never hang it. + try: + await asyncio.wait_for(embed.entered.wait(), timeout=10) + except TimeoutError: # pragma: no cover - diagnostic path + embed.release.set() + exc = task.exception() if task.done() else None + msg = ("the update never reached its embed step, so the window " + f"never opened (task exception: {exc!r})") + raise AssertionError(msg) from exc + # Issued from INSIDE the first update's window, on the SHARED repo. + repo.link_item_category(victim.id, target, user_data={}) + landed = fx.raw_count( + "SELECT count(*) FROM memu_category_items" + " WHERE item_id = ? AND category_id = ?", + victim.id, target, + ) + embed.release.set() + await asyncio.wait_for(task, timeout=30) + return landed + + landed_inside = asyncio.run(drive()) + + # The write took effect immediately, not after the window closed. + assert landed_inside == 1 + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", + victim.id, + ) == 1 + # ... and the parked update still preserved its own links. + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + + def test_the_proxy_does_not_escape_the_delegated_call(self, tmp_path): + """run_steps threads the returned mapping into the LATER steps. + + ``step.run`` returns ``dict(result)`` and ``run_steps`` assigns it to + ``state``, so whatever ``store`` the returned mapping carries flows into + persist_index and build_response. Leaking a proxy past its window is the + class of defect the per-call design exists to remove, so the mapping must + hand back the REAL store. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store) + fx.link_all(store, item.id, cats) + + out = fx.run_update(store, cats, item.id, content="revised") + + assert out["store"] is store + assert out["store"].category_item_repo is store.category_item_repo + # The delegated call must have SEEN a proxy, or the assertion above + # would hold for a build that never neutralised anything. + assert store.category_item_repo.__class__.__name__ != "_KeepRelRepoProxy" + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + + def test_a_dangling_relation_row_is_kept_out_of_category_updates(self, tmp_path): + """Item 2: memU's build_response subscripts the category dict UNGUARDED. + + ``_patch_build_response`` does ``memory_category_repo.categories[c]`` for + every key of ``category_updates``, and patch_update runs it AFTER the + content write has committed - so an id it cannot resolve raises + ``KeyError`` post-commit and ``bridge.update_item`` reports False for an + update that fully applied. memU could not reach that state (it derived + ids through ``_map_category_names_to_ids``); rebuilding from raw relation + rows can, so the rebuild filters to resolvable ids. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(1) + item = fx.add_item(store) + known = cats["procedures"] + store.category_item_repo.link_item_category(item.id, known, user_data={}) + + unknown = "no-such-category-" + uuid.uuid4().hex[:8] + con = sqlite3.connect(fx.path) + con.execute( + "INSERT INTO memu_category_items" + " (id, item_id, category_id, created_at, updated_at)" + " VALUES (?, ?, ?, datetime('now'), datetime('now'))", + (str(uuid.uuid4()), item.id, unknown), + ) + con.commit() + con.close() + assert unknown not in store.memory_category_repo.categories + + out = fx.run_update(store, cats, item.id, content="revised") + + # The content write landed and the call returned normally. + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ? AND summary = ?", + item.id, "revised", + ) == 1 + assert known in out["category_updates"] + assert unknown not in out["category_updates"] + # The dangling row itself is left alone: this fix filters the + # REPORT, it does not delete relations. + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + # And what build_response would subscript is now resolvable. + for cid in out["category_updates"]: + assert cid in store.memory_category_repo.categories + + +class TestUpdateRefreshesContentHash: + """Fix 9: update_item must refresh extra.content_hash. + + create_item_reinforce dedups on json_extract(extra,'$.content_hash'), but + update_item rewrites summary/memory_type without touching it, so an updated + item keeps its OLD text's hash. Measured before the fix: 149 of 149 updated + items were hash-stale against a 400/400 fresh never-updated baseline. + """ + + def test_hash_matches_new_summary(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="original text") + + store.memory_item_repo.update_item(item_id=item.id, summary="corrected text") + + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "corrected text", "knowledge", + ) + + def test_hash_refreshed_after_a_restart(self, tmp_path): + """The discriminating arm: the refresh must read the DB ROW. + + get_item()/list_items() build MemoryItem WITHOUT extra=, so a cached + item has extra == {}. A cache-based implementation finds no + content_hash here and refreshes nothing, while still passing the + create-path test above. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="original text") + store.close() + + restarted = fx.store() + restarted.memory_item_repo.list_items() + assert restarted.memory_item_repo.items[item.id].extra == {} + + restarted.memory_item_repo.update_item( + item_id=item.id, summary="corrected text", + ) + + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "corrected text", "knowledge", + ) + + def test_unpatched_update_leaves_the_hash_stale(self, tmp_path): + """Control arm: memU's own update_item keeps the old text's hash.""" + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="original text") + memu_update = fx.Repo._nerve_memu_update_item + + memu_update(store.memory_item_repo, item_id=item.id, summary="corrected text") + + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "original text", "knowledge", + ) + + def test_type_change_alone_recomputes_from_the_row(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="the text") + + store.memory_item_repo.update_item(item_id=item.id, memory_type="behavior") + + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "the text", "behavior", + ) + + def test_summary_and_type_changed_together_hash_from_both(self, tmp_path): + """The combined shape. Changing summary and memory_type in one call must + hash from BOTH new values: an implementation that keeps the old type + whenever a summary is supplied satisfies the summary-only and type-only + cases above while hashing a combined update wrongly. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig", memory_type="knowledge") + + store.memory_item_repo.update_item( + item_id=item.id, summary="new text", memory_type="behavior", + ) + + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "new text", "behavior", + ) + # Neither half may have been a no-op, or the assertion is free. + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items" + " WHERE id = ? AND summary = ? AND memory_type = ?", + item.id, "new text", "behavior", + ) == 1 + + def test_a_racing_type_change_is_not_hashed_from_the_argument(self, tmp_path): + """The type half of the same property as the summary case below. + + Taking the type from the ARGUMENT rather than from the row passes every + non-racing case (after the delegation commits they are equal), yet + produces a hash for a type the row no longer holds - and the conditional + write cannot catch it, because that predicate binds the row's own value. + Only reading BOTH halves off the row is correct. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig", memory_type="knowledge") + repo = store.memory_item_repo + + sessions = repo._sessions + original_session = sessions.session + raced = [] + + def racing_session(): + session = original_session() + real_commit = session.commit + + def _commit(): + real_commit() + if raced: + return + conn = sqlite3.connect(fx.path) + landed = conn.execute( + "SELECT memory_type FROM memu_memory_items WHERE id = ?", + (item.id,), + ).fetchone() + if landed and landed[0] == "behavior": + conn.execute( + "UPDATE memu_memory_items SET memory_type = ? WHERE id = ?", + ("profile", item.id), + ) + conn.commit() + raced.append(True) + conn.close() + + session.commit = _commit + return session + + sessions.session = racing_session + try: + repo.update_item(item_id=item.id, memory_type="behavior") + finally: + sessions.session = original_session + + assert raced == [True] + final_type = sqlite3.connect(fx.path).execute( + "SELECT memory_type FROM memu_memory_items WHERE id = ?", (item.id,), + ).fetchone()[0] + assert final_type == "profile" + + stored = fx.raw_extra(item.id)["content_hash"] + assert stored in ( + _content_hash("orig", final_type), + _content_hash("orig", "knowledge"), + ) + assert stored != _content_hash("orig", "behavior") + + def test_a_writer_outside_the_memu_loop_cannot_leave_a_stale_hash(self, tmp_path): + """The hash must come from the row the write LEFT BEHIND. + + A pre-read snapshot closes its session before memU's write transaction, + so a writer outside the memU loop thread (the date sweep runs on + _blocking_pool with its own connection; `nerve memory` is a second + process) can land in between and leave a hash of text the row does not + hold. Here a second connection rewrites the summary the instant memU's + write commits - which is AFTER the pre-read snapshot but BEFORE a + post-write read, so the hook is shape-agnostic and lands in the window + either implementation exposes. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + + sessions = repo._sessions + original_session = sessions.session + raced = [] + + def racing_session(): + session = original_session() + real_commit = session.commit + + def _commit(): + real_commit() + if raced: + return + conn = sqlite3.connect(fx.path) + landed = conn.execute( + "SELECT summary FROM memu_memory_items WHERE id = ?", + (item.id,), + ).fetchone() + if landed and landed[0] == "my new text": + conn.execute( + "UPDATE memu_memory_items SET summary = ? WHERE id = ?", + ("text from the other writer", item.id), + ) + conn.commit() + raced.append(True) + conn.close() + + session.commit = _commit + return session + + sessions.session = racing_session + try: + repo.update_item(item_id=item.id, summary="my new text") + finally: + sessions.session = original_session + + # The race really happened, or the test proves nothing. + assert raced == [True] + final_summary = sqlite3.connect(fx.path).execute( + "SELECT summary FROM memu_memory_items WHERE id = ?", (item.id,), + ).fetchone()[0] + assert final_summary == "text from the other writer" + + stored = fx.raw_extra(item.id)["content_hash"] + # Never a hash of text the row does not hold: either it is + # consistent with the row's final summary, or it was left alone. + assert stored in ( + _content_hash(final_summary, "knowledge"), + _content_hash("orig", "knowledge"), + ) + assert stored != _content_hash("my new text", "knowledge") + + def test_a_declined_writeback_leaves_the_cache_agreeing_with_the_row(self, tmp_path): + """When the conditional write does not land, the cache must not claim it did. + + The writeback is a no-op if another writer moved summary/memory_type + between the read and the write. Stamping the refreshed hash into + repo.items anyway would leave recall serving a hash the row does not + hold - the same cache/DB divergence class as the delete failure paths. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + + sessions = repo._sessions + original_session = sessions.session + raced = [] + + def racing_session(): + session = original_session() + real_execute = session.execute + + def _execute(stmt, *args, **kwargs): + # Only the conditional hash write; memU's own update_item + # goes through exec/add/commit, so this lands in exactly the + # read -> write window. + if "json_set" in str(stmt) and not raced: + raced.append(True) + conn = sqlite3.connect(fx.path) + conn.execute( + "UPDATE memu_memory_items SET summary = ? WHERE id = ?", + ("text from the other writer", item.id), + ) + conn.commit() + conn.close() + return real_execute(stmt, *args, **kwargs) + + session.execute = _execute + return session + + sessions.session = racing_session + try: + result = repo.update_item(item_id=item.id, summary="my new text") + finally: + sessions.session = original_session + + assert raced == [True] + row_hash = fx.raw_extra(item.id)["content_hash"] + # Nothing was written, so the row keeps the hash it already had. + assert row_hash == _content_hash("orig", "knowledge") + # ... and neither the cache nor the returned item may disagree. + assert repo.items[item.id].extra["content_hash"] == row_hash + assert result.extra["content_hash"] == row_hash + + def test_a_concurrent_extra_key_survives_in_the_cache(self, tmp_path): + """The refreshed extra must come from the WRITE, not a pre-read snapshot. + + json_set merges into the row's current extra while the conditional write + binds only summary/memory_type, so a writer that changes any OTHER key in + the read -> write window commits and the write still lands. Rebuilding the + cache from the pre-write snapshot would silently revert that key in + memory while the row keeps it - the same cache/DB divergence class as the + declined-writeback case above, one window later. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + result, raced = _race_extra_in_the_write_window( + fx, repo, item.id, + lambda extra: extra.__setitem__("mentioned_at", "2026-08-03"), + ) + + assert raced == [True] + row_extra = fx.raw_extra(item.id) + # The write landed AND the concurrent key survived in the row. + assert row_extra["content_hash"] == _content_hash("my new text", "knowledge") + assert row_extra["mentioned_at"] == "2026-08-03" + # ... and neither the cache nor the returned item may disagree. + assert repo.items[item.id].extra == row_extra + assert result.extra == row_extra + + def test_a_concurrent_reinforce_is_not_reverted_in_the_cache(self, tmp_path): + """The same property for a field that is actually CONSUMED. + + Fix 3's salience ranking reads reinforcement_count and + last_reinforced_at off the cached item, so reverting them in memory + changes recall order, not just bookkeeping. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + fx.add_item(store, summary="orig") # exact-hash reinforce -> rc = 2 + repo = store.memory_item_repo + assert fx.raw_extra(item.id)["reinforcement_count"] == 2 + + def bump(extra): + extra["reinforcement_count"] = 9 + extra["last_reinforced_at"] = "2030-01-01T00:00:00+00:00" + + result, raced = _race_extra_in_the_write_window(fx, repo, item.id, bump) + + assert raced == [True] + row_extra = fx.raw_extra(item.id) + assert row_extra["reinforcement_count"] == 9 + assert row_extra["last_reinforced_at"] == "2030-01-01T00:00:00+00:00" + assert repo.items[item.id].extra == row_extra + assert result.extra == row_extra + + def test_a_decode_failure_after_the_cas_still_syncs_the_cache(self, tmp_path, caplog): + """A failure AFTER the CAS commits must not leave cache and row apart. + + Everything after the commit used to sit under the one best-effort + handler, so a decode failure returned early and left the cache and the + returned item on the OLD hash while the row already held the new one - + the same divergence class as the declined-writeback and concurrent-key + cases, entered from the other side. The reconstruction is sound on this + path only: the CAS bound summary/memory_type and set content_hash alone. + """ + caplog.set_level(logging.WARNING, logger="nerve.memory.memu_bridge") + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + hash_before = fx.raw_extra(item.id)["content_hash"] + want = _content_hash("my new text", "knowledge") + assert want != hash_before + decoded = [] + + def exploding_loads(payload, *args, **kwargs): + decoded.append(payload) + raise ValueError("injected decode failure") + + # Resolve through the MODULE, and patch the decode the refresh path + # actually calls. Measured: update_item performs exactly ONE + # json.loads, so a single injection cannot hit anything else. + import nerve.memory.memu_bridge as bridge_mod + + with patch.object(bridge_mod.json, "loads", exploding_loads): + result = repo.update_item(item_id=item.id, summary="my new text") + + # The injection really fired, on the RETURNING payload. + assert len(decoded) == 1 + assert "content_hash" in str(decoded[0]) + row_extra = fx.raw_extra(item.id) + # The CAS committed: the row carries the NEW hash. + assert row_extra["content_hash"] == want + # The call returned normally - this phase is best-effort. + assert result is not None + # CACHE_EQUALS_DB: neither the cache nor the returned item is stale. + assert repo.items[item.id].extra["content_hash"] == want + assert result.extra["content_hash"] == want + # The swallowed decode is visible, not silent. + warnings = [ + r for r in caplog.records + if r.levelno == logging.WARNING + and "could not be decoded" in r.getMessage() + ] + assert len(warnings) == 1 + assert item.id in warnings[0].getMessage() + + def test_a_failing_hash_writeback_does_not_fail_the_update(self, tmp_path, caplog): + """The hash refresh is derived work: it must never fail a landed update. + + It runs after memU's content write has committed, so letting an + exception escape would leave the content written while the caller is told + the update failed - and with the categories undiffed and the vector index + on the old embedding. A stale hash is base's unconditional behaviour, so + degrading to it is the strictly safer failure. + + The WARNING is the only thing separating "degrade" from "discard + silently", so it is asserted here rather than assumed. + """ + caplog.set_level(logging.WARNING, logger="nerve.memory.memu_bridge") + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + hash_before = fx.raw_extra(item.id)["content_hash"] + + sessions = repo._sessions + original_session = sessions.session + raised_on = [] + + def racing_session(): + session = original_session() + real_execute = session.execute + + def _execute(stmt, *args, **kwargs): + # Only the conditional hash write; memU's own update_item + # goes through exec/add/commit. + if "json_set" in str(stmt): + raised_on.append(True) + raise sqlite3.OperationalError("database is locked") + return real_execute(stmt, *args, **kwargs) + + session.execute = _execute + return session + + sessions.session = racing_session + try: + result = repo.update_item(item_id=item.id, summary="my new text") + finally: + sessions.session = original_session + + # The injection really fired, or the test proves nothing. + assert raised_on == [True] + row = sqlite3.connect(fx.path).execute( + "SELECT summary FROM memu_memory_items WHERE id = ?", (item.id,), + ).fetchone() + # The content write is intact and the call returned normally. + assert row[0] == "my new text" + assert result is not None + # The row keeps its old hash: base's behaviour, not a new state. + assert fx.raw_extra(item.id)["content_hash"] == hash_before + # The cache must not claim a refresh that did not happen. + assert repo.items[item.id].extra["content_hash"] == hash_before + # Swallowed is not silent: exactly one WARNING naming the item and + # the underlying error, so a persistently failing refresh is visible. + warnings = [ + r for r in caplog.records + if r.levelno == logging.WARNING + and "content_hash refresh skipped" in r.getMessage() + ] + assert len(warnings) == 1 + message = warnings[0].getMessage() + assert item.id in message + assert "database is locked" in message + + def test_a_failing_delegation_still_propagates(self, tmp_path): + """The delegation stays OUTSIDE the best-effort guard. + + Swallowing memU's own update failure would report success for an update + that never landed, which is the opposite error to the one above. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + memu_update = fx.Repo._nerve_memu_update_item + + def _boom(self, **kwargs): + raise sqlite3.OperationalError("database is locked") + + fx.Repo._nerve_memu_update_item = _boom + try: + MemUBridge._patch_sqlite_bugs() + with pytest.raises(sqlite3.OperationalError): + store.memory_item_repo.update_item( + item_id=item.id, summary="my new text", + ) + finally: + fx.Repo._nerve_memu_update_item = memu_update + MemUBridge._patch_sqlite_bugs() + + # Nothing was written by the delegation, so the row is untouched. + row = sqlite3.connect(fx.path).execute( + "SELECT summary FROM memu_memory_items WHERE id = ?", (item.id,), + ).fetchone() + assert row[0] == "orig" + assert fx.raw_extra(item.id)["content_hash"] == _content_hash( + "orig", "knowledge", + ) + + def test_item_without_a_hash_is_not_enrolled(self, tmp_path): + """Adding a hash where there was none is a behaviour change, not a fix. + + The cache and the returned item must not be enrolled either: memorize() + and recall read them, so a hash present only in memory is still a hash + the store never agreed to. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="t") + db = sqlite3.connect(fx.path) + db.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = ?", + (json.dumps({"reinforcement_count": 1}), item.id), + ) + db.commit() + db.close() + + restarted = fx.store().memory_item_repo + result = restarted.update_item(item_id=item.id, summary="t2") + + assert "content_hash" not in fx.raw_extra(item.id) + assert "content_hash" not in (result.extra or {}) + assert "content_hash" not in (restarted.items[item.id].extra or {}) + + def test_a_present_but_empty_hash_is_not_enrolled(self, tmp_path): + """The only input where the guard and the SQL predicate disagree. + + `json_extract(extra, '$.content_hash') IS NOT NULL` is TRUE for an empty + string, so the SQL alone would enroll a row whose hash is present but + blank. The Python guard is what keeps "no usable hash" out of hash-dedup, + and this is the case that proves it is load-bearing rather than a + duplicate of the SQL. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="t") + extra = fx.raw_extra(item.id) + extra["content_hash"] = "" + db = sqlite3.connect(fx.path) + db.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = ?", + (json.dumps(extra), item.id), + ) + db.commit() + db.close() + + restarted = fx.store().memory_item_repo + result = restarted.update_item(item_id=item.id, summary="t2") + + assert fx.raw_extra(item.id)["content_hash"] == "" + assert (result.extra or {}).get("content_hash") == "" + + def test_salience_fields_survive_the_extra_merge(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="t") + fx.add_item(store, summary="t") # exact-hash reinforce -> rc = 2 + before = fx.raw_extra(item.id) + assert before["reinforcement_count"] == 2 + + store.memory_item_repo.update_item(item_id=item.id, summary="t2") + + after = fx.raw_extra(item.id) + assert after["reinforcement_count"] == 2 + assert after["last_reinforced_at"] == before["last_reinforced_at"] + assert after["content_hash"] == _content_hash("t2", "knowledge") + # The cache and the returned item are read by memorize()/recall, so + # they must carry the SAME merged extra - not just the new hash. + assert store.memory_item_repo.items[item.id].extra == after + + def test_a_callers_own_extra_is_not_dropped(self, tmp_path): + """The refreshed hash must be MERGED into the caller's extra, not + substituted for it: update_item(summary=..., extra={"ref_id": ...}) must + keep ref_id, which list_items_by_ref_ids filters on. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="t") + + store.memory_item_repo.update_item( + item_id=item.id, summary="t2", extra={"ref_id": "abc"}, + ) + + after = fx.raw_extra(item.id) + assert after["ref_id"] == "abc" + assert after["content_hash"] == _content_hash("t2", "knowledge") + + def test_stale_hash_reinforces_the_corrected_row(self, tmp_path): + """The user-visible consequence, both directions in one test.""" + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="original text") + memu_update = fx.Repo._nerve_memu_update_item + + # Unpatched: correcting the wording, then re-memorizing the OLD + # text, reinforces the corrected row under its new summary. + memu_update(store.memory_item_repo, item_id=item.id, summary="corrected text") + again = fx.add_item(store, summary="original text") + assert again.id == item.id + assert fx.raw_count("SELECT count(*) FROM memu_memory_items") == 1 + assert again.summary == "corrected text" + + with _MemuPatchFixture(tmp_path / "b") as fx: + (tmp_path / "b").mkdir() + store, _ = fx.setup(1) + item = fx.add_item(store, summary="original text") + + # Patched: the same sequence recognises the old text as different. + store.memory_item_repo.update_item(item_id=item.id, summary="corrected text") + again = fx.add_item(store, summary="original text") + assert again.id != item.id + assert fx.raw_count("SELECT count(*) FROM memu_memory_items") == 2 + + def test_a_writer_between_the_two_phases_leaves_the_objects_mismatched( + self, tmp_path, + ): + """Pin the residual window this fix narrows but does not close. + + A writer landing strictly between the delegated update returning and the + wrapper's post-read session moves the row's summary, so the CAS binds - + and correctly writes - the hash of the THIRD PARTY's text. That hash is + then stamped onto the cache entry and the returned item without + re-reading either, so both carry a hash that does not belong to their own + .summary. + + Harmless, and deliberately not closed here: the only SQLite-path consumer + of extra.content_hash is create_item_reinforce, which reads the DB column + and never self.items, so dedup on the row's own current text still finds + the row (asserted below - it is the property a later round must not lose). + Base is worse: it stamps a hash of text the ROW does not hold, in every + arm including the no-writer control. Closing it means making memU's + content write and this refresh atomic, which was built and measured to + store a WRONG hash on two correctness paths. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="orig") + repo = store.memory_item_repo + + sessions = repo._sessions + original_session = sessions.session + opens = [] + injected = [] + + def counting_session(): + opens.append(1) + # Session 1 is memU's own content write; session 2 is the + # wrapper's post-read. Firing here is strictly between them. + if len(opens) == 2 and not injected: + conn = sqlite3.connect(fx.path) + conn.execute( + "UPDATE memu_memory_items SET summary = ? WHERE id = ?", + ("third party text", item.id), + ) + conn.commit() + injected.append( + conn.execute( + "SELECT summary FROM memu_memory_items WHERE id = ?", + (item.id,), + ).fetchone()[0] + ) + conn.close() + return original_session() + + sessions.session = counting_session + try: + result = repo.update_item(item_id=item.id, summary="my new text") + finally: + sessions.session = original_session + + # The writer really landed inside the window, or this proves nothing. + assert injected == ["third party text"] + + row_summary, row_type = sqlite3.connect(fx.path).execute( + "SELECT summary, memory_type FROM memu_memory_items WHERE id = ?", + (item.id,), + ).fetchone() + assert row_summary == "third party text" + row_hash = fx.raw_extra(item.id)["content_hash"] + # The ROW is self-consistent: the CAS bound its own summary/type. + assert row_hash == _content_hash(row_summary, str(row_type)) + + # The measured residue: the cache entry and the returned item carry + # the row's hash on top of the CALLER's summary. + cached = repo.items[item.id] + assert cached.summary == "my new text" + assert cached.extra["content_hash"] == row_hash + assert result.summary == "my new text" + assert result.extra["content_hash"] == row_hash + + # Why it is harmless: dedup reads the COLUMN, so the row's own + # current text still finds the row rather than duplicating it. + again = fx.add_item(store, summary=row_summary) + assert again.id == item.id + assert fx.raw_count("SELECT count(*) FROM memu_memory_items") == 1 + + +class TestReinforceWritebackReadsTheRow: + """Fix 7 correction: the semantic-dedup writeback must seed ``extra`` from + the ROW inside its transaction, not from the item cache. + + Read paths build MemoryItem without extra=, so a cache entry is {} in any + process that did not itself create the item; writing that back replaced the + row's whole extra with just the two salience keys. Measured before the fix: + all 6,473 rows with no content_hash carried reinforcement_count > 1, none + carried rc == 1, and the store holds 0 items of type "tool" (the only + create path that legitimately writes no hash). + """ + + @staticmethod + def _similar_pair(): + return ( + np.array([1.0, 0.0, 0.0], dtype=np.float32), + np.array([1.0, 0.001, 0.0], dtype=np.float32), + ) + + def _cold_cache_reinforce(self, fx, extra_writer=None): + first, second = self._similar_pair() + store, _ = fx.setup(1) + item = fx.add_item(store, summary="cold cache subject", embedding=first) + if extra_writer is not None: + extra_writer(item.id) + store.close() + + # Simulated restart: a fresh store whose cache is filled by a READ. + restarted = fx.store() + restarted.memory_item_repo.list_items() + assert restarted.memory_item_repo.items[item.id].extra == {} + + restarted.memory_item_repo.create_item_reinforce( + resource_id=None, memory_type="knowledge", + summary="cold cache subject!!", embedding=second, user_data={}, + ) + return item, restarted + + def test_content_hash_survives_a_cold_cache_reinforce(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + item, _ = self._cold_cache_reinforce(fx) + + after = fx.raw_extra(item.id) + assert "content_hash" in after + assert after["reinforcement_count"] == 2 + + def test_a_third_party_writers_key_survives(self, tmp_path): + """The arm that distinguishes this from hydrating the read paths. + + _resolve_event_dates_sync adds extra.mentioned_at with raw SQL and never + refreshes the cache, so hydrating reads cannot keep the cache canonical: + with read-path hydration installed, content_hash survived here but + mentioned_at was still destroyed. Reading the row inside the write + transaction is correct for every writer, present and future. + """ + with _MemuPatchFixture(tmp_path) as fx: + def write_mentioned_at(item_id): + db = sqlite3.connect(fx.path) + extra = fx.raw_extra(item_id) + extra["mentioned_at"] = "2026-01-01" + db.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = ?", + (json.dumps(extra), item_id), + ) + db.commit() + db.close() + + item, _ = self._cold_cache_reinforce(fx, write_mentioned_at) + + after = fx.raw_extra(item.id) + assert after["mentioned_at"] == "2026-01-01" + assert "content_hash" in after + + def test_a_key_written_inside_the_writeback_window_is_still_lost(self, tmp_path): + """Pin the window the sibling test does NOT reach. + + test_a_third_party_writers_key_survives writes BEFORE the reinforce, so + it exercises cold-cache seeding. The writeback itself is SELECT -> mutate + -> add -> commit with no CAS, so a writer landing between the SELECT and + the flush is overwritten. This is NOT this fix's creation: base does the + same read-modify-write, and seeding from the row shrinks the loss (base + loses every key, including content_hash) rather than introducing it. The + residual window is pinned here so a later round cannot mistake it for a + closed one, and a CAS for it is tracked separately. + """ + with _MemuPatchFixture(tmp_path) as fx: + first, second = self._similar_pair() + store, _ = fx.setup(1) + item = fx.add_item(store, summary="cold cache subject", embedding=first) + store.close() + + restarted = fx.store() + restarted.memory_item_repo.list_items() + repo = restarted.memory_item_repo + sessions = repo._sessions + original_session = sessions.session + fired = [] + + def spying_session(): + session = original_session() + real_add = session.add + + def add(obj): + # Fires after the writeback's SELECT and before its flush. + if not fired and getattr(obj, "id", None) == item.id: + db = sqlite3.connect(fx.path) + extra = fx.raw_extra(item.id) + extra["mentioned_at"] = "2026-01-01" + db.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = ?", + (json.dumps(extra), item.id), + ) + db.commit() + db.close() + # The race is real only if the key is in the DB already. + fired.append(fx.raw_extra(item.id).get("mentioned_at")) + return real_add(obj) + + session.add = add + return session + + sessions.session = spying_session + try: + repo.create_item_reinforce( + resource_id=None, memory_type="knowledge", + summary="cold cache subject!!", embedding=second, user_data={}, + ) + finally: + sessions.session = original_session + + # The write landed INSIDE the window, or this test proves nothing. + assert fired == ["2026-01-01"] + + after = fx.raw_extra(item.id) + # Measured outcome: the in-window key is lost (the writeback has no + # CAS). What the fix DOES guarantee still holds on this path. + assert "mentioned_at" not in after + assert "content_hash" in after + assert after["reinforcement_count"] == 2 + # And the cache does not claim otherwise. + assert repo.items[item.id].extra == after + + def test_cache_matches_the_row_afterwards(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + item, restarted = self._cold_cache_reinforce(fx) + + assert restarted.memory_item_repo.items[item.id].extra == fx.raw_extra(item.id) + + def test_reinforce_return_value_still_signals_rc_gt_1(self, tmp_path): + """memorize() skips category linking when the returned item's + extra.reinforcement_count > 1. The correction changes what that extra + holds, so pin the DECISION: if it flipped, a reinforced item would stop + getting its category links, i.e. a new orphan source. + """ + first, second = self._similar_pair() + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + fresh = fx.add_item(store, summary="subject", embedding=first) + assert fresh.extra.get("reinforcement_count", 1) == 1 + store.close() + + restarted = fx.store() + restarted.memory_item_repo.list_items() + reinforced = restarted.memory_item_repo.create_item_reinforce( + resource_id=None, memory_type="knowledge", summary="subject!!", + embedding=second, user_data={}, + ) + + assert reinforced.extra.get("reinforcement_count", 1) > 1 + + def test_read_paths_still_omit_extra(self, tmp_path): + """Pin the deliberate scope boundary: the read-path hydration gap is + DOCUMENTED, not fixed here. With the writeback reading the row, an + empty cached extra is no longer destructive, so hydration is a separate + change. If a future PR hydrates the reads, this test should be updated, + not silently kept passing by accident. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + item = fx.add_item(store, summary="t") + store.close() + + restarted = fx.store() + assert restarted.memory_item_repo.get_item(item.id).extra == {} + restarted.memory_item_repo.items.clear() + assert restarted.memory_item_repo.list_items()[item.id].extra == {} + + +class TestCascadeDeleteItem: + """Fix 10: delete_item must not leave the item's category relations behind. + + There is no FK and no ON DELETE CASCADE on memu_category_items, and no + layer owns the dependent rows: _patch_delete_memory_item reads the item's + categories only to build category_updates, then deletes the item row. + Measured before the fix: 6,455 dangling relations, every one of the 5,611 + distinct dangling item_ids present in the item_deleted audit log. They also + inflate memory_expand_category's reported total (which counts relations + while listing through a JOIN) by 3.5-4.4 percent on every category. + """ + + def test_relations_and_caches_are_removed(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store, summary="doomed") + fx.link_all(store, item.id, cats) + + store.memory_item_repo.delete_item(item.id) + + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 0 + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ?", item.id, + ) == 0 + assert item.id not in store.memory_item_repo.items + # DatabaseState.relations is read directly by memu's retrieve path. + assert [r for r in store.category_item_repo.relations + if r.item_id == item.id] == [] + + def test_unpatched_delete_orphans_the_relations(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + memu_delete = fx.Repo._nerve_memu_delete_item + + store, cats = fx.setup(2) + item = fx.add_item(store, summary="doomed") + fx.link_all(store, item.id, cats) + + memu_delete(store.memory_item_repo, item.id) + + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ?", item.id, + ) == 0 + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + + def test_missing_id_and_unlinked_item_are_noops(self, tmp_path): + with _MemuPatchFixture(tmp_path) as fx: + store, _ = fx.setup(1) + store.memory_item_repo.delete_item("no-such-id") + item = fx.add_item(store, summary="t") + store.memory_item_repo.delete_item(item.id) + assert fx.raw_count("SELECT count(*) FROM memu_memory_items") == 0 + + def test_a_failed_item_delete_rolls_the_relations_back(self, tmp_path): + """Atomicity. Deleting relations in their own transaction and then + delegating would leave a SURVIVING item stripped of its links on a + failure - strictly worse than the dangling rows it set out to fix. + SQLiteSessionManager.session() returns a fresh Session per call, so one + session for both deletes is the only way to get this. + + The cache assertions matter as much as the DB ones: Fix 5 serves + self.items unfiltered, so evicting before (or despite) a rolled-back + transaction makes the SURVIVING row invisible to recall. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store, summary="t") + fx.link_all(store, item.id, cats) + # A bystander keeps the cache non-empty so Fix 5 serves the CACHE: + # with an empty cache list_items() falls through to the DB and the + # cache assertion below would be free. + other = fx.add_item(store, summary="bystander") + store.category_item_repo.link_item_category( + other.id, next(iter(cats.values())), user_data={}, + ) + + class _Boom(Exception): + pass + + sessions = store.memory_item_repo._sessions + original_session = sessions.session + + def failing_session(): + session = original_session() + def _raise(_obj): + raise _Boom("forced") + session.delete = _raise + return session + + sessions.session = failing_session + try: + with pytest.raises(_Boom): + store.memory_item_repo.delete_item(item.id) + finally: + sessions.session = original_session + + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ?", item.id, + ) == 1 + # The caches must match the rolled-back DB, or the surviving row is + # unreachable through recall. + assert item.id in store.memory_item_repo.items + assert len([r for r in store.category_item_repo.relations + if r.item_id == item.id]) == 2 + assert item.id in store.memory_item_repo.list_items() + + def test_a_failed_relations_delete_rolls_the_item_back(self, tmp_path): + """The other direction, and the one that pins the ordering. + + session.delete (the ITEM row) is the first statement to raise in BOTH + the one-transaction form and an item-first split, so the sibling case + above cannot see a split. Failing only the RELATIONS delete can: under + a split the item is already committed and gone, leaving 2 dangling rows. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(2) + item = fx.add_item(store, summary="t") + fx.link_all(store, item.id, cats) + # As in the sibling case: a bystander keeps the cache non-empty so + # the list_items() assertion exercises Fix 5's cached path. It is + # created BEFORE the interceptor is installed, and `raised` below + # PROVES the extra item did not change which statement raises rather + # than assuming the statement-text predicate is unaffected. + other = fx.add_item(store, summary="bystander") + store.category_item_repo.link_item_category( + other.id, next(iter(cats.values())), user_data={}, + ) + + class _Boom(Exception): + pass + + sessions = store.memory_item_repo._sessions + original_session = sessions.session + raised = [] + + def failing_session(): + session = original_session() + real_exec = session.exec + + def _exec(stmt, *args, **kwargs): + text = str(stmt).lower().strip() + # Only the relations DELETE; everything else must really run + # or the row lookup fails and the test proves nothing. + if text.startswith("delete") and "category_items" in text: + raised.append(text) + raise _Boom("forced") + return real_exec(stmt, *args, **kwargs) + + session.exec = _exec + return session + + sessions.session = failing_session + try: + with pytest.raises(_Boom): + store.memory_item_repo.delete_item(item.id) + finally: + sessions.session = original_session + + # Exactly one statement raised, and it was the relations DELETE: + # the bystander did not move the failure point. + assert len(raised) == 1 + assert "category_items" in raised[0] + + # BOTH rolled back: the item survives with its links intact. + assert fx.raw_count( + "SELECT count(*) FROM memu_memory_items WHERE id = ?", item.id, + ) == 1 + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 2 + # ... and so must the caches, or recall cannot see the survivor. + assert item.id in store.memory_item_repo.items + assert len([r for r in store.category_item_repo.relations + if r.item_id == item.id]) == 2 + assert item.id in store.memory_item_repo.list_items() + + def test_cache_is_evicted_when_the_row_is_already_gone(self, tmp_path): + """The row can vanish underneath us (another process deleted it). + + Returning early on a missing row would keep the id in self.items, and + Fix 5 serves that cache unfiltered, so list_items() would go on + returning a deleted item while Fix 3 removed it from the vector index. + memU's own delete_item pops the cache unconditionally. + """ + with _MemuPatchFixture(tmp_path) as fx: + store, cats = fx.setup(1) + item = fx.add_item(store, summary="doomed") + fx.link_all(store, item.id, cats) + assert item.id in store.memory_item_repo.items + + # Another connection removes the row underneath us. + conn = sqlite3.connect(fx.path) + conn.execute("DELETE FROM memu_memory_items WHERE id = ?", (item.id,)) + conn.commit() + conn.close() + + store.memory_item_repo.delete_item(item.id) + + assert item.id not in store.memory_item_repo.items + assert [r for r in store.category_item_repo.relations + if r.item_id == item.id] == [] + # list_items() returns the cache dict itself (keyed by id). + assert item.id not in store.memory_item_repo.list_items() + # The DB too, not just the caches: an absent item row is exactly + # when its memu_category_items rows ARE the dangling rows this fix + # exists to prevent, so conditioning the relation delete on the row + # would leave the changelog contract half-kept. + assert fx.raw_count( + "SELECT count(*) FROM memu_category_items WHERE item_id = ?", item.id, + ) == 0 + + def test_fix_3_index_hook_stays_outermost(self, tmp_path): + """Fix 10 REPLACES the base implementation, so it must be installed + BEFORE Fix 3 (the opposite order from the wrapper-style fixes) or the + vector-index remove() hook would be buried and stop firing. + """ + with _MemuPatchFixture(tmp_path) as fx: + assert fx.Repo.delete_item.__qualname__.endswith("_indexed_delete_item") + + store, _ = fx.setup(1) + item = fx.add_item( + store, summary="t", embedding=np.array([1.0, 0.0], dtype=np.float32), + ) + removed = [] + + class _Index: + dirty = False + seen_items_len = 0 + + def remove(self, item_id): + removed.append(item_id) + + store.memory_item_repo._nerve_vec_index = _Index() + store.memory_item_repo.delete_item(item.id) + + assert removed == [item.id] + + +class TestWritePathPatchStructure: + """Structural tripwires for the assumptions the three fixes rest on.""" + + def test_patching_twice_does_not_stack(self, tmp_path): + def chain(fn): + names = [] + while fn is not None and hasattr(fn, "__closure__"): + names.append(fn.__qualname__.rsplit(".", 1)[-1]) + nxt = None + for cell in (fn.__closure__ or ()): + value = cell.cell_contents + if callable(value) and getattr(value, "__qualname__", "").endswith( + ("update_item", "delete_item"), + ): + nxt = value + break + fn = nxt + return names + + with _MemuPatchFixture(tmp_path) as fx: + first_update = chain(fx.Repo.update_item) + first_delete = chain(fx.Repo.delete_item) + handler = fx.CRUDMixin._patch_update_memory_item + + MemUBridge._patch_sqlite_bugs() + MemUBridge._patch_sqlite_bugs() + + assert chain(fx.Repo.update_item) == first_update + assert chain(fx.Repo.delete_item) == first_delete + assert fx.CRUDMixin._patch_update_memory_item is handler + + def test_the_handler_the_service_resolves_is_the_one_patched(self, tmp_path): + """Fix 8 patches a class attribute, and PipelineManager captures the + BOUND method during MemoryService.__init__ - which runs AFTER + _patch_sqlite_bugs() in _initialize_impl, so the patch is picked up. + """ + with _MemuPatchFixture(tmp_path): + from memu.app.crud import CRUDMixin + from memu.app.service import MemoryService + + assert (MemoryService._patch_update_memory_item + is CRUDMixin._patch_update_memory_item) + + def test_map_category_names_to_ids_copies_agree(self): + """_map_category_names_to_ids exists in BOTH CRUDMixin and + MemorizeMixin, and the MRO resolves MemorizeMixin's copy. Fix 8 calls it + late-bound via self. so it follows the MRO; this pins that the two + copies cannot silently diverge. + """ + import inspect + + import memu.app.service # noqa: F401 + from memu.app.crud import CRUDMixin + from memu.app.memorize import MemorizeMixin + from memu.app.service import MemoryService + + assert (MemoryService._map_category_names_to_ids + is MemorizeMixin._map_category_names_to_ids) + assert (inspect.getsource(CRUDMixin._map_category_names_to_ids) + == inspect.getsource(MemorizeMixin._map_category_names_to_ids)) + + def test_patch_mixin_duplicates_are_dead_code(self): + """memu.app.patch.PatchMixin holds byte-equivalent duplicates of both + workflow handlers. Nothing inherits or instantiates it, so it needs no + patch - assert that, so a future memU version wiring it up fails here. + """ + import memu.app.patch as patch_mod + from memu.app.service import MemoryService + + assert patch_mod.PatchMixin.__subclasses__() == [] + assert patch_mod.PatchMixin not in MemoryService.__mro__ + + def test_clear_items_remains_unreachable_from_nerve(self): + """clear_items also leaves dangling relations, but its only memU caller + is CRUDMixin.clear_memory, which nerve never calls. Fixing bulk-delete + semantics is a separate concern; this pins the exemption so it cannot + rot silently. + """ + from pathlib import Path + + package_root = Path(nerve_memory_bridge_file()).parents[1] + assert package_root.name == "nerve", package_root + hits = sorted( + str(path.relative_to(package_root)) + for path in package_root.rglob("*.py") + if "clear_memory" in path.read_text(encoding="utf-8", errors="ignore") + ) + assert hits == [] + + def test_salience_ranking_branch_is_unreachable_today(self): + """_fast_vector_search's salience branch reads extra from the cache. + It is not a live defect either way, because ranking defaults to + "similarity" and nerve never sets it - assert that rather than claiming + this PR fixes or regresses salience ranking. + """ + from memu.app.settings import RetrieveItemConfig + + assert RetrieveItemConfig().ranking == "similarity" + + +def nerve_memory_bridge_file(): + from nerve.memory import memu_bridge + + return memu_bridge.__file__