Hardening/sqlite write atomicity

CHANGELOG.md

@@ -2,6 +2,19 @@
 
 ## Version 2 - MARM Protocol to Universal MCP Server Evolution
 
+<details>
+<summary><strong>Unreleased: SQLite Write Atomicity Hardening (v2.22.x)</strong></summary>
+
+### Multi-Statement Writes Are Now Real Transactions
+
+- MARM's pooled SQLite connections run in full autocommit mode (`isolation_level=None`), so a multi-statement write sequence followed by `conn.commit()` was never actually atomic — each statement committed immediately on its own, and a failure partway through left the earlier statements permanently applied with nothing to roll back. Hardened every meaningful multi-statement mutation path with the project's existing `BEGIN IMMEDIATE` / explicit `COMMIT` / `ROLLBACK`-on-exception pattern (already used by `_store_memory` and `apply_compaction_write`): log-entry creation, session-switch, and whole-session delete (`services/log_entry.py`); session activation (`endpoints/session.py`); notebook add (`services/notebook.py`); the legacy system-notebook cleanup pass (`services/documentation.py`).
+- Fixed a real, independently-discovered bug while auditing `_update_memory` (the consolidation merge path): it used to hold `BEGIN IMMEDIATE`'s write lock open across an `await` for embedding generation, which could stall the whole event loop for any other writer waiting on the same database. Restructured to compute the merge and its embedding before acquiring the lock, then re-verify under the lock (matching content *and* metadata, not just content) that nothing changed concurrently before writing — and folded the memory-chunks cleanup into the same transaction as the content update, so the two can no longer disagree after a partial failure.
+- Fixed a second bug found in independent review: `_update_memory` can legitimately no-op (return without writing) when its target row was deleted or changed concurrently between the duplicate check and the write-lock re-verification, but `_store_memory` still reported the stale `existing_id` as a successful merge — silently dropping the caller's content with no trace. `_update_memory` now returns a bool signaling whether it actually wrote; `_store_memory` falls through and stores the content as a new memory instead of reporting success on a merge that never happened.
+- Every new transaction boundary has a mutation-tested regression test (`tests/test_sqlite_write_atomicity.py`): each one forces a specific SQL statement to fail and asserts the earlier statement(s) in the same block did not durably apply — a happy-path-only test doesn't prove rollback. Coverage now also includes the session-switch, dated-fallback, targeted log-delete, and notebook-delete transaction boundaries.
+- No schema changes, no new endpoints, no tool-surface changes. Pure reliability hardening.
+
+</details>
+
 <details>
 <summary><strong>July 15th, 2026: MARM Console Safe Memory Mutations (v2.22.0)</strong></summary>
 

marm-mcp-server/marm_mcp_server/core/memory.py

@@ -244,7 +244,7 @@ async def auto_classify_content(self, content: str) -> str:
         else:
             return "general"
 
-    async def update_memory(self, memory_id: str, new_content: str) -> None:
+    async def update_memory(self, memory_id: str, new_content: str) -> bool:
         return await _update_memory(self, memory_id, new_content)
 
     async def store_memory(

marm-mcp-server/marm_mcp_server/core/memory_ops.py

@@ -41,83 +41,124 @@
 )
 
 
-async def _update_memory(mem, memory_id: str, new_content: str) -> None:
+async def _update_memory(mem, memory_id: str, new_content: str) -> bool:
     """Append new_content into an existing memory and record the merge in metadata.
 
     Recomputes content_hash and embedding so Layer 1 dedup and semantic recall
-    stay accurate after the merge.
+    stay accurate after the merge. Returns False (no write happened) if the
+    row was deleted or changed concurrently between the pre-read and the
+    write lock -- callers must not assume the merge landed just because this
+    returned without raising.
     """
+    # Unlocked pre-read -- matches _store_memory's duplicate pre-check
+    # convention. This is a read-then-write, but the write lock is only
+    # acquired later, right before the actual UPDATE, so this read must
+    # be re-verified under the lock before it's trusted (see below).
     with mem.get_connection() as conn:
-        conn.execute("BEGIN IMMEDIATE")
         row = conn.execute(
             "SELECT content, metadata FROM memories WHERE id = ?", (memory_id,)
         ).fetchone()
-        if row is None:
-            conn.execute("ROLLBACK")
-            return
-        existing_content, metadata_json = row
-        metadata = json.loads(metadata_json) if metadata_json else {}
-        _MAX = 10000
-        _MARKER = "\n[merged] "
-        _new_budget = _MAX - len(_MARKER)
-        if len(new_content) > _new_budget:
-            new_content = new_content[:_new_budget]
-        _existing_budget = _MAX - len(_MARKER) - len(new_content)
-        existing_content = existing_content[: max(0, _existing_budget)]
-        merged_content = f"{existing_content}{_MARKER}{new_content}"
-        merged_at = datetime.now(timezone.utc).isoformat()
-        if "merge_history" not in metadata:
-            metadata["merge_history"] = []
-        metadata["merge_history"].append(
-            {
-                "merged_at": merged_at,
-                "content_preview": new_content[:100],
-            }
-        )
-
-        merged_hash = compute_content_hash(merged_content)
+    if row is None:
+        return False
+    existing_content, metadata_json = row
+    original_existing_content = existing_content  # unsliced, for the re-check below
+    metadata = json.loads(metadata_json) if metadata_json else {}
+    _MAX = 10000
+    _MARKER = "\n[merged] "
+    _new_budget = _MAX - len(_MARKER)
+    if len(new_content) > _new_budget:
+        new_content = new_content[:_new_budget]
+    _existing_budget = _MAX - len(_MARKER) - len(new_content)
+    existing_content = existing_content[: max(0, _existing_budget)]
+    merged_content = f"{existing_content}{_MARKER}{new_content}"
+    merged_at = datetime.now(timezone.utc).isoformat()
+    if "merge_history" not in metadata:
+        metadata["merge_history"] = []
+    metadata["merge_history"].append(
+        {
+            "merged_at": merged_at,
+            "content_preview": new_content[:100],
+        }
+    )
 
-        merged_embedding_bytes = None
-        encoder_ok = merged_content.strip() and mem._load_encoder_lazily()
-        if encoder_ok:
-            try:
-                merged_vec = await asyncio.to_thread(mem._encode_sync, merged_content)
-                merged_embedding_bytes = _embedding_to_bytes(merged_vec)
-            except Exception as e:
-                _safe_print(f"Failed to regenerate embedding after merge: {e}")
-
-        if merged_embedding_bytes is not None:
-            conn.execute(
-                "UPDATE memories SET content = ?, metadata = ?, content_hash = ?, embedding = ?, timestamp = ? WHERE id = ?",
-                (
-                    merged_content,
-                    json.dumps(metadata),
-                    merged_hash,
-                    merged_embedding_bytes,
-                    merged_at,
-                    memory_id,
-                ),
-            )
-        else:
-            conn.execute(
-                "UPDATE memories SET content = ?, metadata = ?, content_hash = ?, embedding = NULL, timestamp = ? WHERE id = ?",
-                (
-                    merged_content,
-                    json.dumps(metadata),
-                    merged_hash,
-                    merged_at,
-                    memory_id,
-                ),
-            )
+    merged_hash = compute_content_hash(merged_content)
+
+    # Compute the embedding before acquiring the write lock, same as
+    # _store_memory/_replace_memory -- embedding work can be slow, and
+    # holding BEGIN IMMEDIATE's write lock across an await would block
+    # every other writer (sqlite3 calls are synchronous, so a concurrent
+    # BEGIN IMMEDIATE on another connection can stall the whole event
+    # loop) for the duration.
+    merged_embedding_bytes = None
+    encoder_ok = merged_content.strip() and mem._load_encoder_lazily()
+    if encoder_ok:
+        try:
+            merged_vec = await asyncio.to_thread(mem._encode_sync, merged_content)
+            merged_embedding_bytes = _embedding_to_bytes(merged_vec)
+        except Exception as e:
+            _safe_print(f"Failed to regenerate embedding after merge: {e}")
 
     with mem.get_connection() as conn:
-        conn.execute("DELETE FROM memory_chunks WHERE memory_id = ?", (memory_id,))
+        conn.execute("BEGIN IMMEDIATE")
+        try:
+            # Re-check under the lock: the row may have been deleted, or
+            # its content OR metadata changed by a concurrent writer,
+            # since the unlocked read above -- metadata-only writers
+            # exist in this file too (_restore_sources_from_deleted_summary,
+            # _update_summary_metadata_after_delete), so content alone
+            # isn't enough to catch every concurrent write. Bail rather
+            # than risk clobbering a concurrent update with a merge based
+            # on stale data.
+            current = conn.execute(
+                "SELECT content, metadata FROM memories WHERE id = ?", (memory_id,)
+            ).fetchone()
+            if (
+                current is None
+                or current[0] != original_existing_content
+                or current[1] != metadata_json
+            ):
+                conn.execute("ROLLBACK")
+                return False
+
+            if merged_embedding_bytes is not None:
+                conn.execute(
+                    "UPDATE memories SET content = ?, metadata = ?, content_hash = ?, embedding = ?, timestamp = ? WHERE id = ?",
+                    (
+                        merged_content,
+                        json.dumps(metadata),
+                        merged_hash,
+                        merged_embedding_bytes,
+                        merged_at,
+                        memory_id,
+                    ),
+                )
+            else:
+                conn.execute(
+                    "UPDATE memories SET content = ?, metadata = ?, content_hash = ?, embedding = NULL, timestamp = ? WHERE id = ?",
+                    (
+                        merged_content,
+                        json.dumps(metadata),
+                        merged_hash,
+                        merged_at,
+                        memory_id,
+                    ),
+                )
+            # Folded into the same transaction as the content update --
+            # a chunk-delete failure must not leave stale chunks that
+            # disagree with the (already committed) merged content, and
+            # vice versa.
+            conn.execute("DELETE FROM memory_chunks WHERE memory_id = ?", (memory_id,))
+            conn.execute("COMMIT")
+        except Exception:
+            conn.execute("ROLLBACK")
+            raise
 
     chunks = _chunk_text(merged_content)
     if chunks and mem._load_encoder_lazily():
         _chunk_task = asyncio.create_task(  # noqa: RUF006
             _write_chunks(mem, mem.db_path, memory_id, chunks, merged_hash)
         )
+    return True
 
 
 async def _store_memory(
@@ -188,9 +229,15 @@ async def _store_memory(
                 query_vec=pre_embedding,
             )
         if existing_id:
-            await _update_memory(mem, existing_id, sanitized_content)
-            mem._on_memory_written(session)
-            return existing_id
+            merged = await _update_memory(mem, existing_id, sanitized_content)
+            if merged:
+                mem._on_memory_written(session)
+                return existing_id
+            # existing_id's row was deleted or changed concurrently between
+            # the duplicate check above and _update_memory's write-lock
+            # re-verification -- the merge never happened. Fall through and
+            # store sanitized_content as a new memory instead of silently
+            # dropping it and reporting existing_id as if it succeeded.
 
     memory_id = str(uuid.uuid4())
     timestamp = datetime.now(timezone.utc).isoformat()

marm-mcp-server/marm_mcp_server/endpoints/compaction.py

@@ -1,4 +1,20 @@
-"""Compaction MCP endpoint tools — V2 agent-driven summarization, V3 staged apply, V4 write-queue + auto-apply."""
+"""Compaction MCP endpoint tools — V2 agent-driven summarization, V3 staged apply, V4 write-queue + auto-apply.
+
+Audited under the SQLite write-atomicity hardening effort
+(docs/current/sqlite-write-atomicity-hardening.md): no BEGIN IMMEDIATE
+added here. Every mutation on every branch of
+marm_stage_compaction_summaries and marm_apply_compaction's validation
+section is exactly one conn.execute("UPDATE compaction_staging ...")
+call -- a lone statement is already atomic under SQLite regardless of
+isolation_level, so there's no multi-statement sequence to protect.
+Wrapping the whole per-candidate loop in marm_stage_compaction_summaries
+in one transaction was considered and rejected: each candidate is
+designed to succeed or fail independently (see its own per-candidate
+`results` list), and an all-or-nothing batch wrap would change that
+intended semantic, not just add safety. The actual apply write
+(apply_compaction_write, in services/compaction_apply.py) already uses
+BEGIN IMMEDIATE internally and was not touched.
+"""
 
 import json
 from datetime import datetime, timezone

marm-mcp-server/marm_mcp_server/endpoints/session.py

@@ -22,15 +22,20 @@ async def marm_start(request: SessionRequest):
     """
     try:
         with memory.get_connection() as conn:
-            conn.execute("UPDATE sessions SET marm_active = FALSE")
-            conn.execute(
-                """
-                INSERT OR REPLACE INTO sessions (session_name, marm_active, last_accessed)
-                VALUES (?, TRUE, ?)
-            """,
-                (request.session_name, datetime.now(timezone.utc).isoformat()),
-            )
-            conn.commit()
+            conn.execute("BEGIN IMMEDIATE")
+            try:
+                conn.execute("UPDATE sessions SET marm_active = FALSE")
+                conn.execute(
+                    """
+                    INSERT OR REPLACE INTO sessions (session_name, marm_active, last_accessed)
+                    VALUES (?, TRUE, ?)
+                """,
+                    (request.session_name, datetime.now(timezone.utc).isoformat()),
+                )
+                conn.execute("COMMIT")
+            except Exception:
+                conn.execute("ROLLBACK")
+                raise
 
         if not docs_are_loaded():
             await load_marm_documentation()

marm-mcp-server/marm_mcp_server/services/documentation.py

@@ -109,6 +109,14 @@ async def _index_doc(doc: Dict) -> bool:
                     return True
                 print(f"[DOCS] {fname} memory row missing, re-indexing")
 
+        # Audited under the SQLite write-atomicity hardening effort
+        # (docs/current/sqlite-write-atomicity-hardening.md): no BEGIN
+        # IMMEDIATE needed here. Exactly one of the two branches below
+        # runs per call, and each is a single statement -- a lone
+        # statement is already atomic under SQLite regardless of
+        # isolation_level, so there's no multi-statement sequence to
+        # protect. store_memory_queued below intentionally stays outside
+        # any transaction (it awaits and does its own internal locking).
         with memory.get_connection() as conn:
             if row and row[1]:
                 conn.execute("DELETE FROM memories WHERE id = ?", (row[1],))
@@ -263,17 +271,22 @@ async def load_marm_documentation():
             "SELECT value FROM user_settings WHERE key = 'system_notebook_cleanup_v1'"
         ).fetchone()
         if not already_cleaned:
-            for name in _LEGACY_SYSTEM_NOTEBOOK_NAMES:
-                conn.execute("DELETE FROM notebook_entries WHERE name = ?", (name,))
-            conn.execute(
-                "INSERT OR REPLACE INTO user_settings (key, value, updated_at) VALUES (?, ?, ?)",
-                (
-                    "system_notebook_cleanup_v1",
-                    "done",
-                    datetime.now(timezone.utc).isoformat(),
-                ),
-            )
-            conn.commit()
+            conn.execute("BEGIN IMMEDIATE")
+            try:
+                for name in _LEGACY_SYSTEM_NOTEBOOK_NAMES:
+                    conn.execute("DELETE FROM notebook_entries WHERE name = ?", (name,))
+                conn.execute(
+                    "INSERT OR REPLACE INTO user_settings (key, value, updated_at) VALUES (?, ?, ?)",
+                    (
+                        "system_notebook_cleanup_v1",
+                        "done",
+                        datetime.now(timezone.utc).isoformat(),
+                    ),
+                )
+                conn.execute("COMMIT")
+            except Exception:
+                conn.execute("ROLLBACK")
+                raise
             print("[DOCS] Cleaned up legacy system notebook entries")
 
     docs = get_docs_to_load()