"""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 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",
)

# --- 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."""
    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,
        ),
    )
    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
