From 249a2e0f54cefae7e2a449d99fd4c02e7130bc39 Mon Sep 17 00:00:00 2001
From: Steven Huang <134844631+remote-controlled-man@users.noreply.github.com>
Date: Wed, 28 Oct 2026 12:24:17 +0800
Subject: [PATCH 1/2] fix(index): re-embed relation sources after
 forward-reference resolution

Forward references resolved on the write, watcher, and project-index completion paths refreshed relation search rows but left their vector chunks stale, so semantic search kept returning the unresolved link text (#1873). Thread an optional vector sync through relation resolution and pass the embedding backend from each local path; a search-only reindex still builds no vectors.

Signed-off-by: Steven Huang <134844631+remote-controlled-man@users.noreply.github.com>
---
 src/basic_memory/deps/services.py             |   3 +
 src/basic_memory/index/local_project.py       |   3 +
 src/basic_memory/index/local_runtime.py       |   7 +-
 src/basic_memory/index/local_schedulers.py    |  18 ++-
 .../indexing/relation_resolution.py           |  22 +++-
 .../test_local_inline_result_recorder.py      |  33 ++++-
 tests/index/test_local_project_index.py       |  61 +++++++++
 tests/index/test_local_schedulers.py          | 122 ++++++++++++++++++
 tests/indexing/test_relation_resolution.py    |  59 +++++++++
 tests/test_deps.py                            |  26 ++++
 10 files changed, 346 insertions(+), 8 deletions(-)

diff --git a/src/basic_memory/deps/services.py b/src/basic_memory/deps/services.py
index abd0deab4..f44b225ac 100644
--- a/src/basic_memory/deps/services.py
+++ b/src/basic_memory/deps/services.py
@@ -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,
     )
 
 
diff --git a/src/basic_memory/index/local_project.py b/src/basic_memory/index/local_project.py
index 54af62ade..4e728309e 100644
--- a/src/basic_memory/index/local_project.py
+++ b/src/basic_memory/index/local_project.py
@@ -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
diff --git a/src/basic_memory/index/local_runtime.py b/src/basic_memory/index/local_runtime.py
index 431c3473d..25d3b020d 100644
--- a/src/basic_memory/index/local_runtime.py
+++ b/src/basic_memory/index/local_runtime.py
@@ -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,
diff --git a/src/basic_memory/index/local_schedulers.py b/src/basic_memory/index/local_schedulers.py
index d9e1a4fe0..7d03e070d 100644
--- a/src/basic_memory/index/local_schedulers.py
+++ b/src/basic_memory/index/local_schedulers.py
@@ -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,
@@ -225,8 +226,10 @@ 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
@@ -234,6 +237,7 @@ class LocalRelationResolutionScheduler:
     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:
@@ -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,
diff --git a/src/basic_memory/indexing/relation_resolution.py b/src/basic_memory/indexing/relation_resolution.py
index 3d3caa7b6..9e06a214b 100644
--- a/src/basic_memory/indexing/relation_resolution.py
+++ b/src/basic_memory/indexing/relation_resolution.py
@@ -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
@@ -504,6 +505,7 @@ 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.
 
@@ -511,11 +513,16 @@ async def resolve_project_index_completion_relations(
     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(
@@ -557,6 +564,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.
 
@@ -568,6 +576,10 @@ 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 once after every
+    pass settles: resolution rewrites the affected sources' relation search
+    rows, and only changed chunks are re-embedded.
     """
     unresolved_before = await runtime.count_unresolved_relations()
     affected_entities: AffectedEntityIds = set()
@@ -596,4 +608,12 @@ async def resolve_project_relations(
         passes=result.passes,
         affected_entities=result.affected_entities,
     )
+    # 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: one batch re-embeds the affected sources after resolution committed;
+    # a later batch or reindex is not required.
+    if vector_sync is not None and affected_entities:
+        await vector_sync.sync_entity_vectors_batch(sorted(affected_entities))
     return result
diff --git a/tests/index/test_local_inline_result_recorder.py b/tests/index/test_local_inline_result_recorder.py
index 6477d21be..7ac5eec46 100644
--- a/tests/index/test_local_inline_result_recorder.py
+++ b/tests/index/test_local_inline_result_recorder.py
@@ -64,15 +64,16 @@ async def sync_entity_vectors_batch(self, entity_ids, progress_callback=None):
 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(
@@ -133,6 +134,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:
diff --git a/tests/index/test_local_project_index.py b/tests/index/test_local_project_index.py
index 32e6410fc..e3b6dd99e 100644
--- a/tests/index/test_local_project_index.py
+++ b/tests/index/test_local_project_index.py
@@ -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,
@@ -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}
@@ -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,
diff --git a/tests/index/test_local_schedulers.py b/tests/index/test_local_schedulers.py
index 18c6df46d..6fc2dc4ae 100644
--- a/tests/index/test_local_schedulers.py
+++ b/tests/index/test_local_schedulers.py
@@ -6,6 +6,7 @@
 import pytest
 
 from basic_memory.indexing.project_index_coordinator import ProjectIndexCoordinatorResult
+from basic_memory.indexing.relation_resolution import RepositoryRelationResolutionRuntime
 from basic_memory.index.local_schedulers import (
     LocalEntityVectorSyncScheduler,
     LocalProjectIndexScheduler,
@@ -14,7 +15,12 @@
     drain_background_tasks,
 )
 from basic_memory.read_cache import ReadCacheChange, ReadCacheInvalidationStatus
+from basic_memory.repository.note_content_repository import NoteContentRepository
+from basic_memory.repository.semantic_chunking import compose_row_source_text
 from basic_memory.runtime.vector_sync import VectorSyncBatchResult
+from basic_memory.schemas.base import Entity as EntitySchema
+from basic_memory.schemas.search import SearchItemType
+from basic_memory.services.bulk_link_resolver import BulkLinkResolver
 
 PROJECT_EXTERNAL_ID = "00000000-0000-0000-0000-000000000013"
 
@@ -298,6 +304,21 @@ async def resolve_relations(self, entity_id: int | None = None) -> set[int]:
         return self._pending.pop(0) if self._pending else set()
 
 
+class StubVectorSync:
+    """Records entity batches handed to the scheduler's vector-sync step."""
+
+    def __init__(self) -> None:
+        self.batches: list[list[int]] = []
+
+    async def sync_entity_vectors_batch(self, entity_ids: list[int]) -> VectorSyncBatchResult:
+        self.batches.append(list(entity_ids))
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+        )
+
+
 @pytest.mark.asyncio
 async def test_relation_resolution_scheduler_runs_project_resolution(session_maker):
     """A single write schedules one debounced project resolution pass."""
@@ -325,6 +346,107 @@ async def test_relation_resolution_scheduler_runs_project_resolution(session_mak
     assert 13 not in _pending_relation_resolution
 
 
+@pytest.mark.asyncio
+async def test_relation_resolution_scheduler_syncs_resolved_source_vectors(
+    session_maker,
+) -> None:
+    """A scheduled pass re-embeds the source notes whose links it resolved (#1873)."""
+    from basic_memory.index.local_schedulers import _pending_relation_resolution
+
+    _pending_relation_resolution.clear()
+    runtime = StubRelationResolutionRuntime(resolved_sources={22, 11})
+    vector_sync = StubVectorSync()
+
+    scheduler = LocalRelationResolutionScheduler(
+        relation_runtime=runtime,
+        session_maker=session_maker,
+        project_external_id=PROJECT_EXTERNAL_ID,
+        read_cache=None,
+        test_mode=False,
+        debounce_seconds=0.0,
+        vector_sync=vector_sync,
+    )
+    scheduler.schedule_relation_resolution(project_id=13)
+    await drain_background_tasks()
+
+    assert vector_sync.batches == [[11, 22]]
+
+
+@pytest.mark.asyncio
+async def test_relation_resolution_scheduler_reembeds_resolved_source_relations(
+    test_project,
+    entity_service,
+    entity_repository,
+    relation_repository,
+    session_maker,
+    search_service,
+    app_config,
+) -> None:
+    """A write-path resolution pass leaves the source's relation vectors current (#1873).
+
+    The source note is embedded while its link is still dangling; writing the target
+    then lets the scheduler resolve that link. Its refreshed search row must be
+    re-embedded, not left carrying the unresolved relation text.
+    """
+    from basic_memory.index.local_schedulers import _pending_relation_resolution
+
+    _pending_relation_resolution.clear()
+    source = await entity_service.create_entity(
+        EntitySchema(
+            title="Scheduler Source",
+            directory="notes",
+            note_type="note",
+            content="# Scheduler Source\n\n- relates_to [[Scheduler Target]]\n",
+        )
+    )
+    await search_service.index_entity(source)
+    await search_service.sync_entity_vectors_batch([source.id])
+
+    target = await entity_service.create_entity(
+        EntitySchema(
+            title="Scheduler Target",
+            directory="notes",
+            note_type="note",
+            content="# Scheduler Target\n\nTarget body.\n",
+        )
+    )
+
+    runtime = RepositoryRelationResolutionRuntime(
+        session_maker=session_maker,
+        relation_repository=relation_repository,
+        entity_repository=entity_repository,
+        note_content_repository=NoteContentRepository(project_id=test_project.id),
+        target_resolver=BulkLinkResolver(
+            entity_repository=entity_repository,
+            app_config=app_config,
+        ),
+        entity_indexer=search_service,
+    )
+    scheduler = LocalRelationResolutionScheduler(
+        relation_runtime=runtime,
+        session_maker=session_maker,
+        project_external_id=str(test_project.external_id),
+        read_cache=None,
+        test_mode=False,
+        debounce_seconds=0.0,
+        vector_sync=search_service,
+    )
+
+    scheduler.schedule_relation_resolution(project_id=test_project.id)
+    await drain_background_tasks()
+
+    search_rows = await search_service.repository.get_entity_search_rows(source.id)
+    relation_rows = [row for row in search_rows if row.type == SearchItemType.RELATION.value]
+    assert relation_rows
+    assert {row.to_id for row in relation_rows} == {target.id}
+    expected_relation_chunks = {compose_row_source_text(row) for row in relation_rows}
+    manifest = await search_service.repository.get_entity_chunk_manifest(source.id)
+    stored_relation_chunks = {
+        chunk.chunk_text for chunk in manifest if chunk.source_type == SearchItemType.RELATION.value
+    }
+    assert stored_relation_chunks == expected_relation_chunks
+
+
 @pytest.mark.asyncio
 async def test_relation_resolution_scheduler_retires_the_notes_it_resolved(
     session_maker,
diff --git a/tests/indexing/test_relation_resolution.py b/tests/indexing/test_relation_resolution.py
index aa8205b35..384a16ac0 100644
--- a/tests/indexing/test_relation_resolution.py
+++ b/tests/indexing/test_relation_resolution.py
@@ -30,6 +30,7 @@
     ResolvedRelationWrite,
     ResolvedRelationWriteResult,
 )
+from basic_memory.runtime.vector_sync import VectorSyncBatchResult
 
 
 class StubRelationResolutionRuntime:
@@ -52,6 +53,21 @@ async def resolve_relations(self) -> set[int]:
         return self._affected_per_pass[index]
 
 
+class StubVectorSync:
+    """Records entity batches handed to the relation vector-sync step."""
+
+    def __init__(self) -> None:
+        self.batches: list[list[int]] = []
+
+    async def sync_entity_vectors_batch(self, entity_ids: list[int]) -> VectorSyncBatchResult:
+        self.batches.append(list(entity_ids))
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+        )
+
+
 class FakeSession:
     created_count = 0
 
@@ -463,6 +479,25 @@ def test_relation_write_batch_plan_keeps_collision_domains_together() -> None:
     }
 
 
+@pytest.mark.asyncio
+async def test_project_index_completion_relation_resolution_forwards_vector_sync() -> None:
+    runtime = StubRelationResolutionRuntime([1, 0], [{10}, set()])
+    vector_sync = StubVectorSync()
+
+    result = await resolve_project_index_completion_relations(
+        ProjectIndexRelationResolutionContext(
+            project_id=7,
+            project_path="main",
+        ),
+        runtime,
+        vector_sync=vector_sync,
+    )
+
+    assert result is not None
+    assert result.affected_entity_ids == frozenset({10})
+    assert vector_sync.batches == [[10]]
+
+
 @pytest.mark.asyncio
 async def test_resolves_until_a_stable_pass_changes_nothing() -> None:
     runtime = StubRelationResolutionRuntime([3, 1], [{10, 11}, set()])
@@ -493,6 +528,30 @@ async def test_stops_immediately_when_no_relations_resolve() -> None:
     assert runtime.resolve_calls == 1
 
 
+@pytest.mark.asyncio
+async def test_resolve_project_relations_syncs_vectors_once_after_last_pass() -> None:
+    runtime = StubRelationResolutionRuntime([3, 1], [{10, 11}, {12}, set()])
+    vector_sync = StubVectorSync()
+
+    result = await resolve_project_relations(runtime, vector_sync=vector_sync)
+
+    assert result.affected_entity_ids == frozenset({10, 11, 12})
+    assert runtime.resolve_calls == 3
+    # One batch after every pass settled, not one sync per resolution pass.
+    assert vector_sync.batches == [[10, 11, 12]]
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_relations_skips_vector_sync_when_nothing_resolved() -> None:
+    runtime = StubRelationResolutionRuntime([1, 1], [set()])
+    vector_sync = StubVectorSync()
+
+    result = await resolve_project_relations(runtime, vector_sync=vector_sync)
+
+    assert result.affected_entity_ids == frozenset()
+    assert vector_sync.batches == []
+
+
 @pytest.mark.asyncio
 async def test_resolution_loop_is_bounded_by_max_passes() -> None:
     runtime = StubRelationResolutionRuntime([2, 0], [{1}])
diff --git a/tests/test_deps.py b/tests/test_deps.py
index f765b2776..d887f7554 100644
--- a/tests/test_deps.py
+++ b/tests/test_deps.py
@@ -15,6 +15,8 @@
     get_read_cache,
     validate_project_external_id,
 )
+from basic_memory.deps.services import get_relation_resolution_scheduler
+from basic_memory.index.local_schedulers import LocalRelationResolutionScheduler
 from basic_memory.models.project import Project
 from basic_memory.repository.project_repository import ProjectRepository
 from basic_memory.runtime.mode import resolve_runtime_mode
@@ -117,6 +119,30 @@ async def probe(config: AppConfigDep, read_cache: ReadCacheDep) -> dict[str, str
     assert len(_anyio_worker_threads()) <= workers_before
 
 
+@pytest.mark.asyncio
+async def test_relation_resolution_scheduler_provider_reuses_search_service_for_vectors(
+    test_project: Project,
+    session_maker,
+    entity_repository,
+    relation_repository,
+    search_service,
+    app_config,
+):
+    """The write path's resolution pass re-embeds through the same search service."""
+    scheduler = await get_relation_resolution_scheduler(
+        project_external_id=str(test_project.external_id),
+        session_maker=session_maker,
+        entity_repository=entity_repository,
+        relation_repository=relation_repository,
+        search_service=search_service,
+        app_config=app_config,
+        read_cache=None,
+    )
+
+    assert isinstance(scheduler, LocalRelationResolutionScheduler)
+    assert scheduler.vector_sync is search_service
+
+
 @pytest.mark.asyncio
 async def test_validate_project_external_id_success(
     project_repository: ProjectRepository, test_project: Project, session_maker

From c3c238db1ffae2badb37e0a8cbcf970b33ce58a9 Mon Sep 17 00:00:00 2001
From: Steven Huang <134844631+remote-controlled-man@users.noreply.github.com>
Date: Wed, 28 Oct 2026 16:13:49 +0800
Subject: [PATCH 2/2] fix(index): drain deferred relation vector shards

Consume the vector sync batch result instead of discarding it, drain
deferred shards with a no-progress guard, and expose failed/deferred entity
IDs on the relation-resolution result. Failed entities remain marked by the
vector backend for its retry/readiness path.

Signed-off-by: Steven Huang <134844631+remote-controlled-man@users.noreply.github.com>
---
 .../indexing/relation_resolution.py           | 69 ++++++++++++---
 .../test_local_inline_result_recorder.py      |  7 +-
 tests/indexing/test_relation_resolution.py    | 83 +++++++++++++++++++
 3 files changed, 146 insertions(+), 13 deletions(-)

diff --git a/src/basic_memory/indexing/relation_resolution.py b/src/basic_memory/indexing/relation_resolution.py
index 9e06a214b..c16cfc5c6 100644
--- a/src/basic_memory/indexing/relation_resolution.py
+++ b/src/basic_memory/indexing/relation_resolution.py
@@ -553,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:
@@ -577,9 +582,11 @@ async def resolve_project_relations(
     genuine forward references and remain unresolved until their target note
     exists.
 
-    When ``vector_sync`` is set, callers that embed run it once after every
-    pass settles: resolution rewrites the affected sources' relation search
-    rows, and only changed chunks are re-embedded.
+    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()
@@ -593,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",
@@ -608,12 +661,4 @@ async def resolve_project_relations(
         passes=result.passes,
         affected_entities=result.affected_entities,
     )
-    # 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: one batch re-embeds the affected sources after resolution committed;
-    # a later batch or reindex is not required.
-    if vector_sync is not None and affected_entities:
-        await vector_sync.sync_entity_vectors_batch(sorted(affected_entities))
     return result
diff --git a/tests/index/test_local_inline_result_recorder.py b/tests/index/test_local_inline_result_recorder.py
index 7ac5eec46..2b54594b8 100644
--- a/tests/index/test_local_inline_result_recorder.py
+++ b/tests/index/test_local_inline_result_recorder.py
@@ -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,
@@ -58,7 +59,11 @@ 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:
diff --git a/tests/indexing/test_relation_resolution.py b/tests/indexing/test_relation_resolution.py
index 384a16ac0..51537b4c0 100644
--- a/tests/indexing/test_relation_resolution.py
+++ b/tests/indexing/test_relation_resolution.py
@@ -68,6 +68,18 @@ async def sync_entity_vectors_batch(self, entity_ids: list[int]) -> VectorSyncBa
         )
 
 
+class ScriptedVectorSync:
+    """Returns scripted batch results and records requested entity batches."""
+
+    def __init__(self, results: list[VectorSyncBatchResult]) -> None:
+        self.results = list(results)
+        self.batches: list[list[int]] = []
+
+    async def sync_entity_vectors_batch(self, entity_ids: list[int]) -> VectorSyncBatchResult:
+        self.batches.append(list(entity_ids))
+        return self.results.pop(0)
+
+
 class FakeSession:
     created_count = 0
 
@@ -541,6 +553,77 @@ async def test_resolve_project_relations_syncs_vectors_once_after_last_pass() ->
     assert vector_sync.batches == [[10, 11, 12]]
 
 
+@pytest.mark.asyncio
+async def test_resolve_project_relations_drains_deferred_vector_shards() -> None:
+    runtime = StubRelationResolutionRuntime([3, 1], [{10, 11}, set()])
+    vector_sync = ScriptedVectorSync(
+        [
+            VectorSyncBatchResult(
+                entities_total=2,
+                entities_synced=1,
+                entities_failed=0,
+                entities_deferred=1,
+                deferred_entity_ids=(11,),
+                chunks_total=2,
+                chunks_skipped=1,
+            ),
+            VectorSyncBatchResult(
+                entities_total=1,
+                entities_synced=1,
+                entities_failed=0,
+                chunks_total=1,
+                chunks_skipped=1,
+            ),
+        ]
+    )
+
+    result = await resolve_project_relations(runtime, vector_sync=vector_sync)
+
+    assert vector_sync.batches == [[10, 11], [11]]
+    assert result.vector_sync_deferred_entity_ids == frozenset()
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_relations_exposes_failed_vector_entities() -> None:
+    runtime = StubRelationResolutionRuntime([3, 1], [{10, 11}, set()])
+    vector_sync = ScriptedVectorSync(
+        [
+            VectorSyncBatchResult(
+                entities_total=2,
+                entities_synced=1,
+                entities_failed=1,
+                failed_entity_ids=(11,),
+            )
+        ]
+    )
+
+    result = await resolve_project_relations(runtime, vector_sync=vector_sync)
+
+    assert vector_sync.batches == [[10, 11]]
+    assert result.vector_sync_failed_entity_ids == frozenset({11})
+    assert result.vector_sync_deferred_entity_ids == frozenset()
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_relations_stops_draining_without_progress() -> None:
+    runtime = StubRelationResolutionRuntime([3, 1], [{10, 11}, set()])
+    deferred = VectorSyncBatchResult(
+        entities_total=1,
+        entities_synced=0,
+        entities_failed=0,
+        entities_deferred=1,
+        deferred_entity_ids=(11,),
+        chunks_total=2,
+        chunks_skipped=1,
+    )
+    vector_sync = ScriptedVectorSync([deferred, deferred])
+
+    result = await resolve_project_relations(runtime, vector_sync=vector_sync)
+
+    assert vector_sync.batches == [[10, 11], [11]]
+    assert result.vector_sync_deferred_entity_ids == frozenset({11})
+
+
 @pytest.mark.asyncio
 async def test_resolve_project_relations_skips_vector_sync_when_nothing_resolved() -> None:
     runtime = StubRelationResolutionRuntime([1, 1], [set()])