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
3 changes: 3 additions & 0 deletions src/basic_memory/deps/services.py
Original file line number Diff line number Diff line change
Expand Up @@ -504,6 +504,9 @@ async def get_relation_resolution_scheduler(
project_external_id=project_external_id,
read_cache=read_cache,
test_mode=app_config.is_test_env,
# The write path embeds notes; sources whose links this pass resolves
# need their changed relation chunks re-embedded too (#1873).
vector_sync=search_service,
)


Expand Down
3 changes: 3 additions & 0 deletions src/basic_memory/index/local_project.py
Original file line number Diff line number Diff line change
Expand Up @@ -967,6 +967,9 @@ async def run_local_project_index(
project_path=request.project.project_path,
),
runtime.completion_relation_runtime,
# A search-only reindex (``bm reindex --search``) must not build
# vectors, so it passes no sync backend.
vector_sync=runtime.embedding_vector_sync if request.embeddings else None,
)
# Stamped last, after relation resolution, so the marker means "this project
# has a complete index" rather than "a scan started". Anything that raises
Expand Down
7 changes: 6 additions & 1 deletion src/basic_memory/index/local_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,12 @@ async def index_file_completed(
else nullcontext(ReadCacheChangeRecorder())
)
async with invalidation_scope as relation_change:
relation_result = await resolve_project_relations(self.relation_runtime)
relation_result = await resolve_project_relations(
self.relation_runtime,
# Resolution rewrites affected sources' relation rows; re-embed
# them only when this runtime is embedding at all.
vector_sync=self.search_service if self.index_embeddings else None,
)
relation_change.record(
await resolved_relations_read_cache_change(
self.session_maker,
Expand Down
18 changes: 14 additions & 4 deletions src/basic_memory/index/local_schedulers.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

from basic_memory.index.project_indexing import ProjectIndexRunner
from basic_memory.index.schedulers import SearchReindexService
from basic_memory.indexing.embedding_index_planning import EmbeddingBatchVectorSync
from basic_memory.indexing.relation_resolution import (
RelationResolutionRuntime,
resolve_project_relations,
Expand Down Expand Up @@ -225,15 +226,18 @@ class LocalRelationResolutionScheduler:
write is both wasteful and a real write-load cost. Instead each write only
enqueues: the first write of a burst schedules one debounced background pass
and every other write coalesces onto it (at most one pending pass per
project). The accept path stays light; reconciliation runs offline. No-op in
test mode, consistent with the other local schedulers.
project). The accept path stays light; reconciliation runs offline. A caller
that embeds passes ``vector_sync`` so sources whose relation rows the pass
rewrote are re-embedded after it settles. No-op in test mode, consistent
with the other local schedulers.
"""

relation_runtime: RelationResolutionRuntime
session_maker: async_sessionmaker[AsyncSession]
project_external_id: str
read_cache: ReadCacheInvalidator | None
test_mode: bool
vector_sync: EmbeddingBatchVectorSync | None = None
debounce_seconds: float = 0.5

def schedule_relation_resolution(self, *, project_id: int) -> None:
Expand Down Expand Up @@ -265,14 +269,20 @@ async def _resolve_after_debounce(self, project_id: int) -> None:
# A second bump closes the window in which an intermediate entity
# response could have populated the current generation.
if self.read_cache is None:
await resolve_project_relations(self.relation_runtime)
await resolve_project_relations(
self.relation_runtime,
vector_sync=self.vector_sync,
)
else:
# The pass retires the notes whose links it resolved; a raise
# retires the whole project.
async with invalidate_recorded_change(
self.read_cache, self.project_external_id
) as change:
result = await resolve_project_relations(self.relation_runtime)
result = await resolve_project_relations(
self.relation_runtime,
vector_sync=self.vector_sync,
)
change.record(
await resolved_relations_read_cache_change(
self.session_maker,
Expand Down
69 changes: 67 additions & 2 deletions src/basic_memory/indexing/relation_resolution.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

from basic_memory import db
from basic_memory.indexing.accepted_note_search import accepted_search_content_from_markdown
from basic_memory.indexing.embedding_index_planning import EmbeddingBatchVectorSync
from basic_memory.indexing.models import IndexFileJobStatus, RelationTargetRequest
from basic_memory.markdown.path_links import is_path_target
from basic_memory.models import Entity
Expand Down Expand Up @@ -504,18 +505,24 @@ async def resolve_project_index_completion_relations(
runtime: RelationResolutionRuntime,
*,
max_passes: int = 3,
vector_sync: EmbeddingBatchVectorSync | None = None,
) -> ResolveRelationsResult | None:
"""Run the final relation-resolution pass for a completed project index.

The context names the project so queue-based runtimes can plan an enqueue
from the same completion facts; the inline path resolves directly against
the already project-scoped runtime. A context without complete project
identity plans no request, so the resolution pass is skipped.
``vector_sync`` re-embeds sources whose relation rows this pass rewrote.
"""
request = plan_project_index_completion_relation_resolution(context)
if request is None:
return None
return await resolve_project_relations(runtime, max_passes=max_passes)
return await resolve_project_relations(
runtime,
max_passes=max_passes,
vector_sync=vector_sync,
)


def plan_index_file_relation_resolution(
Expand Down Expand Up @@ -546,6 +553,11 @@ class ResolveRelationsResult:
# Source notes whose links this run resolved; see
# services.read_cache_changes.resolved_relations_read_cache_change.
affected_entity_ids: frozenset[int] = frozenset()
# Entities the vector backend could not finish. Failed entities remain marked
# for the backend's retry/readiness path; the result exposes both sets instead
# of silently reporting the relation refresh as fully synced.
vector_sync_failed_entity_ids: frozenset[int] = frozenset()
vector_sync_deferred_entity_ids: frozenset[int] = frozenset()

@property
def resolved(self) -> int:
Expand All @@ -557,6 +569,7 @@ async def resolve_project_relations(
runtime: RelationResolutionRuntime,
*,
max_passes: int = 3,
vector_sync: EmbeddingBatchVectorSync | None = None,
) -> ResolveRelationsResult:
"""Resolve all resolvable forward references for one project runtime.

Expand All @@ -568,6 +581,12 @@ async def resolve_project_relations(
resolver from looping forever. Relations left after a stable pass are
genuine forward references and remain unresolved until their target note
exists.

When ``vector_sync`` is set, callers that embed run it after every pass
settles and then drain any deferred shards. Resolution rewrites the affected
sources' relation search rows, and only changed chunks are re-embedded.
Failed entities remain marked by the backend's retry/readiness state and are
exposed on the result.
"""
unresolved_before = await runtime.count_unresolved_relations()
affected_entities: AffectedEntityIds = set()
Expand All @@ -581,12 +600,58 @@ async def resolve_project_relations(
if not affected:
break

remaining = await runtime.count_unresolved_relations()
failed_entity_ids: set[EntityId] = set()
deferred_entity_ids: set[EntityId] = set()

# Trigger: the resolution pass rewrote relation search rows for these sources
# after their owners had already embedded them.
# Why: chunk keys follow the row text, so those changed relation rows have no
# current vector until something re-syncs them; nothing else revisits them.
# Outcome: re-embed the affected sources after resolution committed. The batch
# backend reports failed and deferred entities instead of raising, so drain
# every deferred shard here; failed entities stay marked by the backend's
# retry/readiness state and both sets are exposed on the result.
if vector_sync is not None and affected_entities:
pending_entity_ids = sorted(affected_entities)
previous_pending_chunks: int | None = None
while pending_entity_ids:
sync_result = await vector_sync.sync_entity_vectors_batch(pending_entity_ids)
failed_entity_ids.update(sync_result.failed_entity_ids)
pending_entity_ids = list(sync_result.deferred_entity_ids)
if not pending_entity_ids:
break

pending_chunks = sync_result.chunks_total - sync_result.chunks_skipped
if previous_pending_chunks is not None and pending_chunks >= previous_pending_chunks:
logger.warning(
"Stopped draining deferred relation vectors: no progress",
deferred_entities=len(pending_entity_ids),
pending_chunks=pending_chunks,
)
break
previous_pending_chunks = pending_chunks

deferred_entity_ids.update(pending_entity_ids)
if failed_entity_ids:
logger.warning(
"Relation vector sync left failed entities",
failed_entities=len(failed_entity_ids),
)
if deferred_entity_ids:
logger.warning(
"Relation vector sync left deferred entities after drain",
deferred_entities=len(deferred_entity_ids),
)

result = ResolveRelationsResult(
unresolved_before=unresolved_before,
remaining=await runtime.count_unresolved_relations(),
remaining=remaining,
passes=passes,
affected_entities=len(affected_entities),
affected_entity_ids=frozenset(affected_entities),
vector_sync_failed_entity_ids=frozenset(failed_entity_ids),
vector_sync_deferred_entity_ids=frozenset(deferred_entity_ids),
)
logger.info(
"Resolved project relations",
Expand Down
40 changes: 37 additions & 3 deletions tests/index/test_local_inline_result_recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from basic_memory.indexing.project_index_maintenance import ProjectIndexMovedEntitySearchRefresher
from basic_memory.indexing.relation_resolution import RelationResolutionRuntime
from basic_memory.runtime.projects import ProjectRuntimeReference
from basic_memory.runtime.vector_sync import VectorSyncBatchResult
from basic_memory.runtime.storage import (
RuntimeStorageEventOperation,
RuntimeStorageEventOperationKind,
Expand Down Expand Up @@ -58,21 +59,26 @@ def __init__(self) -> None:

async def sync_entity_vectors_batch(self, entity_ids, progress_callback=None):
self.batches.append(list(entity_ids))
return None
return VectorSyncBatchResult(
entities_total=len(entity_ids),
entities_synced=len(entity_ids),
entities_failed=0,
)


class StubRelationRuntime:
"""Counts project relation resolution passes triggered by the recorder."""

def __init__(self) -> None:
def __init__(self, resolved_sources: set[int] | None = None) -> None:
self.resolve_calls = 0
self._pending_affected = [resolved_sources] if resolved_sources else []

async def count_unresolved_relations(self) -> int:
return 0

async def resolve_relations(self, entity_id: int | None = None) -> set[int]:
self.resolve_calls += 1
return set()
return self._pending_affected.pop(0) if self._pending_affected else set()


def _recorder(
Expand Down Expand Up @@ -133,6 +139,34 @@ async def test_processed_file_skips_embedding_when_disabled(
assert relations.resolve_calls == 1


async def test_processed_file_embeds_resolved_relation_sources(
session_maker: async_sessionmaker[AsyncSession],
) -> None:
"""Resolving a forward reference refreshes source vectors, then the file's own."""
search = StubSearchService()
relations = StubRelationRuntime(resolved_sources={7})
recorder = _recorder(search, relations, index_embeddings=True, session_maker=session_maker)

await recorder.index_file_completed(_operation(), _result(IndexFileJobStatus.processed, 42))

# The relation pass re-embedded the affected source before this file was embedded.
assert search.batches == [[7], [42]]
assert relations.resolve_calls == 2


async def test_disabled_embeddings_skip_relation_vector_sync(
session_maker: async_sessionmaker[AsyncSession],
) -> None:
search = StubSearchService()
relations = StubRelationRuntime(resolved_sources={7})
recorder = _recorder(search, relations, index_embeddings=False, session_maker=session_maker)

await recorder.index_file_completed(_operation(), _result(IndexFileJobStatus.processed, 42))

assert search.batches == []
assert relations.resolve_calls == 2


async def test_superseded_file_skips_embedding_when_enabled(
session_maker: async_sessionmaker[AsyncSession],
) -> None:
Expand Down
61 changes: 61 additions & 0 deletions tests/index/test_local_project_index.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@
ResolvedRelationWrite,
ResolvedRelationWriteResult,
)
from basic_memory.repository.semantic_chunking import compose_row_source_text
from basic_memory.runtime.jobs import (
RuntimeIndexFileBatchJobRequest,
RuntimeObservedIndexFile,
Expand Down Expand Up @@ -1805,6 +1806,9 @@ async def test_local_project_index_resolves_order_dependent_relations_after_batc
assert entity_b is not None
assert entity_c is not None

entity_a_search_rows = await search_service.repository.get_entity_search_rows(entity_a.id)
entity_a_chunk_manifest = await search_service.repository.get_entity_chunk_manifest(entity_a.id)

a_targets = {relation.to_id for relation in entity_a.outgoing_relations}
b_targets = {relation.to_id for relation in entity_b.outgoing_relations}
c_targets = {relation.to_id for relation in entity_c.outgoing_relations}
Expand All @@ -1822,6 +1826,63 @@ async def test_local_project_index_resolves_order_dependent_relations_after_batc
assert c_incoming_sources == {entity_a.id, entity_b.id}
assert {row.to_id for row in relation_search_rows} == {entity_a.id, entity_b.id, entity_c.id}

# The completion pass refreshed entity_a's relation search rows after the
# coordinator embedded the earlier batch. Those changed rows owe new vector
# chunks; the stored relation chunks must match the resolved rows again (#1873).
expected_relation_chunks = {
compose_row_source_text(row)
for row in entity_a_search_rows
if row.type == SearchItemType.RELATION.value
}
stored_relation_chunks = {
chunk.chunk_text
for chunk in entity_a_chunk_manifest
if chunk.source_type == SearchItemType.RELATION.value
}
assert expected_relation_chunks
assert stored_relation_chunks == expected_relation_chunks


async def test_local_project_index_search_only_reindex_leaves_relation_vectors_unbuilt(
test_project: Project,
project_config,
entity_repository,
session_maker: async_sessionmaker[AsyncSession],
search_service,
config_manager,
) -> None:
"""A search-only project index resolves relations but embeds nothing (#1873)."""
del config_manager

(project_config.home / "source.md").write_text(
"---\ntitle: Source Note\ntype: note\n---\n\n# Source Note\n\n- relates_to [[Target Note]]\n",
encoding="utf-8",
)
(project_config.home / "target.md").write_text(
"---\ntitle: Target Note\ntype: note\n---\n\n# Target Note\n\nTarget body.\n",
encoding="utf-8",
)

result = await run_local_project_index_for_project(
test_project,
runtime_factory=LocalProjectIndexRuntimeFactory(batch_size=1),
force_full=True,
embeddings=False,
)

assert result.enqueued_files == 2
async with db.scoped_session(session_maker) as session:
source = await entity_repository.get_by_file_path(session, "source.md")
target = await entity_repository.get_by_file_path(session, "target.md")
assert source is not None
assert target is not None

search_rows = await search_service.repository.get_entity_search_rows(source.id)
assert {row.to_id for row in search_rows if row.type == SearchItemType.RELATION.value} == {
target.id
}
assert await search_service.repository.get_entity_chunk_manifest(source.id) == []


async def test_local_relation_resolution_refreshes_pending_source_without_markdown_file(
test_project: Project,
Expand Down
Loading
Loading