"""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 asyncio
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 resolve_locations(
    repo: str = "", path: str = "", symbol: str = "",
    locations: list[dict] | None = None,
) -> list[dict]:
    """The location list a caller meant, from either calling convention.

    `locations` is the general form (one entry per call site); repo/path/symbol
    are the single-location shorthand and apply only when `locations` was not
    given — passing both is not a merge, it is the caller having decided.

    Extracted because compose_body, create_snippet and the dedup gate must all
    read the shorthand the SAME way. They each had their own copy of the
    `if locations is None` fallback, which is fine until one of them gains a
    rule the others don't — and the gate (#2518) is the one where a disagreement
    would mean comparing a location the record won't actually be stored with.
    """
    if locations is None:
        locations = [{"repo": repo, "path": path, "symbol": symbol}]
    return _normalize_locations(locations)


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.
    """
    locs = resolve_locations(repo, path, symbol, 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", "provenance",
)

# --- 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 _normalized_code(code: str) -> str:
    """Whitespace normalization shared by the verdict hash and the pull-time
    containment check, so 'unchanged' means the same thing in both places:
    trailing whitespace per line and leading/trailing blank lines dropped."""
    return "\n".join(line.rstrip() for line in (code or "").splitlines()).strip()


def code_sha(code: str) -> str:
    """Stable fingerprint of a snippet's code, for expiring stale verdicts.

    Normalized first (see _normalized_code): a reformat that changes nothing
    shouldn't expire a verdict over an editor's trailing-newline habit.
    """
    return hashlib.sha256(_normalized_code(code).encode("utf-8")).hexdigest()[:32]


def compose_verification(
    *,
    status: str,
    checked_code_sha: str,
    detail: str = "",
    path: str = "",
    checked_at: str = "",
    commit_sha: 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()
    # The repo commit the working tree was at when the check ran (#2688). The
    # code_sha above expires a verdict when the RECORD is edited; this makes
    # "the REPO moved on since the check" computable too, once the forge
    # integration can compare it against the current head.
    if (commit_sha or "").strip():
        out["commit_sha"] = commit_sha.strip()
    return out


def compose_provenance(*, commit_sha: str, fetched_at: str = "") -> dict | None:
    """Build the `data.provenance` record: which commit the cached body was
    read at, and when.

    This is the pointer-model half of decision #2686 — the recorded location
    is the source of truth for the code and the stored body is a CACHE of it.
    Provenance says what that cache is a cache OF, so a reader (and later the
    forge fetch, step 5 of milestone 288) can judge staleness instead of
    guessing. Absent provenance is valid and means exactly what every snippet
    meant before this existed: a body captured by hand at an unknown point.
    """
    sha = (commit_sha or "").strip()
    if not sha:
        return None
    return {
        "commit_sha": sha,
        "fetched_at": (fetched_at or "").strip()
        or datetime.now(timezone.utc).isoformat(),
    }


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"),
        "commit_sha": stored.get("commit_sha"),
        # A push touched the recorded location since this verdict (#2691) —
        # the repo moved under it. Cleared by the next verdict, which builds
        # a fresh dict.
        "invalidated_by": stored.get("invalidated_by"),
        # 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 — and so
        # does a push-invalidated one, for the same reason from the repo side.
        "needs_attention": (
            (not current)
            or stored["status"] in VERIFY_DRIFTED
            or bool(stored.get("invalidated_by"))
        ),
    }


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,
    provenance: 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
    # Also carried: what commit the cached body was read at (#2688). The caller
    # owns the live-or-die rule — update_snippet drops it when the code changes
    # without a fresh SHA, because keeping it would claim the new body came
    # from the old commit.
    if provenance:
        out["provenance"] = provenance
    # 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,
    commit_sha: str = "",
):
    """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.

    ``commit_sha`` stamps the body's provenance — the commit the recording
    session read the code at (#2688). Optional: absent means what it always
    meant, a body captured at an unknown point."""
    locations = resolve_locations(repo, path, symbol, locations)
    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,
            provenance=compose_provenance(commit_sha=commit_sha),
        ),
    )
    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,
    commit_sha: str | 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 the caller can see.

    ``commit_sha`` restamps the body's provenance (#2688). It lives or dies
    with the code: passed → restamped at that commit; code changed without it →
    dropped, because keeping it would claim the new body came from the old
    commit; code untouched → carried.

    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"]

    # Provenance follows the code (#2688): a fresh SHA restamps it; a code
    # change without one drops it; an edit that leaves the code alone carries
    # it. Order matters — the explicit SHA wins even when the code changed,
    # because that is precisely the caller saying where the new body came from.
    if commit_sha is not None and commit_sha.strip():
        provenance = compose_provenance(commit_sha=commit_sha)
    elif code is not None and code != (cur.get("code") or ""):
        provenance = None
    else:
        provenance = cur.get("provenance")

    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"),
            provenance=provenance,
        ),
    }
    # 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 = "",
    commit_sha: 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 "",
        commit_sha=commit_sha,
    )
    # 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,
        # An "ok" verdict at a known commit IS a provenance claim — the checker
        # just established that the cached body matches the source there — so
        # it restamps. Any other verdict carries what was known: a verdict is
        # about the code, not a change to it, and must not erase it (#2688).
        provenance=(
            compose_provenance(commit_sha=commit_sha)
            if status == VERIFY_OK and (commit_sha or "").strip()
            else fields.get("provenance")
        ),
    )
    return await notes_svc.update_note(note.user_id, snippet_id, data=data)


# --- pull-time freshness (#2690) ---------------------------------------------
# A pull is the moment freshness matters: the reader is about to trust the
# cached body. When the record owner's keyring serves a forge, the pull
# fetches the recorded file and answers the one mechanically-answerable
# question — does the cached code still appear in the source, verbatim after
# whitespace normalization? The body is a FRAGMENT of the file, so "serve the fetched
# file" would clobber the record; confirmation + provenance refresh is what
# fetching can honestly deliver, and divergence is reported, not overwritten.
#
# With no forge configured this function attaches NOTHING — the response is
# byte-identical to pre-forge behavior (rule #115's baseline).

# Total budget for the in-pull fetch. Tighter than the adapter's own timeout:
# the pull is the moment a session decides whether pulling is worth it
# (#2663's pull-through finding), so a slow forge must cost bounded time and
# then the cache serves.
PULL_FETCH_BUDGET_S = 2.5


async def _stamp_missing(note, host: str) -> None:
    """Record the mechanically-established 'missing' verdict from a pull-time
    404 — the recorded path is gone at the forge's head. Runs in the
    background; written as the owner, like every metadata write here."""
    await record_verification(
        note.user_id, note.id, status=VERIFY_MISSING,
        detail=f"pull-time forge fetch: recorded path not found on {host}",
    )


async def _refresh_provenance(note, commit_sha: str) -> None:
    """Restamp data.provenance after a pull confirmed the cache matches the
    source at ``commit_sha``. Background write, rebuilt like record_verification
    so nothing else about the record changes."""
    fields = snippet_fields(note)
    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=fields.get("verification"),
        provenance=compose_provenance(commit_sha=commit_sha),
    )
    await notes_svc.update_note(note.user_id, note.id, data=data)


async def attach_live_body(note, data: dict) -> None:
    """Decorate a PULL response with forge-checked freshness (#2690).

    Adds, when (and only when) the record owner's keyring serves a forge
    (#2778):
      - ``body_source``: "forge" (confirmed against the source just now) or
        "cache" (the stored body, for whatever reason follows)
      - ``body_freshness``: "current" | "diverged" | "missing" |
        "unreachable" | "no-recorded-location" | "repo-not-on-this-forge"

    Never raises, never blocks past PULL_FETCH_BUDGET_S, never rewrites the
    body: a freshness probe must not be able to break or slow the pull it
    decorates, and divergence is the READER's information, not license to
    clobber a record mid-read. A confirmed-current pull refreshes provenance
    in the background; a 404 stamps the 'missing' verdict into the same
    attention state verify_snippet uses.
    """
    from scribe.services.background import spawn
    from scribe.services.forge import ForgeError, ForgeNotFound, get_forges

    try:
        # The OWNER's keyring, honoring the project pin (#2778) — freshness
        # for a record is checked with its owner's credential, never the
        # reader's.
        selector = await get_forges(note.user_id, getattr(note, "project_id", None))
    except Exception:
        logger.warning("forge lookup failed during pull", exc_info=True)
        return
    if not selector.configured:
        return

    fields = data.get("snippet") if isinstance(data.get("snippet"), dict) else None
    if fields is None:
        fields = snippet_fields(note)
    loc = next(
        (
            entry
            for entry in (fields.get("locations") or [])
            if entry.get("repo") and entry.get("path")
        ),
        None,
    )
    if loc is None:
        data["body_source"] = "cache"
        data["body_freshness"] = "no-recorded-location"
        return
    # Recorded location repos are free-form names ("Scribe"), which can't
    # address a forge API — the project's repo BINDING is the identity that
    # can (#2691). Try the location string first (it may be a real remote),
    # then fall back to the bindings of the snippet's project.
    resolved = selector.resolve(loc["repo"])
    if resolved is None and getattr(note, "project_id", None):
        from scribe.services.repo_bindings import keys_for_project

        for key in await keys_for_project(note.user_id, note.project_id):
            resolved = selector.resolve(key)
            if resolved is not None:
                break
    if resolved is None:
        data["body_source"] = "cache"
        data["body_freshness"] = "repo-not-on-this-forge"
        return
    forge, repo = resolved

    stored_prov_sha = (fields.get("provenance") or {}).get("commit_sha") or ""

    async def _probe():
        # Cached-SHA short-circuit (#2693): provenance names the commit the
        # cached code was last confirmed at, so one cheap "newest commit
        # touching this path" call can prove the file hasn't moved since —
        # no content transfer. That economy is what fits pull-time freshness
        # inside GitHub's rate limits; it's merely nice on a self-hosted
        # Gitea. Any surprise (error, empty, mismatch) falls through to the
        # full fetch, which stays the authoritative path.
        if stored_prov_sha:
            try:
                head = await forge.latest_commit(repo, loc["path"])
            except ForgeError:
                head = ""
            if head and head == stored_prov_sha:
                return None
        return await forge.read_file(repo, loc["path"])

    try:
        fetched = await asyncio.wait_for(_probe(), timeout=PULL_FETCH_BUDGET_S)
    except ForgeNotFound:
        data["body_source"] = "cache"
        data["body_freshness"] = "missing"
        stored = fields.get("verification") or {}
        # Don't re-stamp what's already stamped — a popular-but-broken record
        # would otherwise be rewritten on every pull.
        if stored.get("status") != VERIFY_MISSING:
            spawn(_stamp_missing(note, forge.host), site="pull missing-verdict")
        return
    except (ForgeError, asyncio.TimeoutError):
        data["body_source"] = "cache"
        data["body_freshness"] = "unreachable"
        return

    if fetched is None:
        # Unchanged since the provenance commit — confirmed against the
        # source without moving the file. Same stamp, so nothing to persist
        # (the same-sha rule); the body already reflects that commit.
        data["body_source"] = "forge"
        data["body_freshness"] = "current"
        return

    cached = _normalized_code(fields.get("code") or "")
    if cached and cached in _normalized_code(fetched.content):
        data["body_source"] = "forge"
        data["body_freshness"] = "current"
        if fetched.commit_sha:
            # Read the stored stamp BEFORE writing the fresh one into the
            # response: `fields` aliases data["snippet"], so the other order
            # makes the staleness check compare the new stamp to itself and
            # the persist never fires (caught by the unit test, run 3811).
            stored_prov = fields.get("provenance") or {}
            stale = stored_prov.get("commit_sha") != fetched.commit_sha
            prov = compose_provenance(commit_sha=fetched.commit_sha)
            # Reflected in THIS response as well as persisted — the reader
            # shouldn't need a second pull to see the stamp they caused.
            if isinstance(data.get("snippet"), dict):
                data["snippet"]["provenance"] = prov
            if stale:
                spawn(
                    _refresh_provenance(note, fetched.commit_sha),
                    site="pull provenance-refresh",
                )
    else:
        data["body_source"] = "cache"
        data["body_freshness"] = "diverged"


# --- push-time drift flagging (#2691) ----------------------------------------

def _path_touches(recorded: str, changed: str) -> bool:
    """The location-path semantics, applied to a pushed file: the recorded path
    is the changed file itself, or a directory above it."""
    recorded = (recorded or "").strip("/")
    changed = (changed or "").strip("/")
    if not recorded or not changed:
        return False
    return changed == recorded or changed.startswith(recorded + "/")


async def invalidate_for_push(
    repo_key: str,
    changed: list[str],
    removed: list[str],
    commit_sha: str,
) -> int:
    """Flag snippets whose recorded location a push just touched (#2691).

    Writes ``verification.invalidated_by = {commit_sha, at, path, removed}``
    onto matched snippets that CARRY a verdict — the flag means "the repo
    moved under this verdict, recheck it", and it clears itself the moment a
    fresh verdict is recorded because compose_verification builds a new dict.
    Unverified snippets are skipped: they are already in the unverified
    bucket, and stacking a second unchecked-flavored flag on them adds noise,
    not information.

    Matching goes through repo BINDINGS (any user's — a webhook has no
    caller): each binding names a project, and that project's snippets are
    path-matched against the pushed files. O(bindings + snippets-in-project +
    changed files); nothing else is scanned. Returns how many records were
    newly flagged (an already-flagged record at the same commit is skipped,
    so replayed deliveries don't churn).
    """
    from scribe.services.repo_bindings import bindings_for_key

    bindings = await bindings_for_key(repo_key)
    if not bindings or not (changed or removed):
        return 0
    touched = [(p, False) for p in changed] + [(p, True) for p in removed]

    flagged = 0
    for binding in bindings:
        async with async_session() as session:
            rows = await session.execute(
                select(Note).where(
                    Note.user_id == binding.user_id,
                    Note.project_id == binding.project_id,
                    Note.note_type == SNIPPET_NOTE_TYPE,
                    Note.deleted_at.is_(None),
                    Note.data.path_exists("$.verification"),
                )
            )
            notes = list(rows.scalars().all())
        for note in notes:
            fields = snippet_fields(note)
            verdict = fields.get("verification") or {}
            if not verdict.get("status"):
                continue
            hit = next(
                (
                    (path, was_removed)
                    for loc in (fields.get("locations") or [])
                    for path, was_removed in touched
                    if _path_touches(loc.get("path") or "", path)
                ),
                None,
            )
            if hit is None:
                continue
            existing = verdict.get("invalidated_by") or {}
            if existing.get("commit_sha") == commit_sha:
                continue  # replayed delivery — already says exactly this
            verdict = dict(verdict)
            verdict["invalidated_by"] = {
                "commit_sha": commit_sha,
                "at": datetime.now(timezone.utc).isoformat(),
                "path": hit[0],
                # A removed file is the strongest signal — the recorded
                # location may simply be gone. Surfaced so the attention row
                # says which kind of look it needs.
                "removed": hit[1],
            }
            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=verdict,
                provenance=fields.get("provenance"),
            )
            await notes_svc.update_note(note.user_id, note.id, data=data)
            flagged += 1
    return flagged


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
