feat(sync): recover materializations stuck in writing/pending on startup

An accepted local note write returns before its markdown file is written:
accept_write marks note_content "pending", the materialization preflight
flips it to "writing", and the publisher records "synced" only after the
file lands. A process crash between those points leaves the row stuck
forever and the source-of-truth file never (re)materialized.

Add a startup recovery sweep that re-drives these rows:

- NoteContentRepository.find_stuck_materializations() returns project-scoped
  rows whose file_write_status is writing/pending.
- recover_stuck_materializations() rebuilds a RuntimeNoteMaterializationJobRequest
  from each stuck row's own db_version/db_checksum and runs it through the
  standard preflight/writer/publisher/status-publisher via
  run_recovery_materialization(). The db_version compare-and-set guard makes
  this safe: an older recovery attempt can never revert a newer accepted write
  or clobber an unexpected file. Non-fatal per row so one poisoned note cannot
  block startup for the rest of the project.
- initialize_file_indexing runs recover_project_materializations() once per
  locally-syncable project before the watch service serves, so files are
  re-written ahead of the initial project index scan.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/cloud/note_content_materialization.py

@@ -30,6 +30,7 @@
     RuntimeAcceptedNoteChange,
     RuntimeAcceptedNoteResponse,
     RuntimeNoteContentResponsePayload,
+    RuntimeNoteMaterializationJobRequest,
     RuntimeNoteMaterializationResult,
     RuntimeNoteMaterializationStatus,
     plan_accepted_note_response,
@@ -125,6 +126,107 @@ async def drain_pending_materializations() -> None:
     await _materialization_pool.join()
 
 
+# --- 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.
+
+# Synthetic provenance stamped on recovered writes so operators can tell a
+# crash-recovery materialization apart from a normal accept-path write in logs
+# and object metadata.
+RECOVERY_NOTE_CHANGE_SOURCE = "note-content-materialization-recovery"
+RECOVERY_NOTE_ACTOR_NAME = "startup-recovery"
+
+
+async def run_recovery_materialization(
+    request: RuntimeNoteMaterializationJobRequest,
+    *,
+    session_maker: async_sessionmaker[AsyncSession],
+    file_service: FileService,
+) -> RuntimeNoteMaterializationResult:
+    """Re-drive one stuck materialization through the standard guarded write path.
+
+    Uses the same preflight/writer/publisher/status-publisher as an accept-path
+    write, so the db_version and file-conflict guards apply unchanged. No cleanup
+    paths: a recovery request carries no old-file move, so nothing is deleted.
+    """
+    storage = LocalNoteContentStorage(file_service)
+    return await run_note_materialization(
+        request,
+        preflight=RepositoryNoteMaterializationPreflight(session_maker=session_maker),
+        writer=ContentStoreNoteMaterializationFileWriter(storage),
+        publisher=RepositoryNoteMaterializationPublisher(session_maker=session_maker),
+        status_publisher=RepositoryNoteMaterializationStatusPublisher(session_maker=session_maker),
+        cleanup_enqueuer=InlineNoteFileDeleteEnqueuer(storage),
+    )
+
+
+async def recover_stuck_materializations(
+    *,
+    session_maker: async_sessionmaker[AsyncSession],
+    file_service: FileService,
+    project_id: int,
+) -> int:
+    """Re-drive every note materialization stuck in writing/pending for a project.
+
+    Meant to run once per project at startup, before serving. Non-fatal per row:
+    a single row that raises is logged and skipped so one poisoned note cannot
+    block startup recovery for the rest of the project. Returns the number of rows
+    that reached a written file state.
+    """
+    async with db.scoped_session(session_maker) as session:
+        stuck_rows = await NoteContentRepository(
+            project_id=project_id
+        ).find_stuck_materializations(session)
+
+    if not stuck_rows:
+        return 0
+
+    logger.info(
+        "Recovering stuck note materializations",
+        project_id=project_id,
+        stuck_count=len(stuck_rows),
+    )
+    recovered = 0
+    for row in stuck_rows:
+        # Rebuild the queue request from the row's own accepted db_version/db_checksum
+        # so the preflight guard matches the current accepted state; if a newer write
+        # has since advanced the row, the guard trips and this attempt no-ops.
+        request = RuntimeNoteMaterializationJobRequest(
+            project_id=project_id,
+            entity_id=row.entity_id,
+            db_version=int(row.db_version),
+            db_checksum=str(row.db_checksum),
+            actor_name=RECOVERY_NOTE_ACTOR_NAME,
+            source=RECOVERY_NOTE_CHANGE_SOURCE,
+        )
+        try:
+            result = await run_recovery_materialization(
+                request,
+                session_maker=session_maker,
+                file_service=file_service,
+            )
+        except Exception:
+            # Trigger: one row's materialization raised (storage/DB error).
+            # Why: recovery is best-effort startup cleanup; the version guard makes
+            # a later retry safe, so one bad row must not abort the whole sweep.
+            # Outcome: log and continue to the next stuck row.
+            logger.exception(
+                "Failed to recover stuck note materialization",
+                project_id=project_id,
+                entity_id=row.entity_id,
+            )
+            continue
+        if result.status is RuntimeNoteMaterializationStatus.written:
+            recovered += 1
+    return recovered
+
+
 def note_content_payload_file_path(
     payload: RuntimeNoteContentResponsePayload,
 ) -> RuntimeFilePath | None:

src/basic_memory/repository/note_content_repository.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Mapping, Optional, cast
+from typing import Any, Mapping, Optional, Sequence, cast
 
 from sqlalchemy import select, update
 from sqlalchemy.engine import CursorResult
@@ -326,6 +326,23 @@ async def update_state_fields(
             raise ValueError(f"Can't find NoteContent for entity {entity_id} after update")
         return updated
 
+    async def find_stuck_materializations(
+        self, session: AsyncSession
+    ) -> Sequence[NoteContent]:
+        """Return accepted notes whose file write never completed.
+
+        ``accept_write`` marks a row ``pending`` and the materialization preflight
+        flips it to ``writing`` before the file is written and the publisher records
+        ``synced``/``written``. A process crash between those points leaves the row
+        stuck in ``writing``/``pending`` forever and the source-of-truth markdown
+        file never (re)materialized. Startup recovery re-drives exactly these rows;
+        the db_version compare-and-set guard in the publisher keeps the retry safe
+        (an older recovery attempt can never revert a newer accepted write).
+        """
+        query = self.select().where(NoteContent.file_write_status.in_(("writing", "pending")))
+        result = await session.execute(query)
+        return list(result.scalars().all())
+
     async def delete_by_entity_id(self, session: AsyncSession, entity_id: int) -> bool:
         """Delete note_content by entity identifier."""
         note_content = await self.select_by_id(session, entity_id)

src/basic_memory/services/initialization.py

@@ -7,6 +7,7 @@
 import asyncio
 import os
 import sys
+from pathlib import Path
 from typing import TYPE_CHECKING, Protocol
 
 
@@ -20,6 +21,8 @@
 )
 
 if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
     from basic_memory.index.local_project import LocalProjectIndexRuntime
 
 
@@ -51,6 +54,43 @@ async def run_initial_project_index(
     )
 
 
+async def recover_project_materializations(
+    project: Project,
+    session_maker: "async_sessionmaker[AsyncSession]",
+) -> None:
+    """Re-drive note materializations a crash left stuck for one project.
+
+    A local note write returns before its markdown file is written; a crash
+    between accepting the DB snapshot and writing the file leaves note_content
+    stuck in writing/pending and the source-of-truth file missing. Runs once at
+    startup, before the watch service serves, so the file is (re)written and then
+    picked up by the initial project index. Non-fatal: a recovery failure must not
+    block startup, so it is logged and startup continues.
+    """
+    from basic_memory.cloud.note_content_materialization import recover_stuck_materializations
+    from basic_memory.services.file_service import FileService
+
+    try:
+        # FileService needs only base_path to write the accepted markdown bytes;
+        # the markdown_processor/app_config are unused on the materialization path.
+        file_service = FileService(Path(project.path))
+        recovered = await recover_stuck_materializations(
+            session_maker=session_maker,
+            file_service=file_service,
+            project_id=project.id,
+        )
+        if recovered:
+            logger.info(
+                "Recovered stuck note materializations on startup",
+                project=project.name,
+                recovered=recovered,
+            )
+    except Exception as e:  # pragma: no cover - defensive startup guard
+        logger.error(
+            f"Error recovering stuck materializations for project {project.name}: {e}"
+        )
+
+
 async def initialize_database(app_config: BasicMemoryConfig) -> None:
     """Initialize database with migrations handled automatically by get_or_create_db.
 
@@ -177,6 +217,12 @@ async def initialize_file_indexing(
         active_projects = [p for p in active_projects if p.name not in skip]
         logger.info(f"Skipping projects that are not locally indexable: {skip}")
 
+    # Recover materializations a crash left stuck (writing/pending) before serving.
+    # Runs synchronously here so the source-of-truth files are re-written before the
+    # initial project index scans them; bounded by the count of stuck rows.
+    for project in active_projects:
+        await recover_project_materializations(project, session_maker)
+
     # Start indexing for all projects as background tasks (non-blocking)
     async def index_project_background(project: Project):
         """Index a single project in the background."""