chore(sync): remove duplicate resolver comment

Signed-off-by: phernandez <paul@basicmachines.co>

CHANGELOG.md

@@ -1,5 +1,21 @@
 # CHANGELOG
 
+## v0.21.5 (2026-05-26)
+
+Workspace/project routing fixes for MCP, plus a SQLite vector reindex stability fix.
+
+### Bug Fixes
+
+- **#854**: MCP project listing now keeps duplicate cloud project rows distinct by
+  workspace, so only the selected workspace row inherits local project state.
+- **#853**: `write_note` returns workspace-qualified permalinks for cloud
+  workspace writes, allowing follow-up `memory://` reads to route back to the
+  correct workspace/project.
+- **#852**: Full SQLite vector reindex now loads `sqlite-vec` before dropping
+  `vec0` virtual tables, preventing reindex crashes.
+- **#838**: Local ASGI database initialization is preloaded so MCP routing can
+  safely enter local project contexts.
+
 ## v0.21.1 (2026-05-16)
 
 CI-only release. No user-facing changes.

server.json

@@ -6,12 +6,12 @@
     "url": "https://github.com/basicmachines-co/basic-memory.git",
     "source": "github"
   },
-  "version": "0.21.4",
+  "version": "0.21.5",
   "packages": [
     {
       "registryType": "pypi",
       "identifier": "basic-memory",
-      "version": "0.21.4",
+      "version": "0.21.5",
       "runtimeHint": "uvx",
       "runtimeArguments": [
         {"type": "positional", "value": "basic-memory"},

src/basic_memory/__init__.py

@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
 # Package version - updated by release automation
-__version__ = "0.21.4"
+__version__ = "0.21.5"
 
 # API version for FastAPI - independent of package version
 __api_version__ = "v0"

src/basic_memory/indexing/batch_indexer.py

@@ -493,8 +493,14 @@ class BatchIndexer:
         async def resolve_relation(relation: Relation) -> int:
             async with semaphore:
                 try:
+                    # strict=True for deferred resolution: only fill in to_id on an
+                    # exact permalink/title/file_path match. Fuzzy fallback would silently
+                    # resolve ambiguous links to whichever entity shares tokens with the
+                    # link text, mismatching this with the sync_service forward-reference
+                    # path and producing confidently-wrong graph edges. See
+                    # sync_service.resolve_forward_references for the same change.
                     resolved_entity = await self.entity_service.link_resolver.resolve_link(
-                        relation.to_name
+                        relation.to_name, strict=True
                     )
                     if resolved_entity is None or resolved_entity.id == relation.from_id:
                         return 0

src/basic_memory/mcp/async_client.py

@@ -1,13 +1,36 @@
 import os
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import AsyncIterator, Callable, Optional
+from asyncio import Lock
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from dataclasses import dataclass
+from threading import RLock
+from typing import TYPE_CHECKING, Annotated, Any, AsyncIterator, Callable, Optional
 
+from fastapi import Depends, FastAPI, Request
 from httpx import ASGITransport, AsyncClient, Timeout
 from loguru import logger
 
 import logfire
 from basic_memory.config import ConfigManager, ProjectMode
 
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+LocalDatabaseState = tuple["AsyncEngine", "async_sessionmaker[AsyncSession]"]
+_MISSING_STATE_VALUE = object()
+
+
+@dataclass
+class _PreparedLocalAsgiDatabase:
+    active_count: int
+    previous_engine: object
+    previous_session_maker: object
+    dependency_context: AbstractAsyncContextManager[LocalDatabaseState]
+
+
+_prepared_local_asgi_database_lock = RLock()
+_prepared_local_asgi_database_prepare_locks: dict[FastAPI, Lock] = {}
+_prepared_local_asgi_databases: dict[FastAPI, _PreparedLocalAsgiDatabase] = {}
+
 
 def _force_local_mode() -> bool:
     """Check if local mode is forced via environment variable."""
@@ -34,15 +57,12 @@ def _build_timeout() -> Timeout:
     )
 
 
-def _asgi_client(timeout: Timeout) -> AsyncClient:
-    """Create a local ASGI client."""
-    # Import on first local-client use so CLI help/version paths can import
-    # routing helpers without constructing the full FastAPI router graph.
-    from basic_memory.api.app import app as fastapi_app
+def _build_asgi_client(app: FastAPI, timeout: Timeout) -> AsyncClient:
+    """Create a local ASGI client for an already-prepared FastAPI app."""
     from basic_memory.workspace_context import workspace_permalink_headers
 
     return AsyncClient(
-        transport=ASGITransport(app=fastapi_app),
+        transport=ASGITransport(app=app),
         base_url="http://test",
         timeout=timeout,
         # Local ASGI calls still cross the HTTP boundary, so request handlers need
@@ -51,6 +71,169 @@ def _asgi_client(timeout: Timeout) -> AsyncClient:
     )
 
 
+def _get_prepared_local_asgi_database_prepare_lock(app: FastAPI) -> Lock:
+    """Get the async lock that serializes first-time DB preparation for an app."""
+    with _prepared_local_asgi_database_lock:
+        prepare_lock = _prepared_local_asgi_database_prepare_locks.get(app)
+        if prepare_lock is None:
+            prepare_lock = Lock()
+            _prepared_local_asgi_database_prepare_locks[app] = prepare_lock
+        return prepare_lock
+
+
+@asynccontextmanager
+async def _resolve_local_asgi_database(app: FastAPI) -> AsyncIterator[LocalDatabaseState]:
+    """Resolve database state for a local ASGI request."""
+    from fastapi.dependencies.utils import get_dependant, solve_dependencies
+
+    from basic_memory.deps import get_engine_factory
+
+    async def resolve_database_state(
+        database_state: Annotated[LocalDatabaseState, Depends(get_engine_factory)],
+    ) -> LocalDatabaseState:
+        return database_state
+
+    scope: dict[str, Any] = {
+        "type": "http",
+        "asgi": {"version": "3.0"},
+        "method": "GET",
+        "scheme": "http",
+        "path": "/",
+        "raw_path": b"/",
+        "root_path": "",
+        "query_string": b"",
+        "headers": [],
+        "client": ("testclient", 50000),
+        "server": ("testserver", 80),
+        "app": app,
+        "path_params": {},
+    }
+
+    async with AsyncExitStack() as request_stack, AsyncExitStack() as function_stack:
+        scope["fastapi_inner_astack"] = request_stack
+        scope["fastapi_function_astack"] = function_stack
+        request = Request(scope)
+        dependant = get_dependant(path="/", call=resolve_database_state)
+        solved = await solve_dependencies(
+            request=request,
+            dependant=dependant,
+            dependency_overrides_provider=app,
+            async_exit_stack=request_stack,
+            embed_body_fields=False,
+        )
+        if solved.errors:
+            raise RuntimeError(f"Failed to resolve local ASGI database dependency: {solved.errors}")
+
+        yield await resolve_database_state(**solved.values)
+
+
+def _retain_prepared_local_asgi_database(app: FastAPI) -> bool:
+    """Retain an active local ASGI database preparation if one exists."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is None:
+            return False
+
+        active.active_count += 1
+        return True
+
+
+def _install_prepared_local_asgi_database(
+    app: FastAPI,
+    database_state: LocalDatabaseState,
+    dependency_context: AbstractAsyncContextManager[LocalDatabaseState],
+) -> None:
+    """Install local ASGI database state after dependency resolution."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is not None:
+            raise RuntimeError("Local ASGI database state installed while another state is active")
+
+        previous_engine = getattr(app.state, "engine", _MISSING_STATE_VALUE)
+        previous_session_maker = getattr(app.state, "session_maker", _MISSING_STATE_VALUE)
+        engine, session_maker = database_state
+
+        app.state.engine = engine
+        app.state.session_maker = session_maker
+        _prepared_local_asgi_databases[app] = _PreparedLocalAsgiDatabase(
+            active_count=1,
+            previous_engine=previous_engine,
+            previous_session_maker=previous_session_maker,
+            dependency_context=dependency_context,
+        )
+
+
+def _restore_local_asgi_state_attribute(app: FastAPI, name: str, previous_value: object) -> None:
+    """Restore a FastAPI app.state attribute captured before local ASGI preparation."""
+    if previous_value is _MISSING_STATE_VALUE:
+        if hasattr(app.state, name):
+            delattr(app.state, name)
+    else:
+        setattr(app.state, name, previous_value)
+
+
+def _release_prepared_local_asgi_database(
+    app: FastAPI,
+) -> AbstractAsyncContextManager[LocalDatabaseState] | None:
+    """Release local ASGI database state after a client context exits."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is None:
+            raise RuntimeError("Local ASGI database state released without a matching retain")
+
+        active.active_count -= 1
+        if active.active_count > 0:
+            return None
+
+        del _prepared_local_asgi_databases[app]
+        _restore_local_asgi_state_attribute(app, "engine", active.previous_engine)
+        _restore_local_asgi_state_attribute(
+            app,
+            "session_maker",
+            active.previous_session_maker,
+        )
+        return active.dependency_context
+
+
+@asynccontextmanager
+async def _prepared_local_asgi_database(app: FastAPI) -> AsyncIterator[None]:
+    """Initialize local ASGI database state before the first request."""
+    prepare_lock = _get_prepared_local_asgi_database_prepare_lock(app)
+    async with prepare_lock:
+        if not _retain_prepared_local_asgi_database(app):
+            database_context = _resolve_local_asgi_database(app)
+            database_state = await database_context.__aenter__()
+            try:
+                _install_prepared_local_asgi_database(app, database_state, database_context)
+            except Exception:
+                await database_context.__aexit__(None, None, None)
+                raise
+
+    try:
+        yield
+    finally:
+        database_context = _release_prepared_local_asgi_database(app)
+        if database_context is not None:
+            await database_context.__aexit__(None, None, None)
+
+
+@asynccontextmanager
+async def _asgi_client(timeout: Timeout) -> AsyncIterator[AsyncClient]:
+    """Create a local ASGI client."""
+    # Import on first local-client use so CLI help/version paths can import
+    # routing helpers without constructing the full FastAPI router graph.
+    from basic_memory.api.app import app as fastapi_app
+
+    # Trigger: local ASGITransport does not execute FastAPI lifespan startup.
+    # Why: letting request dependencies initialize Postgres can run asyncpg DDL
+    # under Starlette's request loop and trigger CPython's empty-ready-queue race.
+    # Outcome: request handling sees the same app.state database objects as API
+    # lifespan startup would have provided.
+    async with _prepared_local_asgi_database(fastapi_app):
+        async with _build_asgi_client(fastapi_app, timeout) as client:
+            yield client
+
+
 async def _resolve_cloud_token(config) -> str:
     """Resolve cloud token with API key preferred, OAuth fallback."""
     with logfire.span(
@@ -260,7 +443,12 @@ def create_client() -> AsyncClient:
 
     if _force_local_mode() or not _force_cloud_mode():
         logger.info("Creating ASGI client for local Basic Memory API")
-        return _asgi_client(timeout)
+        # Deprecated sync path: create_client() cannot await the local ASGI
+        # pre-initialization used by get_client(), so callers that need proper
+        # resource setup should use the async context manager instead.
+        from basic_memory.api.app import app as fastapi_app
+
+        return _build_asgi_client(fastapi_app, timeout)
 
     logger.info("Creating HTTP client for cloud proxy (legacy create_client path)")
     config = ConfigManager().config

src/basic_memory/mcp/tools/project_management.py

@@ -10,7 +10,12 @@ from typing import Literal
 from fastmcp import Context
 from loguru import logger
 
-from basic_memory.config import ConfigManager, has_cloud_credentials
+from basic_memory.config import (
+    BasicMemoryConfig,
+    ConfigManager,
+    ProjectEntry,
+    has_cloud_credentials,
+)
 from basic_memory.mcp.async_client import (
     _explicit_routing,
     _force_local_mode,
@@ -131,9 +136,66 @@ def _merge_projects(
     return merged
 
 
+def _workspace_entry_priority(entry: WorkspaceProjectEntry) -> tuple[bool, int, str, str]:
+    """Prefer default/personal workspaces when duplicate project permalinks exist."""
+    workspace_type_rank = 0 if entry.workspace.workspace_type == "personal" else 1
+    return (
+        # False sorts before True, so the cloud/default workspace comes first.
+        not entry.workspace.is_default,
+        workspace_type_rank,
+        entry.workspace.name.casefold(),
+        entry.workspace.tenant_id,
+    )
+
+
+def _select_attached_cloud_entry(
+    cloud_entries: tuple[WorkspaceProjectEntry, ...],
+    *,
+    config_entry: ProjectEntry | None,
+    config: BasicMemoryConfig | None,
+) -> WorkspaceProjectEntry | None:
+    """Choose the single cloud row that should inherit local project state."""
+    if not cloud_entries:
+        return None
+
+    preferred_workspace_ids: list[str] = []
+    if config_entry and config_entry.workspace_id:
+        preferred_workspace_ids.append(config_entry.workspace_id)
+    if (
+        config
+        and config.default_workspace
+        and config.default_workspace not in preferred_workspace_ids
+    ):
+        preferred_workspace_ids.append(config.default_workspace)
+
+    # The configured default workspace can differ from the cloud-side default.
+    # Use the cloud default only after explicit local config preferences.
+    default_workspace_entry = next(
+        (entry for entry in cloud_entries if entry.workspace.is_default),
+        None,
+    )
+    if (
+        default_workspace_entry is not None
+        and default_workspace_entry.workspace.tenant_id not in preferred_workspace_ids
+    ):
+        preferred_workspace_ids.append(default_workspace_entry.workspace.tenant_id)
+
+    for workspace_id in preferred_workspace_ids:
+        for entry in cloud_entries:
+            if entry.workspace.tenant_id == workspace_id:
+                return entry
+
+    if len(cloud_entries) == 1:
+        return cloud_entries[0]
+
+    return sorted(cloud_entries, key=_workspace_entry_priority)[0]
+
+
 def _merge_workspace_projects(
     local_list: ProjectList | None,
     cloud_entries: tuple[WorkspaceProjectEntry, ...],
+    *,
+    config: BasicMemoryConfig | None = None,
 ) -> list[dict]:
     """Merge local projects with cloud projects from every accessible workspace."""
     local_by_permalink: dict[str, ProjectItem] = {}
@@ -141,20 +203,40 @@ def _merge_workspace_projects(
         for project in local_list.projects:
             local_by_permalink[project.permalink] = project
 
+    config_by_permalink: dict[str, ProjectEntry] = {}
+    if config:
+        config_by_permalink = {
+            generate_permalink(project_name): entry
+            for project_name, entry in config.projects.items()
+        }
+
+    cloud_entries_by_permalink: dict[str, list[WorkspaceProjectEntry]] = {}
+    for entry in cloud_entries:
+        cloud_entries_by_permalink.setdefault(entry.project.permalink, []).append(entry)
+
+    attached_entry_by_permalink: dict[str, WorkspaceProjectEntry | None] = {}
+    for permalink in local_by_permalink:
+        attached_entry_by_permalink[permalink] = _select_attached_cloud_entry(
+            tuple(cloud_entries_by_permalink.get(permalink, ())),
+            config_entry=config_by_permalink.get(permalink),
+            config=config,
+        )
+
     cloud_permalinks = {entry.project.permalink for entry in cloud_entries}
     merged: list[dict] = []
 
     for entry in sorted(
         cloud_entries,
-        key=lambda item: (
-            not item.workspace.is_default,
-            item.workspace.workspace_type != "personal",
-            item.workspace.name.casefold(),
-            item.project.permalink,
-        ),
+        key=lambda item: (*_workspace_entry_priority(item), item.project.permalink),
     ):
         permalink = entry.project.permalink
-        local_proj = local_by_permalink.get(permalink)
+        local_proj = (
+            local_by_permalink.get(permalink)
+            # WorkspaceProjectEntry is a frozen dataclass containing Pydantic
+            # models, so value equality is the intended comparison here.
+            if attached_entry_by_permalink.get(permalink) == entry
+            else None
+        )
         cloud_proj = entry.project
         source = "local+cloud" if local_proj else "cloud"
         local_path = local_proj.path if local_proj else None
@@ -339,7 +421,7 @@ async def list_memory_projects(
                 )
 
     if cloud_entries:
-        merged = _merge_workspace_projects(local_list, cloud_entries)
+        merged = _merge_workspace_projects(local_list, cloud_entries, config=config)
     else:
         merged = _merge_projects(
             local_list,

src/basic_memory/mcp/tools/write_note.py

@@ -12,7 +12,13 @@ from basic_memory.mcp.project_context import get_project_client, add_project_met
 from basic_memory.mcp.server import mcp
 from fastmcp import Context
 from basic_memory.schemas.base import Entity
-from basic_memory.utils import coerce_dict, parse_tags, validate_project_path
+from basic_memory.utils import (
+    build_qualified_permalink_reference,
+    coerce_dict,
+    parse_tags,
+    validate_project_path,
+)
+from basic_memory.workspace_context import current_workspace_permalink_context
 
 # Define TagType as a Union that can accept either a string or a list of strings or None
 TagType = Union[List[str], str, None]
@@ -274,11 +280,20 @@ async def write_note(
                 else:
                     # Re-raise if it's not a conflict error
                     raise  # pragma: no cover
+            response_permalink = result.permalink
+            workspace_context = current_workspace_permalink_context()
+            if response_permalink and workspace_context is not None:
+                response_permalink = build_qualified_permalink_reference(
+                    active_project.permalink,
+                    response_permalink,
+                    workspace_permalink=workspace_context.workspace_slug,
+                )
+
             summary = [
                 f"# {action} note",
                 f"project: {active_project.name}",
                 f"file_path: {result.file_path}",
-                f"permalink: {result.permalink}",
+                f"permalink: {response_permalink}",
                 f"checksum: {result.checksum[:8] if result.checksum else 'unknown'}",
             ]
 
@@ -315,12 +330,12 @@ async def write_note(
 
             # Log the response with structured data
             logger.info(
-                f"MCP tool response: tool=write_note project={active_project.name} action={action} permalink={result.permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved}"
+                f"MCP tool response: tool=write_note project={active_project.name} action={action} permalink={response_permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved}"
             )
             if output_format == "json":
                 return {
                     "title": result.title,
-                    "permalink": result.permalink,
+                    "permalink": response_permalink,
                     "file_path": result.file_path,
                     "checksum": result.checksum,
                     "action": action.lower(),

src/basic_memory/repository/sqlite_search_repository.py

@@ -620,6 +620,25 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             )
             await session.commit()
 
+    async def drop_vector_tables(self) -> None:
+        """Drop SQLite vector tables on a sqlite-vec-enabled connection."""
+        async with db.scoped_session(self.session_maker) as session:
+            vector_sql_result = await session.execute(
+                text(
+                    "SELECT sql FROM sqlite_master "
+                    "WHERE type = 'table' AND name = 'search_vector_embeddings'"
+                )
+            )
+            vector_sql = vector_sql_result.scalar()
+            if vector_sql and "using vec0" in vector_sql.lower():
+                await self._ensure_sqlite_vec_loaded(session)
+
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_embeddings"))
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_chunks"))
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_index"))
+            await session.commit()
+        self._vector_tables_initialized = False
+
     async def delete_stale_vector_rows(self) -> None:
         """Delete vector rows whose source entities no longer exist."""
         await self._ensure_vector_tables()

src/basic_memory/services/search_service.py

@@ -126,19 +126,23 @@ class SearchService:
 
     async def reindex_all(self, background_tasks: Optional[BackgroundTasks] = None) -> None:
         """Reindex all content from database."""
+        from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
 
         logger.info("Starting full reindex")
         # Clear and recreate search index
         await self.repository.execute_query(text("DROP TABLE IF EXISTS search_index"), params={})
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_embeddings"), params={}
-        )
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_chunks"), params={}
-        )
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_index"), params={}
-        )
+        if isinstance(self.repository, SQLiteSearchRepository):
+            await self.repository.drop_vector_tables()
+        else:
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_embeddings"), params={}
+            )
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_chunks"), params={}
+            )
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_index"), params={}
+            )
         await self.init_search_index()
 
         # Reindex all entities

src/basic_memory/sync/sync_service.py

@@ -1447,7 +1447,16 @@ class SyncService:
                 f"to_name={relation.to_name}"
             )
 
-            resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
+            # Use strict=True: deferred resolution should only fill in to_id when an
+            # exact permalink/title/file_path match exists. The fuzzy fallback (search-based
+            # token match) would silently resolve ambiguous links like
+            # `[[overview (state-management/session-execution)]]` to whichever entity shares
+            # the most tokens, polluting the graph with confidently-wrong edges that no
+            # audit catches. Leaving such relations unresolved keeps to_id=NULL so they
+            # surface as forward references and can be fixed by the producer.
+            resolved_entity = await self.entity_service.link_resolver.resolve_link(
+                relation.to_name, strict=True
+            )
 
             # ignore reference to self
             if resolved_entity and resolved_entity.id != relation.from_id:

test-int/mcp/test_workspace_permalink_integration.py

@@ -70,8 +70,12 @@ def team_workspace() -> WorkspaceInfo:
 @pytest.fixture
 def route_workspaces(app):
     @contextmanager
-    def route(*workspaces: WorkspaceInfo):
-        with _workspace_routing(app, workspaces):
+    def route(*workspaces: WorkspaceInfo, forward_permalink_headers: bool = True):
+        with _workspace_routing(
+            app,
+            workspaces,
+            forward_permalink_headers=forward_permalink_headers,
+        ):
             yield
 
     return route
@@ -117,7 +121,12 @@ def _save_permalink_config(
 
 
 @contextmanager
-def _workspace_routing(app, workspaces: Iterable[WorkspaceInfo]):
+def _workspace_routing(
+    app,
+    workspaces: Iterable[WorkspaceInfo],
+    *,
+    forward_permalink_headers: bool = True,
+):
     """Route MCP tool HTTP calls through an ASGI-backed cloud workspace seam."""
     workspace_list = tuple(workspaces)
     workspace_ids = {workspace.tenant_id for workspace in workspace_list}
@@ -128,10 +137,11 @@ def _workspace_routing(app, workspaces: Iterable[WorkspaceInfo]):
     @asynccontextmanager
     async def factory(workspace: str | None = None):
         assert workspace is None or workspace in workspace_ids
+        headers = workspace_permalink_headers() if forward_permalink_headers else {}
         async with HttpxAsyncClient(
             transport=ASGITransport(app=app),
             base_url="http://test",
-            headers=workspace_permalink_headers(),
+            headers=headers,
         ) as inner:
             yield inner
 
@@ -461,3 +471,39 @@ async def test_team_workspace_permalink_routes_to_specific_workspace(
 
         workspace_read = await _read_json(mcp_server, identifier=expected_permalink)
         assert workspace_read["permalink"] == expected_permalink
+
+
+@pytest.mark.asyncio
+async def test_write_note_by_project_id_qualifies_permalink_when_headers_not_forwarded(
+    mcp_server,
+    test_project,
+    app_config,
+    team_workspace,
+    route_workspaces,
+):
+    """MCP writes should return self-routing IDs even if the API omits slug headers."""
+    _save_permalink_config(app_config, include_project=True, default_project=None)
+
+    title = "Project Id Workspace Permalink"
+    short_permalink = "permalink-suite/project-id-workspace-permalink"
+    expected_permalink = f"{team_workspace.slug}/{test_project.name}/{short_permalink}"
+
+    with route_workspaces(team_workspace, forward_permalink_headers=False):
+        write_payload = await _call_json(
+            mcp_server,
+            "write_note",
+            {
+                "project_id": test_project.external_id,
+                "title": title,
+                "directory": "permalink-suite",
+                "content": f"# {title}\n\nProject ID workspace body.",
+                "output_format": "json",
+            },
+        )
+        assert write_payload["permalink"] == expected_permalink
+
+        workspace_read = await _read_json(
+            mcp_server,
+            identifier=f"memory://{write_payload['permalink']}",
+        )
+        assert workspace_read["title"] == title

tests/indexing/test_batch_indexer.py

@@ -688,6 +688,78 @@ async def test_batch_indexer_index_markdown_file_can_defer_relation_resolution(
     assert source.outgoing_relations[0].to_name == "Deferred Target"
 
 
+@pytest.mark.asyncio
+async def test_batch_indexer_uses_strict_link_resolution_for_deferred_relations(
+    app_config,
+    entity_service,
+    entity_repository,
+    relation_repository,
+    search_service,
+    file_service,
+    project_config,
+    monkeypatch,
+):
+    """Regression: batch indexer's deferred relation resolution must call
+    resolve_link with strict=True.
+
+    Mirror of sync_service.resolve_forward_references. Fuzzy fallback in the
+    deferred path silently fills in to_id from BM25/ts_rank results, polluting
+    the graph with confidently-wrong edges. Entity-creation already uses
+    strict=True; this is the other deferred path.
+    """
+    path = "notes/source.md"
+    await _create_file(
+        project_config.home / path,
+        dedent(
+            """
+            ---
+            title: Source
+            type: note
+            ---
+
+            # Source
+
+            - links_to [[never-resolves-target]]
+            """
+        ),
+    )
+
+    batch_indexer = _make_batch_indexer(
+        app_config,
+        entity_service,
+        entity_repository,
+        relation_repository,
+        search_service,
+        file_service,
+    )
+
+    original_resolve_link = entity_service.link_resolver.resolve_link
+    seen_strict: list[object] = []
+
+    async def spy_resolve_link(*args, **kwargs):
+        seen_strict.append(kwargs.get("strict", False))
+        return await original_resolve_link(*args, **kwargs)
+
+    monkeypatch.setattr(entity_service.link_resolver, "resolve_link", spy_resolve_link)
+
+    await batch_indexer.index_files(
+        {path: await _load_input(file_service, path)},
+        max_concurrent=1,
+    )
+
+    assert seen_strict, "batch indexer did not invoke link_resolver.resolve_link"
+    assert all(strict is True for strict in seen_strict), (
+        f"Deferred resolution must call resolve_link(strict=True). Observed: {seen_strict!r}"
+    )
+
+    # The unresolvable relation stayed unresolved.
+    source = await entity_repository.get_by_file_path(path)
+    assert source is not None
+    assert len(source.outgoing_relations) == 1
+    assert source.outgoing_relations[0].to_id is None
+    assert source.outgoing_relations[0].to_name == "never-resolves-target"
+
+
 @pytest.mark.asyncio
 async def test_batch_indexer_strips_frontmatter_from_search_content_when_body_is_empty(
     app_config,

tests/mcp/test_async_client_modes.py

@@ -1,4 +1,5 @@
 from contextlib import asynccontextmanager
+from typing import AsyncIterator
 
 import httpx
 import pytest
@@ -16,11 +17,15 @@ from basic_memory.mcp.async_client import (
 @pytest.fixture(autouse=True)
 def _reset_async_client_state(monkeypatch):
     async_client_module._client_factory = None
+    async_client_module._prepared_local_asgi_databases.clear()
+    async_client_module._prepared_local_asgi_database_prepare_locks.clear()
     monkeypatch.delenv("BASIC_MEMORY_FORCE_LOCAL", raising=False)
     monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)
     monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
     yield
     async_client_module._client_factory = None
+    async_client_module._prepared_local_asgi_databases.clear()
+    async_client_module._prepared_local_asgi_database_prepare_locks.clear()
 
 
 @pytest.mark.asyncio
@@ -51,6 +56,233 @@ async def test_get_client_default_uses_local_asgi_transport(config_manager):
         assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
 
 
+@pytest.mark.asyncio
+async def test_get_client_preinitializes_local_asgi_database(config_manager, monkeypatch):
+    """Local ASGI routing initializes DB state before request handling."""
+    from basic_memory import db
+    from basic_memory.api.app import app as fastapi_app
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    async def fake_get_or_create_db(db_path):
+        calls.append(db_path)
+        return engine, session_maker
+
+    monkeypatch.setattr(db, "get_or_create_db", fake_get_or_create_db)
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == [cfg.database_path]
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.parametrize("async_override", [False, True])
+@pytest.mark.asyncio
+async def test_get_client_uses_existing_local_asgi_database_override(
+    config_manager,
+    async_override,
+):
+    """Local ASGI routing honors FastAPI test dependency overrides."""
+    from basic_memory.api.app import app as fastapi_app
+    from basic_memory.deps import get_engine_factory
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_overrides = dict(fastapi_app.dependency_overrides)
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    if async_override:
+
+        async def override_engine_factory():
+            calls.append("override")
+            return engine, session_maker
+
+    else:
+
+        def override_engine_factory():
+            calls.append("override")
+            return engine, session_maker
+
+    fastapi_app.dependency_overrides[get_engine_factory] = override_engine_factory
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == ["override"]
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        fastapi_app.dependency_overrides = previous_overrides
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.asyncio
+async def test_get_client_resolves_local_asgi_database_override_with_fastapi_di(
+    config_manager,
+):
+    """Local ASGI DB pre-init honors FastAPI dependency injection and cleanup."""
+    from fastapi import Request
+
+    from basic_memory.api.app import app as fastapi_app
+    from basic_memory.deps import get_engine_factory
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_overrides = dict(fastapi_app.dependency_overrides)
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+    cleanup = []
+
+    async def override_engine_factory(
+        request: Request,
+    ) -> AsyncIterator[tuple[object, object]]:
+        calls.append(request.app)
+        try:
+            yield engine, session_maker
+        finally:
+            cleanup.append("closed")
+
+    fastapi_app.dependency_overrides[get_engine_factory] = override_engine_factory
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == [fastapi_app]
+            assert cleanup == []
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert cleanup == ["closed"]
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        fastapi_app.dependency_overrides = previous_overrides
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.asyncio
+async def test_get_client_keeps_local_asgi_database_during_overlapping_contexts(
+    config_manager,
+    monkeypatch,
+):
+    """Local ASGI database state stays installed until the last overlapping client exits."""
+    from basic_memory import db
+    from basic_memory.api.app import app as fastapi_app
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    async def fake_get_or_create_db(db_path):
+        calls.append(db_path)
+        return engine, session_maker
+
+    monkeypatch.setattr(db, "get_or_create_db", fake_get_or_create_db)
+
+    first_context = get_client()
+    second_context = get_client()
+    first_entered = False
+    second_entered = False
+    first_exited = False
+
+    try:
+        first_client = await first_context.__aenter__()
+        first_entered = True
+        second_client = await second_context.__aenter__()
+        second_entered = True
+
+        assert isinstance(first_client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(second_client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+        assert calls == [cfg.database_path]
+
+        await first_context.__aexit__(None, None, None)
+        first_exited = True
+
+        assert fastapi_app.state.engine is engine
+        assert fastapi_app.state.session_maker is session_maker
+
+        await second_context.__aexit__(None, None, None)
+        second_entered = False
+
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        if second_entered:
+            await second_context.__aexit__(None, None, None)
+        if first_entered and not first_exited:
+            await first_context.__aexit__(None, None, None)
+
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
 @pytest.mark.asyncio
 async def test_get_client_explicit_cloud_uses_api_key(config_manager, monkeypatch):
     cfg = config_manager.load_config()

tests/mcp/test_tool_project_management.py

@@ -9,7 +9,8 @@ from sqlalchemy import select
 
 from basic_memory import db
 from basic_memory.mcp.tools import list_memory_projects, create_memory_project, delete_project
-from basic_memory.mcp.tools.project_management import _merge_projects
+from basic_memory.config import BasicMemoryConfig, ProjectEntry
+from basic_memory.mcp.tools.project_management import _merge_projects, _merge_workspace_projects
 from basic_memory.models.project import Project
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 
@@ -909,6 +910,220 @@ def test_merge_projects_overlap():
     assert merged[0]["workspace_tenant_id"] == "org-456"
 
 
+def test_merge_workspace_projects_attaches_local_state_to_one_duplicate_workspace(tmp_path):
+    """A same-name team workspace project should stay cloud-only (#848)."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "local+cloud"
+    assert personal_project["local_path"] == local_path
+    assert personal_project["path"] == local_path
+    assert team_project["source"] == "cloud"
+    assert team_project["local_path"] is None
+    assert team_project["path"] == "/cloud/team-main"
+
+
+def test_merge_workspace_projects_uses_configured_workspace_for_local_state(tmp_path):
+    """Per-project workspace_id should select the attached duplicate row."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(
+        projects={
+            "main": ProjectEntry(
+                path=local_path,
+                workspace_id="team-tenant",
+            )
+        }
+    )
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "cloud"
+    assert personal_project["local_path"] is None
+    assert personal_project["path"] == "/cloud/personal-main"
+    assert team_project["source"] == "local+cloud"
+    assert team_project["local_path"] == local_path
+    assert team_project["path"] == local_path
+
+
+def test_merge_workspace_projects_uses_default_workspace_for_local_state(tmp_path):
+    """Global default_workspace should attach local state before cloud default fallback."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(
+        projects={"main": ProjectEntry(path=local_path)},
+        default_workspace="team-tenant",
+    )
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "cloud"
+    assert personal_project["local_path"] is None
+    assert personal_project["path"] == "/cloud/personal-main"
+    assert team_project["source"] == "local+cloud"
+    assert team_project["local_path"] == local_path
+    assert team_project["path"] == local_path
+
+
+def test_merge_workspace_projects_sorted_fallback_attaches_personal_workspace(tmp_path):
+    """When config has no preference and no cloud default exists, use stable priority."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=False,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (team_ws, [team_main]),
+            (personal_ws, [personal_main]),
+        ]
+    )
+    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "local+cloud"
+    assert personal_project["local_path"] == local_path
+    assert personal_project["path"] == local_path
+    assert team_project["source"] == "cloud"
+    assert team_project["local_path"] is None
+    assert team_project["path"] == "/cloud/team-main"
+
+
 # --- Workspace passthrough tests ---
 
 

tests/services/test_semantic_search.py

@@ -5,7 +5,9 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
+from sqlalchemy import text
 
+from basic_memory import db
 from basic_memory.repository import EntityRepository
 from basic_memory.repository.search_repository_base import VectorSyncBatchResult
 from basic_memory.repository.semantic_errors import (
@@ -256,6 +258,98 @@ async def test_reindex_vectors_respects_embed_opt_out(search_service, monkeypatc
     }
 
 
+@pytest.mark.asyncio
+async def test_reindex_vectors_purges_sqlite_vectors_before_sync(search_service, monkeypatch):
+    """Regression for #829: stale vec0 rows must be purged before batch sync."""
+    repository = _sqlite_repo(search_service)
+    repository._semantic_enabled = True
+    calls: list[str] = []
+
+    monkeypatch.setattr(
+        search_service.entity_repository,
+        "find_all",
+        AsyncMock(return_value=[SimpleNamespace(id=42, entity_metadata={})]),
+    )
+
+    async def delete_stale_vector_rows():
+        calls.append("purge")
+
+    async def sync_entity_vectors_batch(entity_ids, progress_callback=None):
+        assert entity_ids == [42]
+        assert progress_callback is None
+        calls.append("sync")
+        return VectorSyncBatchResult(entities_total=1, entities_synced=1, entities_failed=0)
+
+    monkeypatch.setattr(repository, "delete_stale_vector_rows", delete_stale_vector_rows)
+    monkeypatch.setattr(search_service, "sync_entity_vectors_batch", sync_entity_vectors_batch)
+
+    stats = await search_service.reindex_vectors()
+
+    assert calls == ["purge", "sync"]
+    assert stats == {
+        "total_entities": 1,
+        "embedded": 1,
+        "skipped": 0,
+        "errors": 0,
+    }
+
+
+@pytest.mark.asyncio
+async def test_reindex_all_uses_sqlite_vec_aware_drop(search_service, monkeypatch):
+    """Full service reindex should not drop vec0 tables through a raw connection."""
+    repository = _sqlite_repo(search_service)
+    executed_sql: list[str] = []
+    calls: list[str] = []
+
+    async def execute_query(query, params=None):
+        executed_sql.append(str(query))
+
+    async def drop_vector_tables():
+        calls.append("drop_vector_tables")
+
+    monkeypatch.setattr(repository, "execute_query", execute_query)
+    monkeypatch.setattr(repository, "drop_vector_tables", drop_vector_tables)
+    monkeypatch.setattr(search_service, "init_search_index", AsyncMock())
+    monkeypatch.setattr(search_service.entity_repository, "find_all", AsyncMock(return_value=[]))
+
+    await search_service.reindex_all()
+
+    assert calls == ["drop_vector_tables"]
+    assert all("search_vector_embeddings" not in sql for sql in executed_sql)
+
+
+@pytest.mark.asyncio
+async def test_drop_vector_tables_skips_sqlite_vec_load_for_plain_table(
+    search_service,
+    monkeypatch,
+):
+    """Plain compatibility tables should not require sqlite-vec to be loaded."""
+    repository = _sqlite_repo(search_service)
+    await repository.drop_vector_tables()
+
+    async with db.scoped_session(repository.session_maker) as session:
+        await session.execute(
+            text("CREATE TABLE search_vector_embeddings (rowid INTEGER PRIMARY KEY)")
+        )
+        await session.commit()
+
+    async def fail_if_loaded(_session):
+        raise AssertionError("plain tables should not load sqlite-vec")
+
+    monkeypatch.setattr(repository, "_ensure_sqlite_vec_loaded", fail_if_loaded)
+
+    await repository.drop_vector_tables()
+
+    async with db.scoped_session(repository.session_maker) as session:
+        result = await session.execute(
+            text(
+                "SELECT name FROM sqlite_master "
+                "WHERE type = 'table' AND name = 'search_vector_embeddings'"
+            )
+        )
+        assert result.scalar() is None
+
+
 @pytest.mark.asyncio
 async def test_reindex_vectors_force_full_clears_project_vectors_before_resync(
     search_service, monkeypatch

tests/sync/test_sync_service.py

@@ -108,6 +108,79 @@ Target content
     assert source.relations[0].to_name == target.title
 
 
+@pytest.mark.asyncio
+async def test_resolve_relations_uses_strict_link_resolution(
+    sync_service: SyncService,
+    project_config: ProjectConfig,
+    entity_service: EntityService,
+    monkeypatch,
+):
+    """Regression: deferred forward-reference resolution must call resolve_link
+    with strict=True.
+
+    Producers sometimes emit disambiguator-style links like
+    `[[overview (state-management/session-execution)]]` whose exact text does
+    not match any entity's permalink, title, or file_path. The previous
+    behavior fell through to BM25/ts_rank fuzzy search in
+    LinkResolver._resolve_in_project and silently picked whichever entity
+    shared the most tokens — polluting the graph with confidently-wrong edges
+    that no audit catches.
+
+    Entity-creation already resolves relations with strict=True (see
+    entity_service.update_entity_relations). The deferred sync path must use
+    the same contract; otherwise unresolved relations get silently filled
+    later by fuzzy search.
+    """
+    # Create a source file with a forward reference. The target doesn't exist,
+    # so resolution will fail — which is exactly when fuzzy fallback would
+    # previously silently pick a wrong target.
+    source_content = dedent("""
+        ---
+        type: knowledge
+        ---
+        # Source
+
+        ## Relations
+        - part_of [[never-resolves-target]]
+    """)
+    await create_test_file(project_config.home / "source.md", source_content)
+    await sync_service.sync(project_config.home)
+
+    project_prefix = generate_permalink(project_config.name)
+    source = await entity_service.get_by_permalink(f"{project_prefix}/source")
+    assert len(source.relations) == 1
+    assert source.relations[0].to_id is None  # initial creation already strict
+
+    # Spy on resolve_link to capture the strict flag the deferred resolver uses.
+    original_resolve_link = sync_service.entity_service.link_resolver.resolve_link
+    seen_strict: list[Any] = []
+
+    async def spy_resolve_link(*args, **kwargs):
+        seen_strict.append(kwargs.get("strict", False))
+        return await original_resolve_link(*args, **kwargs)
+
+    monkeypatch.setattr(
+        sync_service.entity_service.link_resolver,
+        "resolve_link",
+        spy_resolve_link,
+    )
+
+    await sync_service.resolve_relations()
+
+    # Deferred resolution invoked resolve_link, and every call passed strict=True.
+    assert seen_strict, "resolve_relations did not invoke link_resolver.resolve_link"
+    assert all(strict is True for strict in seen_strict), (
+        f"Deferred resolution must call resolve_link(strict=True) to avoid silent "
+        f"fuzzy matching. Observed strict values: {seen_strict!r}"
+    )
+
+    # Sanity check: the unresolvable relation stayed unresolved — no silent
+    # fuzzy match polluted it.
+    source = await entity_service.get_by_permalink(f"{project_prefix}/source")
+    assert source.relations[0].to_id is None
+    assert source.relations[0].to_name == "never-resolves-target"
+
+
 @pytest.mark.asyncio
 async def test_resolve_relations_deletes_duplicate_unresolved_relation(
     sync_service: SyncService,