"""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** — the body is the readable form and the thing that gets embedded, 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). Since migration 0070 the same fields are ALSO written to ``notes.data`` (see ``compose_data``) — a queryable mirror, not a second source of truth: it carries no code, and it exists so location/language can be indexed rather than regexed out of every body. Reads prefer it and fall back to the body parse. The public field API (name/language/signature/location/when_to_use/code) lives here, so callers (MCP tool, REST route, UI) never see which of the two a field came from. """ from __future__ import annotations import hashlib import logging import re from datetime import datetime, timezone from sqlalchemy import func, or_, select from scribe.models import async_session from scribe.models.note import Note from scribe.services import knowledge as knowledge_svc from scribe.services import notes as notes_svc logger = logging.getLogger(__name__) 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() # --- 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 _normalize_locations(locations: list[dict] | None) -> list[dict]: """Clean a list of {repo,path,symbol} locations: strip fields, drop wholly empty entries, de-duplicate identical ones (order preserved). A merged snippet carries several locations (one per call site); a fresh one carries at most one.""" out: list[dict] = [] seen: set[tuple[str, str, str]] = set() for loc in locations or []: repo = (loc.get("repo") or "").strip() path = (loc.get("path") or "").strip() symbol = (loc.get("symbol") or "").strip() if not (repo or path or symbol): continue key = (repo, path, symbol) if key in seen: continue seen.add(key) out.append({"repo": repo, "path": path, "symbol": symbol}) return out def _location_str(loc: dict) -> str: """`repo` · `path` · `symbol` — only the non-empty parts.""" parts = [(loc.get(k) or "").strip() for k in ("repo", "path", "symbol")] return " · ".join(f"`{p}`" for p in parts if p) def _render_location_block(locations: list[dict]) -> str | None: """One `**Location:**` line for a single location; a `**Locations:**` bullet list for several. None when there are none.""" locs = [loc for loc in locations if _location_str(loc)] if not locs: return None if len(locs) == 1: return f"**Location:** {_location_str(locs[0])}" lines = "\n".join(f"- {_location_str(loc)}" for loc in locs) return f"**Locations:**\n{lines}" def _normalize_merged_from(entries: list | None) -> list[dict]: """Merge provenance as `[{"id": int, "locations": [...], "tags": [...]}]`. Order is history, not sorting — earlier merges stay first, so the list reads as the sequence of things folded in. Each entry records WHAT THAT SOURCE CONTRIBUTED, which is what makes un-merge (#2165) exact. Two problems it solves at once: - A location can arrive from a source AND genuinely be the survivor's own. Recording only what the source ADDED means reversing it can never strip a call site the survivor already had. - Two sources can bring the same location. Only the first records it, so un-merging the second leaves it in place — correctly, since the first still claims it. A bare int is accepted and normalized to `{"id": n}` with no attribution. That is not legacy tolerance: `snippet_fields` falls back to PARSING THE BODY when a row has no `data`, and the body's `**Merged from:** #ids` line can only ever carry ids. Such an entry still shows provenance; un-merge refuses it rather than guessing, because guessing is exactly the failure above. """ out: list[dict] = [] seen: set[int] = set() for raw in entries or []: if isinstance(raw, dict): ident, locs, tags = raw.get("id"), raw.get("locations"), raw.get("tags") else: ident, locs, tags = raw, None, None try: i = int(ident) except (TypeError, ValueError): continue if i <= 0 or i in seen: continue seen.add(i) entry: dict = {"id": i} norm_locs = _normalize_locations(locs) if locs else [] if norm_locs: entry["locations"] = norm_locs clean_tags = [t for t in (tags or []) if isinstance(t, str) and t.strip()] if clean_tags: entry["tags"] = clean_tags out.append(entry) return out def merged_from_ids(entries: list | None) -> list[int]: """Just the absorbed ids, in history order — for display and containment.""" return [e["id"] for e in _normalize_merged_from(entries)] def compose_body( *, code: str, language: str = "", signature: str = "", when_to_use: str = "", repo: str = "", path: str = "", symbol: str = "", locations: list[dict] | None = None, merged_from: list[int] | None = None, ) -> str: """Render structured fields into the snippet body markdown. Empty fields are omitted so the body stays clean. Locations: pass ``locations`` (a list of {repo,path,symbol}) for the general multi-location case; the single ``repo``/``path``/``symbol`` params remain as a back-compat shorthand for one location and are used only when ``locations`` is not given. """ if locations is None: locations = [{"repo": repo, "path": path, "symbol": symbol}] locs = _normalize_locations(locations) 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_block = _render_location_block(locs) if loc_block: header.append(loc_block) merged = _normalize_merged_from(merged_from) if merged: # Human-readable mirror of data["merged_from"]. Without it, a merge folds # variants in and the record of what was absorbed lives only in the trash # — recoverable only by someone who already knows to go looking. # Ids only — the body is the human-readable mirror, and per-source # attribution belongs in `data` where it can be queried rather than # re-parsed out of prose. header.append( "**Merged from:** " + ", ".join(f"#{e['id']}" for e in merged) ) 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) _LOCS_RE = re.compile( r"^\*\*Locations:\*\*[ \t]*\n((?:[ \t]*-[ \t]*.+\n?)+)", re.MULTILINE ) _MERGED_RE = re.compile(r"^\*\*Merged from:\*\*\s*(.+?)\s*$", re.MULTILINE) _CODE_RE = re.compile(r"```([\w+.#-]*)\n(.*?)\n```", re.DOTALL) _ID_RE = re.compile(r"#(\d+)") def _parse_location_str(s: str) -> dict | None: """Parse a `repo` · `path` · `symbol` fragment back into a location dict, or None if it's empty.""" raw = [p.strip().strip("`").strip() for p in (s or "").split("·")] repo = raw[0] if len(raw) > 0 else "" path = raw[1] if len(raw) > 1 else "" symbol = raw[2] if len(raw) > 2 else "" if not (repo or path or symbol): return None return {"repo": repo, "path": path, "symbol": symbol} 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. ``locations`` is a list of {repo,path,symbol}; ``repo``/``path``/``symbol`` mirror the FIRST location for back-compat with the single-location callers.""" 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": "", "locations": [], "merged_from": [], "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() # Locations: prefer the multi-location `**Locations:**` bullet list, else the # legacy single `**Location:**` line. locations: list[dict] = [] m = _LOCS_RE.search(body) if m: for line in m.group(1).splitlines(): line = line.strip() if line.startswith("-"): loc = _parse_location_str(line[1:]) if loc: locations.append(loc) else: m = _LOC_RE.search(body) if m: loc = _parse_location_str(m.group(1)) if loc: locations.append(loc) fields["locations"] = locations if locations: fields["repo"] = locations[0]["repo"] fields["path"] = locations[0]["path"] fields["symbol"] = locations[0]["symbol"] m = _MERGED_RE.search(body) if m: fields["merged_from"] = _normalize_merged_from( [int(i) for i in _ID_RE.findall(m.group(1))] ) m = _CODE_RE.search(body) if m: fields["language"] = m.group(1).strip() fields["code"] = m.group(2) # 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 # --- the queryable mirror (notes.data, migration 0070) ----------------------- # Fields kept in `data`. Code is deliberately absent: the body already holds it, # and copying a blob into the column we index *around* would be pure weight. _DATA_FIELDS = ( "name", "when_to_use", "signature", "language", "locations", "merged_from", "verification", ) # --- drift check (#2086) ----------------------------------------------------- # A recorded snippet points at a repo · path · symbol that WILL rot: files move, # symbols get renamed, implementations diverge from the copy stored here. Left # undetected, the record degrades from "canonical reference" to "confidently # wrong" — which is worse than having no record, because it is surfaced with the # same authority either way. # # WHERE THE CHECK RUNS. Not here. Scribe has no checkout of the operator's repos # and must not acquire one (rule #115 — the instance stays agnostic about where # code lives; giving the server repo access would make every install a # credential problem). The agent already has the working tree, so IT does the # comparing and reports a verdict; the server's job is to remember the verdict, # make it queryable, and know when it has expired. # # WHY THE VERDICT CARRIES A CODE HASH. A stored verdict describes the code it # was checked against. Edit the snippet afterwards and that verdict is no longer # about anything — but invalidating it on write means deciding which edits count # (a `when_to_use` tweak shouldn't void a code check; a code rewrite must). That # rule is fiddly and easy to get subtly wrong. Recording the hash sidesteps it # entirely: a verdict whose `code_sha` no longer matches the body is self- # evidently expired, computed at read time, with no invalidation logic to # maintain and no way for an edit path to forget to call it. VERIFY_OK = "ok" VERIFY_MISSING = "missing" # the recorded path is gone VERIFY_MOVED = "moved" # path is there, the symbol isn't in it VERIFY_CHANGED = "changed" # both present, but the source no longer matches VERIFY_STATUSES = (VERIFY_OK, VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED) # Everything that isn't a clean bill of health. "Stale" in the UI and the filter # means this set — the operator wants one list of things to look at, not four. VERIFY_DRIFTED = (VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED) def code_sha(code: str) -> str: """Stable fingerprint of a snippet's code, for expiring stale verdicts. Trailing whitespace per line and leading/trailing blank lines are stripped before hashing: those change when a file is reformatted without the code meaning anything different, and a verdict shouldn't expire over an editor's trailing-newline habit. """ normalized = "\n".join(line.rstrip() for line in (code or "").splitlines()).strip() return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:32] def compose_verification( *, status: str, checked_code_sha: str, detail: str = "", path: str = "", checked_at: str = "", ) -> dict: """Build the `data.verification` record. Unknown statuses are rejected here rather than stored, so the filter never has to cope with a typo'd status.""" if status not in VERIFY_STATUSES: raise ValueError( f"unknown verification status {status!r} — expected one of {VERIFY_STATUSES}" ) out = { "status": status, "code_sha": checked_code_sha, "checked_at": checked_at or datetime.now(timezone.utc).isoformat(), } if (detail or "").strip(): out["detail"] = detail.strip() if (path or "").strip(): out["path"] = path.strip() return out def verification_view(note, fields: dict) -> dict: """The verification readout for one snippet, including whether it's expired. `status` is what was last reported; `current` says whether that verdict still describes the code in the record. A verdict that no longer matches reads as unverified, because that is what it is — nobody has checked THIS code. """ stored = (fields.get("verification") or {}) if isinstance(fields, dict) else {} if not stored or not stored.get("status"): return {"status": "unverified", "current": False, "checked_at": None} current = stored.get("code_sha") == code_sha(fields.get("code") or "") return { "status": stored["status"], "current": current, "checked_at": stored.get("checked_at"), "detail": stored.get("detail"), "path": stored.get("path"), # What the operator actually wants to know: is there something to fix? # An expired verdict counts as "needs looking at" even if it said ok, # since the code it blessed is not the code that's there now. "needs_attention": (not current) or stored["status"] in VERIFY_DRIFTED, } def compose_data( *, name: str = "", when_to_use: str = "", signature: str = "", language: str = "", code: str = "", locations: list[dict] | None = None, merged_from: list[int] | None = None, verification: dict | None = None, ) -> dict: """Build the `notes.data` mirror of a snippet's structured fields. Same facts as the body convention, in a shape Postgres can index — so "which snippets live in this path?" is a containment query rather than a regex over every body. Empty values are omitted so the column stays sparse and containment matches don't trip over blanks. """ out: dict = {} for key, value in ( ("name", (name or "").strip()), ("when_to_use", (when_to_use or "").strip()), ("signature", (signature or "").strip()), ("language", (language or "").strip().lower()), ): if value: out[key] = value locs = _normalize_locations(locations) if locs: out["locations"] = locs merged = _normalize_merged_from(merged_from) if merged: out["merged_from"] = merged # Carried, never composed here — like merged_from. An ordinary edit must not # silently drop the last drift check, and it doesn't need to invalidate it # either: the verdict's code_sha expires it on read if the code moved on. if verification: out["verification"] = verification # The current code's fingerprint — NOT the code, which stays in the body # (see _DATA_FIELDS). Its only job is to make "this verdict has expired" # expressible in SQL: a jsonpath can compare `@.verification.code_sha` to # `@.code_sha` within the same row, so "show me everything that needs # looking at" stays one index-served query instead of a post-filter that # would break pagination counts. if (code or "").strip(): out["code_sha"] = code_sha(code) return out def snippet_fields(note) -> dict: """Structured fields for a snippet, preferring the indexed `data` column and falling back to parsing the body. Both paths must agree, because rows written before migration 0070 have no `data` and are never backfilled — a hand-edited body is the authority for those, and there is no deadline by which they must be converted. `code` only ever comes from the body, since `data` doesn't carry it. """ parsed = parse_snippet_fields(note.title, note.body, note.tags) stored = getattr(note, "data", None) if not stored: return parsed merged = dict(parsed) for key in _DATA_FIELDS: if stored.get(key): merged[key] = stored[key] # Keep the single-location back-compat mirror consistent with whichever # location list won. locs = merged.get("locations") or [] merged["repo"] = locs[0]["repo"] if locs else "" merged["path"] = locs[0]["path"] if locs else "" merged["symbol"] = locs[0]["symbol"] if locs else "" return merged async def backfill_snippet_data(*, batch: int = 500) -> int: """Populate `notes.data` for snippet rows that predate migration 0070. Runs once at startup (see app.py). Returns how many rows were filled. Migration 0070 deliberately left `data` NULL on existing rows, because *reading* a snippet never needed it — `snippet_fields` falls back to parsing the body. Querying does: the location reverse lookup (#2083) is a jsonpath over this column, so a NULL-`data` snippet would be reported as "nothing recorded here" and the caller would write the helper again — silently worse than having no reverse lookup at all. Every consumer of the column (reverse lookup, and the write-path trigger built on it) can then assume the mirror is present instead of carrying a second body-regex arm. This does not reverse 0070's actual caution, which was about mangling a hand-edited *body*: the body is never touched here, only the mirror derived from it, by the same parser the read path already trusts. Idempotent — a filled row is skipped forever after, and a snippet with no structured fields at all settles at `{}` rather than staying NULL and being re-scanned. Trashed rows are included so a later restore comes back queryable. "Unfilled" means SQL NULL *or* JSON `null` — two different states in a JSONB column, and only the first is what migration 0070 left behind. SQLAlchemy's JSON types default to ``none_as_null=False``, so assigning Python ``None`` to this column persists the JSON encoding of null rather than SQL NULL; an ``IS NULL`` test alone walks straight past such a row and reports nothing to do. Both mean "no usable mirror", so both are filled. """ filled = 0 unfilled = or_(Note.data.is_(None), func.jsonb_typeof(Note.data) == "null") async with async_session() as session: while True: rows = list( ( await session.execute( select(Note) .where(Note.note_type == SNIPPET_NOTE_TYPE) .where(unfilled) .limit(batch) ) ) .scalars() .all() ) if not rows: break for note in rows: fields = parse_snippet_fields(note.title, note.body, note.tags) note.data = compose_data( name=fields["name"], when_to_use=fields["when_to_use"], signature=fields["signature"], language=fields["language"], code=fields["code"], locations=fields["locations"], merged_from=fields["merged_from"], ) await session.commit() filled += len(rows) if len(rows) < batch: break if filled: logger.info("Snippet data backfill: populated `data` for %d snippet(s)", filled) return filled 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. The raw `data` column is intentionally NOT exposed: it mirrors what ``snippet`` already reports, and shipping both would give API consumers two sources of truth for the same facts.""" data = note.to_dict() fields = snippet_fields(note) data["snippet"] = fields # Promoted out of `snippet` because it is a computed READOUT, not a recorded # field: `current` and `needs_attention` are derived at read time by hashing # the code, and burying them among the stored fields would invite a caller # to try writing them back. data["verification"] = verification_view(note, fields) 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 = "", locations: list[dict] | None = None, tags: list[str] | None = None, project_id: int | None = None, ): """Create a snippet note (embedded on create for immediate recall). Returns the created Note. Pass ``locations`` for the multi-location case; the single ``repo``/``path``/``symbol`` are the one-location shorthand.""" if locations is None: locations = [{"repo": repo, "path": path, "symbol": symbol}] note = 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, locations=locations, ), note_type=SNIPPET_NOTE_TYPE, tags=compose_tags(language, tags), project_id=project_id, # The indexed mirror of the same fields (0070). Written together with the # body so the two can never describe different things. data=compose_data( name=name, when_to_use=when_to_use, signature=signature, language=language, code=code, locations=locations, ), ) return note 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 / isn't readable by this user. Share-aware (rule #78): a fetch by id is an explicit act, so it resolves the caller's full read scope rather than ownership alone. Without this, a snippet that a search legitimately surfaced could not then be opened — see #2093.""" result = await notes_svc.get_note_for_user(user_id, snippet_id) if result is None: return None note, _permission = result if note.note_type != SNIPPET_NOTE_TYPE or note.deleted_at is not None: return None return note async def list_snippets( user_id: int, *, q: str | None = None, tag: str = "", limit: int = 50, offset: int = 0, project_id: int | None = None, repo: str = "", path: str = "", symbol: str = "", verification: str = "", ) -> tuple[list[dict], int]: """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. ``repo`` / ``path`` / ``symbol`` are the reverse lookup: "what canonical helpers already live here?" They narrow to snippets recorded at a matching location, ANDed within one location entry, with ``path`` also matching as a directory prefix. Combinable with ``q`` — search *and* place. ``verification`` narrows on the drift check: ``attention`` is the useful one — everything whose recorded location or code no longer checks out, plus everything whose verdict expired because the snippet was edited since.""" 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), project_id=project_id, locations=knowledge_svc.location_parts(repo=repo, path=path, symbol=symbol) or None, verification=verification, ) 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, locations: list[dict] | None = None, tags: list[str] | 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 the caller can see. Share-aware (rule #47/#78): resolves the read scope, then requires WRITE — so an editor/admin grant lets the holder edit, and a viewer grant does not. Raises PermissionError when the caller can read but not write, because "not found" would be a lie about a record they can plainly open. The write itself is performed as the OWNER, mirroring routes/snippets.py, since the underlying note update is owner-scoped. ``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.""" note = await get_snippet(user_id, snippet_id) if note is None: return None from scribe.services.access import can_write_note if not await can_write_note(user_id, snippet_id): raise PermissionError( f"snippet {snippet_id} is shared with you read-only — ask its owner " f"for edit access, or record your own version" ) cur = snippet_fields(note) overlay = { "name": name, "code": code, "language": language, "signature": signature, "when_to_use": when_to_use, } merged = {**cur, **{k: v for k, v in overlay.items() if v is not None}} if locations is not None: merged_locations = _normalize_locations(locations) elif repo is not None or path is not None or symbol is not None: base = cur["locations"][0] if cur["locations"] else {"repo": "", "path": "", "symbol": ""} merged_locations = _normalize_locations([{ "repo": repo if repo is not None else base["repo"], "path": path if path is not None else base["path"], "symbol": symbol if symbol is not None else base["symbol"], }]) else: merged_locations = cur["locations"] 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"], locations=merged_locations, # Carried, never set here: an ordinary edit must not erase the record # of what was folded in, and only a merge may add to it. merged_from=merged.get("merged_from"), ), # Re-derived from the same merged field set as the body, so an edit can't # leave the indexed mirror describing the previous version. "data": compose_data( name=merged["name"], when_to_use=merged["when_to_use"], signature=merged["signature"], language=merged["language"], code=merged["code"], locations=merged_locations, merged_from=merged.get("merged_from"), # Carried through the edit rather than cleared. If this edit changed # the code, the verdict's code_sha stops matching and it reads as # unverified from here on — no invalidation branch to get wrong. verification=merged.get("verification"), ), } # 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 UNSET: fields["project_id"] = project_id # As the OWNER: update_note is owner-scoped, so a shared editor's own id # would find nothing. The write was authorised by can_write_note above. updated = await notes_svc.update_note(note.user_id, snippet_id, **fields) return updated async def record_verification( user_id: int, snippet_id: int, *, status: str, detail: str = "", path: str = "", ): """Record the result of a drift check against the snippet's source. The CHECK happens agent-side — Scribe has no checkout and shouldn't want one (see the drift-check note above). This just remembers the verdict, stamped with a hash of the code it was checked against so it expires by itself when the snippet is edited. Requires WRITE access: a verdict changes how the record is presented and whether it shows up in the operator's "needs attention" list, so being able to read a shared snippet must not let you mark it broken. Returns the updated note, or None if the id isn't a snippet this user may write. Raises ValueError on an unknown status. """ note = await get_snippet(user_id, snippet_id) if note is None: return None from scribe.services.access import can_write_note if not await can_write_note(user_id, snippet_id): return None fields = snippet_fields(note) verification = compose_verification( status=status, checked_code_sha=code_sha(fields.get("code") or ""), detail=detail, path=path or fields.get("path") or "", ) # Rebuilt from the CURRENT stored fields plus the new verdict, so recording a # check can't quietly rewrite anything else about the record. Note the body # is untouched — a verdict is metadata about the snippet, not part of it, and # writing it into the body would put it into the embedding. data = compose_data( name=fields.get("name", ""), when_to_use=fields.get("when_to_use", ""), signature=fields.get("signature", ""), language=fields.get("language", ""), code=fields.get("code", ""), locations=fields.get("locations") or [], merged_from=fields.get("merged_from") or [], verification=verification, ) return await notes_svc.update_note(note.user_id, snippet_id, data=data) 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 a snippet this user may WRITE. 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 the explicit write check: `get_snippet` resolves the READ scope, which now includes snippets merely shared with this user — being able to see one must not imply being able to bin it. """ note = await get_snippet(user_id, snippet_id) if note is None: return False from scribe.services.access import can_write_note if not await can_write_note(user_id, snippet_id): 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]: """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, contributions) where `contributions` is one `{"locations": [...], "tags": [...]}` per source, positionally aligned with `sources`, holding ONLY what that source actually added — anything the survivor (or an earlier source) already had is not attributed to it. That is what lets un-merge subtract exactly, without stripping a call site the survivor legitimately owns.""" locations = _normalize_locations(target_fields.get("locations") or []) extra = _extra_tags(target_tags, target_fields.get("language", "")) contributions: list[dict] = [] for sfields, stags in sources: before = {_location_str(loc) for loc in locations} added_locs = [] for loc in _normalize_locations(sfields.get("locations") or []): if _location_str(loc) not in before: before.add(_location_str(loc)) locations.append(loc) added_locs.append(loc) added_tags = [] for t in _extra_tags(stags, sfields.get("language", "")): if t not in extra: extra.append(t) added_tags.append(t) contributions.append({"locations": added_locs, "tags": added_tags}) return _normalize_locations(locations), extra, contributions async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]): """Unify source snippets INTO the target: union their locations + tags onto the canonical target, keep the target's scalar fields, trash the sources (recoverable), re-embed the survivor. Share-aware and write-gated, like update: an editor/admin grant is enough, a viewer grant is not. Raises PermissionError when the caller can read the target but not write it. Every source must share the TARGET'S OWNER — cross-owner merge stays out of scope (#231) — and must itself be writable; sources failing either test are skipped rather than silently half-merged. The survivor records what it absorbed as `merged_from` — in the `data` mirror and as a `**Merged from:** #ids` line in the body, written from one value like every other field. It accumulates across merges and ordinary edits carry it forward; only a merge adds to it. Returns (merged_target_note, merged_source_ids), or None if the target isn't a snippet the caller can see. Note the returned ids are the sources actually TRASHED, while `merged_from` is everything folded in — they differ only if a source vanished between the field union and the trash call, which leaves the source visible rather than losing anything.""" from scribe.services.access import can_write_note target = await get_snippet(user_id, target_id) if target is None: return None if not await can_write_note(user_id, target_id): raise PermissionError( f"snippet {target_id} is shared with you read-only — you can't merge " f"into a record you can't edit" ) owner_id = target.user_id sources = [] for sid in source_ids: if sid == target_id: continue s = await get_snippet(user_id, sid) # Same owner as the target, and writable by this caller. Merging trashes # the source, so read access is not enough. if s is None or s.user_id != owner_id: continue if not await can_write_note(user_id, sid): continue sources.append(s) tgt_fields = snippet_fields(target) parsed_sources = [(snippet_fields(s), s.tags) for s in sources] locations, extra_tags, contributions = merge_snippet_fields( tgt_fields, target.tags, parsed_sources ) # Provenance: what this record absorbed, and when it absorbed it, in order. # Merge keeps the target's scalar fields and trashes the sources, so without # this the fact that a variant ever existed survives only in the trash — and # only for someone who already knew to go looking. Accumulated, not replaced: # a target merged twice keeps both histories. # # Each entry also carries what THAT source contributed, which is what makes # un-merge exact (#2165) rather than a blind subtraction that would strip # call sites the survivor legitimately owns. merged_from = _normalize_merged_from( list(tgt_fields.get("merged_from") or []) + [ {"id": s.id, **contrib} for s, contrib in zip(sources, contributions) ] ) # Owner-scoped write, authorised above — same reason as update_snippet. updated = await notes_svc.update_note( owner_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, merged_from=merged_from, ), tags=compose_tags(tgt_fields["language"], extra_tags), # The survivor's location set grew, so its mirror has to grow with it — # otherwise a merged snippet would be unfindable at the very call sites # the merge just recorded. data=compose_data( name=tgt_fields["name"], when_to_use=tgt_fields["when_to_use"], signature=tgt_fields["signature"], language=tgt_fields["language"], code=tgt_fields["code"], locations=locations, merged_from=merged_from, # No verification carried: the survivor's code is a union of several # sources, so no prior verdict describes it. It reads as unverified, # which is the honest answer — nobody has checked THIS code. ), ) 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(owner_id, "note", s.id) if batch is not None: merged_ids.append(s.id) return updated, merged_ids class UnmergeError(Exception): """Un-merge refused — the reason is the message, meant for the operator.""" async def unmerge_snippet(user_id: int, survivor_id: int, source_id: int): """Reverse ONE source out of a merged survivor: restore it, and strip exactly what it contributed. WHY THIS OWNS THE RESTORE. The obvious alternative was to have trash-restore notice that the record it's reviving was merged into something and offer to reverse. That would make the generic trash path learn snippet semantics for one record type. Instead un-merge performs the restore itself, so the inverse is one operation with one authorization check and the trash path stays ignorant. Restoring from the trash directly is still allowed and still leaves both records claiming the same call sites — which is why this exists — but it is no longer the only way back. EXACTNESS. Subtraction uses the contribution recorded at merge time, not the source's current locations. A source that was itself edited after being merged would otherwise strip locations it never contributed, and a location the survivor independently owned would be lost. When an entry carries no attribution (provenance parsed back out of the body, which can only hold ids), this REFUSES rather than guessing. Returns (survivor_note, restored_source_note). Raises UnmergeError with a reason the operator can act on; returns None if the survivor isn't a snippet this caller can see. """ from scribe.services.access import can_write_note survivor = await get_snippet(user_id, survivor_id) if survivor is None: return None if not await can_write_note(user_id, survivor_id): raise PermissionError( f"snippet {survivor_id} is shared with you read-only — you can't " f"un-merge a record you can't edit" ) fields = snippet_fields(survivor) entries = _normalize_merged_from(fields.get("merged_from")) entry = next((e for e in entries if e["id"] == int(source_id)), None) if entry is None: raise UnmergeError( f"snippet {survivor_id} has no record of absorbing #{source_id}" ) if "locations" not in entry and "tags" not in entry: raise UnmergeError( f"#{source_id} was folded into {survivor_id} before per-source " f"provenance was recorded, so what it contributed isn't known. " f"Restore it from the trash and adjust both records by hand — " f"subtracting a guess could strip call sites {survivor_id} owns." ) # Bring the source back FIRST: if it can't be revived there is nothing to # un-merge into, and the survivor is better left whole than stripped of # locations whose other claimant never returned. # # An ALREADY-ALIVE source is the common case, not an error — the operator # restored it from the trash themselves, which is precisely the state that # motivated this feature: both records then claim the same call sites, and # nothing had ever stripped the survivor's copy. Skip the revive and go # straight to the subtraction that fixes it. from scribe.services.trash import restore_entity restored = await get_snippet(user_id, int(source_id)) if restored is None: if await restore_entity(survivor.user_id, "note", int(source_id)) is None: raise UnmergeError( f"#{source_id} could not be restored — it was most likely purged " f"from the trash, and a purged source cannot be brought back" ) restored = await get_snippet(user_id, int(source_id)) drop_locs = {_location_str(loc) for loc in (entry.get("locations") or [])} drop_tags = set(entry.get("tags") or []) kept_locations = [ loc for loc in (fields.get("locations") or []) if _location_str(loc) not in drop_locs ] kept_extra = [ t for t in _extra_tags(survivor.tags, fields.get("language", "")) if t not in drop_tags ] remaining = [e for e in entries if e["id"] != int(source_id)] updated = await notes_svc.update_note( survivor.user_id, survivor_id, body=compose_body( code=fields["code"], language=fields["language"], signature=fields["signature"], when_to_use=fields["when_to_use"], locations=kept_locations, merged_from=remaining, ), tags=compose_tags(fields["language"], kept_extra), data=compose_data( name=fields["name"], when_to_use=fields["when_to_use"], signature=fields["signature"], language=fields["language"], code=fields["code"], locations=kept_locations, merged_from=remaining, # Same reasoning as merge: the survivor's location set just changed, # so any prior drift verdict no longer describes it. Dropped rather # than carried. ), ) if updated is None: return None return updated, restored