feat(mcp): Phase 5 — write-time near-duplicate gate (update-over-create)

#755 Phase 5. create_note / create_task now BLOCK a near-duplicate instead of
silently inserting: they return {"duplicate": true, "existing_id", message}
pointing at the record to UPDATE. Fights store bloat and stale competing copies
that semantic search (RAG) would otherwise resurface for reconciliation. A
force=true override creates anyway for genuinely-distinct records.

- services/dedup.py: find_duplicate_note — two signals, scoped to owner + same
  project + same kind: (1) normalized-title exact match (cheap, always); (2)
  semantic cosine ≥ 0.90 but ONLY when body ≥ 200 chars (short/title-only
  embeddings false-positive — the pre-pivot lesson). Project-less (orphan)
  records compare only to other orphans on BOTH signals (orphan_only on the
  semantic call) — they're not matched across every project.
- Gate wired into the MCP create_note/create_task tools (the LLM write path)
  with force override; _INSTRUCTIONS documents the duplicate response + force.
- Opt-in by design: the service helper is only called from the interactive
  create tools. Internal/programmatic creates (recurrence spawn, imports) go
  straight through services.create_note and are NOT gated — a recurring task
  spawning its next same-titled instance must not be blocked.
- Scope v1: MCP tools only. REST/web (human CRUD, needs a UI affordance) and
  create_rule (not a RAG surface; _INSTRUCTIONS already steer it) are follow-ups.
- tests: dedup service (title/semantic/body-gate/type-filter) + tool gate
  (blocks, force bypasses) for notes and tasks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/scribe/mcp/server.py

@@ -59,6 +59,12 @@ not something you wait to be asked for:
   before you re-derive it or open a duplicate.
 - Before creating a task, search for an existing one (search content_type=
   'task') — don't open a second task for work already tracked.
+- create_note / create_task enforce this: if a title- or meaning-similar record
+  already exists in the same project, the call is BLOCKED and returns
+  {"duplicate": true, "existing_id": ...} instead of creating. UPDATE that
+  record (update_note / update_task / add_task_log) rather than duplicating.
+  Only pass force=true when it's genuinely a distinct record — a duplicate both
+  bloats the store and surfaces as a stale competing copy in later searches.
 - Scope to the project in scope. When a project is active (you called
   enter_project), pass its project_id to search / list_tasks / list_notes so
   results stay inside that project. Querying with no project_id pulls in every

src/scribe/mcp/tools/notes.py

@@ -14,6 +14,7 @@ Sentinel conventions (inherited from existing fable-mcp tools):
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.services import dedup as dedup_svc
 from scribe.services import notes as notes_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
@@ -70,6 +71,7 @@ async def create_note(
     tags: list[str] | None = None,
     project_id: int = 0,
     system_ids: list[int] | None = None,
+    force: bool = False,
 ) -> dict:
     """Create a new note in Scribe.
 
@@ -80,10 +82,24 @@ async def create_note(
         project_id: Associate with a project (use 0 for no project / orphan note).
         system_ids: Ids of the project's Systems to associate this note with
             (e.g. research about a subsystem). See list_systems / create_system.
+        force: Bypass the near-duplicate gate. By default, if a title- or
+            meaning-similar note already exists in the same project, creation is
+            BLOCKED and the existing note's id is returned so you update it
+            instead (no duplicate bloat / no stale RAG copies). Set true only
+            when you're sure this is a genuinely distinct note.
 
-    Returns the created note object including its assigned id.
+    Returns the created note object including its assigned id, OR — when a
+    near-duplicate is found and force is false — {"duplicate": true,
+    "existing_id": ..., "message": ...} and nothing is created.
     """
     uid = current_user_id()
+    if not force:
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body, project_id=project_id or None,
+            is_task=False, note_type="note",
+        )
+        if dup is not None:
+            return dedup_svc.duplicate_response(dup, "note")
     note = await notes_svc.create_note(
         uid,
         title=title,

src/scribe/mcp/tools/tasks.py

@@ -19,6 +19,7 @@ Sentinels (preserved from existing fable-mcp):
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.services import dedup as dedup_svc
 from scribe.services import notes as notes_svc
 from scribe.services import planning as planning_svc
 from scribe.services import rulebooks as rulebooks_svc
@@ -107,6 +108,7 @@ async def create_task(
     kind: str = "work",
     system_ids: list[int] | None = None,
     arose_from_id: int = 0,
+    force: bool = False,
 ) -> dict:
     """Create a new task in Scribe.
 
@@ -127,6 +129,14 @@ async def create_task(
             objects; see list_systems / create_system) to associate this task with.
         arose_from_id: For an issue, the id of the task/feature it arose from
             (provenance). 0 = none.
+        force: Bypass the near-duplicate gate. By default, if a title- or
+            meaning-similar task already exists in the same project, creation is
+            BLOCKED and the existing task's id is returned so you update it
+            instead. Set true only for a genuinely distinct task.
+
+    Returns the created task, OR — when a near-duplicate is found and force is
+    false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing
+    created).
     """
     uid = current_user_id()
     if kind == "plan":
@@ -136,6 +146,13 @@ async def create_task(
             "milestone + seeds the design), then create each step as its own "
             "task with create_task(milestone_id=<that milestone>)."
         )
+    if not force:
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body, project_id=project_id or None,
+            is_task=True, note_type="note",
+        )
+        if dup is not None:
+            return dedup_svc.duplicate_response(dup, "task")
     note = await notes_svc.create_note(
         uid,
         title=title,

src/scribe/services/dedup.py

@@ -0,0 +1,124 @@
+"""Write-time near-duplicate detection — the update-over-create gate.
+
+Goal: stop a second near-identical row from being created when an existing one
+should be UPDATED instead. Duplicates bloat the store and, worse, get surfaced
+by semantic search (RAG) later as competing/stale copies that then have to be
+reconciled. This is the enforcement half of the instruction-level "prefer
+updating over creating" reflex.
+
+OPT-IN by design: the interactive create paths (MCP create tools + REST create
+routes) run this gate; internal/programmatic creates do NOT (e.g. a recurring
+task spawning its next instance, or a bulk import — those legitimately repeat a
+title and must not be blocked). Callers that want the gate call find_duplicate_*
+themselves and act on a hit; nothing here mutates.
+
+Two signals, both scoped to the same owner + project + kind:
+  1. Normalized-title exact match — cheap, always checked.
+  2. Semantic similarity (cosine ≥ _SEMANTIC_THRESHOLD) — only when the incoming
+     body is substantial. Short/title-only embeddings sit in a tight neighborhood
+     and false-positive (the pre-pivot lesson: "Lore: Shell 0" vs
+     "Lore: Reinitialization 0" matched at 0.91 with no body), so we gate it on
+     a minimum body length.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from sqlalchemy import func, select
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.services import embeddings as embeddings_svc
+
+# Run the semantic check only when the incoming body has at least this many
+# characters — below it, embeddings are dominated by the title and false-positive.
+_MIN_BODY_FOR_SEMANTIC = 200
+# Cosine threshold for "this is the same thing, reworded." Deliberately high to
+# keep false positives rare (a hard block with a force-override is unforgiving of
+# noise). Matches the 0.90 the pre-pivot dedup settled on.
+_SEMANTIC_THRESHOLD = 0.90
+
+
+@dataclass
+class DuplicateMatch:
+    """An existing record judged a near-duplicate of an incoming create."""
+    id: int
+    title: str
+    similarity: float  # 1.0 for an exact normalized-title match
+    reason: str        # "title" | "semantic"
+
+
+def duplicate_response(dup: "DuplicateMatch", kind: str) -> dict:
+    """Standard 'blocked — update instead' payload returned by a create tool
+    when the gate finds a near-duplicate. `kind` is 'note' or 'task' (drives the
+    update_<kind> hint)."""
+    return {
+        "duplicate": True,
+        "existing_id": dup.id,
+        "existing_title": dup.title,
+        "similarity": dup.similarity,
+        "match": dup.reason,
+        "message": (
+            f'A {dup.reason}-similar {kind} already exists (id {dup.id}: '
+            f'"{dup.title}"). Prefer UPDATING it (update_{kind}) over creating a '
+            f"near-duplicate. If this really is a distinct {kind}, retry with "
+            f"force=true."
+        ),
+    }
+
+
+async def find_duplicate_note(
+    user_id: int,
+    title: str,
+    body: str = "",
+    project_id: int | None = None,
+    is_task: bool | None = None,
+    note_type: str = "note",
+) -> DuplicateMatch | None:
+    """Best near-duplicate of (title, body) within the same owner + project +
+    kind, or None. Title match first (cheap, exact), then semantic when the body
+    is long enough to be meaningful. Never raises — embedder failure degrades to
+    title-only (callers should still be able to create)."""
+    norm = " ".join((title or "").split()).lower()
+
+    # --- Signal 1: normalized-title exact match (same scope) ---
+    if norm:
+        async with async_session() as session:
+            stmt = select(Note).where(
+                Note.user_id == user_id,
+                Note.deleted_at.is_(None),
+                Note.note_type == note_type,
+                func.lower(func.trim(Note.title)) == norm,
+            )
+            if project_id is not None:
+                stmt = stmt.where(Note.project_id == project_id)
+            else:
+                stmt = stmt.where(Note.project_id.is_(None))
+            if is_task is True:
+                stmt = stmt.where(Note.status.isnot(None))
+            elif is_task is False:
+                stmt = stmt.where(Note.status.is_(None))
+            existing = (await session.execute(stmt.limit(1))).scalars().first()
+            if existing is not None:
+                return DuplicateMatch(existing.id, existing.title, 1.0, "title")
+
+    # --- Signal 2: semantic similarity (only with a substantial body) ---
+    if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
+        query = f"{title}\n{body}".strip()
+        # Scope the semantic check the same way as the title check: a record in
+        # project P compares only to P; a project-less (orphan) record compares
+        # only to other orphans (orphan_only), NOT across every project — without
+        # this, semantic_search_notes applies no project filter when project_id
+        # is None and would match an orphan note against any project's notes.
+        hits = await embeddings_svc.semantic_search_notes(
+            user_id, query, project_id=project_id, is_task=is_task,
+            orphan_only=(project_id is None),
+            limit=3, threshold=_SEMANTIC_THRESHOLD,
+        )
+        for score, note in hits:
+            # semantic_search_notes doesn't filter note_type — enforce it here so
+            # a note doesn't shadow a task of the same wording, etc.
+            if note.note_type == note_type:
+                return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
+
+    return None

tests/test_mcp_tool_notes.py

@@ -25,6 +25,32 @@ def _fake_note(**overrides) -> MagicMock:
     return note
 
 
+@pytest.mark.asyncio
+async def test_create_note_blocked_by_duplicate_gate():
+    from scribe.services.dedup import DuplicateMatch
+    dup = DuplicateMatch(id=88, title="Embeddings notes", similarity=0.94, reason="semantic")
+    create_mock = AsyncMock()
+    with patch("scribe.mcp.tools.notes.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=dup)), \
+         patch("scribe.mcp.tools.notes.notes_svc.create_note", create_mock):
+        out = await create_note(title="embeddings", body="notes about embeddings")
+    assert out["duplicate"] is True
+    assert out["existing_id"] == 88
+    assert out["match"] == "semantic"
+    create_mock.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_create_note_force_bypasses_duplicate_gate():
+    find_mock = AsyncMock()
+    with patch("scribe.mcp.tools.notes.dedup_svc.find_duplicate_note", find_mock), \
+         patch("scribe.mcp.tools.notes.notes_svc.create_note",
+               AsyncMock(return_value=_fake_note(id=3))):
+        out = await create_note(title="dup", force=True)
+    assert out["id"] == 3
+    find_mock.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_list_notes_repackages_tuple_into_dict():
     rows = [_fake_note(id=1), _fake_note(id=2)]

tests/test_mcp_tool_tasks.py

@@ -113,6 +113,33 @@ async def test_create_task_passes_status():
     assert mock.call_args.kwargs["status"] == "todo"
 
 
+@pytest.mark.asyncio
+async def test_create_task_blocked_by_duplicate_gate():
+    """A near-duplicate blocks creation and returns the existing id (no insert)."""
+    from scribe.services.dedup import DuplicateMatch
+    dup = DuplicateMatch(id=42, title="Set up CI", similarity=1.0, reason="title")
+    create_mock = AsyncMock()
+    with patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=dup)), \
+         patch("scribe.mcp.tools.tasks.notes_svc.create_note", create_mock):
+        out = await create_task(title="set up ci")
+    assert out["duplicate"] is True
+    assert out["existing_id"] == 42
+    create_mock.assert_not_called()  # nothing was created
+
+
+@pytest.mark.asyncio
+async def test_create_task_force_bypasses_duplicate_gate():
+    """force=true skips the gate entirely and creates."""
+    find_mock = AsyncMock()
+    with patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note", find_mock), \
+         patch("scribe.mcp.tools.tasks.notes_svc.create_note",
+               AsyncMock(return_value=_fake_task(id=9))):
+        out = await create_task(title="dup", force=True)
+    assert out["id"] == 9
+    find_mock.assert_not_called()  # gate not even consulted
+
+
 @pytest.mark.asyncio
 async def test_create_task_rejects_retired_plan_kind():
     """kind=plan is hard-retired — plans are milestones (start_planning)."""

tests/test_services_dedup.py

@@ -0,0 +1,84 @@
+"""Unit tests for the write-time near-duplicate gate (services/dedup.py)."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services.dedup import DuplicateMatch, duplicate_response, find_duplicate_note
+
+
+def _session_returning(note):
+    """A mocked async_session() whose single execute() yields `note` (or None)."""
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+    result = MagicMock()
+    result.scalars.return_value.first.return_value = note
+    s.execute = AsyncMock(return_value=result)
+    return s
+
+
+def _fake_note(id=1, title="T", note_type="note"):
+    n = MagicMock()
+    n.id, n.title, n.note_type = id, title, note_type
+    return n
+
+
+@pytest.mark.asyncio
+async def test_title_exact_match_returns_title_duplicate():
+    note = _fake_note(id=10, title="Setup CI")
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_returning(note)):
+        # whitespace/case differences are normalized away
+        dup = await find_duplicate_note(7, "  setup   ci ", project_id=2, is_task=True)
+    assert dup is not None
+    assert dup.id == 10
+    assert dup.reason == "title"
+    assert dup.similarity == 1.0
+
+
+@pytest.mark.asyncio
+async def test_short_body_skips_semantic_check():
+    sem = AsyncMock()
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(7, "Unique", body="too short", project_id=2)
+    assert dup is None
+    sem.assert_not_called()  # body under _MIN_BODY_FOR_SEMANTIC
+
+
+@pytest.mark.asyncio
+async def test_semantic_match_when_body_substantial():
+    hit = _fake_note(id=20, title="Existing", note_type="note")
+    sem = AsyncMock(return_value=[(0.93, hit)])
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "Title", body="x" * 250, project_id=2, is_task=False, note_type="note",
+        )
+    assert dup is not None
+    assert dup.id == 20
+    assert dup.reason == "semantic"
+    assert dup.similarity == 0.93
+
+
+@pytest.mark.asyncio
+async def test_semantic_match_of_other_note_type_is_ignored():
+    other = _fake_note(id=21, title="X", note_type="process")
+    sem = AsyncMock(return_value=[(0.97, other)])
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(7, "Title", body="x" * 250, note_type="note")
+    assert dup is None  # type mismatch must not block
+
+
+def test_duplicate_response_shape():
+    dm = DuplicateMatch(id=5, title="Foo", similarity=1.0, reason="title")
+    r = duplicate_response(dm, "task")
+    assert r["duplicate"] is True
+    assert r["existing_id"] == 5
+    assert r["match"] == "title"
+    assert "force=true" in r["message"]
+    assert "update_task" in r["message"]