fix(sync): recover a crash between file write and publish instead of flagging a conflict

The startup recovery sweep (:448) re-drives note_content rows stuck in
'writing'/'pending'. A crash after the file write but before the DB publish
leaves the row 'writing' with the correct accepted content already on disk and
file_checksum still None (new note) or the pre-write checksum (update). On
re-drive the write guard saw a present file whose checksum did not match the
expected previous checksum and raised RuntimeFileConflictError, so recovery
marked the row 'external_change_detected' and never advanced file_version —
recovery only actually worked when the file had never been written, arguably the
rarer crash location.

Add a same-content short-circuit to write_prepared_note_to_content_store: read
the on-disk checksum once, and if it equals the checksum of the accepted content
we would write, skip the redundant write and return the existing file's state so
the publisher advances file_version and marks the row 'synced'. Only raise a
conflict when the present file matches neither the accepted content nor the
expected previous checksum (a genuine external edit). Extract the shared
runtime_file_conflict helper so assert_runtime_file_matches_expected and the
write path apply identical conflict logic from a single checksum read.

Adds a regression test seeding a stuck 'writing' row whose accepted content is
already on disk and asserting recovery reaches 'synced' with file_version
advanced. Found by adversarial review of the recovery sweep.

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/cloud/note_content_materialization.py

@@ -129,12 +129,15 @@ async def drain_pending_materializations() -> None:
 # --- Startup Recovery ---
 # accept_write marks note_content "pending", then the materialization preflight
 # flips it to "writing" before the file is written and the publisher records
-# "synced". If the process dies between those points the row is stuck forever and
-# the source-of-truth markdown file is never written. On the next startup we
-# re-drive every stuck row. The db_version compare-and-set guard in the preflight
-# and publisher makes this unconditionally safe: an older recovery attempt can
-# never overwrite a newer accepted write or its file, so recovery re-materializes
-# without first checking whether some other writer already caught up.
+# "synced". If the process dies anywhere between those points the row is stuck
+# forever: the crash may land before the file write (nothing on disk) or after it
+# but before publish (the correct accepted file is already on disk, row still
+# "writing"). On the next startup we re-drive every stuck row. The write path
+# short-circuits when the accepted content is already on disk, so the
+# crash-after-write case publishes to "synced" instead of tripping the
+# external-change guard. The db_version compare-and-set guard in the preflight and
+# publisher makes recovery unconditionally safe: an older recovery attempt can
+# never overwrite a newer accepted write or its file.
 
 # Synthetic provenance stamped on recovered writes so operators can tell a
 # crash-recovery materialization apart from a normal accept-path write in logs

src/basic_memory/runtime/note_content.py

@@ -1173,19 +1173,37 @@ async def read_runtime_file_checksum(
     return await reader.compute_checksum(file_path)
 
 
+def runtime_file_conflict(
+    actual_checksum: RuntimeFileChecksum | None,
+    expected_checksum: RuntimeFileChecksum | None,
+    file_path: RuntimeFilePath,
+) -> RuntimeFileConflict | None:
+    """Return a conflict when a present file does not match the expected checksum.
+
+    An absent file (``actual_checksum is None``) never conflicts. A present file
+    conflicts unless the caller expected exactly that checksum — a ``None``
+    expectation (a fresh note that assumes no file) always conflicts with a
+    present file.
+    """
+    if actual_checksum is None:
+        return None
+    if expected_checksum is None or actual_checksum != expected_checksum:
+        return RuntimeFileConflict(
+            file_path=file_path,
+            expected_checksum=expected_checksum,
+            actual_checksum=actual_checksum,
+        )
+    return None
+
+
 async def assert_runtime_file_matches_expected(
     reader: RuntimeFileChecksumReader,
     expected: RuntimeExpectedFileState,
 ) -> None:
     """Raise when a guarded write would overwrite an unexpected runtime file."""
     actual_checksum = await read_runtime_file_checksum(reader, expected.file_path)
-    if actual_checksum is None:
-        return
-    if expected.expected_checksum is None or actual_checksum != expected.expected_checksum:
-        raise RuntimeFileConflictError(
-            RuntimeFileConflict(
-                file_path=expected.file_path,
-                expected_checksum=expected.expected_checksum,
-                actual_checksum=actual_checksum,
-            )
-        )
+    conflict = runtime_file_conflict(
+        actual_checksum, expected.expected_checksum, expected.file_path
+    )
+    if conflict is not None:
+        raise RuntimeFileConflictError(conflict)

src/basic_memory/runtime/note_materialization.py

@@ -6,11 +6,13 @@
 from datetime import datetime
 from typing import Protocol
 
+from basic_memory.file_utils import compute_checksum
 from basic_memory.runtime.note_content import (
-    RuntimeExpectedFileState,
     RuntimeFileChecksumReader,
+    RuntimeFileConflictError,
     RuntimeNoteMaterializationJobRequest,
-    assert_runtime_file_matches_expected,
+    read_runtime_file_checksum,
+    runtime_file_conflict,
 )
 from basic_memory.runtime.storage import RuntimeFileChecksum, RuntimeFilePath
 from basic_memory.runtime.note_object_metadata import RuntimeNoteObjectMetadata
@@ -92,13 +94,34 @@ async def write_prepared_note_to_content_store(
     prepared_write: RuntimePreparedNoteWrite,
 ) -> RuntimeWrittenFileState:
     """Write one prepared accepted note after checking the expected file state."""
-    await assert_runtime_file_matches_expected(
-        content_store,
-        RuntimeExpectedFileState(
+    accepted_checksum = await compute_checksum(prepared_write.markdown_content)
+    actual_checksum = await read_runtime_file_checksum(content_store, prepared_write.file_path)
+
+    # Trigger: the correct accepted content is already on disk — e.g. a crash
+    #   after the file write but before the DB publish left the note_content row
+    #   'writing', and startup recovery re-drives the same write.
+    # Why: the previous_file_checksum guard would otherwise misread that
+    #   already-correct file as an external conflict (previous_file_checksum is
+    #   None for a new note, or the pre-write checksum for an update), stranding
+    #   the row in 'external_change_detected' and never advancing file_version.
+    # Outcome: skip the redundant write and return the existing file's state so
+    #   the publisher advances file_version and marks the row 'synced'.
+    if actual_checksum is not None and actual_checksum == accepted_checksum:
+        file_metadata = await content_store.get_file_metadata(prepared_write.file_path)
+        return RuntimeWrittenFileState(
             file_path=prepared_write.file_path,
-            expected_checksum=prepared_write.previous_file_checksum,
-        ),
+            file_checksum=actual_checksum,
+            file_updated_at=file_metadata.modified_at,
+        )
+
+    # A present file that matches neither the accepted content nor the expected
+    # previous checksum is a genuine external edit: refuse to overwrite it.
+    conflict = runtime_file_conflict(
+        actual_checksum, prepared_write.previous_file_checksum, prepared_write.file_path
     )
+    if conflict is not None:
+        raise RuntimeFileConflictError(conflict)
+
     file_checksum = await content_store.write_file(
         prepared_write.file_path,
         prepared_write.markdown_content,

tests/cloud/test_note_content_materialization.py

@@ -547,6 +547,54 @@ async def test_recover_stuck_materializations_does_not_overwrite_unexpected_file
     assert row.file_write_status == "external_change_detected"
 
 
+@pytest.mark.asyncio
+async def test_recover_stuck_materializations_publishes_already_written_file(
+    session_maker,
+    test_project: Project,
+    sample_entity,
+    file_service: FileService,
+) -> None:
+    """A crash after the file write but before publish is recovered, not a conflict.
+
+    The row is left 'writing' with file_checksum None while the correct accepted
+    content is already on disk. Recovery must recognise the same content, skip the
+    redundant write, and publish to 'synced' instead of raising an external-change
+    conflict — the arguably-more-common crash location the guard alone mishandles.
+    """
+    markdown_content = "# Recovered\n\nThe file was written before the crash.\n"
+    await _seed_stuck_note_content(
+        session_maker,
+        project_id=test_project.id,
+        entity_id=sample_entity.id,
+        markdown_content=markdown_content,
+        db_version=1,
+        db_checksum="db-checksum-1",
+        file_write_status="writing",
+    )
+    # Simulate the crash-after-write-before-publish window: the accepted content
+    # is already on disk while the row still reads 'writing' (file_checksum None).
+    target = file_service.base_path / sample_entity.file_path
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(markdown_content, encoding="utf-8")
+
+    recovered = await recover_stuck_materializations(
+        session_maker=session_maker,
+        file_service=file_service,
+        project_id=test_project.id,
+    )
+
+    assert recovered == 1
+    assert target.read_text(encoding="utf-8") == markdown_content
+
+    repository = NoteContentRepository(project_id=test_project.id)
+    async with db.scoped_session(session_maker) as session:
+        row = await repository.get_by_entity_id(session, sample_entity.id)
+    assert row is not None
+    assert row.file_write_status == "synced"
+    assert row.file_checksum is not None
+    assert row.file_version == 1
+
+
 @pytest.mark.asyncio
 async def test_run_recovery_materialization_does_not_revert_newer_accepted_version(
     session_maker,