diff --git a/src/basic_memory/indexing/accepted_note_mutation_runner.py b/src/basic_memory/indexing/accepted_note_mutation_runner.py index 82c8763e5..7337c88d9 100644 --- a/src/basic_memory/indexing/accepted_note_mutation_runner.py +++ b/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, diff --git a/src/basic_memory/indexing/models.py b/src/basic_memory/indexing/models.py index 3d97e841d..fcf1df9c5 100644 --- a/src/basic_memory/indexing/models.py +++ b/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,16 @@ 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. + # Contract: this is a HINT, not ground truth. Object metadata carries + # the version of the write that produced the object, so it can LAG the + # accepted row (a content-identical accept bumps db_version without a + # new materialization) but can never exceed it. A lagging version on + # identical content reads as an echo downstream, which is the correct + # skip; consumers must take authoritative reads (which return the DB + # row's version) as the decision-grade value (PR #1144 review). + db_version=None if result.content_superseded else result.db_version, ) @@ -537,6 +556,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 +621,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 +680,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), ) diff --git a/src/basic_memory/runtime/note_object_metadata.py b/src/basic_memory/runtime/note_object_metadata.py index 2a05288e6..7ee8f2713 100644 --- a/src/basic_memory/runtime/note_object_metadata.py +++ b/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), ) diff --git a/tests/indexing/test_accepted_note_mutation_runner.py b/tests/indexing/test_accepted_note_mutation_runner.py index b7884241f..8461ddb1e 100644 --- a/tests/indexing/test_accepted_note_mutation_runner.py +++ b/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 diff --git a/tests/indexing/test_index_file_runner.py b/tests/indexing/test_index_file_runner.py index 689e7967a..74ab047b0 100644 --- a/tests/indexing/test_index_file_runner.py +++ b/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"] diff --git a/tests/indexing/test_models.py b/tests/indexing/test_models.py index b780bcf1c..a7d99ef90 100644 --- a/tests/indexing/test_models.py +++ b/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, ) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 9b303b56f..ecf47df40 100644 --- a/tests/test_runtime.py +++ b/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