feat: add overwrite guard to write_note tool (#632)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

src/basic_memory/config.py

@@ -245,6 +245,16 @@ class BasicMemoryConfig(BaseSettings):
         description="Disable automatic permalink generation in frontmatter. When enabled, new notes won't have permalinks added and sync won't update permalinks. Existing permalinks will still work for reading.",
     )
 
+    write_note_overwrite_default: bool = Field(
+        default=False,
+        description=(
+            "Default value for write_note's overwrite parameter. "
+            "When False (default), write_note errors if note already exists. "
+            "Set to True to restore pre-v0.20 upsert behavior. "
+            "Env: BASIC_MEMORY_WRITE_NOTE_OVERWRITE_DEFAULT"
+        ),
+    )
+
     ensure_frontmatter_on_sync: bool = Field(
         default=True,
         description="Ensure markdown files have frontmatter during sync by adding derived title/type/permalink when missing. When combined with disable_permalinks=True, this setting takes precedence for missing-frontmatter files and still writes permalinks.",

src/basic_memory/mcp/tools/write_note.py

@@ -1,9 +1,11 @@
 """Write note tool for Basic Memory MCP server."""
 
+import textwrap
 from typing import List, Union, Optional, Literal
 
 from loguru import logger
 
+from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import get_project_client, add_project_metadata
 from basic_memory.mcp.server import mcp
 from fastmcp import Context
@@ -15,8 +17,8 @@ TagType = Union[List[str], str, None]
 
 
 @mcp.tool(
-    description="Create or update a markdown note. Returns a markdown formatted summary of the semantic content.",
-    annotations={"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
+    description="Create a markdown note. If the note already exists, returns an error by default — pass overwrite=True to replace.",
+    annotations={"destructiveHint": True, "idempotentHint": False, "openWorldHint": False},
 )
 async def write_note(
     title: str,
@@ -27,12 +29,15 @@ async def write_note(
     tags: list[str] | str | None = None,
     note_type: str = "note",
     metadata: dict | None = None,
+    overwrite: bool | None = None,
     output_format: Literal["text", "json"] = "text",
     context: Context | None = None,
 ) -> str | dict:
     """Write a markdown note to the knowledge base.
 
-    Creates or updates a markdown note with semantic observations and relations.
+    Creates a markdown note with semantic observations and relations.
+    If the note already exists, returns an error by default. Pass overwrite=True
+    to replace the existing note. For incremental updates, use edit_note instead.
 
     Project Resolution:
     Server resolves projects using a unified priority chain (same in local and cloud modes):
@@ -74,6 +79,8 @@ async def write_note(
         metadata: Optional dict of extra frontmatter fields merged into entity_metadata.
                   Useful for schema notes or any note that needs custom YAML frontmatter
                   beyond title/type/tags. Nested dicts are supported.
+        overwrite: If True, replace existing note on conflict. If False, error on conflict.
+                   If None (default), consult write_note_overwrite_default config setting.
         output_format: "text" returns the existing markdown summary. "json" returns
                        machine-readable metadata.
         context: Optional FastMCP context for performance caching.
@@ -106,12 +113,13 @@ async def write_note(
             note_type="guide"
         )
 
-        # Update existing note (same title/directory)
+        # Overwrite an existing note explicitly
         write_note(
             project="my-research",
             title="Meeting Notes",
             directory="meetings",
-            content="# Weekly Standup\\n\\n- [decision] Use PostgreSQL instead #tech"
+            content="# Weekly Standup\\n\\n- [decision] Use PostgreSQL instead #tech",
+            overwrite=True
         )
 
         # Create a schema note with custom frontmatter via metadata
@@ -132,6 +140,14 @@ async def write_note(
         HTTPError: If project doesn't exist or is inaccessible
         SecurityError: If directory path attempts path traversal
     """
+    # Resolve overwrite flag: explicit parameter > config default
+    # Trigger: caller omitted the parameter (None)
+    # Why: lets users set a global default without breaking per-call overrides
+    effective_overwrite = (
+        overwrite if overwrite is not None
+        else ConfigManager().config.write_note_overwrite_default
+    )
+
     async with get_project_client(project, workspace, context) as (client, active_project):
         logger.info(
             f"MCP tool call tool=write_note project={active_project.name} directory={directory}, title={title}, tags={tags}"
@@ -199,6 +215,25 @@ async def write_note(
                 or "conflict" in str(e).lower()
                 or "already exists" in str(e).lower()
             ):
+                # Guard: block overwrite unless explicitly enabled
+                if not effective_overwrite:
+                    logger.warning(
+                        f"write_note blocked: note already exists (overwrite not enabled) "
+                        f"permalink={entity.permalink}"
+                    )
+                    if output_format == "json":
+                        return {
+                            "title": title,
+                            "permalink": entity.permalink,
+                            "file_path": None,
+                            "checksum": None,
+                            "action": "conflict",
+                            "error": "NOTE_ALREADY_EXISTS",
+                        }
+                    return _format_overwrite_error(
+                        title, entity.permalink, active_project.name
+                    )
+
                 logger.debug(f"Entity exists, updating instead permalink={entity.permalink}")
                 try:
                     if not entity.permalink:
@@ -270,3 +305,23 @@ async def write_note(
 
         summary_result = "\n".join(summary)
         return add_project_metadata(summary_result, active_project.name)
+
+
+def _format_overwrite_error(title: str, permalink: str | None, project_name: str) -> str:
+    """Format a helpful error when write_note is blocked by the overwrite guard."""
+    return textwrap.dedent(f"""\
+        # Error: Note already exists
+
+        **"{title}"** already exists (permalink: `{permalink}`).
+
+        `write_note` does not overwrite by default. Choose an option:
+
+        | Goal | Action |
+        |------|--------|
+        | Append content | `edit_note("{permalink}", operation="append", content="...")` |
+        | Prepend content | `edit_note("{permalink}", operation="prepend", content="...")` |
+        | Replace a section | `edit_note("{permalink}", operation="replace_section", section="...", content="...")` |
+        | Full replace | `write_note("{title}", ..., overwrite=True)` |
+        | Inspect first | `read_note("{permalink}")` |
+
+        Project: {project_name}""")

test-int/mcp/test_write_note_integration.py

@@ -88,7 +88,7 @@ async def test_write_note_update_existing(mcp_server, app, test_project):
 
         assert "# Created note" in result1.content[0].text  # pyright: ignore [reportAttributeAccessIssue]
 
-        # Update the same note
+        # Update the same note (explicit overwrite)
         result2 = await client.call_tool(
             "write_note",
             {
@@ -97,6 +97,7 @@ async def test_write_note_update_existing(mcp_server, app, test_project):
                 "directory": "test",
                 "content": "# Update Test\n\nUpdated content with changes.",
                 "tags": "updated,modified",
+                "overwrite": True,
             },
         )
 
@@ -475,3 +476,49 @@ async def test_write_note_project_path_validation(mcp_server, app, test_project)
         # Should successfully create without path validation errors
         assert "# Created note" in response_text
         assert "not allowed" not in response_text
+
+
+@pytest.mark.asyncio
+async def test_write_note_overwrite_guard_via_mcp_client(mcp_server, app, test_project):
+    """End-to-end test: overwrite guard works through the MCP Client protocol."""
+
+    async with Client(mcp_server) as client:
+        # Create initial note
+        result1 = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "MCP Guard Test",
+                "directory": "guard",
+                "content": "# MCP Guard Test\n\nOriginal content via MCP.",
+            },
+        )
+        assert "# Created note" in result1.content[0].text  # pyright: ignore [reportAttributeAccessIssue]
+
+        # Second write without overwrite should be blocked
+        result2 = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "MCP Guard Test",
+                "directory": "guard",
+                "content": "# MCP Guard Test\n\nReplacement content via MCP.",
+            },
+        )
+        response_text = result2.content[0].text  # pyright: ignore [reportAttributeAccessIssue]
+        assert "# Error: Note already exists" in response_text
+        assert "edit_note" in response_text
+
+        # Overwrite with explicit flag should succeed
+        result3 = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "MCP Guard Test",
+                "directory": "guard",
+                "content": "# MCP Guard Test\n\nReplacement content via MCP.",
+                "overwrite": True,
+            },
+        )
+        response_text3 = result3.content[0].text  # pyright: ignore [reportAttributeAccessIssue]
+        assert "# Updated note" in response_text3

tests/mcp/test_obsidian_yaml_formatting.py

@@ -143,6 +143,7 @@ async def test_write_note_update_preserves_yaml_format(app, project_config, test
         directory="test",
         content="Updated content",
         tags=["updated", "new-tag", "format"],
+        overwrite=True,
     )
 
     # Should be an update, not a new creation

tests/mcp/test_permalink_collision_file_overwrite.py

@@ -168,6 +168,7 @@ async def test_notes_with_similar_titles_maintain_separate_files(app, test_proje
             title=title,
             directory=folder,
             content=f"# {title}\n\nUnique content for {title}",
+            overwrite=True,
         )
 
         permalink = None

tests/mcp/test_tool_contracts.py

@@ -99,6 +99,7 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
         "tags",
         "note_type",
         "metadata",
+        "overwrite",
         "output_format",
     ],
 }

tests/mcp/test_tool_json_output_modes.py

@@ -38,6 +38,7 @@ async def test_write_note_text_and_json_modes(app, test_project):
         directory="mode-tests",
         content="# Mode Write Note\n\nupdated",
         output_format="json",
+        overwrite=True,
     )
     assert isinstance(json_result, dict)
     assert json_result["title"] == "Mode Write Note"

tests/mcp/test_tool_write_note.py

@@ -1,8 +1,10 @@
 """Tests for note tools that exercise the full stack with SQLite."""
 
 from textwrap import dedent
+
 import pytest
 
+from basic_memory import config as config_module
 from basic_memory.mcp.tools import write_note, read_note, delete_note
 from basic_memory.utils import normalize_newlines
 
@@ -120,6 +122,7 @@ async def test_write_note_update_existing(app, test_project):
         directory="test",
         content="# Test\nThis is an updated note",
         tags=["test", "documentation"],
+        overwrite=True,
     )
     assert "# Updated note" in result
     assert f"project: {test_project.name}" in result
@@ -229,6 +232,7 @@ async def test_issue_93_write_note_respects_custom_permalink_existing_note(app,
         title="Existing Note",
         directory="test",
         content=updated_content,
+        overwrite=True,
     )
 
     # Verify the custom permalink is respected
@@ -385,6 +389,7 @@ async def test_write_note_preserves_custom_metadata(app, project_config, test_pr
         directory="test",
         content="# Updated content",
         tags=["test", "updated"],
+        overwrite=True,
     )
 
     # Verify the update was successful
@@ -501,6 +506,7 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
         title="Note 1",  # Same title as first note
         directory="test",  # Same folder as first note
         content="Replacement content for note 1",  # Different content
+        overwrite=True,
     )
 
     # This should not raise a UNIQUE constraint failure error
@@ -681,6 +687,7 @@ async def test_write_note_update_existing_with_different_note_type(app, test_pro
         content="# Updated Content\nThis is now a guide",
         tags=["guide"],
         note_type="guide",
+        overwrite=True,
     )
 
     assert result2
@@ -1143,3 +1150,148 @@ class TestWriteNoteSecurityEdgeCases:
             if ".." in attack_folder.strip() or "~" in attack_folder.strip():
                 assert "# Error" in result
                 assert "paths must stay within project boundaries" in result
+
+
+class TestWriteNoteOverwriteGuard:
+    """Test the write_note overwrite guard feature (Issue #625)."""
+
+    @pytest.mark.asyncio
+    async def test_write_note_blocks_overwrite_by_default(self, app, test_project):
+        """Second write_note to same title/directory returns error, original content untouched."""
+        # Create initial note
+        result1 = await write_note(
+            project=test_project.name,
+            title="Guard Test",
+            directory="guard",
+            content="# Guard Test\n\nOriginal content",
+        )
+        assert "# Created note" in result1
+
+        # Second write without overwrite should be blocked
+        result2 = await write_note(
+            project=test_project.name,
+            title="Guard Test",
+            directory="guard",
+            content="# Guard Test\n\nReplacement content",
+        )
+        assert "# Error: Note already exists" in result2
+        assert "Guard Test" in result2
+        assert "edit_note" in result2
+
+        # Original content should be untouched
+        content = await read_note("guard/guard-test", project=test_project.name)
+        assert "Original content" in content
+        assert "Replacement content" not in content
+
+    @pytest.mark.asyncio
+    async def test_write_note_overwrite_false_explicit(self, app, test_project):
+        """Explicit overwrite=False behaves same as default."""
+        await write_note(
+            project=test_project.name,
+            title="Explicit False",
+            directory="guard",
+            content="# Explicit False\n\nOriginal",
+        )
+
+        result = await write_note(
+            project=test_project.name,
+            title="Explicit False",
+            directory="guard",
+            content="# Explicit False\n\nReplacement",
+            overwrite=False,
+        )
+        assert "# Error: Note already exists" in result
+
+    @pytest.mark.asyncio
+    async def test_write_note_overwrite_true_replaces(self, app, test_project):
+        """Explicit overwrite=True performs the update."""
+        await write_note(
+            project=test_project.name,
+            title="Overwrite True",
+            directory="guard",
+            content="# Overwrite True\n\nOriginal content",
+        )
+
+        result = await write_note(
+            project=test_project.name,
+            title="Overwrite True",
+            directory="guard",
+            content="# Overwrite True\n\nReplacement content",
+            overwrite=True,
+        )
+        assert "# Updated note" in result
+
+        # Verify content was replaced
+        content = await read_note("guard/overwrite-true", project=test_project.name)
+        assert "Replacement content" in content
+        assert "Original content" not in content
+
+    @pytest.mark.asyncio
+    async def test_write_note_overwrite_error_json_format(self, app, test_project):
+        """JSON output returns structured error with NOTE_ALREADY_EXISTS."""
+        await write_note(
+            project=test_project.name,
+            title="JSON Guard",
+            directory="guard",
+            content="# JSON Guard\n\nOriginal",
+        )
+
+        result = await write_note(
+            project=test_project.name,
+            title="JSON Guard",
+            directory="guard",
+            content="# JSON Guard\n\nReplacement",
+            output_format="json",
+        )
+        assert isinstance(result, dict)
+        assert result["error"] == "NOTE_ALREADY_EXISTS"
+        assert result["action"] == "conflict"
+        assert result["title"] == "JSON Guard"
+        assert result["permalink"] is not None
+
+    @pytest.mark.asyncio
+    async def test_write_note_config_overwrite_default_true(
+        self, app, test_project, app_config, config_manager
+    ):
+        """Config write_note_overwrite_default=True restores old upsert behavior."""
+        # Set config to allow overwrites by default
+        app_config.write_note_overwrite_default = True
+        config_module._CONFIG_CACHE = app_config
+
+        try:
+            await write_note(
+                project=test_project.name,
+                title="Config Default",
+                directory="guard",
+                content="# Config Default\n\nOriginal",
+            )
+
+            result = await write_note(
+                project=test_project.name,
+                title="Config Default",
+                directory="guard",
+                content="# Config Default\n\nReplacement via config default",
+            )
+            # Should succeed as update because config default is True
+            assert "# Updated note" in result
+
+            content = await read_note("guard/config-default", project=test_project.name)
+            assert "Replacement via config default" in content
+        finally:
+            # Restore config
+            app_config.write_note_overwrite_default = False
+            config_module._CONFIG_CACHE = app_config
+
+    @pytest.mark.asyncio
+    async def test_write_note_new_note_unaffected(self, app, test_project):
+        """Guard only triggers on conflict — new notes are created normally."""
+        result = await write_note(
+            project=test_project.name,
+            title="Brand New Note",
+            directory="guard",
+            content="# Brand New Note\n\nFresh content",
+            tags=["new"],
+        )
+        assert "# Created note" in result
+        assert f"project: {test_project.name}" in result
+        assert "file_path: guard/Brand New Note.md" in result

tests/mcp/test_tool_write_note_metadata.py

@@ -114,6 +114,7 @@ async def test_metadata_survives_update(app, test_project):
         directory="meta-tests",
         content="# Version 2",
         metadata={"author": "Bob", "version": 2},
+        overwrite=True,
     )
 
     assert "# Updated note" in result