"""Snippet MCP tools: record reusable functions/components for later recall. A snippet is a Note with note_type='snippet' (see services/snippets.py). Because snippets are ordinary embedded notes, once recorded they surface through the same semantic search + title-first auto-inject as everything else — so a reusable thing recorded once can be recalled before it's re-written as a one-off. The tools wrap services/snippets.py, mirroring the note/process tools (dedup gate on create, System association passthrough). """ from __future__ import annotations from scribe.mcp._context import current_user_id from scribe.services import access as access_svc from scribe.services import dedup as dedup_svc from scribe.services import snippets as snippets_svc from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes from scribe.services import systems as systems_svc async def list_snippets( q: str = "", tag: str = "", limit: int = 50, project_id: int = 0, repo: str = "", path: str = "", symbol: str = "", verification: str = "", ) -> dict: """List recorded snippets (reusable functions/components). Two ways to ask, usable together: by MEANING (`q` — "what do I need this code to do?") and by PLACE (`repo`/`path`/`symbol` — "what canonical helpers already live in the file I'm about to edit?"). Reach for the place form before you write or change code in a file: it is the cheap way to find prior art you'd otherwise duplicate a few lines down. Args: q: Free-text search across name + body (optional). Matches on meaning as well as wording, so describe what you need the code to DO. tag: Filter to a single tag, e.g. a language like "python" (optional). limit: Max results (1-100). project_id: Narrow to one project. 0 (default) searches every project — usually what you want, since a helper you need here may well have been written somewhere else. repo: Narrow to snippets recorded in this repo, matched exactly — the same repo string used when recording, e.g. "Scribe". path: Narrow to snippets recorded at this path. Matches the exact file OR anything beneath it, so "frontend/src" finds "frontend/src/lib/x.ts" as well as itself. symbol: Narrow to snippets recorded under this symbol name, exactly. verification: Narrow on the drift check (see verify_snippet). "attention" is the one to reach for — everything whose recorded location or code no longer checks out, plus everything whose verdict expired because the snippet was edited after it was checked. Also accepts "ok", "unverified", "drifted", or a specific failure: "missing", "moved", "changed". `repo`/`path`/`symbol` must all match the SAME recorded location, so a snippet that lives in repo A and, separately, at path B in another repo is not returned for repo=A + path=B. Returns {"snippets": [{id, title, tags, preview, usage}], "total": int}. The title reads "name — when to reach for it"; open one in full with get_snippet(id). `usage` is {surfaced_count, pull_count, last_surfaced_at, last_pulled_at}: how often the entry has been put in front of an agent versus actually opened. Treat a high surfaced_count with a zero pull_count as a prompt to fix the record — usually its "when to reach for it" doesn't say when — or to delete it. Such an entry is not harmless: it takes a slot in every future auto-inject menu and crowds out something useful. An entry marked `shared: true` with an `owner` belongs to someone else — one person's suggestion, not settled practice here. Weigh it on its merits and attribute it when you use it. Searching (passing `q`) also reaches snippets shared directly with the operator; browsing without a query deliberately does not, so those stay out of ambient results until asked for. """ uid = current_user_id() items, total = await snippets_svc.list_snippets( uid, q=q or None, tag=tag, limit=max(1, min(limit, 100)), project_id=project_id or None, repo=repo, path=path, symbol=symbol, verification=verification, ) labeled = await access_svc.label_shared_items(uid, items) usage = await usage_for_notes([int(it["id"]) for it in labeled]) for it in labeled: it["usage"] = usage.get(int(it["id"]), empty_usage()) return {"snippets": labeled, "total": total} async def create_snippet( name: str, code: str, language: str = "", signature: str = "", when_to_use: str = "", repo: str = "", path: str = "", symbol: str = "", locations: list[dict] | None = None, tags: list[str] | None = None, project_id: int = 0, system_ids: list[int] | None = None, force: bool = False, ) -> dict: """Record a reusable function/component so future sessions can RECALL it instead of writing a fresh one-off. Reach for this the moment you build (or notice) something reusable: a helper, a hook, a component, a pattern worth repeating. Recording it once makes it surface automatically when a similar problem comes up later. Before writing a new utility, search first — a snippet may already exist. Args: name: Short name of the function/component, e.g. "useDebouncedRef". code: The code itself (required). language: Language/format, e.g. "python", "vue", "sql". Becomes a tag and the code-fence language. signature: One-line signature/interface, e.g. "debounce(fn, ms) -> fn". when_to_use: One line on when to reach for it — this becomes part of the title, so it's what a recall menu shows. Keep it sharp. repo/path/symbol: Canonical location of the reference implementation. locations: Several locations at once, as [{"repo","path","symbol"}, ...], when you already know the thing lives in more than one place. Takes precedence over the single repo/path/symbol shorthand. tags: Extra plain-string tags (language + "snippet" are added for you). project_id: Associate with a project (0 = no project). Snippets surface proactively within their project; search finds them across projects. system_ids: Ids of the project's Systems to associate this snippet with. force: Bypass the near-duplicate gate (see below). 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. 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") uid = current_user_id() title = snippets_svc.compose_title(name, when_to_use) body = snippets_svc.compose_body( code=code, language=language, signature=signature, when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, locations=locations, ) if not force: dup = await dedup_svc.find_duplicate_note( uid, title, body, project_id=project_id or None, is_task=False, note_type=snippets_svc.SNIPPET_NOTE_TYPE, ) if dup is not None: return dedup_svc.duplicate_response(dup, "snippet") note = await snippets_svc.create_snippet( uid, name=name, code=code, language=language, signature=signature, when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, locations=locations, tags=tags, project_id=project_id or None, ) if system_ids: await systems_svc.set_record_systems(uid, note.id, system_ids) data = snippets_svc.snippet_to_dict(note) if system_ids: data["systems"] = [ s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) ] return data async def get_snippet(snippet_id: int) -> dict: """Fetch a snippet by id — the full record: code, signature, location, and a parsed `snippet` field of its structured parts. If the record belongs to someone else it carries `shared: true` with the `owner` and your `permission`. Read that as ONE PERSON'S SUGGESTION, not as established practice here: judge it on its merits, say whose it is when you reference it, and don't adopt it as the house pattern without checking. """ uid = current_user_id() note = await snippets_svc.get_snippet(uid, snippet_id) if note is None: raise ValueError(f"snippet {snippet_id} not found") data = snippets_svc.snippet_to_dict(note) data.update(await access_svc.describe_provenance(uid, note)) # A "pull" is an explicit open, so it's recorded HERE rather than in # snippets_svc.get_snippet — the service is also reached by update/merge # paths, and counting those would inflate exactly the number that is # supposed to mean "someone chose to look at this" (#2085). record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_snippet") return data async def unmerge_snippet(survivor_id: int, source_id: int) -> dict: """Reverse ONE source out of a merged snippet — the inverse of merge_snippets. Restores the source record and strips exactly what it contributed from the survivor: the locations and tags it ADDED at merge time, never the ones the survivor already had. Reach for it when a merge turns out to have unified two things that only looked alike. Also the fix for a half-undone merge. Restoring a merged-in source from the trash by hand brings the record back but leaves the survivor still claiming its call sites, so both records claim the same places and the reverse lookup reads the duplicate claims as real. Running this on an already-restored source repairs that: it skips the restore and does the subtraction. Args: survivor_id: The snippet that absorbed the other. source_id: The snippet to pull back out of it. Returns {"survivor": {...}, "restored": {...}}. Refuses, with the reason, when: the survivor has no record of absorbing that id; the source was purged from the trash; or the merge predates per-source provenance, in which case what it contributed isn't known and subtracting a guess could strip call sites the survivor genuinely owns — restore it from the trash and adjust both records by hand instead. """ uid = current_user_id() try: result = await snippets_svc.unmerge_snippet(uid, survivor_id, source_id) except snippets_svc.UnmergeError as exc: raise ValueError(str(exc)) from exc if result is None: raise ValueError(f"snippet {survivor_id} not found") survivor, restored = result return { "survivor": snippets_svc.snippet_to_dict(survivor), "restored": snippets_svc.snippet_to_dict(restored) if restored else None, } async def find_duplicate_snippets(threshold: float = 0.0) -> dict: """Find snippets already recorded that look like duplicates of each other. The create gate PREVENTS a new duplicate and merge_snippets CURES one you point it at — this is the missing third piece: it FINDS the ones already in the record, so nobody has to notice them by hand. Results are grouped into candidate merge SETS, not just pairs. Grouping is transitive: if A resembles B and B resembles C, all three land in one set even when A and C don't directly clear the bar. That mirrors what merge does (it folds every source into one survivor), but it means a chain of mild resemblances can rope in a member that isn't really alike — so read a set as a proposal and check the members before acting. Reports only YOUR snippets. merge_snippets requires one owner across the whole set, so surfacing someone else's would propose a merge that can't be performed. Acting on a group: pick the best record as the canonical target, then `merge_snippets(target_id, [other ids])`. Merge unions the fields and folds every source's location in, so the survivor is findable at all their call sites; the sources are trashed, recoverably. Prefer as target the one with the clearest "when to reach for it" — merge keeps the target's title. Args: threshold: Similarity floor, 0-1. 0 (default) uses the configured setting. Raise it if the report is noisy, lower it to catch more. Returns {"groups": [{"note_ids", "snippets", "top_score"}], "pairs", "threshold"}. An empty `groups` means nothing resembles anything else that closely — the common and desirable case. """ uid = current_user_id() return await dedup_svc.find_duplicate_snippets( uid, threshold=threshold if threshold > 0 else None ) async def verify_snippet( snippet_id: int, status: str, detail: str = "", path: str = "", ) -> dict: """Record whether a snippet's recorded location and code still match source. YOU do the checking — Scribe has no copy of the repo and deliberately never gets one. This tool only remembers your verdict so it becomes queryable and so the operator can see what has rotted. The procedure, once per snippet you're checking: 1. `get_snippet(id)` — read its `snippet.locations` and `snippet.code`. 2. Does the recorded path still exist in the working tree? If not → status="missing". 3. Does the recorded symbol still appear in that file? If not → status="moved" (the file is there, the thing isn't). 4. Does the source still match the recorded code, allowing for formatting? Judge whether it still does the same thing — an added parameter or a changed branch is "changed"; a reindent is not. If it diverged → status="changed". 5. All three hold → status="ok". Put what you actually found in `detail` ("renamed to parse_location_str", "moved to services/knowledge.py"). It's what makes the record fixable later by someone who wasn't here, so write it for them, not as a status echo. A verdict expires automatically if the snippet is edited afterwards: it is stamped with a hash of the code it was checked against, so it can never go on vouching for code nobody checked. Re-verify after fixing a record. Args: snippet_id: The snippet you checked. status: "ok" | "missing" | "moved" | "changed". detail: What you found — free text, shown to the operator. path: The path you actually checked, if it differs from the recorded one (e.g. you found the symbol at its new home). Defaults to the recorded path. Requires write access: a verdict changes how the record is presented, so being able to read a snippet someone shared with you doesn't let you mark it broken. """ uid = current_user_id() note = await snippets_svc.record_verification( uid, snippet_id, status=status, detail=detail, path=path, ) if note is None: raise ValueError( f"snippet {snippet_id} not found, or you don't have write access to it" ) return snippets_svc.snippet_to_dict(note) async def update_snippet( snippet_id: int, name: str | None = None, code: str | None = None, language: str | None = None, signature: str | None = None, when_to_use: str | None = None, repo: str | None = None, path: str | None = None, symbol: str | None = None, locations: list[dict] | None = None, tags: list[str] | None = None, project_id: int = 0, system_ids: list[int] | None = None, ) -> dict: """Update a snippet. Only the fields you pass change. An omitted field is left alone; an EMPTY STRING clears it — so a stale signature, a wrong "when to use", or an obsolete location can be removed, not just overwritten. A snippet that surfaces in recall with wrong details is worse than none, so correcting downward has to be possible. Args: locations: Replace the whole location set, as [{"repo","path","symbol"}, ...]. Pass [] to clear every location. The single repo/path/symbol args instead overlay onto the FIRST location, leaving the rest. tags: Replaces the extra-tag set (language + "snippet" are re-derived). project_id: 0 leaves it unchanged, -1 detaches it from its project, a positive id moves it. Editing someone else's snippet requires an editor or admin share from them. A read-only share is refused with a message saying so — record your own version instead of trying to force it. """ uid = current_user_id() if project_id == 0: project = snippets_svc.UNSET elif project_id < 0: project = None else: project = project_id try: note = await snippets_svc.update_snippet( uid, snippet_id, name=name, code=code, language=language, signature=signature, when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, locations=locations, tags=tags, project_id=project, ) except PermissionError as exc: # Readable but not writable — surface the real reason, not "not found". raise ValueError(str(exc)) from exc if note is None: raise ValueError(f"snippet {snippet_id} not found") if system_ids is not None: await systems_svc.set_record_systems(note.user_id, snippet_id, system_ids) data = snippets_svc.snippet_to_dict(note) data.update(await access_svc.describe_provenance(uid, note)) if system_ids is not None: data["systems"] = [ s.to_dict() for s in await systems_svc.list_record_systems(note.user_id, snippet_id) ] return data async def delete_snippet(snippet_id: int) -> dict: """Retire a snippet you recorded — it moves to the trash and is recoverable. Reach for this when a snippet is wrong, obsolete, or was never worth keeping. A recorded snippet is offered as prior art on every matching turn, so a bad one costs more than a missing one. If instead it's a duplicate of something that should survive, prefer merge_snippets — that keeps the call sites. """ uid = current_user_id() if not await snippets_svc.delete_snippet(uid, snippet_id): raise ValueError(f"snippet {snippet_id} not found") return {"deleted": True, "id": snippet_id} 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. The survivor records what it absorbed as `merged_from` (in its `snippet` fields and as a "Merged from: #ids" line in the body), so a variant that got folded in leaves a trace outside the trash. It accumulates across merges. Reversible: each entry also records what that source contributed, so `unmerge_snippet(target_id, source_id)` can restore it and strip exactly those locations back off — never the ones the target already had. 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") try: result = await snippets_svc.merge_snippets(uid, target_id, ids) except PermissionError as exc: raise ValueError(str(exc)) from exc 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, delete_snippet, merge_snippets, verify_snippet, find_duplicate_snippets, unmerge_snippet, ): mcp.tool(name=fn.__name__)(fn)