diff --git a/nerve/config.py b/nerve/config.py index feefae8f..b70b17cf 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -1419,7 +1419,19 @@ class MemoryCategoryConfig: @classmethod @_coerced def from_dict(cls, d: dict) -> MemoryCategoryConfig: - return cls(name=d["name"], description=d.get("description", "")) + # Strip at the origin: this name becomes a persistent category row and a + # lookup key, and memU strips before writing the row. Reject a blank or a + # non-string rather than coercing it: a bare ``name:`` in YAML parses as + # None, and str() would turn that typo into a category literally named + # "None". A non-str name has never worked (both nerve and memU call str + # methods on it), so rejecting it removes no reachable behaviour. + raw = d["name"] + if not isinstance(raw, str): + raise ValueError("memory category name must be a string") + name = raw.strip() + if not name: + raise ValueError("memory category name must not be blank") + return cls(name=name, description=d.get("description", "")) @dataclass diff --git a/nerve/gateway/routes/memory.py b/nerve/gateway/routes/memory.py index d38a3665..fe316c83 100644 --- a/nerve/gateway/routes/memory.py +++ b/nerve/gateway/routes/memory.py @@ -223,10 +223,15 @@ class CategoryUpdateRequest(BaseModel): async def create_memu_category(req: CategoryCreateRequest, user: dict = Depends(require_auth)): """Create a new memU memory category at runtime.""" bridge = _require_memu() - success = await bridge.create_category(req.name, req.description, source="web_ui") + # Checked here rather than on the model: a pydantic validator fails with 422 + # before the body runs, and this route reports bad input as 400. + name = req.name.strip() + if not name: + raise HTTPException(status_code=400, detail="Category name must not be blank") + success = await bridge.create_category(name, req.description, source="web_ui") if not success: raise HTTPException(status_code=500, detail="Failed to create category") - return {"name": req.name, "created": True} + return {"name": name, "created": True} @router.patch("/api/memory/memu/items/{item_id}") diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 9642175b..c81d1bbb 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -93,6 +93,25 @@ def _is_sqlite_locked_error(exc: BaseException) -> bool: _CATEGORY_BREADCRUMB_MAXLEN = 200 +def _norm_category_name(name: str) -> str: + """Normalize a category name to memU's lookup key: ``strip().lower()``. + + memU's contract is *strip at creation, strip+lowercase at lookup*: it strips + before writing a row, and all three consumer copies of the reverse lookup key + on ``name.strip().lower()`` (``memu/app/crud.py``, ``patch.py``, + ``memorize.py``). ``ctx.category_name_to_id`` is the only name-to-id channel + between nerve and memU, so every nerve site that writes a key into it must + derive that key the same way, or the id is unreachable. Keep ``.lower()``: + ``.casefold()`` is a *different* function (``'ss'`` vs ``'ß'``) and would + silently desync nerve from memU again. + + Because the map holds at most one id per normalized key, a *distinct* + category sharing a key is unrepresentable -- which is why creation resolves + against this key instead of the raw name. + """ + return name.strip().lower() + + def _category_breadcrumb(name: str, description: str, summary: str) -> str: """Build a short one-line breadcrumb for a category recall hit. @@ -1564,9 +1583,6 @@ async def _initialize_impl(self) -> bool: if not self.config.openai_api_key else {}), }, ) - self._available = True - self._metrics.service_available = True - self._metrics.initialized_at = datetime.now(timezone.utc).isoformat() logger.info("memU service initialized with SQLite at %s", sqlite_dsn) # Per-connection pragmas (busy_timeout, synchronous=NORMAL). @@ -1579,6 +1595,13 @@ async def _initialize_impl(self) -> bool: if is_bedrock: self._inject_bedrock_clients() + # Hydrate the category cache BEFORE seeding: _ensure_categories and + # the rebuild below both read repo.categories, which starts empty in + # a fresh process. Call for the side effect only -- the returned dict + # is a snapshot that get_or_create_category never updates, so the + # live repo.categories is the collection to read. + self._service.database.memory_category_repo.list_categories() + await self._ensure_categories() # Mark categories as ready so memU's _initialize_categories @@ -1595,7 +1618,7 @@ async def _initialize_impl(self) -> bool: ctx.category_name_to_id = {} for cat_id, cat in self._service.database.memory_category_repo.categories.items(): ctx.category_ids.append(cat.id) - ctx.category_name_to_id[cat.name.lower()] = cat.id + ctx.category_name_to_id[_norm_category_name(cat.name)] = cat.id if ctx.category_name_to_id: logger.info( "Populated category mapping: %d categories", @@ -1851,6 +1874,12 @@ def _numpy_create_item(self, *args, **kwargs): _numpy_create_item._nerve_numpy_wrapped = True # type: ignore[attr-defined] SQLiteMemoryItemRepo.create_item = _numpy_create_item + # Publish success state only on success: an exception above must not + # leave a half-initialized bridge reporting itself available, since + # the primary caller ignores this return value. + self._available = True + self._metrics.service_available = True + self._metrics.initialized_at = datetime.now(timezone.utc).isoformat() return True except ImportError: @@ -1865,17 +1894,11 @@ async def _ensure_categories(self) -> None: if not self._service or not self.config.memory.categories: return - existing: set[str] = set() - try: - cats = self._service.database.memory_category_repo.list_categories() - for _, cat in cats.items(): - existing.add(getattr(cat, "name", "")) - except Exception: - pass - + # No pre-filter: the impl resolves create-vs-reuse itself, on the + # normalized name. An exact-match skip here shadow-creates a second row + # for any case or whitespace variant, and skips the config registration + # the variant needs. Reuse is cheap (it returns before embedding). for cat_cfg in self.config.memory.categories: - if cat_cfg.name in existing: - continue try: # Already on the memU loop (called from _initialize_impl) — # invoke the impl directly instead of re-submitting. @@ -3008,11 +3031,99 @@ async def create_category(self, name: str, description: str, source: str = "brid return False return await self._submit(self._create_category_impl(name, description, source)) + def _find_category_by_norm_name(self, name: str) -> Any | None: + """Return the cached category whose normalized name matches, else None. + + ``get_or_create_category`` filters on the *exact* name and the table has + no unique index on it, so it cannot do this itself. + + Tie-break: on a store that ALREADY holds several rows sharing one + normalized key (written before this fix, or by another writer), prefer the + row ``ctx.category_name_to_id`` currently points at. The rebuild above + assigns that key row-by-row, so the LAST matching row wins there; a plain + first-match scan here would disagree, and registering its answer would + silently flip the mapping to a different row -- orphaning every item + linked to the one that just lost the key. Resolution must therefore agree + with the rebuild rather than race it. Do not "simplify" this back to a + bare scan. The scan remains the fallback for the two cases with no + mapping yet: a fresh process before the rebuild, and seeding. + """ + key = _norm_category_name(name) + cats = self._service.database.memory_category_repo.categories + ctx = self._service._get_context() + mapped_id = (getattr(ctx, "category_name_to_id", None) or {}).get(key) + if mapped_id is not None: + mapped = cats.get(mapped_id) + # The name re-check only ever narrows: all three writers key on the + # row's own normalized name, so a live mapping always agrees. It costs + # nothing and stops a hypothetically stale entry from redirecting a + # create onto an unrelated row, which would be worse than the flip. + if mapped is not None and _norm_category_name( + getattr(mapped, "name", "") + ) == key: + return mapped + for cat in cats.values(): + if _norm_category_name(getattr(cat, "name", "")) == key: + return cat + return None + + def _register_category(self, cat: Any, description: str) -> None: + """Point nerve's name-keyed channels at ``cat``, keyed on its STORED name. + + memU reads ``category_config_map`` by the stored row's name + (``memorize.py``), while ``MemoryService.__init__`` pre-keys it by the + *config* names -- so a row stored under a different spelling is never a key + unless registered here. + + ``category_configs`` (the LLM prompt's offered set) is seeded only from + ``config.memory.categories`` at ``MemoryService`` construction, and the + create path below appends to it precisely "so new memorizations can assign + to this category". A reuse returns before reaching that append, so a + persisted row with no configured counterpart -- exactly the web-UI case -- + would get both maps and still never be offered to the LLM, making it + unassignable. Registering it here restores the repair the unconditional + base create always performed. The membership test is on the NORMALIZED + name, not the exact one: a configured ``procedures`` and a stored + ``PROCEDURES`` are one category, and appending both would advertise it + twice. + """ + from memu.app.service import CategoryConfig + cfg = CategoryConfig(name=cat.name, description=description) + self._service.category_config_map[cat.name] = cfg + key = _norm_category_name(cat.name) + if not any( + _norm_category_name(getattr(existing, "name", "")) == key + for existing in self._service.category_configs + ): + self._service.category_configs.append(cfg) + self._service._category_prompt_str = self._service._format_categories_for_prompt( + self._service.category_configs + ) + ctx = self._service._get_context() + if getattr(ctx, "category_name_to_id", None) is not None: + ctx.category_name_to_id[key] = cat.id + async def _create_category_impl(self, name: str, description: str, source: str = "bridge") -> bool: """Create a category — repo write + context mutation (memU loop).""" if not self._service: return False try: + name = name.strip() + + # Resolve BEFORE embedding, not after: seeding calls this once per + # configured category on every start, so resolving below the embed + # would cost one embedding API call per category per restart -- the + # cost categories_ready=True exists to avoid. Returning here is also + # what keeps the create-only side effects below untouched. + existing = self._find_category_by_norm_name(name) + if existing is not None: + if existing.name != name: + logger.info( + "Reusing category %r for requested name %r", existing.name, name, + ) + self._register_category(existing, description) + return True + # Generate embedding for the category (requires OpenAI key) embedding = None if self._has_embeddings: @@ -3023,6 +3134,14 @@ async def _create_category_impl(self, name: str, description: str, source: str = except Exception as e: logger.warning("Could not embed category %s: %s", name, e) + # Re-check after the embed's await: the memU loop runs each create as + # its own coroutine with no lock, so a concurrent create of the same + # normalized name can have inserted the row while we were suspended. + existing = self._find_category_by_norm_name(name) + if existing is not None: + self._register_category(existing, description) + return True + # Create in the DB repo cat = self._service.database.memory_category_repo.get_or_create_category( name=name, description=description, embedding=embedding, user_data={}, @@ -3030,9 +3149,9 @@ async def _create_category_impl(self, name: str, description: str, source: str = # Update in-memory config so new memorizations can assign to this category from memu.app.service import CategoryConfig - cfg = CategoryConfig(name=name, description=description) + cfg = CategoryConfig(name=cat.name, description=description) self._service.category_configs.append(cfg) - self._service.category_config_map[name] = cfg + self._service.category_config_map[cat.name] = cfg self._service._category_prompt_str = self._service._format_categories_for_prompt( self._service.category_configs ) @@ -3042,7 +3161,7 @@ async def _create_category_impl(self, name: str, description: str, source: str = if hasattr(ctx, 'category_ids') and ctx.category_ids is not None: ctx.category_ids.append(cat.id) if hasattr(ctx, 'category_name_to_id') and ctx.category_name_to_id is not None: - ctx.category_name_to_id[name.lower()] = cat.id + ctx.category_name_to_id[_norm_category_name(cat.name)] = cat.id logger.info("Created category: %s", name) await self._audit("category_created", "category", name, source, {"description": description}) diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 92037666..d7a89de2 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -2,9 +2,11 @@ import asyncio import json +import logging import sqlite3 from datetime import datetime, timedelta, timezone from pathlib import Path +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -1134,3 +1136,988 @@ 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 + + +# --------------------------------------------------------------------------- +# memU category-name normalization +# +# ctx.category_name_to_id is the only name-to-id channel between nerve and memU. +# memU's contract is strip at creation, strip+lowercase at lookup, and all three +# of its consumer copies key on name.strip().lower(). nerve substitutes its own +# rebuild and creation paths, so these tests pin that its keys agree, and that at +# most one category exists per normalized key. +# --------------------------------------------------------------------------- + + +_CATEGORY_MODELS_CACHE = {} + + +def _category_models(): + """Build the SQLAlchemy models ONCE per process. + + nerve's Fix 6 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 is what lets these tests use + several isolated stores, and reopen one file to simulate a restart. + """ + if "models" in _CATEGORY_MODELS_CACHE: + return _CATEGORY_MODELS_CACHE["models"] + # memu has an import-order circularity (database/__init__ -> factory -> + # app/__init__ -> service -> factory), so app.service must come first. + import memu.app.service # noqa: F401 + import memu.database.sqlite.schema as schema_mod + + # Building the models needs the patches applied (Fix 6 renames memu's own + # "sqlite_*" tables, which SQLite reserves). Resolve through the MODULE: + # Fix 6 replaces this attribute. + MemUBridge._patch_sqlite_bugs() + models = schema_mod.get_sqlite_sqlalchemy_models(scope_model=None) + _CATEGORY_MODELS_CACHE["models"] = models + return models + + +class _CategoryCtx: + """Stand-in for memU's workflow context (only the category fields).""" + + def __init__(self): + self.category_ids = [] + self.category_name_to_id = {} + self.categories_ready = False + + +class _EmbedSpy: + """Embedding client that counts calls. + + The resolution's PLACEMENT is load-bearing, not cosmetic: seeding invokes the + create path once per configured category on every start, so a check placed + below the embed costs one embedding API call per category per restart. + """ + + def __init__(self): + self.calls = 0 + self.gate = None + + async def embed(self, texts): + self.calls += 1 + if self.gate is not None: + await self.gate.wait() + return [[0.0, 1.0] for _ in texts] + + +class _CategoryFixture: + """Isolated memU stores over one temp file, with the patches applied. + + The SQLiteMemoryItemRepo methods are restored on exit; the other globals + _patch_sqlite_bugs() installs are not, so do not rely on full isolation. + """ + + def __init__(self, tmp_path): + # Snapshot before _category_models(), which patches too -- a snapshot + # taken in __enter__ would already hold the patched methods. + import memu.app.service # noqa: F401 - import-order circularity + from memu.database.sqlite.repositories.memory_item_repo import ( + SQLiteMemoryItemRepo, + ) + + self._repo = SQLiteMemoryItemRepo + self._saved = { + n: SQLiteMemoryItemRepo.__dict__.get(n) + for n in ("update_item", "delete_item", "clear_items", "list_items", + "create_item", "create_item_reinforce", "vector_search_items") + } + self.models = _category_models() + self.path = str(tmp_path / "memu.sqlite") + self._stores = [] + + def __enter__(self): + MemUBridge._patch_sqlite_bugs() + return self + + def __exit__(self, *exc): + for store in self._stores: + try: + store.close() + except Exception: + pass + for name, fn in self._saved.items(): + if fn is None: + if name in self._repo.__dict__: + delattr(self._repo, name) + else: + setattr(self._repo, name, fn) + 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 write_row(self, store, name, description="from web UI"): + """Insert a row directly through the repo, bypassing nerve. + + Simulates a row created before this fix, or by another writer. + """ + return store.memory_category_repo.get_or_create_category( + name=name, description=description, embedding=None, user_data={}, + ) + + def bridge(self, store, config_names=(), embeddings=True): + """A MemUBridge whose _service is a stub carrying the real collaborators. + + Only what the category paths touch is stubbed; the code under test is + nerve's own. + """ + from memu.app.service import CategoryConfig + + from nerve.config import MemoryCategoryConfig + + bridge = MemUBridge.__new__(MemUBridge) + config = NerveConfig() + config.memory = MemoryConfig( + sqlite_dsn=f"sqlite:///{self.path}", + categories=[ + MemoryCategoryConfig(name=n, description=f"desc {n}") + for n in config_names + ], + ) + bridge.config = config + bridge._audit_db = None + bridge._main_loop = None + bridge._available = False + + spy = _EmbedSpy() + ctx = _CategoryCtx() + service = SimpleNamespace( + database=store, + _context=ctx, + _get_context=lambda: ctx, + category_configs=[ + CategoryConfig(name=n, description=f"desc {n}") for n in config_names + ], + category_config_map={ + n: CategoryConfig(name=n, description=f"desc {n}") for n in config_names + }, + _category_prompt_str="", + _format_categories_for_prompt=lambda cats: " | ".join(c.name for c in cats), + _get_llm_client=lambda _kind: spy, + ) + bridge._service = service + # _has_embeddings is a property on the class; category_fixture replaces it + # with one reading this flag, so each test chooses whether the embedding + # branch runs. + bridge._category_test_embeddings = embeddings + return bridge, service, ctx, spy + + +@pytest.fixture +def category_fixture(tmp_path, monkeypatch): + """A _CategoryFixture with _has_embeddings driven by the stub's flag.""" + monkeypatch.setattr( + MemUBridge, + "_has_embeddings", + property(lambda self: getattr(self, "_category_test_embeddings", False)), + ) + with _CategoryFixture(tmp_path) as fx: + yield fx + + +def _rebuild_map(bridge, ctx): + """Rebuild the name-to-id map the way _initialize_impl does. + + The key is spelled out here rather than imported from the bridge, so this + helper works unchanged against a tree that lacks the fix: a helper that + ImportErrors would make every arm fail in its harness, which discriminates + nothing. This is the memU consumer contract these tests hold nerve to. + """ + ctx.category_ids = [] + ctx.category_name_to_id = {} + for _cat_id, cat in bridge._service.database.memory_category_repo.categories.items(): + ctx.category_ids.append(cat.id) + ctx.category_name_to_id[cat.name.strip().lower()] = cat.id + + +def _run_real_rebuild_block(bridge, ctx): + """Execute the REAL rebuild block, extracted from _initialize_impl's source. + + _rebuild_map above is an independent oracle; this drives nerve's own key + derivation so the rebuild arm cannot pass by re-implementing the fix. Running + the whole of _initialize_impl would need the entire memU service constructed, + which is what makes extracting this block the cheaper honest option. It is + keyed on structure, not line numbers, and asserts its own match. + """ + import inspect + import re + import sys + import textwrap + + src = inspect.getsource(type(bridge)._initialize_impl) + match = re.search( + r"^(\s*)ctx\.category_ids = \[\]\n(.*?)\n(?=\1if ctx\.category_name_to_id:)", + src, + re.S | re.M, + ) + assert match, "could not locate the rebuild block in _initialize_impl" + block = textwrap.dedent(match.group(1) + "ctx.category_ids = []\n" + match.group(2)) + assert "category_name_to_id[" in block, f"extracted the wrong block: {block!r}" + namespace = dict(vars(sys.modules[type(bridge).__module__])) + namespace.update({"ctx": ctx, "self": bridge}) + exec(compile(block, "", "exec"), namespace) + + +def _unreachable_ids(store, ctx): + """Category ids no name-based lookup can reach -- the damage this fix prevents.""" + live = {c.id for c in store.memory_category_repo.categories.values()} + return live - set(ctx.category_name_to_id.values()) + + +class TestCategoryNameNormalization: + """nerve's name-to-id keys must match memU's strip().lower() contract.""" + + def test_fixture_restores_the_item_repo_on_exit(self, tmp_path): + """The fixture must not leak its SQLiteMemoryItemRepo patches to its neighbours. + + TestIndexedUpdateItemForwarding introspects update_item's signature, so a + leaked wrapper turns it red from ~200 lines away. Order-independent: the + leak is observable within one test body. + """ + import inspect + + import memu.app.service # noqa: F401 - import-order circularity + from memu.database.sqlite.repositories.memory_item_repo import ( + SQLiteMemoryItemRepo as Repo, + ) + + names = ( + "update_item", "delete_item", "clear_items", "list_items", + "create_item", "create_item_reinforce", "vector_search_items", + ) + + def methods(): + return {n: Repo.__dict__.get(n) for n in names} + + def keyword_only(): + param = inspect.signature(Repo.update_item).parameters.get("item_id") + return param is not None and param.kind is inspect.Parameter.KEYWORD_ONLY + + before = methods() + assert keyword_only(), "precondition: update_item unpatched on entry" + + with _CategoryFixture(tmp_path) as fx: + fx.store() + assert "item_id" not in inspect.signature(Repo.update_item).parameters, ( + "the patch is meant to be applied inside the fixture" + ) + + # Every SQLiteMemoryItemRepo name the patch reassigns, not just the one + # the neighbour reads: a restore list narrowed to update_item must fail here. + leaked = sorted(n for n, fn in methods().items() if fn is not before[n]) + assert not leaked, f"_CategoryFixture leaked on exit: {leaked}" + assert keyword_only(), "_CategoryFixture leaked _patch_sqlite_bugs() on exit" + + @pytest.mark.asyncio + async def test_padded_category_name_is_reachable_by_its_stripped_key( + self, category_fixture + ): + """A padded name is unreachable at base with NO collision anywhere. + + The worse of the two shapes, and invisible to a collision query: one row, + one map key, and every consumer's stripped lookup misses it. + """ + store = category_fixture.store() + bridge, _svc, ctx, _spy = category_fixture.bridge(store) + + assert await bridge._create_category_impl(" procedures ", "d") is True + _rebuild_map(bridge, ctx) + + cats = list(store.memory_category_repo.categories.values()) + assert [c.name for c in cats] == ["procedures"] + assert ctx.category_name_to_id["procedures"] == cats[0].id + assert _unreachable_ids(store, ctx) == set() + + @pytest.mark.asyncio + async def test_case_variant_creation_reuses_the_existing_row(self, category_fixture): + """Two spellings must not become two rows sharing one map key.""" + store = category_fixture.store() + bridge, _svc, ctx, _spy = category_fixture.bridge(store) + + assert await bridge._create_category_impl("procedures", "FIRST") is True + assert await bridge._create_category_impl("PROCEDURES", "SECOND") is True + _rebuild_map(bridge, ctx) + + cats = list(store.memory_category_repo.categories.values()) + assert len(cats) == 1 + assert len(ctx.category_name_to_id) == 1 + assert _unreachable_ids(store, ctx) == set() + + @pytest.mark.asyncio + async def test_exact_repeat_creation_is_still_idempotent(self, category_fixture): + """Control: the restart seed path depends on the exact repeat staying cheap.""" + store = category_fixture.store() + bridge, _svc, _ctx, _spy = category_fixture.bridge(store) + + assert await bridge._create_category_impl("procedures", "d") is True + assert await bridge._create_category_impl("procedures", "d") is True + + assert len(store.memory_category_repo.categories) == 1 + + def test_rebuild_uses_the_normalized_key(self, category_fixture): + """Pins the rebuild key independently of the creation path. + + Without this arm the creation fix alone makes the suite green while the + rebuild key stays unguarded -- and the rebuild is what maps rows written + before this fix, or by any other writer. + """ + store = category_fixture.store() + row = category_fixture.write_row(store, " Procedures ") + bridge, _svc, ctx, _spy = category_fixture.bridge(store) + + _run_real_rebuild_block(bridge, ctx) + + assert ctx.category_name_to_id["procedures"] == row.id + assert _unreachable_ids(store, ctx) == set() + + def test_the_key_is_lower_not_casefold(self, category_fixture): + """memU keys on ``.lower()``, so nerve must too -- they are not the same. + + Every other arm here uses an ASCII name, where ``lower()`` and + ``casefold()`` are byte-identical, so none of them can tell the two apart: + a mutant switching to ``casefold()`` passes the whole suite. 'Straße' is + the cheapest name where they diverge ('straße' vs 'strasse'), and the + lookup asserted is exactly what memU's three consumers compute, so this + arm fails the moment nerve's key stops agreeing with theirs. + """ + name = "Straße" + assert name.lower() != name.casefold(), "the fixture name must discriminate" + + store = category_fixture.store() + row = category_fixture.write_row(store, f" {name} ") + bridge, _svc, ctx, _spy = category_fixture.bridge(store) + + _run_real_rebuild_block(bridge, ctx) + + # memU's own lookup key, spelled out rather than imported from the bridge. + assert ctx.category_name_to_id[f" {name} ".strip().lower()] == row.id + assert _unreachable_ids(store, ctx) == set() + + @pytest.mark.asyncio + async def test_reuse_does_NOT_call_the_embedding_client(self, category_fixture): + """The resolution must sit ABOVE the embed, not below it. + + A resolution placed after the embed satisfies every other arm here; only + this one distinguishes it. Below the embed, seeding would cost one + embedding API call per configured category on every restart. + """ + store = category_fixture.store() + bridge, _svc, _ctx, spy = category_fixture.bridge(store, embeddings=True) + + assert await bridge._create_category_impl("procedures", "d") is True + assert spy.calls == 1, "a genuinely new category must still embed" + + assert await bridge._create_category_impl("PROCEDURES", "d") is True + assert spy.calls == 1, "a reuse must not embed" + + @pytest.mark.asyncio + async def test_two_concurrent_creates_of_the_same_name_make_ONE_row( + self, category_fixture + ): + """The post-embed recheck, and this is its only guard. + + The memU loop runs each create as its own coroutine with no lock, so + without a recheck after the embed's await both creates pass the first + existence check and insert -- reintroducing the very collision this change + removes, through the fix. + """ + store = category_fixture.store() + bridge, _svc, ctx, spy = category_fixture.bridge(store, embeddings=True) + spy.gate = asyncio.Event() + + first = asyncio.create_task(bridge._create_category_impl("procedures", "d")) + second = asyncio.create_task(bridge._create_category_impl("PROCEDURES", "d")) + # Both must be suspended inside embed() before either may insert. + while spy.calls < 2: + await asyncio.sleep(0) + spy.gate.set() + assert await first is True + assert await second is True + + _rebuild_map(bridge, ctx) + assert len(store.memory_category_repo.categories) == 1 + assert len(ctx.category_name_to_id) == 1 + assert _unreachable_ids(store, ctx) == set() + + @pytest.mark.asyncio + async def test_reuse_does_not_duplicate_the_prompt_or_the_id_or_audit_a_create( + self, category_fixture + ): + """The early return keeps the create-only side effects untouched.""" + store = category_fixture.store() + bridge, service, ctx, _spy = category_fixture.bridge(store) + # The bridge mutates the ctx the fixture handed back (_get_context returns + # this object), so an id appended by the reuse path is observable here. + assert bridge._service._get_context() is ctx + audited = [] + bridge._audit = AsyncMock(side_effect=lambda *a, **k: audited.append(a)) + + await bridge._create_category_impl("procedures", "d") + configs_after_create = len(service.category_configs) + prompt_after_create = service._category_prompt_str + ids_after_create = list(ctx.category_ids) + assert ids_after_create, "the create arm must have appended its id" + audited.clear() + + await bridge._create_category_impl("PROCEDURES", "d") + + # BEFORE any rebuild: _rebuild_map resets category_ids and refills it from + # the single-row DB, which erases a spurious append before it can be seen. + # This is the only oracle for the "a reuse never re-appends an id" claim. + assert list(ctx.category_ids) == ids_after_create, ( + "a reuse must not append an id" + ) + assert len(service.category_configs) == configs_after_create + assert service._category_prompt_str == prompt_after_create + assert audited == [], "a reuse must not audit a create that did not happen" + _rebuild_map(bridge, ctx) + assert len(ctx.category_ids) == len(set(ctx.category_ids)) + + @pytest.mark.asyncio + async def test_a_create_on_an_ALREADY_COLLIDED_store_does_not_flip_the_mapping( + self, category_fixture + ): + """Resolution must pick the row the rebuild mapped, not merely the first. + + No other arm here writes two rows before acting, so a pre-collided store -- + the state this change explicitly promises to leave alone -- is otherwise + untested. The rebuild assigns the shared key row-by-row, so the LAST match + wins; a first-match scan answers the OTHER row, and registering that answer + repoints the key without touching either row, orphaning everything linked + to the row that just lost it. + """ + seeded = category_fixture.store() + first = category_fixture.write_row(seeded, "procedures", "FIRST") + second = category_fixture.write_row(seeded, "PROCEDURES", "SECOND") + assert first.id != second.id, "the fixture must produce two distinct rows" + + restarted = category_fixture.store() + bridge, _svc, ctx, _spy = category_fixture.bridge(restarted) + restarted.memory_category_repo.list_categories() + _run_real_rebuild_block(bridge, ctx) + + # One key for two rows is the pre-existing damage; the arm is about what a + # later create does to it, so pin the starting point rather than assume it. + assert len(ctx.category_name_to_id) == 1 + snapshot = dict(ctx.category_name_to_id) + rows_before = {c.id: c.name for c in restarted.memory_category_repo.categories.values()} + assert len(rows_before) == 2 + + assert await bridge._create_category_impl("Procedures", "d") is True + + assert dict(ctx.category_name_to_id) == snapshot, ( + "a create must not repoint an already-collided key at the other row" + ) + # Not by merging, renaming or deleting either row -- the no-migration + # promise -- and not by inserting a third one, which is the base behaviour + # this change already fixes and which must stay fixed. + assert { + c.id: c.name for c in restarted.memory_category_repo.categories.values() + } == rows_before + + +class TestEnsureCategoriesNormalization: + """Seeding must adopt a normalized-equal row, never shadow-create one.""" + + @pytest.mark.asyncio + async def test_ensure_categories_does_not_shadow_a_case_variant_row( + self, category_fixture + ): + """The amplifier: a one-off UI typo orphans a category on the next start. + + At base the exact-match skip misses 'PROCEDURES' for config 'procedures', + so seeding creates a second row and the rebuild orphans one of them. + """ + seeded = category_fixture.store() + row = category_fixture.write_row(seeded, "PROCEDURES") + + restarted = category_fixture.store() + bridge, _svc, ctx, _spy = category_fixture.bridge( + restarted, config_names=("procedures",) + ) + restarted.memory_category_repo.list_categories() + await bridge._ensure_categories() + _rebuild_map(bridge, ctx) + + cats = list(restarted.memory_category_repo.categories.values()) + assert [c.id for c in cats] == [row.id] + assert _unreachable_ids(restarted, ctx) == set() + + @pytest.mark.asyncio + async def test_config_map_is_registered_for_a_REUSED_row(self, category_fixture): + """memU reads category_config_map by the STORED row's name. + + MemoryService pre-keys that map by the *config* names, so a row stored + under another spelling is never a key unless the reuse path adds it -- + which is why this arm must go through reuse, and must assert the lookup + by the stored name rather than by the requested one. + """ + seeded = category_fixture.store() + category_fixture.write_row(seeded, "PROCEDURES") + + restarted = category_fixture.store() + bridge, service, _ctx, _spy = category_fixture.bridge( + restarted, config_names=("procedures",) + ) + restarted.memory_category_repo.list_categories() + await bridge._ensure_categories() + + assert service.category_config_map.get("PROCEDURES") is not None + + @pytest.mark.asyncio + async def test_a_reused_UNCONFIGURED_row_becomes_assignable_in_the_prompt( + self, category_fixture + ): + """A reuse must still register the row in the LLM's offered set. + + ``category_configs`` is seeded only from config at service construction + and is never rebuilt from the DB, and the create path's own append exists + "so new memorizations can assign to this category". The reuse branch returns + before it, so a persisted row with no configured counterpart -- the web-UI + case -- got both name maps and was still never offered to the LLM, i.e. + nothing could ever be filed under it. The base create was unconditional and + did register it, so the early return removed that repair. + """ + seeded = category_fixture.store() + row = category_fixture.write_row(seeded, "PROCEDURES") + + restarted = category_fixture.store() + # No config_names: the row exists in the DB and in NO config entry. + bridge, service, _ctx, _spy = category_fixture.bridge(restarted) + restarted.memory_category_repo.list_categories() + assert service.category_configs == [], "the arm needs an unconfigured start" + + assert await bridge._create_category_impl("procedures", "d") is True + + assert [c.name for c in service.category_configs] == [row.name], ( + "a reused row must be offered to the LLM exactly once, under its " + "STORED name" + ) + assert row.name in service._category_prompt_str + # Still a reuse, not a create: one row, no second spelling inserted. + assert len(restarted.memory_category_repo.categories) == 1 + + @pytest.mark.asyncio + async def test_prompt_registration_is_stable_across_repeated_seeding( + self, category_fixture + ): + """Idempotence must hold across a restart, not merely within one call. + + The arm above says "register it"; this one says "and never twice". Both are + needed: an unconditional append satisfies the first and grows the offered + set once per restart, which is the duplicate-in-prompt shape a separately + filed defect already covers and which this change must not create. + """ + seeded = category_fixture.store() + category_fixture.write_row(seeded, "PROCEDURES") + + restarted = category_fixture.store() + # Configured under a DIFFERENT spelling than the stored row, so exact-name + # membership would not find it and only a normalized test can. + bridge, service, _ctx, _spy = category_fixture.bridge( + restarted, config_names=("procedures",) + ) + restarted.memory_category_repo.list_categories() + before = [c.name for c in service.category_configs] + assert before == ["procedures"] + + for _ in range(3): + await bridge._ensure_categories() + + assert [c.name for c in service.category_configs] == before, ( + "a normalized-equal config entry already offers this category" + ) + + @pytest.mark.asyncio + async def test_repeated_ensure_categories_makes_no_embedding_calls( + self, category_fixture + ): + """The cost property categories_ready=True exists to protect. + + Seeding now invokes the create path unconditionally, so reuse must stay + free of embedding calls or every restart pays for the whole config. + """ + store = category_fixture.store() + bridge, _svc, _ctx, spy = category_fixture.bridge( + store, config_names=("procedures", "patterns"), embeddings=True + ) + store.memory_category_repo.list_categories() + + await bridge._ensure_categories() + first_run = spy.calls + assert first_run == 2, "the first seed embeds each new category once" + + await bridge._ensure_categories() + assert spy.calls == first_run, "a second seed must make no embedding calls" + assert len(store.memory_category_repo.categories) == 2 + + @pytest.mark.asyncio + async def test_ensure_categories_is_quiet_and_stable_across_repeated_runs( + self, category_fixture, caplog + ): + """Three runs over a case-variant row: no growth, no warning, no orphan.""" + seeded = category_fixture.store() + category_fixture.write_row(seeded, "PROCEDURES") + + restarted = category_fixture.store() + bridge, _svc, ctx, _spy = category_fixture.bridge( + restarted, config_names=("procedures",) + ) + restarted.memory_category_repo.list_categories() + + with caplog.at_level(logging.WARNING, logger="nerve.memory.memu_bridge"): + for _ in range(3): + await bridge._ensure_categories() + + _rebuild_map(bridge, ctx) + assert len(restarted.memory_category_repo.categories) == 1 + assert _unreachable_ids(restarted, ctx) == set() + assert [r.message for r in caplog.records] == [] + + +# --------------------------------------------------------------------------- +# The production initialization path +# +# The arms above drive _create_category_impl / _ensure_categories with a stubbed +# _service, so they never observe _initialize_impl's own ORDERING. Two of its +# properties exist only there: +# +# * the category cache is hydrated BEFORE seeding (a fresh process starts with +# an empty cache, so without it the seed path cannot see pre-existing rows +# and shadow-creates a duplicate); +# * the availability flags are published only AFTER every step succeeded. +# +# Both need the REAL function, and _initialize_impl builds a memU MemoryService, +# which can only happen ONCE per process: memU caches its SQLAlchemy models +# module-globally, and nerve's Fix 6 clears that cache on every call, so a second +# build raises ArgumentError("Column object 'url' already assigned"). Measured: +# a second _initialize_impl in the same interpreter returns False, and merely +# having built the fixture models above is enough to make the FIRST one return +# False. An in-process arm would therefore pass on rc=False without ever +# reaching the code it names -- so each arm runs the real function in a fresh +# interpreter and asserts rc is True first. The in-repo precedent for this shape +# is tests/test_config_sources.py. +# --------------------------------------------------------------------------- + + +_CATEGORY_TABLE_DDL = """ +CREATE TABLE memu_memory_categories ( + id VARCHAR NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, + updated_at DATETIME NOT NULL, + name VARCHAR NOT NULL, + description TEXT NOT NULL, + summary TEXT, + embedding_json TEXT, + user_id VARCHAR, + PRIMARY KEY (id) +); +""" +# Verbatim from the live schema memU's create_all produces, and the point of the +# whole fix: PRIMARY KEY (id) only, no unique index on name, so 'procedures' and +# 'PROCEDURES' are two perfectly legal rows. + + +def _run_init_script(tmp_path, body: str, seed_name: str | None = None): + """Run ``body`` against a real _initialize_impl in a fresh interpreter. + + Seeds ``seed_name`` with plain sqlite3 rather than through memU: importing + memU here to write one row would build the models and poison the init under + test. memU's create_all adds the remaining tables to the same file. + """ + import os + import subprocess + import sys + import uuid + + home = tmp_path / "nerve-home" + home.mkdir(parents=True, exist_ok=True) + db = home / "memu.sqlite" + seeded_id = "" + if seed_name is not None: + seeded_id = str(uuid.uuid4()) + conn = sqlite3.connect(db) + conn.executescript(_CATEGORY_TABLE_DDL) + conn.execute( + "INSERT INTO memu_memory_categories (id, updated_at, name, description)" + " VALUES (?, CURRENT_TIMESTAMP, ?, ?)", + (seeded_id, seed_name, "from another writer"), + ) + conn.commit() + conn.close() + + preamble = ( + "import asyncio, json, logging, os, sqlite3, sys\n" + "logging.disable(logging.CRITICAL)\n" + "from nerve.config import NerveConfig, MemoryConfig, MemoryCategoryConfig\n" + "from nerve.memory.memu_bridge import MemUBridge\n" + "DB = sys.argv[1]\n" + "SEEDED_ID = sys.argv[2]\n" + "config = NerveConfig()\n" + "config.memory = MemoryConfig(sqlite_dsn='sqlite:///' + DB,\n" + " categories=[MemoryCategoryConfig(name='procedures', description='d')])\n" + "config.openai_api_key = ''\n" + "bridge = MemUBridge(config)\n" + "out = {}\n" + ) + epilogue = "print('NERVE_RESULT ' + json.dumps(out))\n" + proc = subprocess.run( + [sys.executable, "-c", preamble + body + epilogue, str(db), seeded_id], + capture_output=True, + text=True, + # NERVE_HOME redirects every machine-local path (memu-resources etc.) into + # tmp_path, so the arm cannot touch the developer's real ~/.nerve. + env={**os.environ, "NERVE_HOME": str(home)}, + timeout=300, + ) + marker = [ln for ln in proc.stdout.splitlines() if ln.startswith("NERVE_RESULT ")] + assert marker, ( + f"init subprocess produced no result\nSTDOUT:\n{proc.stdout}\n" + f"STDERR:\n{proc.stderr}" + ) + result = json.loads(marker[-1][len("NERVE_RESULT "):]) + result["_seeded_id"] = seeded_id + result["_db"] = str(db) + return result + + +class TestInitializeImplCategoryOrdering: + """_initialize_impl's own ordering guarantees, driven through the real function.""" + + def test_seeding_sees_a_preexisting_row_because_the_cache_is_hydrated_first( + self, tmp_path + ): + """A restart over a case-variant row must adopt it, not shadow it. + + The seed path resolves against the repo's in-memory cache, which is empty + in a fresh process, so the hydration before _ensure_categories is what + lets it see the row at all. Asserted on the DATABASE rather than the + cache: the damage is a second persisted row, and a cache-only assertion + cannot tell an adopted row from a freshly created one. + """ + result = _run_init_script( + tmp_path, + "async def main():\n" + " out['rc'] = await bridge._initialize_impl()\n" + " conn = sqlite3.connect(DB)\n" + " rows = conn.execute(\n" + " 'select id, name from memu_memory_categories').fetchall()\n" + " conn.close()\n" + " out['db_ids'] = [r[0] for r in rows]\n" + " out['db_names'] = sorted(r[1] for r in rows)\n" + " ctx = bridge._service._get_context()\n" + " out['map'] = dict(ctx.category_name_to_id)\n" + "asyncio.run(main())\n", + seed_name="PROCEDURES", + ) + + assert result["rc"] is True, "the arm must observe a real successful init" + assert result["db_names"] == ["PROCEDURES"], ( + "seeding shadow-created a second row for the case variant" + ) + assert result["db_ids"] == [result["_seeded_id"]] + assert result["map"] == {"procedures": result["_seeded_id"]} + unreachable = set(result["db_ids"]) - set(result["map"].values()) + assert unreachable == set() + + def test_a_failure_after_the_service_is_built_publishes_no_availability( + self, tmp_path + ): + """All three flags, because initialized_at is never reset anywhere. + + _attach_engine_pragmas is the first step after the MemoryService + construction the flags used to sit beside, so raising there lands + strictly between the old site and the new one. A bridge that reports + itself available after a failed init is what the relocation prevents; + `initialized_at` has no reset path, so an except-clause fix could not + achieve this and the assertion must cover it explicitly. + """ + result = _run_init_script( + tmp_path, + "def boom(self):\n" + " raise RuntimeError('injected failure after the old flag site')\n" + "MemUBridge._attach_engine_pragmas = boom\n" + "async def main():\n" + " out['rc'] = await bridge._initialize_impl()\n" + " out['available'] = bridge._available\n" + " out['service_available'] = bridge._metrics.service_available\n" + " out['initialized_at'] = bridge._metrics.initialized_at\n" + "asyncio.run(main())\n", + ) + + assert result["rc"] is False, "the injection must actually fail the init" + assert result["available"] is False + assert result["service_available"] is False + assert result["initialized_at"] == "" + + def test_a_clean_init_does_publish_availability(self, tmp_path): + """Control for the arm above: the relocation must not withhold the flags. + + Without this, moving the flags to somewhere unreachable would satisfy the + failure arm perfectly. + """ + result = _run_init_script( + tmp_path, + "async def main():\n" + " out['rc'] = await bridge._initialize_impl()\n" + " out['available'] = bridge._available\n" + " out['service_available'] = bridge._metrics.service_available\n" + " out['initialized_at_set'] = bridge._metrics.initialized_at != ''\n" + "asyncio.run(main())\n", + ) + + assert result["rc"] is True + assert result["available"] is True + assert result["service_available"] is True + assert result["initialized_at_set"] is True + + +class TestBlankCategoryNameRejection: + """A blank name must be refused at its origin, never coerced. + + memU's prompt formatter renders a blank name as a placeholder, and nothing + can then assign to it -- so coercing downstream would turn a typo into a + permanent category row. + """ + + def test_blank_config_name_is_rejected_by_from_dict(self): + from nerve.config import MemoryCategoryConfig + + for blank in ("", " ", "\t\n"): + with pytest.raises(ValueError, match="must not be blank"): + MemoryCategoryConfig.from_dict({"name": blank}) + + def test_well_formed_config_name_is_stripped_and_accepted(self): + from nerve.config import MemoryCategoryConfig + + cfg = MemoryCategoryConfig.from_dict( + {"name": " procedures ", "description": "d"} + ) + assert cfg.name == "procedures" + assert cfg.description == "d" + + def test_no_category_is_ever_named_untitled(self): + """Pins the reject-not-coerce policy against a future placeholder.""" + from nerve.config import MemoryCategoryConfig + + try: + cfg = MemoryCategoryConfig.from_dict({"name": " "}) + except ValueError: + return + raise AssertionError(f"blank name was coerced to {cfg.name!r}, not rejected") + + def test_yaml_null_name_is_rejected_not_coerced_to_the_string_None(self): + """A bare ``name:`` in YAML is the realistic typo, and str() hides it. + + ``str(None)`` is the truthy literal ``'None'``, so a blank guard placed + after stringification never fires and seeding persists a category named + ``None``. Asserts the parse too, so the arm cannot pass because PyYAML + stopped producing None. + """ + import yaml + + from nerve.config import MemoryCategoryConfig + + parsed = yaml.safe_load("categories:\n - name:\n description: x\n") + assert parsed["categories"][0]["name"] is None + + with pytest.raises(ValueError, match="must be a string"): + MemoryCategoryConfig.from_dict(parsed["categories"][0]) + + def test_non_string_config_name_is_rejected(self): + """The same silent-coercion shape as YAML null, for every non-str scalar. + + A non-str name has never been reachable: nerve's own key derivation + (``name.lower()``) and memU's (``cfg.name.strip()``) both raise + AttributeError on it, so rejecting here removes no working case and only + moves the failure to the config boundary that can name the offender. + """ + from nerve.config import MemoryCategoryConfig + + for bad in (12, 1.5, True, ["a"], {"k": 1}): + with pytest.raises(ValueError, match="must be a string"): + MemoryCategoryConfig.from_dict({"name": bad, "description": "d"}) + + def test_no_config_can_produce_a_category_named_None(self): + """Independent oracle: assert the OUTCOME, not the exception type. + + A future revision that reverts to ``str(d["name"])`` while keeping a + blank guard passes the raises-arms' sibling shapes but fails here. + """ + from nerve.config import MemoryCategoryConfig + + for bad in (None, 12, True): + try: + cfg = MemoryCategoryConfig.from_dict({"name": bad}) + except ValueError: + continue + raise AssertionError( + f"name={bad!r} was coerced to {cfg.name!r}, not rejected" + ) + + +class TestCreateCategoryRoute: + """POST /api/memory/memu/categories input validation.""" + + @pytest.fixture + def client(self): + from fastapi import FastAPI + from fastapi.testclient import TestClient + + import nerve.config as cfg_mod + from nerve.gateway.routes._deps import init_deps + from nerve.gateway.routes.memory import router + + config = NerveConfig() + # require_auth reads get_config().auth.jwt_secret -- empty makes it a no-op. + config.auth.jwt_secret = "" + cfg_mod._config = config + + bridge = SimpleNamespace( + available=True, + calls=[], + ) + + async def _create_category(name, description, source="bridge"): + bridge.calls.append((name, description, source)) + return True + + bridge.create_category = _create_category + init_deps(engine=SimpleNamespace(_memory_bridge=bridge), db=None) # type: ignore[arg-type] + + app = FastAPI() + app.include_router(router) + try: + yield TestClient(app), bridge + finally: + cfg_mod._config = None + + def test_blank_name_is_rejected_with_400(self, client): + """400, not 422: a pydantic validator would fail before the body runs, + and this route reports bad input as 400 (as its sibling handlers do).""" + http, bridge = client + + response = http.post("/api/memory/memu/categories", json={"name": " "}) + + assert response.status_code == 400 + assert bridge.calls == [], "a rejected name must never reach the bridge" + + def test_padded_name_is_stripped_before_reaching_the_bridge(self, client): + http, bridge = client + + response = http.post( + "/api/memory/memu/categories", json={"name": " procedures "} + ) + + assert response.status_code == 200 + assert response.json()["name"] == "procedures" + assert bridge.calls == [("procedures", "", "web_ui")]