diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json index 2f925e7..7dc0459 100644 --- a/plugin/.claude-plugin/plugin.json +++ b/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": { diff --git a/plugin/skills/reusing-code/SKILL.md b/plugin/skills/reusing-code/SKILL.md index cf1cccd..dd904bf 100644 --- a/plugin/skills/reusing-code/SKILL.md +++ b/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 diff --git a/src/scribe/mcp/server.py b/src/scribe/mcp/server.py index 5c1f8ca..3b92a5d 100644 --- a/src/scribe/mcp/server.py +++ b/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 diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py index 8f15541..528cc66 100644 --- a/src/scribe/mcp/tools/snippets.py +++ b/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) diff --git a/src/scribe/routes/snippets.py b/src/scribe/routes/snippets.py index 56d0aac..a52963f 100644 --- a/src/scribe/routes/snippets.py +++ b/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("//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("/", methods=["DELETE"]) @login_required async def delete_snippet_route(snippet_id: int): diff --git a/src/scribe/services/snippets.py b/src/scribe/services/snippets.py index e1cdcac..1085927 100644 --- a/src/scribe/services/snippets.py +++ b/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 diff --git a/tests/test_mcp_tool_snippets.py b/tests/test_mcp_tool_snippets.py index 7ea6284..fcc15ac 100644 --- a/tests/test_mcp_tool_snippets.py +++ b/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", } diff --git a/tests/test_routes_snippets.py b/tests/test_routes_snippets.py index 79b0387..ca05cac 100644 --- a/tests/test_routes_snippets.py +++ b/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)) diff --git a/tests/test_services_snippets.py b/tests/test_services_snippets.py index 839a096..722510b 100644 --- a/tests/test_services_snippets.py +++ b/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"