diff --git a/CHANGELOG.md b/CHANGELOG.md index e511f839..384c07cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## Version 2 - MARM Protocol to Universal MCP Server Evolution +
+Unreleased: SQLite Write Atomicity Hardening (v2.22.x) + +### 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. + +
+
July 15th, 2026: MARM Console Safe Memory Mutations (v2.22.0) diff --git a/marm-mcp-server/marm_mcp_server/core/memory.py b/marm-mcp-server/marm_mcp_server/core/memory.py index 8ffdd29f..207a55cb 100644 --- a/marm-mcp-server/marm_mcp_server/core/memory.py +++ b/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( diff --git a/marm-mcp-server/marm_mcp_server/core/memory_ops.py b/marm-mcp-server/marm_mcp_server/core/memory_ops.py index f6193b84..b2cb8328 100644 --- a/marm-mcp-server/marm_mcp_server/core/memory_ops.py +++ b/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() diff --git a/marm-mcp-server/marm_mcp_server/endpoints/compaction.py b/marm-mcp-server/marm_mcp_server/endpoints/compaction.py index 50adde1f..c41b59b6 100644 --- a/marm-mcp-server/marm_mcp_server/endpoints/compaction.py +++ b/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 diff --git a/marm-mcp-server/marm_mcp_server/endpoints/session.py b/marm-mcp-server/marm_mcp_server/endpoints/session.py index 0ca9e90f..960edde7 100644 --- a/marm-mcp-server/marm_mcp_server/endpoints/session.py +++ b/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() diff --git a/marm-mcp-server/marm_mcp_server/services/documentation.py b/marm-mcp-server/marm_mcp_server/services/documentation.py index 71f70e76..bf63a612 100644 --- a/marm-mcp-server/marm_mcp_server/services/documentation.py +++ b/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() diff --git a/marm-mcp-server/marm_mcp_server/services/log_entry.py b/marm-mcp-server/marm_mcp_server/services/log_entry.py index d31ed349..477e152a 100644 --- a/marm-mcp-server/marm_mcp_server/services/log_entry.py +++ b/marm-mcp-server/marm_mcp_server/services/log_entry.py @@ -46,42 +46,47 @@ async def create_log_entry( new_session = f"{base_name}-{date_tag}" marker_id = str(uuid.uuid4()) with memory.get_connection() as conn: - conn.execute("UPDATE sessions SET marm_active = FALSE") - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed, marm_active) - VALUES (?, ?, TRUE) - ON CONFLICT(session_name) DO UPDATE SET - last_accessed = excluded.last_accessed, - marm_active = TRUE - """, - (new_session, datetime.now(timezone.utc).isoformat()), - ) - conn.execute( - """ - INSERT INTO log_entries - (id, session_name, entry_date, topic, summary, full_entry, project, platform) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - marker_id, - new_session, - date_tag, - "session_start", - base_name, - formatted_entry, - MARM_PROJECT or None, - MARM_PLATFORM or None, - ), - ) + conn.execute("BEGIN IMMEDIATE") try: + conn.execute("UPDATE sessions SET marm_active = FALSE") + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed, marm_active) + VALUES (?, ?, TRUE) + ON CONFLICT(session_name) DO UPDATE SET + last_accessed = excluded.last_accessed, + marm_active = TRUE + """, + (new_session, datetime.now(timezone.utc).isoformat()), + ) conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (datetime.now(timezone.utc).isoformat(), new_session), + """ + INSERT INTO log_entries + (id, session_name, entry_date, topic, summary, full_entry, project, platform) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + marker_id, + new_session, + date_tag, + "session_start", + base_name, + formatted_entry, + MARM_PROJECT or None, + MARM_PLATFORM or None, + ), ) + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + (datetime.now(timezone.utc).isoformat(), new_session), + ) + except Exception: + pass + conn.execute("COMMIT") except Exception: - pass - conn.commit() + conn.execute("ROLLBACK") + raise memory.active_log_session = new_session # The session row and marker entry above are already durably # committed -- an event-publish failure must not turn that @@ -108,18 +113,23 @@ async def create_log_entry( date_tag = datetime.now(timezone.utc).strftime("%Y-%m-%d") session = f"session-{date_tag}" with memory.get_connection() as conn: - conn.execute("UPDATE sessions SET marm_active = FALSE") - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed, marm_active) - VALUES (?, ?, TRUE) - ON CONFLICT(session_name) DO UPDATE SET - last_accessed = excluded.last_accessed, - marm_active = TRUE - """, - (session, datetime.now(timezone.utc).isoformat()), - ) - conn.commit() + conn.execute("BEGIN IMMEDIATE") + try: + conn.execute("UPDATE sessions SET marm_active = FALSE") + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed, marm_active) + VALUES (?, ?, TRUE) + ON CONFLICT(session_name) DO UPDATE SET + last_accessed = excluded.last_accessed, + marm_active = TRUE + """, + (session, datetime.now(timezone.utc).isoformat()), + ) + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise memory.active_log_session = session # Chunk boundary check @@ -153,38 +163,43 @@ async def create_log_entry( entry_id = str(uuid.uuid4()) now_iso = datetime.now(timezone.utc).isoformat() with memory.get_connection() as conn: - conn.execute( - """ - INSERT INTO log_entries (id, session_name, entry_date, topic, summary, full_entry, project, platform) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - entry_id, - session, - entry_date, - topic, - summary, - formatted_entry, - MARM_PROJECT or None, - MARM_PLATFORM or None, - ), - ) - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed) - VALUES (?, ?) - ON CONFLICT(session_name) DO UPDATE SET last_accessed = excluded.last_accessed - """, - (session, now_iso), - ) + conn.execute("BEGIN IMMEDIATE") try: conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (now_iso, session), + """ + INSERT INTO log_entries (id, session_name, entry_date, topic, summary, full_entry, project, platform) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + entry_id, + session, + entry_date, + topic, + summary, + formatted_entry, + MARM_PROJECT or None, + MARM_PLATFORM or None, + ), + ) + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed) + VALUES (?, ?) + ON CONFLICT(session_name) DO UPDATE SET last_accessed = excluded.last_accessed + """, + (session, now_iso), ) + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + (now_iso, session), + ) + except Exception: + pass + conn.execute("COMMIT") except Exception: - pass - conn.commit() + conn.execute("ROLLBACK") + raise # Dual-write into semantic memory so marm_smart_recall can find it; # a store failure must never fail the log write itself. @@ -306,56 +321,70 @@ async def delete_log_or_notebook_entry( if type == "log": memories_deleted = 0 if session_name: - # Dual-written semantic memories must not outlive their log entries - rows = conn.execute( - "SELECT id FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", - (session_name, target, target), - ).fetchall() - entry_ids = [r[0] for r in rows] - cursor = conn.execute( - "DELETE FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", - (session_name, target, target), - ) - deleted = cursor.rowcount - if entry_ids: - placeholders = ",".join("?" * len(entry_ids)) - memories_deleted = conn.execute( - "DELETE FROM memories WHERE json_extract(metadata, '$.source') = 'log_entry' " - f"AND json_extract(metadata, '$.log_entry_id') IN ({placeholders})", - entry_ids, - ).rowcount - if deleted: + conn.execute("BEGIN IMMEDIATE") + try: + # Dual-written semantic memories must not outlive their log entries + rows = conn.execute( + "SELECT id FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", + (session_name, target, target), + ).fetchall() + entry_ids = [r[0] for r in rows] + cursor = conn.execute( + "DELETE FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", + (session_name, target, target), + ) + deleted = cursor.rowcount + if entry_ids: + placeholders = ",".join("?" * len(entry_ids)) + memories_deleted = conn.execute( + "DELETE FROM memories WHERE json_extract(metadata, '$.source') = 'log_entry' " + f"AND json_extract(metadata, '$.log_entry_id') IN ({placeholders})", + entry_ids, + ).rowcount + if deleted: + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + ( + datetime.now(timezone.utc).isoformat(), + session_name, + ), + ) + except Exception: + pass + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise + else: + conn.execute("BEGIN IMMEDIATE") + try: + conn.execute( + "DELETE FROM sessions WHERE session_name = ?", (target,) + ) + cursor = conn.execute( + "DELETE FROM log_entries WHERE session_name = ?", (target,) + ) + deleted = cursor.rowcount + # Guarded like every other session_summary_cache touch in + # this module -- a missing/locked cache row must not + # abort the rest of the whole-session delete. try: conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (datetime.now(timezone.utc).isoformat(), session_name), + "DELETE FROM session_summary_cache WHERE session_name = ?", + (target,), ) except Exception: pass - else: - conn.execute( - "DELETE FROM sessions WHERE session_name = ?", (target,) - ) - cursor = conn.execute( - "DELETE FROM log_entries WHERE session_name = ?", (target,) - ) - deleted = cursor.rowcount - # Guarded like every other session_summary_cache touch in - # this module -- a missing/locked cache row must not - # abort the rest of the whole-session delete. - try: - conn.execute( - "DELETE FROM session_summary_cache WHERE session_name = ?", + memories_deleted = conn.execute( + "DELETE FROM memories WHERE session_name = ? " + "AND json_extract(metadata, '$.source') = 'log_entry'", (target,), - ) + ).rowcount + conn.execute("COMMIT") except Exception: - pass - memories_deleted = conn.execute( - "DELETE FROM memories WHERE session_name = ? " - "AND json_extract(metadata, '$.source') = 'log_entry'", - (target,), - ).rowcount - conn.commit() + conn.execute("ROLLBACK") + raise # Flip the runtime pointer only after the delete durably # commits -- otherwise a failed commit leaves the process # thinking the active session is "main" while the target @@ -369,11 +398,16 @@ async def delete_log_or_notebook_entry( "memories_deleted": memories_deleted, } else: # type == "notebook" - cursor = conn.execute( - "DELETE FROM notebook_entries WHERE name = ?", (target,) - ) - deleted = cursor.rowcount - conn.commit() + conn.execute("BEGIN IMMEDIATE") + try: + cursor = conn.execute( + "DELETE FROM notebook_entries WHERE name = ?", (target,) + ) + deleted = cursor.rowcount + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise if deleted > 0: memory.remove_active_notebook_entry(target) return { diff --git a/marm-mcp-server/marm_mcp_server/services/notebook.py b/marm-mcp-server/marm_mcp_server/services/notebook.py index 841348e3..b94f20f5 100644 --- a/marm-mcp-server/marm_mcp_server/services/notebook.py +++ b/marm-mcp-server/marm_mcp_server/services/notebook.py @@ -28,24 +28,29 @@ async def _add(name: Optional[str], data: Optional[str], **_) -> dict: platform = MARM_PLATFORM or None now = datetime.now(timezone.utc).isoformat() with memory.get_connection() as conn: - cursor = conn.execute( - """ - UPDATE notebook_entries - SET data = ?, embedding = ?, updated_at = ? - WHERE name = ? AND project IS ? AND platform IS ? - """, - (data, embedding_bytes, now, name, project, platform), - ) - if cursor.rowcount == 0: - conn.execute( + conn.execute("BEGIN IMMEDIATE") + try: + cursor = conn.execute( """ - INSERT INTO notebook_entries - (name, data, embedding, updated_at, project, platform) - VALUES (?, ?, ?, ?, ?, ?) + UPDATE notebook_entries + SET data = ?, embedding = ?, updated_at = ? + WHERE name = ? AND project IS ? AND platform IS ? """, - (name, data, embedding_bytes, now, project, platform), + (data, embedding_bytes, now, name, project, platform), ) - conn.commit() + if cursor.rowcount == 0: + conn.execute( + """ + INSERT INTO notebook_entries + (name, data, embedding, updated_at, project, platform) + VALUES (?, ?, ?, ?, ?, ?) + """, + (name, data, embedding_bytes, now, project, platform), + ) + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise await events.emit("notebook_entry_added", {"name": name, "data": data}) return { "status": "success", diff --git a/marm-mcp-server/tests/test_consolidation_write_time.py b/marm-mcp-server/tests/test_consolidation_write_time.py index 83e8c717..b90b79c9 100644 --- a/marm-mcp-server/tests/test_consolidation_write_time.py +++ b/marm-mcp-server/tests/test_consolidation_write_time.py @@ -54,8 +54,10 @@ async def test_update_memory_records_merge_history_in_metadata(tmp_path): @pytest.mark.asyncio async def test_update_memory_is_silent_for_missing_id(tmp_path): mem = MARMMemory(str(tmp_path / "memory.db")) - # Should not raise — returns None gracefully - await mem.update_memory("nonexistent-id", "some content") + # Should not raise -- returns False (no write happened) rather than + # raising or silently pretending the merge landed. + result = await mem.update_memory("nonexistent-id", "some content") + assert result is False @pytest.mark.asyncio @@ -254,6 +256,52 @@ async def mock_semantic_dup( assert len(metadata["merge_history"]) == 1 +@pytest.mark.asyncio +async def test_semantic_merge_falls_through_to_new_row_if_target_deleted_concurrently( + monkeypatch, tmp_path +): + """If the semantic-duplicate target is deleted or changed between + find_semantic_duplicate's read and _update_memory's write-lock + re-verification, _update_memory bails and returns False. _store_memory + must not silently report the (now-stale) existing_id as if the merge + succeeded -- it must store the content as a new memory instead, or the + second writer's content is lost with no trace.""" + from marm_mcp_server.core import memory as memory_module + from marm_mcp_server.core import memory_ops as memory_ops_module + + monkeypatch.setattr(memory_ops_module, "CONSOLIDATION_ENABLED", True) + mem = memory_module.MARMMemory(str(tmp_path / "memory.db")) + mem._encoder_failed = True + + first_id = await mem.store_memory("fixed the authentication bug", "session-a") + + async def mock_semantic_dup( + memory, content, session_name, threshold, query_vec=None + ): + # Simulate a concurrent delete landing between the duplicate check + # and _update_memory's own pre-read -- e.g. a racing marm_delete. + with mem.get_connection() as conn: + conn.execute("DELETE FROM memories WHERE id = ?", (first_id,)) + conn.commit() + return first_id + + monkeypatch.setattr(memory_ops_module, "find_semantic_duplicate", mock_semantic_dup) + + second_id = await mem.store_memory("auth error resolved in login flow", "session-a") + + assert second_id != first_id, ( + "content was silently dropped -- _store_memory reported the " + "deleted row's id as if the merge succeeded" + ) + + with mem.get_connection() as conn: + row = conn.execute( + "SELECT content FROM memories WHERE id = ?", (second_id,) + ).fetchone() + assert row is not None, "the second writer's content was never stored anywhere" + assert row[0] == "auth error resolved in login flow" + + @pytest.mark.asyncio async def test_dissimilar_content_stores_as_new_row(monkeypatch, tmp_path): from marm_mcp_server.core import memory_ops as memory_ops_module diff --git a/marm-mcp-server/tests/test_sqlite_write_atomicity.py b/marm-mcp-server/tests/test_sqlite_write_atomicity.py new file mode 100644 index 00000000..9d3f3c77 --- /dev/null +++ b/marm-mcp-server/tests/test_sqlite_write_atomicity.py @@ -0,0 +1,784 @@ +"""Rollback regression coverage for the SQLite write-atomicity hardening +effort (docs/current/sqlite-write-atomicity-hardening.md). + +Every test here forces a specific SQL statement inside an already-wrapped +BEGIN IMMEDIATE/COMMIT/ROLLBACK block to fail, then asserts the *earlier* +statement(s) in that same block did not durably apply -- proving the +transaction boundary is real, not just present. A test that only checks +the happy path does not prove rollback (see the spec's own Testing +Checklist). Coordinator-owned per the spec's Test Ownership section -- +packet agents did not add tests here. +""" + +import asyncio +import contextlib +import sqlite3 +import uuid +from datetime import datetime, timezone + +import pytest +from fastapi.testclient import TestClient + +from conftest import load_isolated_server, local_client + + +# --- shared failure-injection helpers --- + + +class _FailOnStatement: + """Wraps a real sqlite3 connection. conn.execute() raises the first + time it sees SQL containing `trigger` (case-insensitive substring + match); every other call -- before and after -- passes through to the + real connection untouched.""" + + def __init__(self, real, trigger): + self._real = real + self._trigger = trigger.upper() + self.fired = False + + def execute(self, sql, *args, **kwargs): + if not self.fired and isinstance(sql, str) and self._trigger in sql.upper(): + self.fired = True + raise sqlite3.OperationalError(f"forced failure: {self._trigger}") + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + +def _fail_on(monkeypatch, mem, trigger): + """Monkeypatch mem.get_connection so the next connection it hands out + raises the first time SQL containing `trigger` runs.""" + real_get_connection = mem.get_connection + + @contextlib.contextmanager + def _patched(): + with real_get_connection() as real_conn: + yield _FailOnStatement(real_conn, trigger) + + monkeypatch.setattr(mem, "get_connection", _patched) + + +class _RecordingConn: + """Wraps a real connection and appends a label to `events` the first + time SQL matching a key in `label_map` executes. Never raises -- + used to verify call *order*, not to force failures.""" + + def __init__(self, real, events, label_map): + self._real = real + self._events = events + self._label_map = label_map + + def execute(self, sql, *args, **kwargs): + if isinstance(sql, str): + upper = sql.upper() + for substr, label in self._label_map.items(): + if substr in upper: + self._events.append(label) + break + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + +def _record_statements(monkeypatch, mem, events, label_map): + real_get_connection = mem.get_connection + + @contextlib.contextmanager + def _patched(): + with real_get_connection() as real_conn: + yield _RecordingConn(real_conn, events, label_map) + + monkeypatch.setattr(mem, "get_connection", _patched) + + +# --- Packet A: services/log_entry.py --- + + +def test_create_log_entry_rollback_no_partial_row_on_second_statement_failure( + monkeypatch, tmp_path +): + """create_log_entry's final insert block: log_entries insert, then a + sessions upsert, then a guarded cache update. Force the sessions + upsert to fail after the log_entries insert already ran -- the whole + block must roll back, so no log_entries row should exist afterward.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + # load_isolated_server wipes and re-imports the whole marm_mcp_server + # package tree, so service modules must be imported *after* it runs -- + # importing earlier would capture a stale module the live app no + # longer uses, and the patch below would silently target nothing. + import marm_mcp_server.services.log_entry as log_entry + + # Seed the target session first so create_log_entry's explicit + # session_name path is used (skips the session-switch and + # dated-fallback blocks, which also do "INSERT INTO sessions" -- + # without this, the trigger could fire on the wrong block). + client.post( + "/marm_log_entry", + json={"session_name": "atomicity-a1", "entry": "2026-01-01-seed-first entry"}, + ) + + _fail_on(monkeypatch, log_entry.memory, "INSERT INTO sessions") + + resp = client.post( + "/marm_log_entry", + json={ + "session_name": "atomicity-a1", + "entry": "2026-01-02-forced-should not persist", + }, + ) + assert resp.json()["status"] == "error" + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + count = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE full_entry LIKE ?", + ("%should not persist%",), + ).fetchone()[0] + assert count == 0, "log_entries row survived a rolled-back transaction" + + +def test_whole_session_delete_rollback_keeps_rows_on_late_failure( + monkeypatch, tmp_path +): + """delete_log_or_notebook_entry's whole-session branch: sessions + delete, log_entries delete, guarded cache delete, memories delete. + Force the memories delete (the last statement) to fail -- sessions + and log_entries rows must still exist afterward.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + import marm_mcp_server.services.log_entry as log_entry + + client.post( + "/marm_log_entry", + json={"session_name": "atomicity-a2", "entry": "2026-01-01-seed-entry to keep"}, + ) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + before_sessions = conn.execute( + "SELECT COUNT(*) FROM sessions WHERE session_name = ?", ("atomicity-a2",) + ).fetchone()[0] + before_entries = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE session_name = ?", + ("atomicity-a2",), + ).fetchone()[0] + assert before_sessions == 1 + assert before_entries >= 1 + + _fail_on(monkeypatch, log_entry.memory, "DELETE FROM memories") + + resp = client.post("/marm_delete", json={"type": "log", "target": "atomicity-a2"}) + assert resp.json()["status"] == "error" + + with sqlite3.connect(db_path) as conn: + after_sessions = conn.execute( + "SELECT COUNT(*) FROM sessions WHERE session_name = ?", ("atomicity-a2",) + ).fetchone()[0] + after_entries = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE session_name = ?", + ("atomicity-a2",), + ).fetchone()[0] + assert after_sessions == before_sessions, "sessions row deleted despite rollback" + assert after_entries == before_entries, "log_entries row deleted despite rollback" + + +def test_marm_start_rollback_keeps_previous_active_session(monkeypatch, tmp_path): + """marm_start: UPDATE sessions SET marm_active = FALSE, then an + INSERT OR REPLACE for the requested session. Force the insert to + fail -- the previous active session's marm_active flag must not have + been cleared.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + first = client.post("/marm_start", json={"session_name": "atomicity-a3-old"}) + assert first.status_code == 200 + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + active_before = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a3-old",), + ).fetchone()[0] + assert active_before in (1, True) + + import marm_mcp_server.endpoints.session as session_endpoint + + _fail_on(monkeypatch, session_endpoint.memory, "INSERT OR REPLACE INTO sessions") + + second = client.post("/marm_start", json={"session_name": "atomicity-a3-new"}) + # marm_start's error contract differs from marm_log_entry/marm_delete's + # 200-with-error-dict shape: its except blocks raise HTTPException(500) + # rather than returning a dict. + assert second.status_code == 500 + + with sqlite3.connect(db_path) as conn: + active_after = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a3-old",), + ).fetchone()[0] + assert active_after in (1, True), ( + "previous active session was cleared even though the new " + "session's insert (and the whole transaction) failed" + ) + + +def test_session_switch_rollback_keeps_previous_active_session(monkeypatch, tmp_path): + """create_log_entry's session-switch block (triggered by a "Session: "/ + "Topic: " prefixed entry): UPDATE sessions SET marm_active = FALSE, + then an INSERT for the new session, then an INSERT for the marker log + entry. Force the marker insert (the third statement) to fail -- the + previously active session's marm_active flag must not have been + cleared, and neither the new session row nor the marker entry should + exist.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + first = client.post("/marm_start", json={"session_name": "atomicity-a4-old"}) + assert first.status_code == 200 + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + active_before = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a4-old",), + ).fetchone()[0] + assert active_before in (1, True) + + import marm_mcp_server.services.log_entry as log_entry + + _fail_on(monkeypatch, log_entry.memory, "INSERT INTO log_entries") + + resp = client.post( + "/marm_log_entry", json={"entry": "Session: atomicity-a4-new-topic"} + ) + assert resp.json()["status"] == "error" + + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + new_session = f"atomicity-a4-new-topic-{today}" + with sqlite3.connect(db_path) as conn: + active_after = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a4-old",), + ).fetchone()[0] + new_session_exists = conn.execute( + "SELECT COUNT(*) FROM sessions WHERE session_name = ?", (new_session,) + ).fetchone()[0] + marker_exists = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE session_name = ? AND topic = 'session_start'", + (new_session,), + ).fetchone()[0] + assert active_after in (1, True), ( + "previous active session was cleared despite the marker insert (and " + "the whole transaction) failing" + ) + assert new_session_exists == 0, "new session row survived a rolled-back transaction" + assert marker_exists == 0, ( + "session_start marker row survived a rolled-back transaction" + ) + + +def test_dated_fallback_rollback_keeps_previous_active_session(monkeypatch, tmp_path): + """create_log_entry's dated-fallback block (no session_name given and + memory.active_log_session is still "main"): UPDATE sessions SET + marm_active = FALSE, then an INSERT for the new dated session. Force + that insert to fail -- the previously active session's marm_active + flag must not have been cleared, and the dated fallback session row + must not exist.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + first = client.post("/marm_start", json={"session_name": "atomicity-a5-old"}) + assert first.status_code == 200 + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + active_before = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a5-old",), + ).fetchone()[0] + assert active_before in (1, True) + + import marm_mcp_server.services.log_entry as log_entry + + # First occurrence of "INSERT INTO sessions" in this call path is the + # dated-fallback block's own insert (the later per-entry upsert further + # down in create_log_entry is a second, distinct occurrence that never + # gets reached once this one raises). + _fail_on(monkeypatch, log_entry.memory, "INSERT INTO sessions") + + resp = client.post( + "/marm_log_entry", json={"entry": "plain entry with no session prefix"} + ) + assert resp.json()["status"] == "error" + + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + fallback_session = f"session-{today}" + with sqlite3.connect(db_path) as conn: + active_after = conn.execute( + "SELECT marm_active FROM sessions WHERE session_name = ?", + ("atomicity-a5-old",), + ).fetchone()[0] + fallback_exists = conn.execute( + "SELECT COUNT(*) FROM sessions WHERE session_name = ?", + (fallback_session,), + ).fetchone()[0] + assert active_after in (1, True), ( + "previous active session was cleared despite the dated-fallback " + "insert (and the whole transaction) failing" + ) + assert fallback_exists == 0, ( + "dated fallback session row survived a rolled-back transaction" + ) + + +def test_targeted_log_delete_rollback_keeps_entry_and_memory(monkeypatch, tmp_path): + """delete_log_or_notebook_entry's targeted (session_name given) log + branch: DELETE FROM log_entries, then a conditional DELETE FROM + memories for the dual-written semantic copy. Force the memories + delete to fail after the log_entries delete already ran -- both rows + must still exist afterward.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + create = client.post( + "/marm_log_entry", + json={ + "session_name": "atomicity-a6", + "entry": "2026-01-01-topic6-entry that must survive targeted delete", + }, + ) + assert create.json()["status"] == "success" + entry_id = create.json()["entry_id"] + memory_id = create.json()["memory_id"] + assert memory_id is not None, ( + "dual-write to semantic memory did not happen -- test can't exercise " + "the memories-delete branch without it" + ) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + before_entry = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE id = ?", (entry_id,) + ).fetchone()[0] + before_memory = conn.execute( + "SELECT COUNT(*) FROM memories WHERE id = ?", (memory_id,) + ).fetchone()[0] + assert before_entry == 1 + assert before_memory == 1 + + import marm_mcp_server.services.log_entry as log_entry + + _fail_on(monkeypatch, log_entry.memory, "DELETE FROM memories") + + resp = client.post( + "/marm_delete", + json={"type": "log", "target": "topic6", "session_name": "atomicity-a6"}, + ) + assert resp.json()["status"] == "error" + + with sqlite3.connect(db_path) as conn: + after_entry = conn.execute( + "SELECT COUNT(*) FROM log_entries WHERE id = ?", (entry_id,) + ).fetchone()[0] + after_memory = conn.execute( + "SELECT COUNT(*) FROM memories WHERE id = ?", (memory_id,) + ).fetchone()[0] + assert after_entry == before_entry, "log_entries row deleted despite rollback" + assert after_memory == before_memory, ( + "dual-written memory row deleted despite rollback" + ) + + +def test_notebook_delete_rollback_keeps_entry_on_commit_failure(monkeypatch, tmp_path): + """delete_log_or_notebook_entry's notebook branch: DELETE FROM + notebook_entries, then COMMIT. Force the commit itself to fail -- the + entry's row must survive, proving the delete didn't durably apply + without a successful commit.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "INSERT INTO notebook_entries (name, data, updated_at, project, platform) " + "VALUES (?, ?, ?, NULL, NULL)", + ( + "atomicity-a7-entry", + "must survive rollback", + "2026-01-01T00:00:00+00:00", + ), + ) + conn.commit() + + import marm_mcp_server.services.log_entry as log_entry + + _fail_on(monkeypatch, log_entry.memory, "COMMIT") + + resp = client.post( + "/marm_delete", json={"type": "notebook", "target": "atomicity-a7-entry"} + ) + assert resp.json()["status"] == "error" + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + "SELECT data FROM notebook_entries WHERE name = ?", + ("atomicity-a7-entry",), + ).fetchone() + assert row is not None, "notebook entry was deleted despite the commit failing" + assert row[0] == "must survive rollback" + + +# --- Packet B: services/notebook.py, services/documentation.py --- + + +def test_notebook_add_rollback_no_partial_update(monkeypatch, tmp_path): + """_add's update-then-conditional-insert is an UPDATE-XOR-INSERT + upsert -- only one of the two statements ever actually writes data + per call, so forcing the INSERT branch to fail proves nothing (a + single failing statement can't leave a partial row; there's no + earlier write in that branch to roll back). The real atomicity case + is the UPDATE branch: seed an existing row so the UPDATE performs a + real write, then force the commit itself to fail -- the row's + original data must survive, not the new data.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "INSERT INTO notebook_entries (name, data, updated_at, project, platform) " + "VALUES (?, ?, ?, NULL, NULL)", + ("atomicity-b1-existing", "original data", "2026-01-01T00:00:00+00:00"), + ) + conn.commit() + + import marm_mcp_server.services.notebook as notebook_service + + _fail_on(monkeypatch, notebook_service.memory, "COMMIT") + + resp = client.post( + "/marm_notebook", + json={ + "action": "add", + "name": "atomicity-b1-existing", + "data": "replacement data that must not persist", + }, + ) + assert resp.json()["status"] == "error" + + with sqlite3.connect(db_path) as conn: + data = conn.execute( + "SELECT data FROM notebook_entries WHERE name = ?", + ("atomicity-b1-existing",), + ).fetchone()[0] + assert data == "original data", ( + "notebook_entries row was updated despite the commit failing" + ) + + +def test_legacy_docs_cleanup_rollback_keeps_legacy_entries(monkeypatch, tmp_path): + """load_marm_documentation's legacy-cleanup block: a loop of DELETEs + over _LEGACY_SYSTEM_NOTEBOOK_NAMES, then an INSERT OR REPLACE marker + row. Force the marker insert to fail -- the legacy notebook rows + must still exist (the cleanup must not have partially applied).""" + load_isolated_server(monkeypatch, tmp_path) + + import marm_mcp_server.services.documentation as documentation + + db_path = tmp_path / "marm_memory.db" + legacy_name = "marm_protocol" + assert legacy_name in documentation._LEGACY_SYSTEM_NOTEBOOK_NAMES + with sqlite3.connect(db_path) as conn: + conn.execute( + "INSERT INTO notebook_entries (name, data, updated_at) VALUES (?, ?, ?)", + (legacy_name, "legacy payload", "2026-01-01T00:00:00+00:00"), + ) + conn.commit() + + _fail_on( + monkeypatch, + documentation.memory, + "INSERT OR REPLACE INTO user_settings", + ) + + with pytest.raises(sqlite3.OperationalError): + asyncio.run(documentation.load_marm_documentation()) + + with sqlite3.connect(db_path) as conn: + remaining = conn.execute( + "SELECT COUNT(*) FROM notebook_entries WHERE name = ?", (legacy_name,) + ).fetchone()[0] + marker = conn.execute( + "SELECT COUNT(*) FROM user_settings WHERE key = 'system_notebook_cleanup_v1'" + ).fetchone()[0] + assert remaining == 1, ( + "legacy notebook row was deleted despite the marker insert failing" + ) + assert marker == 0, "cleanup marker was written despite the transaction failing" + + +# --- Packet C: core/memory_ops.py --- + + +def test_update_memory_computes_embedding_before_acquiring_lock(monkeypatch, tmp_path): + """The whole point of the _update_memory fix: embedding generation + must complete before BEGIN IMMEDIATE runs, never during an open + transaction. Verified by call order, not just absence of a crash -- + a wrong order wouldn't raise, it would just silently reintroduce the + stall risk this fix exists to close.""" + from marm_mcp_server.core.memory import MARMMemory + from marm_mcp_server.core.memory_ops import _update_memory + + mem = MARMMemory(str(tmp_path / "update-order.db")) + mem._encoder_failed = False + mem.encoder = None + + memory_id = str(uuid.uuid4()) + with mem.get_connection() as conn: + conn.execute( + "INSERT INTO memories (id, session_name, content, content_hash, timestamp) " + "VALUES (?, ?, ?, ?, ?)", + (memory_id, "s1", "original content", "hash1", "2026-01-01T00:00:00+00:00"), + ) + conn.commit() + + events = [] + + def _fake_load_encoder_lazily(): + return True + + def _fake_encode_sync(text): + events.append("encode") + return [0.1, 0.2, 0.3] + + monkeypatch.setattr(mem, "_load_encoder_lazily", _fake_load_encoder_lazily) + monkeypatch.setattr(mem, "_encode_sync", _fake_encode_sync) + _record_statements(monkeypatch, mem, events, {"BEGIN IMMEDIATE": "begin_immediate"}) + + asyncio.run(_update_memory(mem, memory_id, "appended content")) + + assert events == ["encode", "begin_immediate"], ( + f"expected embedding computation before BEGIN IMMEDIATE, got {events}" + ) + + +def test_update_memory_rollback_no_partial_content_or_chunk_update( + monkeypatch, tmp_path +): + """_update_memory's write block: UPDATE memories, then a folded-in + DELETE FROM memory_chunks. Force the chunk delete to fail -- the + memory row's content must remain the pre-merge original, and any + pre-existing chunk rows must survive untouched.""" + from marm_mcp_server.core.memory import MARMMemory + from marm_mcp_server.core.memory_ops import _update_memory + + mem = MARMMemory(str(tmp_path / "update-rollback.db")) + mem._encoder_failed = True + + memory_id = str(uuid.uuid4()) + with mem.get_connection() as conn: + conn.execute( + "INSERT INTO memories (id, session_name, content, content_hash, timestamp) " + "VALUES (?, ?, ?, ?, ?)", + ( + memory_id, + "s1", + "original content before merge", + "hash-original", + "2026-01-01T00:00:00+00:00", + ), + ) + conn.execute( + "INSERT INTO memory_chunks (memory_id, chunk_index, chunk_text, embedding) " + "VALUES (?, 0, 'pre-existing chunk', ?)", + (memory_id, b"\x00\x01"), + ) + conn.commit() + + _fail_on(monkeypatch, mem, "DELETE FROM memory_chunks") + + with pytest.raises(sqlite3.OperationalError): + asyncio.run(_update_memory(mem, memory_id, "new content to append")) + + with mem.get_connection() as conn: + row = conn.execute( + "SELECT content, content_hash FROM memories WHERE id = ?", (memory_id,) + ).fetchone() + chunk_count = conn.execute( + "SELECT COUNT(*) FROM memory_chunks WHERE memory_id = ?", (memory_id,) + ).fetchone()[0] + + assert row[0] == "original content before merge", ( + "memory content was partially updated despite the chunk delete failing" + ) + assert row[1] == "hash-original" + assert chunk_count == 1, "pre-existing chunk row was lost despite the rollback" + + +def test_console_replace_memory_rollback_regression(monkeypatch, tmp_path): + """_replace_memory already had BEGIN IMMEDIATE before this hardening + effort (confirmed during Packet C's audit) -- this is a regression + test proving that pre-existing atomicity still holds, not proof of a + newly-added transaction. Force the compaction_staging stale-marking + update (the second statement) to fail after the content UPDATE ran -- + the memory's content must remain unchanged.""" + concept_db_path = tmp_path / "marm_index.db" + monkeypatch.setenv("MARM_CONCEPT_DB_PATH", str(concept_db_path)) + server = load_isolated_server( + monkeypatch, tmp_path, api_key="test-key", write_queue_enabled=True + ) + headers = {"Authorization": "Bearer test-key"} + + with TestClient(server.app) as client: + create = client.post( + "/internal/memories", + headers=headers, + json={ + "content": "original console memory content", + "session_name": "atomicity-c3", + "context_type": "decision", + }, + ) + assert create.status_code == 201 + memory_id = create.json()["id"] + + import marm_mcp_server.core.memory as memory_module + + _fail_on(monkeypatch, memory_module.memory, "UPDATE compaction_staging") + + # console_replace_memory's endpoint only catches RuntimeError, and + # TestClient re-raises unhandled server exceptions by default + # (rather than turning them into a 500 response) -- the forced + # sqlite3.OperationalError propagates all the way out here. The + # point of this test is the DB-level rollback, not this + # endpoint's error-response contract. + with pytest.raises(sqlite3.OperationalError): + client.put( + f"/internal/memories/{memory_id}", + headers=headers, + json={ + "content": "replacement content that must not persist", + "session_name": "atomicity-c3", + "context_type": "decision", + }, + ) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + content = conn.execute( + "SELECT content FROM memories WHERE id = ?", (memory_id,) + ).fetchone()[0] + assert content == "original console memory content", ( + "memory content was replaced despite the transaction's second " + "statement failing -- _replace_memory's existing atomicity regressed" + ) + + +def test_stage_compaction_summaries_processes_candidates_independently( + monkeypatch, tmp_path +): + """endpoints/compaction.py audit (no code changes made there): every + per-candidate mutation in marm_stage_compaction_summaries is a single + UPDATE statement, so there's no multi-statement atomicity gap to + wrap -- and wrapping the whole per-request loop in one transaction + would be a *regression*, not a hardening, since each candidate is + designed to succeed or fail independently. Proves that intended + semantic directly: one expired candidate in the same request as one + valid candidate must not block the valid one from staging.""" + server = load_isolated_server(monkeypatch, tmp_path) + client = local_client(server.app) + + db_path = tmp_path / "marm_memory.db" + with sqlite3.connect(db_path) as conn: + for mem_id, content in ( + ("mem-valid", "valid source"), + ("mem-expired", "expired source"), + ): + conn.execute( + "INSERT INTO memories (id, session_name, content, content_hash, timestamp) " + "VALUES (?, ?, ?, ?, ?)", + (mem_id, "atomicity-c4", content, mem_id, "2026-01-01T00:00:00+00:00"), + ) + conn.execute( + """ + INSERT INTO compaction_staging ( + id, session_name, source_memory_ids, preview, suggested_summary, + status, candidate_hash, source_updated_at_snapshot, + expires_at, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "candidate-valid", + "atomicity-c4", + '["mem-valid"]', + "valid source", + "", + "pending_summary", + "hash-valid", + "{}", + "2099-01-01T00:00:00+00:00", + "2026-01-01T00:00:00+00:00", + "2026-01-01T00:00:00+00:00", + ), + ) + conn.execute( + """ + INSERT INTO compaction_staging ( + id, session_name, source_memory_ids, preview, suggested_summary, + status, candidate_hash, source_updated_at_snapshot, + expires_at, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "candidate-expired", + "atomicity-c4", + '["mem-expired"]', + "expired source", + "", + "pending_summary", + "hash-expired", + "{}", + "2020-01-01T00:00:00+00:00", # already past + "2026-01-01T00:00:00+00:00", + "2026-01-01T00:00:00+00:00", + ), + ) + conn.commit() + + resp = client.post( + "/marm_stage_compaction_summaries", + json={ + "summaries": [ + { + "candidate_id": "candidate-valid", + "suggested_summary": "a real summary", + }, + { + "candidate_id": "candidate-expired", + "suggested_summary": "should not matter, candidate is expired", + }, + ] + }, + ) + results = {r["candidate_id"]: r for r in resp.json()["results"]} + assert results["candidate-valid"]["status"] == "summary_staged", ( + "the valid candidate was blocked by the expired one being in the " + "same request -- per-candidate independence regressed" + ) + assert results["candidate-expired"]["status"] == "error" + + with sqlite3.connect(db_path) as conn: + valid_status = conn.execute( + "SELECT status FROM compaction_staging WHERE id = ?", ("candidate-valid",) + ).fetchone()[0] + expired_status = conn.execute( + "SELECT status FROM compaction_staging WHERE id = ?", + ("candidate-expired",), + ).fetchone()[0] + assert valid_status == "summary_staged" + assert expired_status == "stale" diff --git a/marm-mcp-server/tests/test_stdio_transport.py b/marm-mcp-server/tests/test_stdio_transport.py index f0c5b9e0..d2c3cfca 100644 --- a/marm-mcp-server/tests/test_stdio_transport.py +++ b/marm-mcp-server/tests/test_stdio_transport.py @@ -323,13 +323,20 @@ def test_stdio_delete_whole_session_keeps_active_session_if_commit_fails( monkeypatch, tmp_path ): """PR #94 CodeRabbit finding: active_log_session must flip to "main" - only after the whole-session delete's conn.commit() actually succeeds - -- otherwise a failed commit leaves the runtime pointer saying "main" + only after the whole-session delete's commit actually succeeds -- + otherwise a failed commit leaves the runtime pointer saying "main" while the target session's rows are still intact in the DB. - sqlite3.Connection is an immutable C type (can't monkeypatch .commit - on it directly), so this wraps the real connection in a thin proxy - that forwards everything except commit(), which raises.""" + services/log_entry.py now commits via an explicit + conn.execute("BEGIN IMMEDIATE") / conn.execute("COMMIT") / + conn.execute("ROLLBACK") transaction (SQLite write-atomicity + hardening) instead of relying on ConnectionContext's implicit + commit-on-clean-exit -- so the forced failure has to intercept the + execute("COMMIT") call specifically, not the .commit() DB-API method + (sqlite3.Connection is also an immutable C type, so .commit() can't + be monkeypatched directly anyway). This wraps the real connection in + a thin proxy that forwards everything except an execute("COMMIT"), + which raises.""" import contextlib import sqlite3 @@ -343,6 +350,11 @@ class _CommitFailsConn: def __init__(self, real): self._real = real + def execute(self, sql, *args, **kwargs): + if isinstance(sql, str) and sql.strip().upper() == "COMMIT": + raise sqlite3.OperationalError("disk I/O error (forced)") + return self._real.execute(sql, *args, **kwargs) + def commit(self): raise sqlite3.OperationalError("disk I/O error (forced)")