fix(snippets): close the recall-surface gaps found reviewing the Drafter
CI & Build / Python lint (push) Successful in 3s
CI & Build / TypeScript typecheck (push) Successful in 11s
CI & Build / integration (push) Successful in 19s
CI & Build / Python tests (push) Successful in 43s
CI & Build / Build & push image (push) Successful in 1m7s

Four defects from the 2026-07-25 review of the recall (#227) and merge (#231)
milestones. The theme: a snippet could be recorded but not fully corrected, and
the agent and web surfaces had drifted apart.

- #2076 language was mis-derived from the first caller tag, so a snippet created
  with tags and no language read that tag back as its language — corrupting the
  tag set and the code fence on the next update. Only the FIRST tag can carry
  the language, since compose_tags emits [language, "snippet", *caller].
- #2077 MCP update_snippet mapped "" to "unchanged", so no field could ever be
  cleared and no snippet detached from its project. Now an omitted field is left
  alone, an empty string clears, and project_id follows the -1 = detach
  convention. A service-level UNSET sentinel keeps None available as the clear.
- #2078 surface parity: adds delete_snippet (MCP had none, so a wrong snippet
  could not be retired by the agent that recorded it), locations on MCP create
  and update, system_ids through the REST routes and the editor, and the
  near-duplicate gate on REST create with a "record it anyway" escape.
- #2079 project scoping: list_snippets takes project_id through the service, the
  MCP tool and the REST route, defaulting to every project — reaching across
  projects is the point when the helper you need was written elsewhere.

Sharing the list across owners is deliberately NOT in here: query_knowledge is
shared with the Knowledge browse surface, so widening it changes behaviour well
beyond snippets. Left open on #2079 for a scope decision.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RLwAaV4DQEmVyn496HnEvt
This commit is contained in:
2026-07-25 19:00:05 -04:00
parent 7a81b7333e
commit b33e2a79c6
12 changed files with 492 additions and 43 deletions
+4 -1
View File
@@ -223,7 +223,10 @@ project_id, system_ids) so a later session is offered it. Make when_to_use sharp
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).
duplicates). Keep the record honest: a snippet whose details have gone stale can
be corrected with update_snippet (an empty string clears a field), and one that
is wrong or obsolete should be retired with delete_snippet — a bad snippet keeps
being offered as prior art, which costs more than none at all.
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
+67 -21
View File
@@ -16,13 +16,19 @@ 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:
async def list_snippets(
q: str = "", tag: str = "", limit: int = 50, project_id: int = 0,
) -> dict:
"""List recorded snippets (reusable functions/components).
Args:
q: Free-text search across name + body (optional).
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.
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).
@@ -30,6 +36,7 @@ async def list_snippets(q: str = "", tag: str = "", limit: int = 50) -> dict:
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,
)
return {"snippets": items, "total": total}
@@ -43,6 +50,7 @@ async def create_snippet(
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,
@@ -65,6 +73,9 @@ async def create_snippet(
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.
@@ -87,6 +98,7 @@ async def create_snippet(
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(
@@ -99,7 +111,7 @@ async def create_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,
locations=locations, tags=tags, project_id=project_id or None,
)
if system_ids:
await systems_svc.set_record_systems(uid, note.id, system_ids)
@@ -123,31 +135,48 @@ async def get_snippet(snippet_id: int) -> dict:
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 = "",
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 provided fields change — empty strings leave that
field unchanged; pass tags to replace the extra-tag set."""
uid = current_user_id()
"""Update a snippet. Only the fields you pass change.
def _n(v: str) -> str | None:
return v if v != "" else None
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.
"""
uid = current_user_id()
if project_id == 0:
project = snippets_svc.UNSET
elif project_id < 0:
project = None
else:
project = project_id
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,
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,
)
if note is None:
raise ValueError(f"snippet {snippet_id} not found")
@@ -161,6 +190,20 @@ async def update_snippet(
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.
@@ -194,5 +237,8 @@ async def merge_snippets(target_id: int, source_ids: list[int]) -> dict:
def register(mcp) -> None:
for fn in (list_snippets, create_snippet, get_snippet, update_snippet, merge_snippets):
for fn in (
list_snippets, create_snippet, get_snippet, update_snippet,
delete_snippet, merge_snippets,
):
mcp.tool(name=fn.__name__)(fn)
+56 -6
View File
@@ -16,7 +16,9 @@ from quart import Blueprint, jsonify, request
from scribe.auth import get_current_user_id, login_required
from scribe.routes.utils import not_found, parse_pagination
from scribe.services import dedup as dedup_svc
from scribe.services import snippets as snippets_svc
from scribe.services import systems as systems_svc
from scribe.services.access import can_write_note
from scribe.services.notes import get_note_for_user
@@ -46,9 +48,13 @@ async def list_snippets_route():
uid = get_current_user_id()
q = request.args.get("q") or None
tag = request.args.get("tag", "")
try:
project_id = int(request.args.get("project_id", 0) or 0) or None
except (TypeError, ValueError):
project_id = None
limit, offset = parse_pagination()
items, total = await snippets_svc.list_snippets(
uid, q=q, tag=tag, limit=limit, offset=offset,
uid, q=q, tag=tag, limit=limit, offset=offset, project_id=project_id,
)
return jsonify({"snippets": items, "total": total})
@@ -63,6 +69,31 @@ async def create_snippet_route():
if not name or not code:
return jsonify({"error": "name and code are required"}), 400
project_id = data.get("project_id") or None
# Same near-duplicate gate the MCP create path applies: recording the same
# reusable thing twice is what merge then has to undo, so catch it here too.
# `force` is the deliberate override once the operator has seen the warning.
if not data.get("force"):
dup = await dedup_svc.find_duplicate_note(
uid,
snippets_svc.compose_title(name, data.get("when_to_use", "")),
snippets_svc.compose_body(
code=data.get("code", ""),
language=data.get("language", ""),
signature=data.get("signature", ""),
when_to_use=data.get("when_to_use", ""),
repo=data.get("repo", ""),
path=data.get("path", ""),
symbol=data.get("symbol", ""),
locations=data.get("locations"),
),
project_id=project_id,
is_task=False,
note_type=snippets_svc.SNIPPET_NOTE_TYPE,
)
if dup is not None:
return jsonify(dedup_svc.duplicate_response(dup, "snippet")), 409
note = await snippets_svc.create_snippet(
uid,
name=name,
@@ -77,7 +108,13 @@ async def create_snippet_route():
tags=data.get("tags"),
project_id=project_id,
)
return jsonify(snippets_svc.snippet_to_dict(note)), 201
if data.get("system_ids") is not None:
await systems_svc.set_record_systems(uid, note.id, data["system_ids"])
out = snippets_svc.snippet_to_dict(note)
out["systems"] = [
s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
]
return jsonify(out), 201
@snippets_bp.route("/<int:snippet_id>", methods=["GET"])
@@ -90,6 +127,13 @@ async def get_snippet_route(snippet_id: int):
note, permission = loaded
data = snippets_svc.snippet_to_dict(note)
data["permission"] = permission
# Read the association as the OWNER: a shared reader isn't scoped to the
# owner's project, so their own id would come back empty (mirrors the
# write-as-owner pattern this module already uses).
data["systems"] = [
s.to_dict()
for s in await systems_svc.list_record_systems(note.user_id, snippet_id)
]
return jsonify(data)
@@ -115,12 +159,20 @@ async def update_snippet_route(snippet_id: int):
if "tags" in data:
kwargs["tags"] = data["tags"]
if "project_id" in data:
# A present-but-empty project_id is a deliberate detach, not "unchanged"
# — the service distinguishes the two via its UNSET sentinel.
kwargs["project_id"] = data["project_id"] or None
updated = await snippets_svc.update_snippet(owner_uid, snippet_id, **kwargs)
if updated is None:
return not_found("Snippet")
return jsonify(snippets_svc.snippet_to_dict(updated))
if data.get("system_ids") is not None:
await systems_svc.set_record_systems(owner_uid, snippet_id, data["system_ids"])
out = snippets_svc.snippet_to_dict(updated)
out["systems"] = [
s.to_dict() for s in await systems_svc.list_record_systems(owner_uid, snippet_id)
]
return jsonify(out)
@snippets_bp.route("/<int:snippet_id>/merge", methods=["POST"])
@@ -173,8 +225,6 @@ async def delete_snippet_route(snippet_id: int):
note, _ = loaded
if not await can_write_note(uid, snippet_id):
return jsonify({"error": "Permission denied"}), 403
from scribe.services.trash import delete as trash_delete
batch = await trash_delete(note.user_id, "note", snippet_id)
if batch is None:
if not await snippets_svc.delete_snippet(note.user_id, snippet_id):
return not_found("Snippet")
return "", 204
+11 -1
View File
@@ -59,21 +59,27 @@ async def query_knowledge(
q: str | None,
limit: int,
offset: int,
project_id: int | None = None,
) -> tuple[list[dict], int]:
"""Query knowledge objects (non-task notes) with filters.
`project_id` narrows to one project (None = every project).
Returns (items, total_count).
"""
# Semantic search path — scores take priority over sort
if q:
return await _semantic_knowledge_search(
user_id, q, note_type=note_type, tags=tags, limit=limit, offset=offset
user_id, q, note_type=note_type, tags=tags, limit=limit,
offset=offset, project_id=project_id,
)
async with async_session() as session:
base = select(Note).where(Note.user_id == user_id)
base = _apply_type_filter(base, note_type)
if project_id is not None:
base = base.where(Note.project_id == project_id)
for tag in tags:
base = base.where(Note.tags.contains([tag]))
@@ -104,6 +110,7 @@ async def _semantic_knowledge_search(
tags: list[str],
limit: int,
offset: int,
project_id: int | None = None,
) -> tuple[list[dict], int]:
"""Hybrid search: keyword matches first (title/body ILIKE), then semantic results.
@@ -130,6 +137,8 @@ async def _semantic_knowledge_search(
.where(Note.title.ilike(pattern) | Note.body.ilike(pattern))
)
base = _apply_type_filter(base, note_type)
if project_id is not None:
base = base.where(Note.project_id == project_id)
for tag in tags:
base = base.where(Note.tags.contains([tag]))
# Title matches first, then body-only matches, newest first within each
@@ -152,6 +161,7 @@ async def _semantic_knowledge_search(
limit=min(200, limit * 4),
threshold=0.3,
is_task=is_task_filter,
project_id=project_id,
)
for _score, note in candidates:
if note.deleted_at is not None:
+39 -8
View File
@@ -32,6 +32,11 @@ from scribe.services import notes as notes_svc
SNIPPET_NOTE_TYPE = "snippet"
SNIPPET_TAG = "snippet"
# Sentinel for "argument not supplied" on update, so None stays available as a
# real value meaning "clear this". Needed for project_id, where 0 is not a valid
# id and None is the clear — the two can't share one default.
UNSET: object = object()
def _embed_snippet(note) -> None:
"""Fire-and-forget embedding refresh for a snippet.
@@ -235,11 +240,13 @@ def parse_snippet_fields(
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
# Language fallback for a body whose code fence lost its language. Only the
# FIRST tag can be trusted: compose_tags emits [language, "snippet", *caller],
# so a leading tag that isn't the marker is the language — while a leading
# marker means no language was recorded. Scanning for "first tag that isn't
# the marker" instead would promote a caller's plain tag to the language.
if not fields["language"] and tags and tags[0] != SNIPPET_TAG:
fields["language"] = tags[0]
return fields
@@ -302,8 +309,13 @@ async def list_snippets(
tag: str = "",
limit: int = 50,
offset: int = 0,
project_id: int | None = None,
) -> tuple[list[dict], int]:
"""List snippets (id/title/tags/preview dicts), most-recently-updated first."""
"""List snippets (id/title/tags/preview dicts), most-recently-updated first.
``project_id`` narrows to one project; omit it to reach across every project
— which is the point when the thing you're about to write was already solved
somewhere else."""
return await knowledge_svc.query_knowledge(
user_id=user_id,
note_type=SNIPPET_NOTE_TYPE,
@@ -312,6 +324,7 @@ async def list_snippets(
q=q,
limit=max(1, min(limit, 100)),
offset=max(0, offset),
project_id=project_id,
)
@@ -329,12 +342,15 @@ async def update_snippet(
symbol: str | None = None,
locations: list[dict] | None = None,
tags: list[str] | None = None,
project_id: int | None = None,
project_id: int | None | object = UNSET,
):
"""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.
``project_id``: omit to leave unchanged, pass None to detach from its
project, pass an id to move it.
Locations: ``locations`` replaces the whole set; else a legacy single
``repo``/``path``/``symbol`` overlays onto the first existing location; else
the existing locations are kept."""
@@ -377,7 +393,7 @@ async def update_snippet(
fields["tags"] = compose_tags(
merged["language"], tags if tags is not None else existing_extra
)
if project_id is not None:
if project_id is not UNSET:
fields["project_id"] = project_id
updated = await notes_svc.update_note(user_id, snippet_id, **fields)
@@ -387,6 +403,21 @@ async def update_snippet(
return updated
async def delete_snippet(user_id: int, snippet_id: int) -> bool:
"""Retire a snippet to the trash (recoverable). Returns False if the id isn't
the user's snippet.
Recall makes this corrective, not merely tidy: a wrong or obsolete snippet
doesn't sit quietly — it keeps being offered as prior art. Removing it has to
be reachable from wherever it was recorded.
"""
note = await get_snippet(user_id, snippet_id)
if note is None:
return False
from scribe.services.trash import delete as trash_delete
return await trash_delete(note.user_id, "note", snippet_id) is not None
# --- merge: unify found one-offs into one canonical snippet ------------------
def _extra_tags(tags: list[str] | None, language: str = "") -> list[str]: