feat(indexing): relay self-supersede on stale base + db_version provenance (#1589)

Two changes for the basic-memory-cloud #1589 collab persistence redesign:

1. Relay self-supersede: the accepted-note base-checksum precondition now
   accepts a stale base when BOTH the current accepted version's last_source
   and the incoming request source are collaboration_relay. A relay persist
   that times out client-side after committing leaves the relay's recorded
   base one version behind its own write; rejecting that wedges every
   subsequent store in a 409 loop (observed live in cloud production
   2026-07-23: zero persists for 10+ minutes while the user typed). The
   relay's next snapshot always supersedes its own prior write - the live
   Y.Doc is the merge of everything the relay ever persisted. Foreign
   writers keep full guarded semantics; the deleted-entity 409 also stays.

2. db_version provenance on index results: RuntimeNoteObjectProvenance now
   parses bm-db-version alongside actor/source; it threads through
   IndexedFileLiveUpdatePlan, IndexFileJobResult, and
   IndexFileNoteLiveUpdatePlan (withheld for superseded content, like
   content_checksum) so cloud's index-completion live updates can carry the
   monotonic version. Consumers then decide echo-vs-out-of-band by version
   arithmetic instead of checksum guessing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013wFpjrd126rVe6VisiYVDD

Author: phernandez

src/basic_memory/indexing/accepted_note_mutation_runner.py

@@ -41,6 +41,7 @@
     plan_accepted_note_write_change,
 )
 from basic_memory.runtime.note_move import normalize_note_move_destination_path
+from basic_memory.runtime.note_object_metadata import NOTE_SOURCE_COLLABORATION_RELAY
 from basic_memory.runtime.storage import (
     NoteExternalId,
     ProjectExternalId,
@@ -607,7 +608,23 @@ async def _run_accepted_note_update(
             request.base_checksum is not None
             and current_note_content.db_checksum != request.base_checksum
         ):
-            reject_stale_base_checksum(current_db_checksum=current_note_content.db_checksum)
+            # Relay self-supersede (#1589): when the current accepted version was
+            # itself written by the collaboration relay AND this request is the
+            # relay again, a stale base can only mean a lost ack — a persist that
+            # timed out client-side after committing here. The relay's next
+            # snapshot always supersedes its own prior write (the live Y.Doc is
+            # the merge of everything the relay ever persisted), so rejecting
+            # would wedge every subsequent store against our own committed
+            # version (2026-07-23 production incident). Foreign writers keep the
+            # full guarded semantics.
+            relay_self_supersede = (
+                request.source == NOTE_SOURCE_COLLABORATION_RELAY
+                and current_note_content.last_source == NOTE_SOURCE_COLLABORATION_RELAY
+            )
+            if not relay_self_supersede:
+                reject_stale_base_checksum(
+                    current_db_checksum=current_note_content.db_checksum
+                )
         try:
             prepared_write = await prepare_accepted_note_replace(
                 preparer,

src/basic_memory/indexing/models.py

@@ -237,6 +237,11 @@ class IndexFileJobResult:
     actor_kind: str | None = None
     actor_name: str | None = None
     live_update_source: str | None = None
+    # Accepted DB version mirrored on the trusted object metadata (#1589):
+    # carried into note-level live updates so consumers can decide
+    # echo-vs-out-of-band by version arithmetic. None when the metadata was
+    # absent or failed checksum validation.
+    db_version: int | None = None
     # True when the object carried our own bm-file-checksum metadata that no
     # longer matches what this job indexed: a newer own-stack write landed
     # mid-job, so this result describes superseded content (issue #1445).
@@ -310,6 +315,9 @@ class IndexFileNoteLiveUpdatePlan:
     actor_user_profile_id: str | None = None
     actor_kind: RuntimeNoteActorKind | None = None
     actor_name: RuntimeNoteActorName | None = None
+    # Accepted DB version from trusted object metadata (#1589); None when the
+    # metadata was absent or failed checksum validation.
+    db_version: int | None = None
 
 
 DEFAULT_INDEX_FILE_NOTE_LIVE_UPDATE_SOURCE: RuntimeNoteChangeSource = "s3_webhook"
@@ -359,6 +367,7 @@ def index_file_job_result_from_indexed_file(
         live_update_source=(
             live_update_plan.live_update_source if live_update_plan is not None else None
         ),
+        db_version=live_update_plan.db_version if live_update_plan is not None else None,
         content_superseded=(
             live_update_plan.content_superseded if live_update_plan is not None else False
         ),
@@ -436,6 +445,9 @@ def plan_index_file_note_live_update(
         actor_user_profile_id=result.actor_user_profile_id,
         actor_kind=result.actor_kind,
         actor_name=result.actor_name,
+        # Superseded content also withholds the version: like content_checksum
+        # above, a stale version must not invite consumers to reconcile to it.
+        db_version=None if result.content_superseded else result.db_version,
     )
 
 
@@ -537,6 +549,8 @@ class IndexedFileLiveUpdatePlan:
     actor_user_profile_id: str | None = None
     actor_kind: str | None = None
     actor_name: str | None = None
+    # Trusted-branch only, like the actor fields (#1589).
+    db_version: int | None = None
     live_update_source: RuntimeNoteChangeSource | None = None
     operation: FileIndexOperation | None = None
 
@@ -600,6 +614,7 @@ def plan_current_materialized_note_result(
             actor_kind=provenance.actor_kind,
             actor_name=provenance.actor_name,
             live_update_source=provenance.source,
+            db_version=provenance.db_version,
         ),
         object_checksum_source=plan.object_checksum_source,
         object_checksum=plan.object_checksum,
@@ -658,6 +673,7 @@ def plan_indexed_file_live_update_metadata(
         actor_kind=provenance.actor_kind,
         actor_name=provenance.actor_name,
         live_update_source=provenance.source,
+        db_version=provenance.db_version,
         operation=file_index_operation_from_note_object_metadata(object_metadata),
     )
 

src/basic_memory/runtime/note_object_metadata.py

@@ -43,6 +43,9 @@
 VALID_NOTE_OBJECT_SOURCES: frozenset[RuntimeNoteChangeSource] = frozenset(
     {"api", "collaboration_relay", "mcp", "s3_webhook", "web_v2"}
 )
+# Named because the accepted-note write path special-cases relay writes: the
+# relay superseding its own prior write is never a real conflict (#1589).
+NOTE_SOURCE_COLLABORATION_RELAY: RuntimeNoteChangeSource = "collaboration_relay"
 _SAFE_ACTOR_NAME_CHARS = re.compile(r"[^A-Za-z0-9 ._()+/:-]+")
 _WHITESPACE = re.compile(r"\s+")
 _MAX_ACTOR_NAME_LENGTH = 120
@@ -227,6 +230,10 @@ class RuntimeNoteObjectProvenance:
     actor_kind: RuntimeNoteActorKind | None = None
     actor_name: RuntimeNoteActorName | None = None
     source: RuntimeNoteChangeSource | None = None
+    # The accepted DB version mirrored onto the object (#1589): lets index-path
+    # live updates carry the monotonic version so consumers can decide
+    # echo-vs-out-of-band by arithmetic instead of checksum comparison.
+    db_version: RuntimeNoteContentVersion | None = None
 
     @classmethod
     def from_object_metadata(cls, metadata: RuntimeNoteObjectMetadataMap | None) -> Self:
@@ -237,6 +244,7 @@ def from_object_metadata(cls, metadata: RuntimeNoteObjectMetadataMap | None) ->
             actor_kind=actor_kind,
             actor_name=actor_name,
             source=source_from_object_metadata(metadata),
+            db_version=db_version_from_object_metadata(metadata),
         )
 
 

tests/indexing/test_accepted_note_mutation_runner.py

@@ -571,7 +571,7 @@ def _entity(
     )
 
 
-def _note_content(entity: Entity) -> NoteContent:
+def _note_content(entity: Entity, last_source: str | None = None) -> NoteContent:
     return NoteContent(
         entity_id=entity.id,
         project_id=entity.project_id,
@@ -583,7 +583,7 @@ def _note_content(entity: Entity) -> NoteContent:
         file_version=1,
         file_checksum="file-checksum",
         file_write_status="pending",
-        last_source=None,
+        last_source=last_source,
     )
 
 
@@ -927,6 +927,102 @@ async def test_run_accepted_note_update_rejects_stale_base_checksum() -> None:
     assert search_repository.calls == []
 
 
+@pytest.mark.asyncio
+async def test_run_accepted_note_update_accepts_relay_self_supersede_on_stale_base() -> None:
+    # Lost-ack wedge regression (#1589, 2026-07-23 production incident): a relay
+    # persist timed out client-side AFTER committing, so the accepted row is the
+    # relay's own write while the relay's recorded base is one version behind.
+    # The relay superseding its own prior write is never a real conflict - the
+    # live Y.Doc is the merge of everything the relay ever persisted - so the
+    # stale base must be accepted, not 409-wedged forever.
+    session = _MutationSession()
+    schema = _schema()
+    project = _project()
+    prepared = _prepared_replacement()
+    entity = _entity(file_path="notes/accepted.md")
+    note_content = _note_content(entity, last_source="collaboration_relay")
+    project_repository = _ProjectRepository(project)
+    entity_lookup_repository = _EntityLookupRepository(by_external_id=entity)
+    note_content_lookup_repository = _NoteContentLookupRepository(note_content)
+    preparer = _CreatePreparer(prepared)
+    preparer_factory = _PreparerFactory(preparer)
+    pending_entity_repository = _PendingEntityRepository(entity)
+    note_content_accept_repository = _NoteContentAcceptRepository(note_content)
+    search_repository = _SearchRepository()
+
+    change = await run_accepted_note_update(
+        cast(AsyncSession, session),
+        request=AcceptedNoteUpdateMutation(
+            project_external_id="project-123",
+            entity_external_id="note-123",
+            data=schema,
+            actor=AcceptedNoteMutationActor(user_profile_id=_ACTOR_ID),
+            source="collaboration_relay",
+            base_checksum="stale-checksum",
+        ),
+        dependencies=_dependencies(
+            project_repository=project_repository,
+            entity_lookup_repository=entity_lookup_repository,
+            note_content_lookup_repository=note_content_lookup_repository,
+            preparer_factory=preparer_factory,
+            pending_entity_repository=pending_entity_repository,
+            note_content_accept_repository=note_content_accept_repository,
+            search_repository=search_repository,
+        ),
+    )
+
+    assert change.status_code == 200
+    assert note_content_accept_repository.calls[0][1].db_version == 2
+    assert note_content_accept_repository.calls[0][1].markdown_content == "# Replacement\n"
+
+
+@pytest.mark.asyncio
+async def test_run_accepted_note_update_relay_stale_base_still_rejects_foreign_writes() -> None:
+    # The self-supersede rule is scoped to relay-over-relay only: when the
+    # current accepted version came from a FOREIGN writer (MCP here), a stale
+    # relay base is a genuine conflict and keeps the full guarded semantics.
+    session = _MutationSession()
+    schema = _schema()
+    project = _project()
+    prepared = _prepared_replacement()
+    entity = _entity(file_path="notes/accepted.md")
+    note_content = _note_content(entity, last_source="mcp")
+    project_repository = _ProjectRepository(project)
+    entity_lookup_repository = _EntityLookupRepository(by_external_id=entity)
+    note_content_lookup_repository = _NoteContentLookupRepository(note_content)
+    preparer = _CreatePreparer(prepared)
+    preparer_factory = _PreparerFactory(preparer)
+    pending_entity_repository = _PendingEntityRepository(entity)
+    note_content_accept_repository = _NoteContentAcceptRepository(note_content)
+    search_repository = _SearchRepository()
+
+    with pytest.raises(AcceptedNoteMutationRejected) as exc_info:
+        await run_accepted_note_update(
+            cast(AsyncSession, session),
+            request=AcceptedNoteUpdateMutation(
+                project_external_id="project-123",
+                entity_external_id="note-123",
+                data=schema,
+                actor=AcceptedNoteMutationActor(user_profile_id=_ACTOR_ID),
+                source="collaboration_relay",
+                base_checksum="stale-checksum",
+            ),
+            dependencies=_dependencies(
+                project_repository=project_repository,
+                entity_lookup_repository=entity_lookup_repository,
+                note_content_lookup_repository=note_content_lookup_repository,
+                preparer_factory=preparer_factory,
+                pending_entity_repository=pending_entity_repository,
+                note_content_accept_repository=note_content_accept_repository,
+                search_repository=search_repository,
+            ),
+        )
+
+    rejection = exc_info.value.rejection
+    assert rejection.kind is AcceptedNoteMutationRejectKind.conflict
+    assert note_content_accept_repository.calls == []
+
+
 @pytest.mark.asyncio
 async def test_run_accepted_note_update_rejects_base_checksum_when_entity_missing() -> None:
     # A base_checksum with no addressed entity means the note was deleted after

tests/indexing/test_index_file_runner.py

@@ -281,6 +281,7 @@ async def test_run_index_file_preserves_current_materialized_note_metadata() ->
         operation=FileIndexOperation.created,
         actor_user_profile_id="33333333-3333-3333-3333-333333333333",
         live_update_source="mcp",
+        db_version=1,
     )
     assert metadata_source.paths == ["notes/a.md"]
     assert materialized_source.paths == ["notes/a.md"]

tests/indexing/test_models.py

@@ -775,6 +775,7 @@ def test_plan_current_materialized_note_result_preserves_trusted_live_update_met
             actor_kind=NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT,
             actor_name="Claude Code",
             live_update_source="mcp",
+            db_version=1,
         ),
         object_checksum_source=RuntimeStorageObjectChecksumSource.note_file_checksum,
         object_checksum="checksum-1",
@@ -874,6 +875,7 @@ def test_plan_indexed_file_live_update_metadata_preserves_matching_metadata():
         actor_kind=NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT,
         actor_name="Claude Code",
         live_update_source="mcp",
+        db_version=2,
         operation=FileIndexOperation.updated,
     )
 

tests/test_runtime.py

@@ -1119,6 +1119,7 @@ def test_note_object_provenance_parses_trusted_actor_and_source(self):
                 NOTE_OBJECT_ACTOR_KIND_METADATA: NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT,
                 NOTE_OBJECT_ACTOR_NAME_METADATA: " Pat\t\n<script>! ",
                 NOTE_OBJECT_SOURCE_METADATA: "mcp",
+                NOTE_OBJECT_DB_VERSION_METADATA: "7",
             }
         )
 
@@ -1127,6 +1128,7 @@ def test_note_object_provenance_parses_trusted_actor_and_source(self):
             actor_kind=NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT,
             actor_name="Pat script",
             source="mcp",
+            db_version=7,
         )
 
         untrusted_actor_name = RuntimeNoteObjectProvenance.from_object_metadata(