fix(mcp): edit_note reports a refused write as an error

src/basic_memory/cli/commands/tool.py

@@ -959,6 +959,8 @@ def edit_note(
     bm tool edit-note my-note --operation replace_section --section "## Notes" --content "updated" --no-replace-subsections
     """
     # Deferred: loading the MCP tool stack at module import slows CLI startup (#886).
+    from fastmcp.exceptions import ToolError
+
     from basic_memory.mcp.tools import edit_note as mcp_edit_note
 
     try:
@@ -986,6 +988,11 @@ def edit_note(
             raise typer.Exit(1)
 
         _print_json(result)
+    except ToolError as e:
+        # A failed edit is a tool error whose message, in JSON mode, is the
+        # structured result; report it the same way as an error field.
+        typer.echo(f"Error: {_tool_error_payload(e).get('error') or e}", err=True)
+        raise typer.Exit(1)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)

src/basic_memory/mcp/tools/edit_note.py

@@ -1,6 +1,7 @@
 """Edit note tool for Basic Memory MCP server."""
 
-from typing import Any, TYPE_CHECKING, Annotated, Literal, Optional
+import json
+from typing import Any, TYPE_CHECKING, Annotated, Literal, NoReturn, Optional
 
 import frontmatter
 import logfire
@@ -233,6 +234,15 @@ def _format_cross_project_entity_response(
 Retry with `project_id="{target_project_id}"`, or use `list_memory_projects()` to confirm the intended project before editing."""
 
 
+def _raise_edit_failure(output_format: str, payload: dict[str, Any], text: str) -> NoReturn:
+    """Report a failed edit as a tool error, keeping the guidance for the caller.
+
+    A returned "Edit Failed" string reads as success to MCP clients; raising makes
+    the result an error (isError) while the message still carries the same help.
+    """
+    raise ToolError(json.dumps(payload) if output_format == "json" else text)
+
+
 def _format_error_response(
     error_message: str,
     operation: str,
@@ -517,6 +527,9 @@ async def edit_note(
                    metadata={"status": "resolved", "closed_at": "2026-06-18T10:42:00Z"})
 
     Raises:
+        ToolError: If the edit fails or is refused (for example, the note was modified
+            concurrently). The message carries the troubleshooting guidance, or the
+            structured result in JSON mode.
         HTTPError: If project doesn't exist or is inaccessible
         ValueError: If operation is invalid or required parameters are missing
         SecurityError: If identifier attempts path traversal
@@ -888,8 +901,9 @@ async def edit_note(
             except Exception as e:
                 logger.error(f"Error editing note: {e}")
                 if isinstance(e, UnresolvedProjectRouteError):
-                    if output_format == "json":
-                        return {
+                    _raise_edit_failure(
+                        output_format,
+                        {
                             "title": None,
                             "permalink": None,
                             "file_path": None,
@@ -899,26 +913,31 @@ async def edit_note(
                             "error": "UNRESOLVED_PROJECT_ROUTE",
                             "project": active_project.name,
                             "projectRoute": e.project_prefix,
-                        }
-                    return _format_unresolved_project_route_response(
-                        error=e,
-                        active_project=active_project.name,
+                        },
+                        _format_unresolved_project_route_response(
+                            error=e,
+                            active_project=active_project.name,
+                        ),
                     )
-                if output_format == "json":
-                    return {
+                # A refused edit (for example a 409 when the note was modified
+                # concurrently) must not read as success, or the caller never retries it.
+                _raise_edit_failure(
+                    output_format,
+                    {
                         "title": None,
                         "permalink": None,
                         "file_path": None,
                         "checksum": None,
                         "operation": operation,
                         "fileCreated": False,
                         "error": str(e),
-                    }
-                return _format_error_response(
-                    str(e),
-                    operation,
-                    identifier,
-                    find_text,
-                    effective_replacements,
-                    active_project.name,
+                    },
+                    _format_error_response(
+                        str(e),
+                        operation,
+                        identifier,
+                        find_text,
+                        effective_replacements,
+                        active_project.name,
+                    ),
                 )

test-int/mcp/test_concurrent_write_integration.py

@@ -582,3 +582,71 @@ async def read_one(index: int):
             assert f"Volume body {index}." in payload["content"], (
                 f"note {index} content missing under load: {payload}"
             )
+
+
+@pytest.mark.asyncio
+async def test_concurrent_edit_append_reports_every_refusal(mcp_server, app, test_project) -> None:
+    """Every append reported as written is in the note, and every refused one is not.
+
+    Two clients append to one note at once. On SQLite the db_version compare-and-set
+    refuses an append that loses the race ("modified concurrently"); on Postgres the
+    row lock usually serializes them and nothing is refused. Either way, a caller that
+    trusts ``is_error`` must be able to account for every line: an append refused but
+    returned as an ordinary result would be counted as written and never retried.
+    """
+    appends_per_writer = 40
+
+    async with Client(mcp_server) as setup:
+        note = _parse(
+            await setup.call_tool(
+                "write_note",
+                {
+                    "project": test_project.name,
+                    "title": "Shared Page",
+                    "directory": "shared",
+                    "content": "start",
+                    "output_format": "json",
+                },
+            )
+        )
+
+    outcomes: list[tuple[str, bool]] = []
+
+    async def writer(tag: str) -> None:
+        async with Client(mcp_server) as client:
+            for index in range(appends_per_writer):
+                result = await client.call_tool(
+                    "edit_note",
+                    {
+                        "project": test_project.name,
+                        "identifier": note["permalink"],
+                        "operation": "append",
+                        "content": f"\n{tag}-{index}\n",
+                        "output_format": "json",
+                    },
+                    raise_on_error=False,
+                )
+                outcomes.append((f"{tag}-{index}", result.is_error))
+
+    await asyncio.gather(writer("A"), writer("B"))
+
+    async with Client(mcp_server) as reader:
+        body = _parse(
+            await reader.call_tool(
+                "read_note",
+                {
+                    "project": test_project.name,
+                    "identifier": note["permalink"],
+                    "output_format": "json",
+                },
+            )
+        )["content"]
+
+    kept = {line for line in body.splitlines() if line.startswith(("A-", "B-"))}
+    reported_written = {tag for tag, is_error in outcomes if not is_error}
+    refused = {tag for tag, is_error in outcomes if is_error}
+
+    assert len(outcomes) == 2 * appends_per_writer
+    missing = sorted(reported_written - kept)
+    assert missing == [], f"appends reported as written but absent from the note: {missing}"
+    assert kept.isdisjoint(refused), f"refused appends present in the note: {kept & refused}"

test-int/mcp/test_edit_note_integration.py

@@ -4,12 +4,17 @@
 Tests the complete edit note workflow: MCP client -> MCP server -> FastAPI -> database
 """
 
+import json
 from pathlib import Path
 
 import pytest
 from fastmcp import Client
 
 from basic_memory.file_utils import parse_frontmatter
+from basic_memory.repository.note_content_repository import (
+    NoteContentRepository,
+    NoteContentVersionConflict,
+)
 
 
 @pytest.mark.asyncio
@@ -379,9 +384,11 @@ async def test_edit_note_error_handling_note_not_found(mcp_server, app, test_pro
                 "content": "replacement",
                 "find_text": "old text",
             },
+            raise_on_error=False,
         )
 
-        # Should return helpful error message
+        # A failed edit is an MCP error result, and its text keeps the guidance
+        assert edit_result.is_error is True
         assert len(edit_result.content) == 1
         error_text = edit_result.content[0].text
         assert "Edit Failed" in error_text
@@ -588,8 +595,11 @@ async def test_edit_note_rejects_blank_metadata_type(mcp_server, app, test_proje
                 "content": "",
                 "metadata": {"type": ""},
             },
+            raise_on_error=False,
         )
 
+        # A failed edit is an MCP error result, and its text keeps the guidance
+        assert edit_result.is_error is True
         assert "Edit Failed" in edit_result.content[0].text
         assert "at least 1 item" in edit_result.content[0].text
         read_result = await client.call_tool(
@@ -626,9 +636,11 @@ async def test_edit_note_error_handling_text_not_found(mcp_server, app, test_pro
                 "content": "replacement text",
                 "find_text": "non-existent text",
             },
+            raise_on_error=False,
         )
 
-        # Should return helpful error message
+        # A failed edit is an MCP error result, and its text keeps the guidance
+        assert edit_result.is_error is True
         assert len(edit_result.content) == 1
         error_text = edit_result.content[0].text
         assert "Edit Failed - Text Not Found" in error_text
@@ -637,6 +649,62 @@ async def test_edit_note_error_handling_text_not_found(mcp_server, app, test_pro
         assert "read_note(" in error_text
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("output_format", ["text", "json"])
+async def test_edit_note_refused_concurrent_write_is_an_error(
+    mcp_server, app, test_project, monkeypatch, output_format
+):
+    """An edit the engine refuses with a 409 must reach the client as an MCP error.
+
+    Two writers racing on one note can make the db_version compare-and-set refuse the
+    second edit. Forcing that conflict deterministically: the refused append changed
+    nothing, so a result that is not is_error would tell the caller it was written.
+    """
+
+    async with Client(mcp_server) as client:
+        created = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Refused Edit Note",
+                "directory": "test",
+                "content": "# Refused Edit Note\n\nOriginal body.",
+                "output_format": "json",
+            },
+        )
+        note = json.loads(created.content[0].text)
+        path = Path(test_project.path) / note["file_path"]
+        original = path.read_text(encoding="utf-8")
+
+        async def lose_the_race(self, session, write):
+            raise NoteContentVersionConflict(f"db_version advanced for {write.entity_id}")
+
+        monkeypatch.setattr(NoteContentRepository, "accept_write", lose_the_race)
+
+        edit_result = await client.call_tool(
+            "edit_note",
+            {
+                "project": test_project.name,
+                "identifier": note["permalink"],
+                "operation": "append",
+                "content": "\nThis append was refused.",
+                "output_format": output_format,
+            },
+            raise_on_error=False,
+        )
+
+        assert edit_result.is_error is True
+        text = edit_result.content[0].text
+        assert "modified concurrently" in text
+        if output_format == "json":
+            payload = json.loads(text)
+            assert payload["checksum"] is None
+            assert payload["fileCreated"] is False
+        else:
+            assert "# Edit Failed" in text
+        assert path.read_text(encoding="utf-8") == original
+
+
 @pytest.mark.asyncio
 async def test_edit_note_error_handling_wrong_replacement_count(mcp_server, app, test_project):
     """Test error handling when expected_replacements doesn't match actual occurrences."""
@@ -669,9 +737,11 @@ async def test_edit_note_error_handling_wrong_replacement_count(mcp_server, app,
                 "find_text": "test",
                 "expected_replacements": 5,
             },
+            raise_on_error=False,
         )
 
-        # Should return helpful error message about count mismatch
+        # A failed edit is an MCP error result, and its text keeps the guidance
+        assert edit_result.is_error is True
         assert len(edit_result.content) == 1
         error_text = edit_result.content[0].text
         assert "Edit Failed - Wrong Replacement Count" in error_text
@@ -965,8 +1035,10 @@ async def test_edit_note_append_autocreate_does_not_fuzzy_match(mcp_server, app,
                 "content": "replaced",
                 "find_text": "Content",
             },
+            raise_on_error=False,
         )
 
+        assert edit_result2.is_error is True
         error_text = edit_result2.content[0].text
         assert "Edit Failed" in error_text
 

tests/cli/test_cli_tool_json_output.py

@@ -586,6 +586,37 @@ def test_edit_note_error_response(mock_mcp_edit):
     assert result.exit_code == 1
 
 
+@patch(
+    "basic_memory.mcp.tools.edit_note",
+    new_callable=AsyncMock,
+    side_effect=ToolError(
+        json.dumps(
+            {
+                "title": None,
+                "permalink": None,
+                "file_path": None,
+                "checksum": None,
+                "operation": "append",
+                "fileCreated": False,
+                "error": "The note was modified concurrently. Reload the latest content and retry.",
+            }
+        )
+    ),
+)
+def test_edit_note_reports_a_tool_error_payload_and_exits_nonzero(
+    mock_mcp_edit: AsyncMock,
+) -> None:
+    """A failed edit raised as a tool error prints its error field, not raw JSON."""
+    result = runner.invoke(
+        cli_app,
+        ["tool", "edit-note", "test-note", "--operation", "append", "--content", "content"],
+    )
+
+    assert result.exit_code == 1
+    assert "Error: The note was modified concurrently." in result.output
+    assert '"fileCreated"' not in result.output
+
+
 # --- build-context ---