From 1942913366c7e20887fd80675cc6c82677e7c432 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 25 Jul 2026 11:00:33 -0400 Subject: [PATCH] =?UTF-8?q?feat(scribe):=20add=20snippet=20recall=20?= =?UTF-8?q?=E2=80=94=20note=5Ftype=3D'snippet'=20service=20+=20MCP=20tools?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Record reusable functions/components once so they surface via the existing semantic search + title-first auto-inject, instead of re-solving as one-offs. A snippet is a Note with note_type='snippet' (no schema change): note_type is free-text, and semantic_search_notes never filters by type, so snippets join the recall/auto-inject pool the moment they're embedded. Structured fields (name/language/signature/location/when_to_use/code) are stored via a body convention — title = "name — when to use" (what auto-inject surfaces), language + "snippet" as tags, templated markdown body — keeping storage swappable later without changing the tool/UI contract. - services/snippets.py: compose/parse helpers + create/get/list/update wrappers over notes_svc (dedup + System association reused). - mcp/tools/snippets.py: list_snippets / create_snippet / get_snippet / update_snippet, registered in tools/__init__.py. - unit tests for the serialize/parse round-trip and the MCP tool surface. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Pa2EsuB54BuWQ8GfJq9c7t --- src/scribe/mcp/tools/__init__.py | 3 +- src/scribe/mcp/tools/snippets.py | 162 +++++++++++++++++++ src/scribe/services/snippets.py | 257 +++++++++++++++++++++++++++++++ tests/test_mcp_tool_snippets.py | 100 ++++++++++++ tests/test_services_snippets.py | 77 +++++++++ 5 files changed, 598 insertions(+), 1 deletion(-) create mode 100644 src/scribe/mcp/tools/snippets.py create mode 100644 src/scribe/services/snippets.py create mode 100644 tests/test_mcp_tool_snippets.py create mode 100644 tests/test_services_snippets.py diff --git a/src/scribe/mcp/tools/__init__.py b/src/scribe/mcp/tools/__init__.py index 5e9161c..b9b8dc3 100644 --- a/src/scribe/mcp/tools/__init__.py +++ b/src/scribe/mcp/tools/__init__.py @@ -5,7 +5,7 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called from `mcp.server.build_mcp_server`. """ from scribe.mcp.tools import ( - milestones, notes, processes, projects, recent, repos, rulebooks, search, systems, tags, tasks, trash, + milestones, notes, processes, projects, recent, repos, rulebooks, search, snippets, systems, tags, tasks, trash, ) @@ -21,5 +21,6 @@ def register_all(mcp) -> None: recent.register(mcp) repos.register(mcp) processes.register(mcp) + snippets.register(mcp) rulebooks.register(mcp) trash.register(mcp) diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py new file mode 100644 index 0000000..8f15541 --- /dev/null +++ b/src/scribe/mcp/tools/snippets.py @@ -0,0 +1,162 @@ +"""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 dedup as dedup_svc +from scribe.services import snippets as snippets_svc +from scribe.services import systems as systems_svc + + +async def list_snippets(q: str = "", tag: str = "", limit: int = 50) -> dict: + """List recorded snippets (reusable functions/components). + + Args: + q: Free-text search across name + body (optional). + tag: Filter to a single tag, e.g. a language like "python" (optional). + limit: Max results (1-100). + + Returns {"snippets": [{id, title, tags, preview}], "total": int}. The title + reads "name — when to reach for it"; open one in full with get_snippet(id). + """ + uid = current_user_id() + items, total = await snippets_svc.list_snippets( + uid, q=q or None, tag=tag, limit=max(1, min(limit, 100)), + ) + return {"snippets": items, "total": total} + + +async def create_snippet( + name: str, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + 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. + 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. + """ + 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, + ) + 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, + 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.""" + 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") + return snippets_svc.snippet_to_dict(note) + + +async def update_snippet( + snippet_id: int, + name: str = "", + code: str = "", + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + tags: list[str] | None = None, + project_id: int = 0, + system_ids: list[int] | None = None, +) -> dict: + """Update a snippet. Only provided fields change — empty strings leave that + field unchanged; pass tags to replace the extra-tag set.""" + uid = current_user_id() + + def _n(v: str) -> str | None: + return v if v != "" else None + + note = await snippets_svc.update_snippet( + uid, snippet_id, + name=_n(name), code=_n(code), language=_n(language), + signature=_n(signature), when_to_use=_n(when_to_use), + repo=_n(repo), path=_n(path), symbol=_n(symbol), + tags=tags, project_id=project_id or None, + ) + if note is None: + raise ValueError(f"snippet {snippet_id} not found") + if system_ids is not None: + await systems_svc.set_record_systems(uid, snippet_id, system_ids) + data = snippets_svc.snippet_to_dict(note) + if system_ids is not None: + data["systems"] = [ + s.to_dict() for s in await systems_svc.list_record_systems(uid, snippet_id) + ] + return data + + +def register(mcp) -> None: + for fn in (list_snippets, create_snippet, get_snippet, update_snippet): + mcp.tool(name=fn.__name__)(fn) diff --git a/src/scribe/services/snippets.py b/src/scribe/services/snippets.py new file mode 100644 index 0000000..e6ef253 --- /dev/null +++ b/src/scribe/services/snippets.py @@ -0,0 +1,257 @@ +"""Snippet service — reusable functions/components recorded for recall. + +A *snippet* is a Note with ``note_type='snippet'``: a named, reusable function or +component recorded once so a later session can recall it before writing a +one-off. It carries a name, language, signature, canonical location +(repo · path · symbol), a one-line "when to reach for it", and the code itself. + +Structured fields are stored as a **body-convention** — no dedicated column, so +a snippet inherits everything a note has (embeddings, ACL, project/System +association, dedup) and, crucially, becomes eligible for semantic recall the +moment it's embedded: + + - ``title`` = ``"{name} — {when_to_use}"``. The title is exactly what the + title-first auto-inject surfaces, so this one line self-describes the snippet + in a recall menu. + - ``tags`` = ``[language, "snippet", *caller_tags]``. + - ``body`` = templated markdown (When to use / Signature / Location, then a + fenced code block). + +The public field API (name/language/signature/location/when_to_use/code) lives +here, so the underlying storage could later move to a structured column without +changing any caller (MCP tool, REST route, UI). +""" +from __future__ import annotations + +import re + +from scribe.services import knowledge as knowledge_svc +from scribe.services import notes as notes_svc + +SNIPPET_NOTE_TYPE = "snippet" +SNIPPET_TAG = "snippet" + + +# --- serialize: structured fields -> note (title/body/tags) ------------------ + +def compose_title(name: str, when_to_use: str = "") -> str: + """`name — when to use` (or just `name` when no usage note is given).""" + name = (name or "").strip() + when = (when_to_use or "").strip() + return f"{name} — {when}" if when else name + + +def compose_tags(language: str = "", tags: list[str] | None = None) -> list[str]: + """Language (lowercased) first, then the `snippet` marker, then caller tags — + de-duplicated, order preserved.""" + out: list[str] = [] + lang = (language or "").strip().lower() + if lang: + out.append(lang) + out.append(SNIPPET_TAG) + for t in tags or []: + t = (t or "").strip() + if t and t not in out: + out.append(t) + return out + + +def _location_line(repo: str = "", path: str = "", symbol: str = "") -> str: + parts = [p.strip() for p in (repo, path, symbol) if (p or "").strip()] + return " · ".join(f"`{p}`" for p in parts) + + +def compose_body( + *, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", +) -> str: + """Render structured fields into the snippet body markdown. Empty fields are + omitted so the body stays clean.""" + header: list[str] = [] + if (when_to_use or "").strip(): + header.append(f"**When to use:** {when_to_use.strip()}") + if (signature or "").strip(): + header.append(f"**Signature:** `{signature.strip()}`") + loc = _location_line(repo, path, symbol) + if loc: + header.append(f"**Location:** {loc}") + fence_lang = (language or "").strip().lower() + code_block = f"```{fence_lang}\n{(code or '').rstrip()}\n```" + if header: + return "\n\n".join(header) + "\n\n" + code_block + "\n" + return code_block + "\n" + + +# --- parse: note -> structured fields (best-effort, never raises) ------------ + +_WHEN_RE = re.compile(r"^\*\*When to use:\*\*\s*(.+?)\s*$", re.MULTILINE) +_SIG_RE = re.compile(r"^\*\*Signature:\*\*\s*(.+?)\s*$", re.MULTILINE) +_LOC_RE = re.compile(r"^\*\*Location:\*\*\s*(.+?)\s*$", re.MULTILINE) +_CODE_RE = re.compile(r"```([\w+.#-]*)\n(.*?)\n```", re.DOTALL) + + +def parse_snippet_fields( + title: str, body: str, tags: list[str] | None = None +) -> dict: + """Recover structured fields from a snippet note. Tolerant by design: a field + that isn't present comes back empty and this never raises, so a hand-edited + body can't break the edit form.""" + title = title or "" + body = body or "" + name, _, when_from_title = title.partition(" — ") + fields = { + "name": name.strip(), + "when_to_use": when_from_title.strip(), + "signature": "", + "language": "", + "repo": "", + "path": "", + "symbol": "", + "code": "", + } + + m = _WHEN_RE.search(body) + if m: + fields["when_to_use"] = m.group(1).strip() + m = _SIG_RE.search(body) + if m: + fields["signature"] = m.group(1).strip().strip("`").strip() + m = _LOC_RE.search(body) + if m: + loc_parts = [p.strip().strip("`").strip() for p in m.group(1).split("·")] + for key, val in zip(("repo", "path", "symbol"), loc_parts): + fields[key] = val + m = _CODE_RE.search(body) + if m: + fields["language"] = m.group(1).strip() + fields["code"] = m.group(2) + + if not fields["language"]: + for t in tags or []: + if t and t != SNIPPET_TAG: + fields["language"] = t + break + return fields + + +def snippet_to_dict(note) -> dict: + """Note serialization plus a parsed ``snippet`` sub-object of structured + fields, so callers get both the raw record and the typed view.""" + data = note.to_dict() + data["snippet"] = parse_snippet_fields(note.title, note.body, note.tags) + return data + + +# --- service wrappers over notes_svc ---------------------------------------- + +async def create_snippet( + user_id: int, + *, + name: str, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + tags: list[str] | None = None, + project_id: int | None = None, +): + """Create a snippet note. Returns the created Note.""" + return await notes_svc.create_note( + user_id, + title=compose_title(name, when_to_use), + body=compose_body( + code=code, language=language, signature=signature, + when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, + ), + note_type=SNIPPET_NOTE_TYPE, + tags=compose_tags(language, tags), + project_id=project_id, + ) + + +async def get_snippet(user_id: int, snippet_id: int): + """Fetch a snippet by id, or None if it doesn't exist / isn't a snippet.""" + note = await notes_svc.get_note(user_id, snippet_id) + if note is None or note.note_type != SNIPPET_NOTE_TYPE: + return None + return note + + +async def list_snippets( + user_id: int, + *, + q: str | None = None, + tag: str = "", + limit: int = 50, + offset: int = 0, +) -> tuple[list[dict], int]: + """List snippets (id/title/tags/preview dicts), most-recently-updated first.""" + return await knowledge_svc.query_knowledge( + user_id=user_id, + note_type=SNIPPET_NOTE_TYPE, + tags=[tag] if tag else [], + sort="modified", + q=q, + limit=max(1, min(limit, 100)), + offset=max(0, offset), + ) + + +async def update_snippet( + user_id: int, + 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, + tags: list[str] | None = None, + project_id: int | None = None, +): + """Partial update: only fields passed (not None) change. Re-serializes the + merged field set back into title/body/tags. Returns the Note, or None if the + id isn't a snippet.""" + note = await notes_svc.get_note(user_id, snippet_id) + if note is None or note.note_type != SNIPPET_NOTE_TYPE: + return None + + cur = parse_snippet_fields(note.title, note.body, note.tags) + overlay = { + "name": name, "code": code, "language": language, "signature": signature, + "when_to_use": when_to_use, "repo": repo, "path": path, "symbol": symbol, + } + merged = {**cur, **{k: v for k, v in overlay.items() if v is not None}} + + fields: dict = { + "title": compose_title(merged["name"], merged["when_to_use"]), + "body": compose_body( + code=merged["code"], language=merged["language"], + signature=merged["signature"], when_to_use=merged["when_to_use"], + repo=merged["repo"], path=merged["path"], symbol=merged["symbol"], + ), + } + # Recompute tags: keep any non-language, non-marker tags the note already had + # (or the caller's replacement set), then re-derive language + marker. + existing_extra = [ + t for t in (note.tags or []) if t not in (SNIPPET_TAG, cur.get("language", "")) + ] + fields["tags"] = compose_tags( + merged["language"], tags if tags is not None else existing_extra + ) + if project_id is not None: + fields["project_id"] = project_id + + return await notes_svc.update_note(user_id, snippet_id, **fields) diff --git a/tests/test_mcp_tool_snippets.py b/tests/test_mcp_tool_snippets.py new file mode 100644 index 0000000..7ea6284 --- /dev/null +++ b/tests/test_mcp_tool_snippets.py @@ -0,0 +1,100 @@ +"""Tests for MCP snippet tools — patches the service layer.""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from scribe.mcp._context import _user_id_ctx + + +@pytest.fixture(autouse=True) +def _bind_user(): + token = _user_id_ctx.set(7) + yield + _user_id_ctx.reset(token) + + +def _fake_snippet(): + n = MagicMock() + n.id = 1 + n.title = "debounce — rate-limit a callback" + n.body = "```js\nreturn 1\n```\n" + n.tags = ["js", "snippet"] + n.note_type = "snippet" + n.to_dict.return_value = { + "id": 1, "title": n.title, "note_type": "snippet", "tags": n.tags, + } + return n + + +@pytest.mark.asyncio +async def test_create_snippet_requires_name_and_code(): + from scribe.mcp.tools.snippets import create_snippet + with pytest.raises(ValueError): + await create_snippet(name="", code="x") + with pytest.raises(ValueError): + await create_snippet(name="x", code=" ") + + +@pytest.mark.asyncio +async def test_create_snippet_records_and_returns_parsed(): + created = _fake_snippet() + with patch("scribe.services.dedup.find_duplicate_note", AsyncMock(return_value=None)), \ + patch("scribe.services.snippets.create_snippet", + AsyncMock(return_value=created)) as mock_create: + from scribe.mcp.tools.snippets import create_snippet + out = await create_snippet( + name="debounce", code="return 1", language="js", + when_to_use="rate-limit a callback", + ) + assert out["note_type"] == "snippet" + assert out["snippet"]["name"] == "debounce" + assert out["snippet"]["language"] == "js" + assert mock_create.await_args.kwargs["name"] == "debounce" + + +@pytest.mark.asyncio +async def test_create_snippet_dedup_blocks_and_labels_snippet(): + dup = MagicMock(id=99) + with patch("scribe.services.dedup.find_duplicate_note", AsyncMock(return_value=dup)), \ + patch("scribe.services.dedup.duplicate_response", + MagicMock(return_value={"duplicate": True, "existing_id": 99})) as mock_resp, \ + patch("scribe.services.snippets.create_snippet", AsyncMock()) as mock_create: + from scribe.mcp.tools.snippets import create_snippet + out = await create_snippet(name="debounce", code="return 1") + assert out["duplicate"] is True + mock_create.assert_not_awaited() + assert mock_resp.call_args.args[1] == "snippet" + + +@pytest.mark.asyncio +async def test_get_snippet_not_found_raises(): + with patch("scribe.services.snippets.get_snippet", AsyncMock(return_value=None)): + from scribe.mcp.tools.snippets import get_snippet + with pytest.raises(ValueError): + await get_snippet(123) + + +@pytest.mark.asyncio +async def test_update_snippet_missing_raises(): + with patch("scribe.services.snippets.update_snippet", AsyncMock(return_value=None)): + from scribe.mcp.tools.snippets import update_snippet + with pytest.raises(ValueError): + await update_snippet(123, name="x") + + +def test_register_attaches_four_tools(): + from scribe.mcp.tools import snippets + names: list[str] = [] + + class FakeMcp: + def tool(self, name): + names.append(name) + + def deco(fn): + return fn + return deco + + snippets.register(FakeMcp()) + assert set(names) == { + "list_snippets", "create_snippet", "get_snippet", "update_snippet", + } diff --git a/tests/test_services_snippets.py b/tests/test_services_snippets.py new file mode 100644 index 0000000..3371484 --- /dev/null +++ b/tests/test_services_snippets.py @@ -0,0 +1,77 @@ +"""Unit tests for the snippet serialize/parse helpers (pure functions, no DB).""" +from scribe.services import snippets as s + + +def test_compose_title_with_and_without_usage(): + assert s.compose_title("debounce", "rate-limit a callback") == "debounce — rate-limit a callback" + assert s.compose_title(" debounce ", "") == "debounce" + assert s.compose_title("debounce") == "debounce" + + +def test_compose_tags_lowercases_language_and_dedups(): + assert s.compose_tags("Python", ["util", "python"]) == ["python", "snippet", "util"] + assert s.compose_tags("", None) == ["snippet"] + assert s.compose_tags("vue", ["snippet"]) == ["vue", "snippet"] + + +def test_compose_body_includes_fields_and_fence(): + body = s.compose_body( + code="return 1", language="python", signature="f() -> int", + when_to_use="always", repo="scribe", path="a.py", symbol="f", + ) + assert "**When to use:** always" in body + assert "**Signature:** `f() -> int`" in body + assert "`scribe` · `a.py` · `f`" in body + assert "```python\nreturn 1\n```" in body + assert body.rstrip().endswith("```") + + +def test_compose_body_bare_code_only(): + body = s.compose_body(code="x = 1") + assert body.strip() == "```\nx = 1\n```" + + +def test_parse_round_trips_a_composed_snippet(): + title = s.compose_title("useDebouncedRef", "debounce a reactive ref") + body = s.compose_body( + code="const x = 1", language="ts", signature="useDebouncedRef(v, ms)", + when_to_use="debounce a reactive ref", repo="scribe", + path="frontend/src/composables/x.ts", symbol="useDebouncedRef", + ) + got = s.parse_snippet_fields(title, body, ["ts", "snippet"]) + assert got["name"] == "useDebouncedRef" + assert got["when_to_use"] == "debounce a reactive ref" + assert got["signature"] == "useDebouncedRef(v, ms)" + assert got["language"] == "ts" + assert got["code"] == "const x = 1" + assert got["repo"] == "scribe" + assert got["path"] == "frontend/src/composables/x.ts" + assert got["symbol"] == "useDebouncedRef" + + +def test_parse_is_tolerant_of_plain_body(): + got = s.parse_snippet_fields("just a name", "no structure here", None) + assert got["name"] == "just a name" + assert got["when_to_use"] == "" + assert got["signature"] == "" + assert got["code"] == "" # never raises on an unstructured body + + +def test_parse_falls_back_to_tag_for_language(): + got = s.parse_snippet_fields("n — u", "```\ncode\n```", ["ruby", "snippet"]) + assert got["language"] == "ruby" + + +def test_snippet_to_dict_includes_parsed_fields(): + class FakeNote: + title = "debounce — rate-limit" + body = "```js\ncode\n```\n" + tags = ["js", "snippet"] + + def to_dict(self): + return {"id": 1, "title": self.title, "note_type": "snippet", "tags": self.tags} + + data = s.snippet_to_dict(FakeNote()) + assert data["snippet"]["name"] == "debounce" + assert data["snippet"]["language"] == "js" + assert data["snippet"]["code"] == "code"