test(mcp): cover write-note overwrite conflict integration

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

src/basic_memory/indexing/batch_indexer.py

@@ -493,14 +493,8 @@ 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, strict=True
+                        relation.to_name
                     )
                     if resolved_entity is None or resolved_entity.id == relation.from_id:
                         return 0

src/basic_memory/mcp/tools/write_note.py

@@ -1,6 +1,7 @@
 """Write note tool for Basic Memory MCP server."""
 
 import textwrap
+from pathlib import Path
 from typing import Annotated, List, Union, Optional, Literal
 
 import logfire
@@ -269,7 +270,19 @@ async def write_note(
                             raise ValueError(
                                 "Entity permalink is required for updates"
                             )  # pragma: no cover
-                        entity_id = await knowledge_client.resolve_entity(entity.permalink)
+                        # Resolve the conflicting entity by file_path with strict=True.
+                        # The 409 came from a file_service.exists(file_path) check, so this
+                        # file_path is the authoritative key for the canonical row. Resolving
+                        # by permalink with fuzzy fallback (the previous behavior) could pick
+                        # an orphan with a similar permalink — especially in workspace-prefixed
+                        # palaces where the client-built permalink omits the workspace slug —
+                        # causing the update to write to the wrong row and the next call to
+                        # mint a -1/-2 suffix on the canonical entity.
+                        # POSIX-normalize so Windows clients send the same form the server stores.
+                        file_path_identifier = Path(entity.file_path).as_posix()
+                        entity_id = await knowledge_client.resolve_entity(
+                            file_path_identifier, strict=True
+                        )
                         result = await knowledge_client.update_entity(
                             entity_id, entity.model_dump()
                         )

src/basic_memory/sync/sync_service.py

@@ -1447,16 +1447,7 @@ class SyncService:
                 f"to_name={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
-            )
+            resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
 
             # ignore reference to self
             if resolved_entity and resolved_entity.id != relation.from_id:

test-int/mcp/test_write_note_integration.py

@@ -5,14 +5,25 @@ Comprehensive tests covering all scenarios including note creation, content form
 tag handling, error conditions, and edge cases from bug reports.
 """
 
+import json
+from pathlib import Path
 from textwrap import dedent
+from typing import Any
 
 import pytest
 from fastmcp import Client
 
 from basic_memory.config import ConfigManager
 from basic_memory.schemas.project_info import ProjectItem
-from pathlib import Path
+
+
+def _json_content(tool_result) -> dict[str, Any]:
+    """Parse a FastMCP tool result content block into a JSON object."""
+    assert len(tool_result.content) == 1
+    assert tool_result.content[0].type == "text"
+    payload = json.loads(tool_result.content[0].text)  # pyright: ignore [reportAttributeAccessIssue]
+    assert isinstance(payload, dict)
+    return payload
 
 
 @pytest.mark.asyncio
@@ -113,6 +124,83 @@ async def test_write_note_update_existing(mcp_server, app, test_project):
         assert f"[Session: Using project '{test_project.name}']" in response_text
 
 
+@pytest.mark.asyncio
+async def test_write_note_overwrite_resolves_conflict_by_file_path_when_permalink_changes(
+    mcp_server, app, test_project, monkeypatch
+):
+    """Overwrite resolves the conflict by strict file path through the MCP client stack."""
+    from basic_memory.mcp.clients import knowledge as knowledge_mod
+
+    original_resolve = knowledge_mod.KnowledgeClient.resolve_entity
+    captured_resolve: dict[str, Any] = {}
+
+    async def spy_resolve(self, identifier: str, *, strict: bool = False) -> str:
+        captured_resolve["identifier"] = identifier
+        captured_resolve["strict"] = strict
+        return await original_resolve(self, identifier, strict=strict)
+
+    monkeypatch.setattr(knowledge_mod.KnowledgeClient, "resolve_entity", spy_resolve)
+
+    async with Client(mcp_server) as client:
+        created = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Overwrite Permalink Change",
+                "directory": "overwrite-conflicts",
+                "content": "# Overwrite Permalink Change\n\nOriginal body.",
+                "output_format": "json",
+            },
+        )
+        created_payload = _json_content(created)
+        assert created_payload["permalink"] == (
+            f"{test_project.name}/overwrite-conflicts/overwrite-permalink-change"
+        )
+
+        replacement = dedent("""
+            ---
+            permalink: overwrite-conflicts/custom-overwrite-permalink
+            ---
+
+            # Overwrite Permalink Change
+
+            Replacement body.
+        """).strip()
+
+        updated = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Overwrite Permalink Change",
+                "directory": "overwrite-conflicts",
+                "content": replacement,
+                "overwrite": True,
+                "output_format": "json",
+            },
+        )
+        updated_payload = _json_content(updated)
+        assert updated_payload["action"] == "updated"
+        assert updated_payload["permalink"] == "overwrite-conflicts/custom-overwrite-permalink"
+        assert updated_payload["file_path"] == "overwrite-conflicts/Overwrite Permalink Change.md"
+        assert captured_resolve == {
+            "identifier": "overwrite-conflicts/Overwrite Permalink Change.md",
+            "strict": True,
+        }
+
+        read_updated = await client.call_tool(
+            "read_note",
+            {
+                "project": test_project.name,
+                "identifier": "overwrite-conflicts/custom-overwrite-permalink",
+                "output_format": "json",
+            },
+        )
+        read_payload = _json_content(read_updated)
+        assert read_payload["permalink"] == "overwrite-conflicts/custom-overwrite-permalink"
+        assert "Replacement body." in read_payload["content"]
+        assert "Original body." not in read_payload["content"]
+
+
 @pytest.mark.asyncio
 async def test_write_note_tag_array(mcp_server, app, test_project):
     """Test creating a note with tag array (Issue #38 regression test)."""

tests/indexing/test_batch_indexer.py

@@ -688,78 +688,6 @@ 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_tool_write_note.py

@@ -1311,3 +1311,76 @@ class TestWriteNoteOverwriteGuard:
         assert "# Created note" in result
         assert f"project: {test_project.name}" in result
         assert "file_path: guard/Brand New Note.md" in result
+
+    @pytest.mark.asyncio
+    async def test_write_note_overwrite_resolves_by_file_path_strictly(
+        self, app, test_project, entity_repository, monkeypatch
+    ):
+        """Regression: overwrite=True must resolve the conflicting entity by
+        file_path with strict=True, not by permalink with fuzzy fallback.
+
+        Bug shape: in workspace-prefixed palaces the client-built permalink
+        omits the workspace slug, so resolve_entity(permalink) with the default
+        strict=False would fall through to fuzzy search and could pick an
+        orphan row sharing tokens with the canonical permalink. The update
+        then wrote to the orphan, the canonical row stayed stale, and the
+        next overwrite minted a -1/-2 suffix because the permalink uniqueness
+        check found duplicate rows.
+
+        The 409 we catch came from a file_service.exists(file_path) check,
+        so file_path is the authoritative key — strict resolution against it
+        is safe even when permalinks are workspace-prefixed elsewhere.
+        """
+        # Spy on the resolve_entity call to assert the identifier and strict flag.
+        from basic_memory.mcp.clients import knowledge as knowledge_mod
+
+        original_resolve = knowledge_mod.KnowledgeClient.resolve_entity
+        captured: dict[str, Any] = {}
+
+        async def spy_resolve(self, identifier, *, strict=False):
+            captured["identifier"] = identifier
+            captured["strict"] = strict
+            return await original_resolve(self, identifier, strict=strict)
+
+        monkeypatch.setattr(knowledge_mod.KnowledgeClient, "resolve_entity", spy_resolve)
+
+        # Create then overwrite the canonical note.
+        await write_note(
+            project=test_project.name,
+            title="Overview",
+            directory="features/foo",
+            content="# Overview\n\nVersion A",
+        )
+        canonical_permalink = f"{test_project.name}/features/foo/overview"
+        canonical = await entity_repository.get_by_permalink(canonical_permalink)
+        assert canonical is not None
+        canonical_id = canonical.id
+
+        result = await write_note(
+            project=test_project.name,
+            title="Overview",
+            directory="features/foo",
+            content="# Overview\n\nVersion B",
+            overwrite=True,
+        )
+        assert "# Updated note" in result
+
+        # The overwrite path resolved by file_path with strict=True — not by
+        # permalink with the default fuzzy fallback.
+        assert captured.get("identifier") == "features/foo/Overview.md"
+        assert captured.get("strict") is True
+
+        # And the canonical row was updated in place — no duplicate -1/-2 row.
+        canonical_after = await entity_repository.get_by_permalink(canonical_permalink)
+        assert canonical_after is not None
+        assert canonical_after.id == canonical_id
+
+        content = await read_note(canonical_permalink, project=test_project.name)
+        assert "Version B" in content
+        assert "Version A" not in content
+
+        for suffix in ("-1", "-2"):
+            stray = await entity_repository.get_by_permalink(f"{canonical_permalink}{suffix}")
+            assert stray is None, (
+                f"overwrite=True minted a stray '{suffix}' suffix on the canonical permalink"
+            )

tests/sync/test_sync_service.py

@@ -108,79 +108,6 @@ 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,