feat(scribe): merge_snippets — unify found one-off snippets into one canonical

Step 2 of the snippet-merge milestone (#231). The dedup gate only PREVENTS
new near-duplicates; merge is the CURE for the ones already scattered.

- services/snippets.py: merge_snippets(user_id, target_id, source_ids) —
  keep the target's scalar fields (name/when_to_use/signature/language/
  code), union the sources' locations + extra tags onto it (so the survivor
  carries every call site as a location), trash the sources (recoverable),
  re-embed the survivor. Pure merge_snippet_fields() factored out for unit
  testing. Returns (survivor_note, merged_ids).
- mcp/tools/snippets.py: merge_snippets(target_id, source_ids) tool (5th),
  and a create_snippet dedup-path nudge toward merge over a forced copy.
- routes/snippets.py: POST /api/snippets/<id>/merge {source_ids} — share-
  aware (can_write target + every source, rule #78) with a same-owner guard
  (cross-owner merge is out of scope).
- plugin reusing-code skill + MCP _INSTRUCTIONS: point found-duplicates at
  merge as the cure (rule #119 surfaces, not a Scribe rule). plugin.json
  0.1.13 -> 0.1.14 in the same change (the #1040 marketplace-ship lesson).
- Tests: pure merge-helper union/dedup; MCP tool (requires a source,
  survivor+merged_ids, not-found); route handler + 5-tool registration.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Pa2EsuB54BuWQ8GfJq9c7t

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.13",
+  "version": "0.1.14",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/skills/reusing-code/SKILL.md

@@ -37,6 +37,16 @@ through recall/auto-inject; this skill is the active reflex around that.
   per reusable thing. If it already exists, `update_snippet` it instead of
   recording a second copy (the create gate will flag a near-duplicate anyway).
 
+## Found the same thing in several places — unify it
+
+When you notice the same reusable thing recorded (or written) as several
+one-offs, don't leave the duplicates competing in recall — **merge them**.
+`merge_snippets(canonical_id, [other_ids])` keeps one canonical record, folds in
+the others' call sites as locations, and retires the duplicates to the trash.
+The result is a single entry that shows every place the thing is used — which is
+exactly the signal that it was worth consolidating. This is the cure the create
+gate only hints at when it blocks a near-duplicate.
+
 ## Why this pays off
 
 A one-off written a second time is the cost this avoids. Recording a snippet

src/scribe/mcp/server.py

@@ -220,7 +220,10 @@ one-off; (2) the moment you build or notice something reusable, record it with
 create_snippet(name, code, when_to_use, language, signature, repo, path, symbol,
 project_id, system_ids) so a later session is offered it. Make when_to_use sharp
 — it becomes the title, which is what a recall menu shows. Edit an existing one
-with update_snippet rather than recording a second copy.
+with update_snippet rather than recording a second copy; when the same reusable
+thing already exists as several one-offs, unify them into one canonical record
+with merge_snippets (it folds every call site in as a location and trashes the
+duplicates).
 
 When developing Scribe itself, honor its multi-user sharing ACL: scope every
 read and mutation of user data by owner + shares — never assume a single

src/scribe/mcp/tools/snippets.py

@@ -73,7 +73,11 @@ async def create_snippet(
 
     Returns the created snippet (including a parsed `snippet` field), OR — when a
     near-duplicate snippet already exists and force is false — {"duplicate": true,
-    "existing_id": ..., "message": ...} and nothing is created.
+    "existing_id": ..., "message": ...} and nothing is created. When that happens
+    and it really is the same reusable thing found in another place, prefer
+    merge_snippets(existing_id, [new...]) — or record then merge — to unify them
+    into ONE canonical record (which then carries every call site as a location),
+    rather than forcing a second copy with force=true.
     """
     if not (name or "").strip() or not (code or "").strip():
         raise ValueError("create_snippet requires a non-empty name and code")
@@ -157,6 +161,38 @@ async def update_snippet(
     return data
 
 
+async def merge_snippets(target_id: int, source_ids: list[int]) -> dict:
+    """Unify duplicate/variant snippets INTO one canonical record — the cure for
+    the same reusable thing recorded as several one-offs.
+
+    Keeps the target as the canonical: its name, when-to-use, signature, language
+    and code win. The sources' locations and extra tags are folded in — so the
+    survivor ends up carrying EVERY call site as a location (which is itself the
+    "this is duplicated N times" signal) — and the source records are moved to the
+    trash (recoverable). The survivor is re-embedded so recall stops surfacing the
+    now-merged duplicates.
+
+    Args:
+        target_id: The snippet to keep (the canonical record).
+        source_ids: Snippet ids to fold into the target and retire. Ids that
+            aren't your snippets are skipped; target_id in the list is ignored.
+
+    Returns the merged canonical snippet (with a parsed `snippet` field) plus
+    `merged_ids` — the source ids actually merged and trashed.
+    """
+    uid = current_user_id()
+    ids = [s for s in (source_ids or []) if s != target_id]
+    if not ids:
+        raise ValueError("merge_snippets requires at least one other source_id")
+    result = await snippets_svc.merge_snippets(uid, target_id, ids)
+    if result is None:
+        raise ValueError(f"snippet {target_id} not found")
+    note, merged_ids = result
+    data = snippets_svc.snippet_to_dict(note)
+    data["merged_ids"] = merged_ids
+    return data
+
+
 def register(mcp) -> None:
-    for fn in (list_snippets, create_snippet, get_snippet, update_snippet):
+    for fn in (list_snippets, create_snippet, get_snippet, update_snippet, merge_snippets):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/routes/snippets.py

@@ -120,6 +120,46 @@ async def update_snippet_route(snippet_id: int):
     return jsonify(snippets_svc.snippet_to_dict(updated))
 
 
+@snippets_bp.route("/<int:snippet_id>/merge", methods=["POST"])
+@login_required
+async def merge_snippet_route(snippet_id: int):
+    """Unify source snippets into this one (the canonical target). Body:
+    {"source_ids": [int, ...]}. Requires write on the target and every source,
+    and all must share the target's owner (cross-owner merge is out of scope)."""
+    uid = get_current_user_id()
+    loaded = await _load_snippet(uid, snippet_id)
+    if loaded is None:
+        return not_found("Snippet")
+    target, _ = loaded
+    if not await can_write_note(uid, snippet_id):
+        return jsonify({"error": "Permission denied"}), 403
+
+    data = await request.get_json() or {}
+    raw = data.get("source_ids") or []
+    source_ids = [s for s in raw if isinstance(s, int) and s != snippet_id]
+    if not source_ids:
+        return jsonify({"error": "source_ids (non-empty list of other snippet ids) is required"}), 400
+
+    owner_uid = target.user_id
+    for sid in source_ids:
+        sloaded = await _load_snippet(uid, sid)
+        if sloaded is None:
+            return not_found(f"Snippet {sid}")
+        snote, _ = sloaded
+        if snote.user_id != owner_uid:
+            return jsonify({"error": "can only merge snippets with the same owner"}), 400
+        if not await can_write_note(uid, sid):
+            return jsonify({"error": f"Permission denied for snippet {sid}"}), 403
+
+    result = await snippets_svc.merge_snippets(owner_uid, snippet_id, source_ids)
+    if result is None:
+        return not_found("Snippet")
+    note, merged_ids = result
+    out = snippets_svc.snippet_to_dict(note)
+    out["merged_ids"] = merged_ids
+    return jsonify(out)
+
+
 @snippets_bp.route("/<int:snippet_id>", methods=["DELETE"])
 @login_required
 async def delete_snippet_route(snippet_id: int):

src/scribe/services/snippets.py

@@ -382,3 +382,78 @@ async def update_snippet(
         # Title/body changed → refresh the embedding so recall reflects the edit.
         _embed_snippet(updated)
     return updated
+
+
+# --- merge: unify found one-offs into one canonical snippet ------------------
+
+def _extra_tags(tags: list[str] | None, language: str = "") -> list[str]:
+    """A snippet's caller tags — everything except the language + `snippet`
+    markers that compose_tags re-derives."""
+    lang = (language or "").strip().lower()
+    return [t for t in (tags or []) if t and t != SNIPPET_TAG and t != lang]
+
+
+def merge_snippet_fields(
+    target_fields: dict, target_tags: list[str] | None, sources: list[tuple[dict, list]]
+) -> tuple[list[dict], list[str]]:
+    """Pure merge: union locations (target's first, then each source in order)
+    and union extra tags. The target's scalar fields (name/when_to_use/signature/
+    language/code) win — only locations and tags accumulate. ``sources`` is a
+    list of (parsed_fields, tags). Returns (locations, extra_tags)."""
+    locations = list(target_fields.get("locations") or [])
+    extra = _extra_tags(target_tags, target_fields.get("language", ""))
+    for sfields, stags in sources:
+        locations.extend(sfields.get("locations") or [])
+        for t in _extra_tags(stags, sfields.get("language", "")):
+            if t not in extra:
+                extra.append(t)
+    return _normalize_locations(locations), extra
+
+
+async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
+    """Unify source snippets INTO the target (all owned by user_id): union their
+    locations + tags onto the canonical target, keep the target's scalar fields,
+    trash the sources (recoverable), re-embed the survivor.
+
+    Returns (merged_target_note, merged_source_ids), or None if the target isn't
+    the user's snippet. Source ids that aren't the user's snippets are skipped."""
+    target = await notes_svc.get_note(user_id, target_id)
+    if target is None or target.note_type != SNIPPET_NOTE_TYPE:
+        return None
+
+    sources = []
+    for sid in source_ids:
+        if sid == target_id:
+            continue
+        s = await notes_svc.get_note(user_id, sid)
+        if s is not None and s.note_type == SNIPPET_NOTE_TYPE:
+            sources.append(s)
+
+    tgt_fields = parse_snippet_fields(target.title, target.body, target.tags)
+    parsed_sources = [
+        (parse_snippet_fields(s.title, s.body, s.tags), s.tags) for s in sources
+    ]
+    locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_sources)
+
+    updated = await notes_svc.update_note(
+        user_id, target_id,
+        body=compose_body(
+            code=tgt_fields["code"], language=tgt_fields["language"],
+            signature=tgt_fields["signature"], when_to_use=tgt_fields["when_to_use"],
+            locations=locations,
+        ),
+        tags=compose_tags(tgt_fields["language"], extra_tags),
+    )
+    if updated is None:
+        return None
+
+    # Retire the merged-in sources to the trash (recoverable).
+    from scribe.services.trash import delete as trash_delete
+    merged_ids: list[int] = []
+    for s in sources:
+        batch = await trash_delete(user_id, "note", s.id)
+        if batch is not None:
+            merged_ids.append(s.id)
+
+    _embed_snippet(updated)
+    return updated, merged_ids

tests/test_mcp_tool_snippets.py

@@ -82,7 +82,38 @@ async def test_update_snippet_missing_raises():
             await update_snippet(123, name="x")
 
 
-def test_register_attaches_four_tools():
+@pytest.mark.asyncio
+async def test_merge_snippets_requires_a_source():
+    from scribe.mcp.tools.snippets import merge_snippets
+    with pytest.raises(ValueError):
+        await merge_snippets(target_id=1, source_ids=[])
+    with pytest.raises(ValueError):
+        await merge_snippets(target_id=1, source_ids=[1])  # only self → nothing to merge
+
+
+@pytest.mark.asyncio
+async def test_merge_snippets_returns_survivor_and_merged_ids():
+    survivor = _fake_snippet()
+    with patch("scribe.services.snippets.merge_snippets",
+               AsyncMock(return_value=(survivor, [2, 3]))) as mock_merge:
+        from scribe.mcp.tools.snippets import merge_snippets
+        out = await merge_snippets(target_id=1, source_ids=[2, 3, 1])
+    assert out["merged_ids"] == [2, 3]
+    assert out["snippet"]["name"] == "debounce"
+    # target_id passed through; self-id filtered out of the source list.
+    assert mock_merge.await_args.args[1] == 1
+    assert mock_merge.await_args.args[2] == [2, 3]
+
+
+@pytest.mark.asyncio
+async def test_merge_snippets_not_found_raises():
+    with patch("scribe.services.snippets.merge_snippets", AsyncMock(return_value=None)):
+        from scribe.mcp.tools.snippets import merge_snippets
+        with pytest.raises(ValueError):
+            await merge_snippets(target_id=1, source_ids=[2])
+
+
+def test_register_attaches_all_tools():
     from scribe.mcp.tools import snippets
     names: list[str] = []
 
@@ -97,4 +128,5 @@ def test_register_attaches_four_tools():
     snippets.register(FakeMcp())
     assert set(names) == {
         "list_snippets", "create_snippet", "get_snippet", "update_snippet",
+        "merge_snippets",
     }

tests/test_routes_snippets.py

@@ -19,7 +19,7 @@ def test_snippet_handlers_callable():
     from scribe.routes import snippets as routes
     for name in (
         "list_snippets_route", "create_snippet_route", "get_snippet_route",
-        "update_snippet_route", "delete_snippet_route",
+        "update_snippet_route", "delete_snippet_route", "merge_snippet_route",
     ):
         assert callable(getattr(routes, name))
 

tests/test_services_snippets.py

@@ -122,6 +122,22 @@ def test_parse_legacy_single_location_line_still_works():
     assert (got["repo"], got["path"], got["symbol"]) == ("scribe", "a.py", "f")
 
 
+def test_merge_snippet_fields_unions_locations_and_tags():
+    target_fields = {"language": "py", "locations": [{"repo": "a", "path": "a.py", "symbol": "f"}]}
+    src1 = ({"language": "py", "locations": [{"repo": "b", "path": "b.py", "symbol": "g"}]},
+            ["py", "snippet", "helper"])
+    src2 = ({"language": "py", "locations": [{"repo": "a", "path": "a.py", "symbol": "f"}]},  # dup loc
+            ["snippet"])
+    locs, extra = s.merge_snippet_fields(target_fields, ["py", "snippet", "core"], [src1, src2])
+    # target location first, then unique source locations; dup dropped.
+    assert locs == [
+        {"repo": "a", "path": "a.py", "symbol": "f"},
+        {"repo": "b", "path": "b.py", "symbol": "g"},
+    ]
+    # extra tags unioned, language + "snippet" markers excluded.
+    assert extra == ["core", "helper"]
+
+
 def test_snippet_to_dict_includes_parsed_fields():
     class FakeNote:
         title = "debounce — rate-limit"