diff --git a/src/basic_memory/indexing/accepted_note_mutation_runner.py b/src/basic_memory/indexing/accepted_note_mutation_runner.py index 7337c88d9..e54b7957b 100644 --- a/src/basic_memory/indexing/accepted_note_mutation_runner.py +++ b/src/basic_memory/indexing/accepted_note_mutation_runner.py @@ -608,20 +608,29 @@ async def _run_accepted_note_update( request.base_checksum is not None and current_note_content.db_checksum != request.base_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 + # Hot-doc canonical (#1589 Phase G): while a live session exists the + # Y.Doc is canonical, so a relay persist supersedes the current + # head. The invariant that makes this safe is "the superseded + # version survives as a storage object version", so a FOREIGN head + # may only be superseded once it is provably IN storage: 'synced' + # and nothing else. 'external_change_detected' explicitly means the + # accepted DB markdown did NOT materialize (the guard protected an + # unexpected external file), and pending/writing/failed heads have + # no object version yet — superseding any of them would erase the + # only copy, because their queued materialization preflights as + # stale and never writes (Codex review, PR #1146). Rejecting keeps + # the relay's next store retrying (seconds) until materialization + # lands. Relay-over-relay stays unconditional: the live Y.Doc is + # the merge of everything the relay ever persisted, which is what + # closes the lost-ack wedge (2026-07-23 production incident). + # Non-relay writers keep the full guarded semantics; the + # deleted-entity 409 above and the db_version CAS both remain. + current_head_in_storage = current_note_content.file_write_status == "synced" + relay_supersede = request.source == NOTE_SOURCE_COLLABORATION_RELAY and ( + current_note_content.last_source == NOTE_SOURCE_COLLABORATION_RELAY + or current_head_in_storage ) - if not relay_self_supersede: + if not relay_supersede: reject_stale_base_checksum( current_db_checksum=current_note_content.db_checksum ) diff --git a/tests/indexing/test_accepted_note_mutation_runner.py b/tests/indexing/test_accepted_note_mutation_runner.py index 8461ddb1e..2e4b15223 100644 --- a/tests/indexing/test_accepted_note_mutation_runner.py +++ b/tests/indexing/test_accepted_note_mutation_runner.py @@ -571,7 +571,11 @@ def _entity( ) -def _note_content(entity: Entity, last_source: str | None = None) -> NoteContent: +def _note_content( + entity: Entity, + last_source: str | None = None, + file_write_status: str = "pending", +) -> NoteContent: return NoteContent( entity_id=entity.id, project_id=entity.project_id, @@ -582,7 +586,7 @@ def _note_content(entity: Entity, last_source: str | None = None) -> NoteContent db_checksum="old-checksum", file_version=1, file_checksum="file-checksum", - file_write_status="pending", + file_write_status=file_write_status, last_source=last_source, ) @@ -977,16 +981,72 @@ async def test_run_accepted_note_update_accepts_relay_self_supersede_on_stale_ba @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. +async def test_run_accepted_note_update_relay_supersedes_foreign_head() -> None: + # Hot-doc canonical (#1589 Phase G): a relay persist is an unconditional + # versioned export, superseding even a FOREIGN current head (MCP here). + # The foreign version survives as file history and the reconciler surfaces + # the conflict from the live-update event; nothing is destroyed. session = _MutationSession() schema = _schema() project = _project() prepared = _prepared_replacement() entity = _entity(file_path="notes/accepted.md") - note_content = _note_content(entity, last_source="mcp") + # The foreign head is materialized ('synced'): its object version exists, + # so superseding it destroys nothing. + note_content = _note_content(entity, last_source="mcp", file_write_status="synced") + 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 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "file_write_status", + ["pending", "writing", "failed", "external_change_detected"], +) +async def test_run_accepted_note_update_relay_keeps_rejecting_unmaterialized_foreign_head( + file_write_status: str, +) -> None: + # Only 'synced' proves the foreign head's accepted markdown is in storage. + # pending/writing/failed have no object version yet, and + # external_change_detected explicitly means the accepted markdown did NOT + # materialize (the guard protected an unexpected external file) — + # superseding any of them would erase the only copy (Codex, PR #1146). + session = _MutationSession() + schema = _schema() + project = _project() + prepared = _prepared_replacement() + entity = _entity(file_path="notes/accepted.md") + note_content = _note_content(entity, last_source="mcp", file_write_status=file_write_status) project_repository = _ProjectRepository(project) entity_lookup_repository = _EntityLookupRepository(by_external_id=entity) note_content_lookup_repository = _NoteContentLookupRepository(note_content) @@ -1023,6 +1083,52 @@ async def test_run_accepted_note_update_relay_stale_base_still_rejects_foreign_w assert note_content_accept_repository.calls == [] +@pytest.mark.asyncio +async def test_run_accepted_note_update_non_relay_stale_base_still_rejects() -> None: + # The unconditional export is scoped to the relay writer only: any other + # source with a stale base keeps the full guarded 409 semantics. + 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() + + 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="api", + 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