fix(memory): fail closed on the remaining read-modify-write add paths

The strict loader landed with the routes, the backup import and the extractor
converted, but three read-modify-write sinks still called load_all(), which
degrades an unreadable store to []. Two of them are the paths users actually
reach, so the data loss in #5673 stayed reproducible:

- src/ai_interaction.py do_manage_memory, action "add" — reached from ordinary
  chat via src/tool_execution.py:793 -> dispatch_ai_tool. "Remember that I
  prefer X" against an unreadable store wrote a one-entry file over it and
  reported success.
- mcp_servers/memory_server.py, action "add" — the same shape through
  _scope_entries(), registered as a built-in in src/builtin_mcp.py.
- src/memory_provider.py NativeMemoryProvider.remember and .delete — wired
  into app state in src/app_initializer.py but not consumed outside tests yet,
  converted here so the pattern is uniform before it goes live.

The MCP server takes _scope_entries(for_update=True) so list keeps the lenient
read. The edit and delete branches on both tool paths were already fail-closed
by accident — an empty view matches nothing and returns before the save — so
they are left alone.

The three new tests drive the real entry points rather than replaying the
shape, and use a truncated store, which is the case that reads back fine so
nothing stops the save. Each asserts memory.json is byte-identical afterwards;
all three fail on the previous commit with the store overwritten.

mcp_servers/memory_server.py

@@ -17,6 +17,8 @@ from mcp.types import Tool, TextContent
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
+from src.memory import MemoryStoreUnreadable
+
 server = Server("memory")
 
 # Late-initialized managers (set during first tool call)
@@ -29,6 +31,10 @@ _OWNER_SCOPE_ERROR = (
     "Error: Memory MCP owner is not configured for an owner-scoped memory store. "
     "Set ODYSSEUS_MCP_MEMORY_OWNER for this server or use the owner-aware native memory tool."
 )
+_UNREADABLE_STORE_ERROR = (
+    "Error: Memory store is temporarily unreadable — nothing was saved. "
+    "Repair or restore memory.json, then retry."
+)
 
 
 def _configured_owner() -> str | None:
@@ -51,9 +57,21 @@ def _owner_scoped_store(entries: list[dict]) -> bool:
     return any(_entry_owner(entry) for entry in entries if isinstance(entry, dict))
 
 
-def _scope_entries() -> tuple[str | None, list[dict], list[dict], str | None]:
-    """Return configured owner, all entries, visible entries, and optional error."""
-    entries = _memory_manager.load_all()
+def _scope_entries(for_update: bool = False) -> tuple[str | None, list[dict], list[dict], str | None]:
+    """Return configured owner, all entries, visible entries, and optional error.
+
+    ``for_update=True`` is for read-modify-write callers. They save the ``all
+    entries`` list back, so an unreadable store must be reported as an error
+    instead of degrading to ``[]`` — otherwise the save writes their one new
+    entry over the whole store (issue #5673).
+    """
+    if for_update:
+        try:
+            entries = _memory_manager.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            return None, [], [], f"{_UNREADABLE_STORE_ERROR} ({e})"
+    else:
+        entries = _memory_manager.load_all()
     owner = _configured_owner()
     if owner is None and _owner_scoped_store(entries):
         return None, entries, [], _OWNER_SCOPE_ERROR
@@ -161,7 +179,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
         category = arguments.get("category", "fact")
         if not text:
             return _text_result("Error: Memory text cannot be empty")
-        owner, memories, _visible, scope_error = _scope_entries()
+        owner, memories, _visible, scope_error = _scope_entries(for_update=True)
         if scope_error:
             return _text_result(scope_error)
         entry = _memory_manager.add_entry(text, source="ai_agent", category=category, owner=owner)

src/ai_interaction.py

@@ -22,6 +22,7 @@ import time
 from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
 
 from src.constants import GENERATED_IMAGES_DIR
+from src.memory import MemoryStoreUnreadable
 
 logger = logging.getLogger(__name__)
 
@@ -384,7 +385,15 @@ async def do_manage_memory(content: str, session_id: Optional[str] = None, owner
             return {"error": "Memory text cannot be empty"}
 
         entry = _memory_manager.add_entry(text, source="ai_agent", category=category, owner=owner)
-        memories = _memory_manager.load_all()
+        # Strict load: this is a read-modify-write, and it is the path an
+        # ordinary "remember that I prefer X" takes. Degrading to [] here would
+        # save just this one entry over a store we only failed to read,
+        # atomically destroying every memory in it (issue #5673).
+        try:
+            memories = _memory_manager.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            logger.error("Refusing to add memory, store unreadable: %s", e)
+            return {"error": "Memory store is temporarily unreadable — nothing was saved."}
         memories.append(entry)
         _memory_manager.save(memories)
 

src/memory_provider.py

@@ -157,7 +157,11 @@ class NativeMemoryProvider(MemoryProvider):
         if metadata:
             entry["metadata"] = dict(metadata)
 
-        memories = self.memory_manager.load_all()
+        # Strict load: read-modify-write. `load_all` degrades an unreadable
+        # store to [], which would save this single entry over everything
+        # already stored (issue #5673). The provider API has no error channel,
+        # so MemoryStoreUnreadable propagates to the caller.
+        memories = self.memory_manager.load_all_for_update()
         memories.append(entry)
         self.memory_manager.save(memories)
 
@@ -223,7 +227,10 @@ class NativeMemoryProvider(MemoryProvider):
         ]
 
     async def delete(self, memory_id: str, *, owner: Optional[str] = None) -> bool:
-        memories = self.memory_manager.load_all()
+        # Strict load for the same reason: `remaining` is derived from this
+        # list and saved back, so it must never be built from a store we
+        # failed to read.
+        memories = self.memory_manager.load_all_for_update()
         remaining = []
         deleted_id = None
 

tests/test_memory_store_unreadable_no_wipe.py

@@ -24,6 +24,7 @@ end-to-end — clean dev returns 500 there and loses nothing).
 `MemoryStoreUnreadable` rather than reporting an empty store.
 """
 
+import asyncio
 import builtins
 import json
 import os
@@ -159,6 +160,87 @@ def test_claim_ownerless_skips_write_when_unreadable(tmp_path, monkeypatch):
     assert {e["id"] for e in m.load_all()} == {"m1", "m2", "m3"}
 
 
+# ── the add sinks users actually reach ────────────────────────────────────
+#
+# The tests above replay the read-modify-write shape. These drive the real
+# entry points end to end, because those are what #5673 reports: "remember
+# that I prefer X" in ordinary chat (src/ai_interaction.py do_manage_memory,
+# routed from src/tool_execution.py) and the built-in memory MCP server
+# (mcp_servers/memory_server.py, registered in src/builtin_mcp.py).
+#
+# They use a truncated store rather than a read error on purpose: it reads
+# fine, so nothing stops the save, which is the case that silently destroyed
+# stores. The assertion is that the file is left byte-identical — still broken,
+# but still holding the user's memories, so it can be repaired by hand.
+
+
+def _truncated_store(tmp_path):
+    """Seed a store that reads back fine but no longer parses."""
+    m = _seeded(tmp_path)
+    good = json.dumps([dict(e) for e in _SEED], indent=2)
+    with open(m.memory_file, "w", encoding="utf-8") as f:
+        f.write(good[:good.rindex("]")])  # drop the closing bracket only
+    with open(m.memory_file, "rb") as f:
+        return m, f.read()
+
+
+def _on_disk(manager) -> bytes:
+    with open(manager.memory_file, "rb") as f:
+        return f.read()
+
+
+def test_agent_memory_add_does_not_overwrite_unreadable_store(tmp_path, monkeypatch):
+    """src/ai_interaction.py do_manage_memory, action "add"."""
+    from src import ai_interaction
+
+    manager, before = _truncated_store(tmp_path)
+    monkeypatch.setattr(ai_interaction, "_memory_manager", manager)
+    monkeypatch.setattr(ai_interaction, "_memory_vector", None)
+
+    result = asyncio.run(ai_interaction.do_manage_memory("add\nuser prefers tabs"))
+
+    assert _on_disk(manager) == before, "the unreadable store was overwritten"
+    assert b"m3" in _on_disk(manager)
+    assert "error" in result, "the add reported success over an unreadable store"
+
+
+def test_mcp_memory_add_does_not_overwrite_unreadable_store(tmp_path, monkeypatch):
+    """mcp_servers/memory_server.py, action "add"."""
+    import mcp_servers.memory_server as memory_server
+
+    manager, before = _truncated_store(tmp_path)
+    monkeypatch.setattr(memory_server, "_memory_manager", manager)
+    monkeypatch.setattr(memory_server, "_memory_vector", None)
+    monkeypatch.setattr(memory_server, "_initialized", True)
+    for key in memory_server._OWNER_ENV_KEYS:
+        monkeypatch.delenv(key, raising=False)
+
+    result = asyncio.run(memory_server.call_tool(
+        "manage_memory", {"action": "add", "text": "user prefers tabs"}
+    ))
+
+    assert _on_disk(manager) == before, "the unreadable store was overwritten"
+    assert b"m3" in _on_disk(manager)
+    assert result[0].text.startswith("Error:")
+
+
+def test_native_provider_remember_does_not_overwrite_unreadable_store(tmp_path):
+    """src/memory_provider.py NativeMemoryProvider.remember.
+
+    Registered into app state in src/app_initializer.py but not yet consumed
+    outside tests, so this is the pattern held in place before it goes live.
+    """
+    from src.memory_provider import NativeMemoryProvider
+
+    manager, before = _truncated_store(tmp_path)
+    provider = NativeMemoryProvider(manager)
+
+    with pytest.raises(MemoryStoreUnreadable):
+        asyncio.run(provider.remember("user prefers tabs", owner="alice"))
+
+    assert _on_disk(manager) == before
+
+
 # ── the legacy memory.txt migration is preserved ──────────────────────────
 
 def test_corrupt_store_still_migrates_from_legacy_txt(tmp_path):