Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion src/basic_memory/indexing/accepted_note_mutation_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
23 changes: 23 additions & 0 deletions src/basic_memory/indexing/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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
),
Expand Down Expand Up @@ -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,
)


Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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,

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Verify db_version against accepted state before publishing

When an accepted write advances db_version without changing the markdown (for example a relay retry/no-op save with the same db_checksum), an older materialized object still has the same bm-file-checksum, so this checksum-only trust check treats the object's stale bm-db-version as current and publishes it to live-update consumers. Since the cloud relay is expected to use db_version arithmetic for echo-vs-out-of-band decisions, please compare the metadata version with the current NoteContent.db_version (or omit it unless that version is verified), not just with the content checksum.

Useful? React with 👍 / 👎.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The scenario is real, but the staleness has one direction that fails safe: an object's metadata version is the version of the write that produced that object, so it can LAG the accepted row (content-identical accept bumps db_version without a new materialization) but can never exceed it. A lagging version rides identical content, so the consumer's version arithmetic reads it as an echo and skips - exactly the right outcome for identical bytes. The cloud reconciler additionally treats event fields as a pre-filter only: any apply path goes through the authoritative /read, whose response carries the DB row's current db_version and is the decision-grade value.

Threading NoteContent.db_version verification into the index/materialized-source contracts would add a tenant-DB read to a path that deliberately works from object state alone, to guard a direction that cannot mislead. Encoded the hint-vs-ground-truth contract in the code comment at the plan site instead (4993e71). Happy to revisit if a consumer ever wants to treat event versions as authoritative - that would be the contract change to block.

),
object_checksum_source=plan.object_checksum_source,
object_checksum=plan.object_checksum,
Expand Down Expand Up @@ -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),
)

Expand Down
8 changes: 8 additions & 0 deletions src/basic_memory/runtime/note_object_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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),
)


Expand Down
100 changes: 98 additions & 2 deletions tests/indexing/test_accepted_note_mutation_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
)


Expand Down Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions tests/indexing/test_index_file_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down
2 changes: 2 additions & 0 deletions tests/indexing/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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,
)

Expand Down
2 changes: 2 additions & 0 deletions tests/test_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
}
)

Expand All @@ -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(
Expand Down
Loading