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

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

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()
                         )

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/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"
+            )