fix(core): remove entity read locks from materialization publish

src/basic_memory/indexing/note_materialization_runner.py

@@ -57,6 +57,7 @@
     RuntimeFilePath,
 )
 from basic_memory.models import Entity, NoteContent
+from basic_memory.repository.entity_repository import EntityRepository
 from basic_memory.repository.note_file_vacate_repository import NoteFileVacateRepository
 
 type NoteMaterializationPreflightOutcome = (
@@ -427,25 +428,31 @@ async def publish_written_file_state(
                 entity_id=request.entity_id,
             )
 
-            # Materialization publishes NoteContent and Entity in one transaction.
-            # The canonical order lives in current_relation_generation_statement:
-            # claim NoteContent before any Entity lock so publication, accepted
-            # mutation, deletion, and materialization cannot form a lock cycle.
+            # These are plain snapshots for planning. This transaction's first row
+            # lock is the NoteContent CAS UPDATE, preserving the canonical order in
+            # current_relation_generation_statement. With no read locks held, the
+            # publisher cannot anchor the lock cycle reported in #1224.
+            #
+            # Everything this publisher writes — the file on disk, Entity
+            # mtime/size, NoteContent file lineage — is derived, eventually
+            # consistent state. A concurrent accepted write or move may
+            # invalidate these snapshots at any point; when it does, the CAS
+            # below no-ops and the newer generation's own publish converges the
+            # projections. Do not "fix" an observed race here by reintroducing
+            # SELECT-time locks: every such lock rebuilds a #1224-class
+            # deadlock, while the drift it would prevent is transient and
+            # repaired by the next write's index pass.
             note_content = await session.scalar(
-                select(NoteContent)
-                .where(
+                select(NoteContent).where(
                     NoteContent.entity_id == request.entity_id,
                     NoteContent.project_id == request.project_id,
                 )
-                .with_for_update()
             )
             entity = await session.scalar(
-                select(Entity)
-                .where(
+                select(Entity).where(
                     Entity.id == request.entity_id,
                     Entity.project_id == request.project_id,
                 )
-                .with_for_update()
             )
             publish_plan = plan_written_note_materialization_publish(
                 request=request,
@@ -509,9 +516,20 @@ async def publish_written_file_state(
                 expected_db_version=expected_db_version,
             )
             if not applied:
-                # A newer accepted write (and its own materialization) superseded
-                # this one between our read and write; skip the stale file_version
-                # publish and the entity metadata update rather than reverting them.
+                # Trigger: a move can commit after the plain planning reads but
+                # before the CAS observes the newer NoteContent row.
+                # Why: that observation guarantees this later plain read sees the
+                # move's committed Entity path.
+                # Outcome: clean the just-written vacated path while keeping a
+                # same-path superseded write in place for its newer materialization.
+                current_entity = await session.scalar(
+                    select(Entity)
+                    .where(
+                        Entity.id == request.entity_id,
+                        Entity.project_id == request.project_id,
+                    )
+                    .execution_options(populate_existing=True)
+                )
                 return RuntimeNoteMaterializationResult(
                     entity_id=request.entity_id,
                     status=RuntimeNoteMaterializationStatus.stale,
@@ -521,6 +539,31 @@ async def publish_written_file_state(
                     ),
                     file_path=written_file.file_path,
                     file_checksum=written_file.file_checksum,
+                    written_file_orphaned=(
+                        current_entity is None or current_entity.file_path != written_file.file_path
+                    ),
+                )
+
+            updated = await EntityRepository(request.project_id).update_fields(
+                session,
+                request.entity_id,
+                {
+                    "mtime": written_file.file_updated_at.timestamp(),
+                    "size": len(prepared_write.markdown_content.encode("utf-8")),
+                },
+            )
+            if not updated:
+                # Trigger: the guarded Entity update found no row after the CAS.
+                # Why: this is a portable fail-safe; on PostgreSQL the successful
+                # CAS holds NoteContent while every Entity metadata producer crosses
+                # that row first, so the miss is unreachable under row locking.
+                # Outcome: report the missing Entity without clearing its vacate path.
+                return RuntimeNoteMaterializationResult(
+                    entity_id=request.entity_id,
+                    status=RuntimeNoteMaterializationStatus.missing,
+                    reason=f"entity disappeared after file write: {request.entity_id}",
+                    file_path=written_file.file_path,
+                    file_checksum=written_file.file_checksum,
                 )
 
             # Trigger: this current materialization made its destination path live.
@@ -530,9 +573,6 @@ async def publish_written_file_state(
                 session,
                 file_path=written_file.file_path,
             )
-            entity.mtime = written_file.file_updated_at.timestamp()
-            entity.size = len(prepared_write.markdown_content.encode("utf-8"))
-            await session.flush()
             return publish_plan.result
 
 

src/basic_memory/repository/note_content_repository.py

@@ -80,13 +80,9 @@ async def _load_entity_identity(
         self,
         session: AsyncSession,
         entity_id: int,
-        *,
-        lock_for_update: bool = False,
     ) -> Entity:
         """Load the owning entity so duplicated identity fields stay aligned."""
         query = select(Entity).where(Entity.id == entity_id)
-        if lock_for_update:
-            query = query.with_for_update()
         result = await session.execute(query)
         entity = result.scalar_one_or_none()
         if entity is None:
@@ -304,14 +300,18 @@ async def update_state_fields(
             # from the entity (rather than mutating the ORM row) so the whole
             # write is the single conditional UPDATE whose rowcount decides the
             # race, portably across SQLite and Postgres.
-            # Materialization publishes NoteContent state and then updates Entity
-            # file metadata in the same transaction. Lock Entity first so that
-            # path cannot invert an Entity -> NoteContent indexing transaction.
-            entity = await self._load_entity_identity(
-                session,
-                entity_id,
-                lock_for_update=True,
-            )
+            # This identity read is deliberately unlocked. Reconciler and
+            # materialization callers hold no prior NoteContent claim, so an
+            # Entity lock here would invert the NoteContent-first order documented
+            # by current_relation_generation_statement and recreate the #1224
+            # deadlock. The conditional UPDATE rowcount is the only guard needed.
+            # A project-index move can repoint entity and note_content paths
+            # without advancing db_version, so this copy can lose that race and
+            # go briefly stale. That is accepted: the identity columns are a
+            # denormalized convenience, planning always prefers Entity.file_path,
+            # and every subsequent write refreshes the copy. Serializing here
+            # would trade a self-healing drift for a deadlock class.
+            entity = await self._load_entity_identity(session, entity_id)
             result = cast(
                 CursorResult[Any],
                 await session.execute(

test-int/test_note_materialization_lock_order.py

@@ -1,4 +1,4 @@
-"""Postgres regression coverage for NoteContent-Entity materialization lock order."""
+"""Postgres coverage that materialization publish cannot anchor a row-lock cycle."""
 
 from __future__ import annotations
 
@@ -132,8 +132,9 @@ async def test_materialization_and_accepted_mutation_share_note_content_first_or
                 )
             )
             await asyncio.wait_for(session_lock.started.wait(), timeout=2)
-            # Let PostgreSQL enqueue the materializer's NoteContent lock. With
-            # the old Entity-first order, the next lock request closes a cycle.
+            # The publisher holds no row locks while it waits at its NoteContent
+            # CAS, so this Entity lock probe cannot close a lock cycle. Keep the
+            # timeout as a loud regression if publish-span read locks return.
             await asyncio.sleep(0.1)
 
             locked_entity = await asyncio.wait_for(
@@ -149,7 +150,7 @@ async def test_materialization_and_accepted_mutation_share_note_content_first_or
                 await mutation_session.execute(
                     update(NoteContent)
                     .where(NoteContent.entity_id == entity_id)
-                    .values(file_path=locked_entity.file_path)
+                    .values(file_path=locked_entity.file_path, db_version=2)
                 )
             else:
                 locked_entity.title = "Accepted mutation completed"
@@ -179,6 +180,7 @@ async def test_materialization_and_accepted_mutation_share_note_content_first_or
         assert result.status is RuntimeNoteMaterializationStatus.stale
         assert result.written_file_orphaned
         assert note_content.file_path == "notes/moved-during-materialization.md"
+        assert note_content.db_version == 2
         assert note_content.file_version is None
         assert note_content.file_checksum == "previous-file-checksum"
         assert entity.file_path == "notes/moved-during-materialization.md"
@@ -191,3 +193,127 @@ async def test_materialization_and_accepted_mutation_share_note_content_first_or
         assert note_content.file_write_status == "synced"
         assert entity.mtime == written_file.file_updated_at.timestamp()
         assert entity.size == len(markdown.encode("utf-8"))
+
+
+@pytest.mark.asyncio
+async def test_publish_cas_loss_never_reverts_newer_accepted_write(
+    engine_factory,
+    test_project: Project,
+) -> None:
+    """A publisher blocked at CAS must preserve a newer same-path accepted write."""
+    engine, session_maker = engine_factory
+    if engine.dialect.name != "postgresql":
+        pytest.skip("row-lock ordering requires PostgreSQL")
+
+    file_path = "notes/cas-loss.md"
+    original_markdown = "# Original\n"
+    async with db.scoped_session(session_maker) as session:
+        entity = Entity(
+            project_id=test_project.id,
+            title="CAS loss",
+            note_type="note",
+            content_type="text/markdown",
+            file_path=file_path,
+            checksum="previous-file-checksum",
+        )
+        session.add(entity)
+        await session.flush()
+        entity_id = entity.id
+        original_mtime = entity.mtime
+        original_size = entity.size
+        await NoteContentRepository(project_id=test_project.id).create(
+            session,
+            NoteContent(
+                entity_id=entity_id,
+                markdown_content=original_markdown,
+                db_version=1,
+                db_checksum="original-db-checksum",
+                file_version=None,
+                file_checksum="previous-file-checksum",
+                file_write_status="writing",
+            ),
+        )
+
+    request = RuntimeNoteMaterializationJobRequest(
+        project_id=test_project.id,
+        entity_id=entity_id,
+        db_version=1,
+        db_checksum="original-db-checksum",
+        source="api",
+    )
+    prepared_write = plan_prepared_note_write(
+        request=request,
+        file_path=file_path,
+        markdown_content=original_markdown,
+        previous_file_checksum="previous-file-checksum",
+        attempted_at=datetime(2026, 8, 5, 2, 0, tzinfo=UTC),
+    )
+    written_file = RuntimeWrittenFileState(
+        file_path=file_path,
+        file_checksum="stale-materialized-checksum",
+        file_updated_at=datetime(2026, 8, 5, 2, 1, tzinfo=UTC),
+    )
+    session_lock = StartedMaterializationLock()
+    publisher = RepositoryNoteMaterializationPublisher(
+        session_maker=session_maker,
+        session_lock=session_lock,
+    )
+
+    publish_task: asyncio.Task[Any] | None = None
+    async with session_maker() as mutation_session:
+        await mutation_session.begin()
+        try:
+            await lock_accepted_note_content_for_entity_mutation(
+                mutation_session,
+                project_id=test_project.id,
+                entity_id=entity_id,
+            )
+
+            publish_task = asyncio.create_task(
+                publisher.publish_written_file_state(
+                    request,
+                    prepared_write,
+                    written_file,
+                )
+            )
+            await asyncio.wait_for(session_lock.started.wait(), timeout=2)
+            await asyncio.sleep(0.1)
+            assert not publish_task.done()
+
+            await mutation_session.execute(
+                update(NoteContent)
+                .where(NoteContent.entity_id == entity_id)
+                .values(
+                    db_version=2,
+                    db_checksum="newer-db-checksum",
+                    markdown_content="# Newer accepted write\n",
+                    file_write_status="pending",
+                )
+            )
+            await mutation_session.commit()
+
+            result = await asyncio.wait_for(publish_task, timeout=2)
+        finally:
+            if mutation_session.in_transaction():
+                await mutation_session.rollback()
+            if publish_task is not None and not publish_task.done():
+                publish_task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await publish_task
+
+    async with session_maker() as verification_session:
+        note_content = await verification_session.get(NoteContent, entity_id)
+        entity = await verification_session.get(Entity, entity_id)
+
+    assert result.status is RuntimeNoteMaterializationStatus.stale
+    assert result.written_file_orphaned is False
+    assert note_content is not None
+    assert note_content.db_version == 2
+    assert note_content.db_checksum == "newer-db-checksum"
+    assert note_content.markdown_content == "# Newer accepted write\n"
+    assert note_content.file_version is None
+    assert note_content.file_checksum == "previous-file-checksum"
+    assert note_content.file_write_status == "pending"
+    assert entity is not None
+    assert entity.mtime == original_mtime
+    assert entity.size == original_size