feat(snippets): un-merge — reverse one source out of a merged survivor

Closes the last of milestone #232. The task said to settle the design
before coding; here is what was settled and why.

THE HAZARD. Restoring a merged-in source from the trash brought the record
back but never stripped its locations off the survivor, so both claimed
the same call sites and the reverse lookup read the duplicate claims as
real. Subtracting blindly is not a fix: a location can arrive from a
source AND genuinely be the survivor's own, and _normalize_locations dedups
them into one, so blind subtraction would strip a call site the survivor
owns. Same problem defeated partial un-merge — `merged_from` recorded ids,
not which locations came from which source.

THE ANSWER. Record per-source attribution AT MERGE TIME, where it is known
exactly: each entry keeps only what that source ADDED, computed
incrementally as sources fold in. Anything the survivor already had, or an
earlier source already brought, is attributed to nobody. Both open
questions fall out of that one change — partial un-merge is exact, and a
survivor-owned location can never be stripped, because it was never
attributed in the first place.

The shape moved from [id] to [{id, locations, tags}]. Free to do: the
corpus holds one snippet and zero merges, so there is no legacy data (rule
#22). A bare int still normalizes to {"id": n} — not legacy tolerance, but
because snippet_fields falls back to PARSING THE BODY when a row has no
`data`, and the body's provenance line can only carry ids. Such an entry
shows history and refuses un-merge with a reason rather than guessing.

WHICH SURFACE. Neither option in the task, quite. Making trash-restore
notice the merge would teach the generic trash path snippet semantics for
one record type. Instead un-merge OWNS the restore: one operation, one
authorization check, trash stays ignorant. Restoring by hand is still
allowed and still leaves both records claiming the same places — so
un-merge treats an already-alive source as the normal case and goes
straight to the subtraction that repairs it. That is the state that
motivated the feature, not an error.

Adds trash.restore_entity(user_id, type, id) — the missing inverse of
delete(), which returns a batch id callers don't keep. Restores the whole
batch, since the batch is the entity plus its cascade.

Refs #2165

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

frontend/src/api/snippets.ts

@@ -20,9 +20,13 @@ export interface SnippetFields {
   path: string;
   symbol: string;
   locations: SnippetLocation[];
-  /** Ids of the snippets folded into this one by merge, oldest first. Read-only:
-   *  merge is the only thing that adds to it, and an edit carries it forward. */
-  merged_from: number[];
+  /** The snippets folded into this one by merge, oldest first. Read-only: merge
+   *  is the only thing that adds to it, and an edit carries it forward.
+   *
+   *  Each entry records what THAT source contributed — never what the survivor
+   *  already had — which is what lets un-merge subtract exactly. An entry with
+   *  no `locations`/`tags` predates that attribution and cannot be un-merged. */
+  merged_from: { id: number; locations?: SnippetLocation[]; tags?: string[] }[];
   code: string;
 }
 
@@ -183,6 +187,16 @@ export async function verifySnippet(
   return apiPost(`/api/snippets/${id}/verify`, verdict);
 }
 
+/** Pull one source back out of a merged survivor: restores it and strips exactly
+ *  what it contributed. Also repairs a half-undone merge — a source restored
+ *  from the trash by hand leaves the survivor still claiming its call sites. */
+export async function unmergeSnippet(
+  survivorId: number,
+  sourceId: number,
+): Promise<{ survivor: Snippet; restored: Snippet | null }> {
+  return apiPost(`/api/snippets/${survivorId}/unmerge`, { source_id: sourceId });
+}
+
 /** Unify `sourceIds` into the canonical snippet `targetId`. Returns the merged
  *  survivor plus `merged_ids` — the sources actually folded in and trashed. */
 export async function mergeSnippets(

frontend/src/views/SnippetDetailView.vue

@@ -1,7 +1,12 @@
 <script setup lang="ts">
 import { ref, computed, onMounted } from "vue";
 import { useRoute, useRouter } from "vue-router";
-import { getSnippet, deleteSnippet, type Snippet } from "@/api/snippets";
+import {
+  getSnippet,
+  deleteSnippet,
+  unmergeSnippet,
+  type Snippet,
+} from "@/api/snippets";
 import { useToastStore } from "@/stores/toast";
 import ConfirmDialog from "@/components/ConfirmDialog.vue";
 
@@ -31,6 +36,32 @@ function locParts(loc: { repo: string; path: string; symbol: string }): string[]
   return [loc.repo, loc.path, loc.symbol].filter((p) => p && p.trim());
 }
 
+// Un-merge (#2165)
+const unmerging = ref<number | null>(null);
+
+/** Only entries carrying what the source contributed can be reversed exactly.
+ *  Without that, subtraction would be a guess that could strip call sites the
+ *  survivor owns in its own right — so the control isn't offered. */
+function canUnmerge(entry: { locations?: unknown[]; tags?: unknown[] }): boolean {
+  return canWrite.value && (!!entry.locations?.length || !!entry.tags?.length);
+}
+
+async function doUnmerge(sourceId: number) {
+  unmerging.value = sourceId;
+  try {
+    await unmergeSnippet(id.value, sourceId);
+    toast.show(`#${sourceId} pulled back out and restored`);
+    await load();
+  } catch (e: unknown) {
+    // 409 carries the reason the record's state makes it impossible — show it
+    // rather than a generic failure, since it's the actionable part.
+    const detail = (e as { body?: { error?: string } }).body?.error;
+    toast.show(detail || `Couldn't un-merge #${sourceId}`, "error");
+  } finally {
+    unmerging.value = null;
+  }
+}
+
 async function load() {
   loading.value = true;
   error.value = null;
@@ -117,7 +148,27 @@ async function confirmDelete() {
         <template v-if="mergedFrom.length">
           <dt>Merged from</dt>
           <dd class="merged-from">
-            <span v-for="mid in mergedFrom" :key="mid">#{{ mid }}</span>
+            <span v-for="m in mergedFrom" :key="m.id" class="merged-entry">
+              <span>#{{ m.id }}</span>
+              <button
+                v-if="canUnmerge(m)"
+                class="unmerge-btn"
+                :disabled="unmerging === m.id"
+                :title="`Pull #${m.id} back out: restore it and remove the ${(m.locations ?? []).length} location(s) it contributed`"
+                @click="doUnmerge(m.id)"
+              >
+                {{ unmerging === m.id ? "…" : "un-merge" }}
+              </button>
+              <!-- Says why rather than hiding the control: a disabled thing with
+                   no explanation reads as a bug. -->
+              <span
+                v-else-if="canWrite"
+                class="unmerge-na"
+                :title="`This merge predates per-source provenance, so what #${m.id} contributed isn't recorded. Restore it from the trash and adjust both records by hand — subtracting a guess could strip call sites this record genuinely owns.`"
+              >
+                (not reversible)
+              </span>
+            </span>
             <span class="merged-hint">
               folded in here — the originals are in the trash
             </span>
@@ -297,6 +348,34 @@ async function confirmDelete() {
   color: var(--color-text-muted);
   font-size: 0.8rem;
 }
+.merged-entry {
+  display: inline-flex;
+  align-items: baseline;
+  gap: 0.3rem;
+}
+.unmerge-btn {
+  font-size: 0.72rem;
+  padding: 0.05rem 0.35rem;
+  border: 1px solid var(--color-border);
+  border-radius: 4px;
+  background: transparent;
+  color: var(--color-text-muted);
+  cursor: pointer;
+}
+.unmerge-btn:hover:not(:disabled) {
+  color: var(--color-text);
+  border-color: var(--color-text-muted);
+}
+.unmerge-btn:disabled {
+  opacity: 0.6;
+  cursor: default;
+}
+.unmerge-na {
+  font-size: 0.72rem;
+  color: var(--color-text-muted);
+  /* Cursor cues that the explanation is in the tooltip. */
+  cursor: help;
+}
 
 .code-block {
   border: 1px solid var(--color-border);

src/scribe/mcp/tools/snippets.py

@@ -190,6 +190,46 @@ async def get_snippet(snippet_id: int) -> dict:
     return data
 
 
+async def unmerge_snippet(survivor_id: int, source_id: int) -> dict:
+    """Reverse ONE source out of a merged snippet — the inverse of merge_snippets.
+
+    Restores the source record and strips exactly what it contributed from the
+    survivor: the locations and tags it ADDED at merge time, never the ones the
+    survivor already had. Reach for it when a merge turns out to have unified two
+    things that only looked alike.
+
+    Also the fix for a half-undone merge. Restoring a merged-in source from the
+    trash by hand brings the record back but leaves the survivor still claiming
+    its call sites, so both records claim the same places and the reverse lookup
+    reads the duplicate claims as real. Running this on an already-restored
+    source repairs that: it skips the restore and does the subtraction.
+
+    Args:
+        survivor_id: The snippet that absorbed the other.
+        source_id: The snippet to pull back out of it.
+
+    Returns {"survivor": {...}, "restored": {...}}.
+
+    Refuses, with the reason, when: the survivor has no record of absorbing that
+    id; the source was purged from the trash; or the merge predates per-source
+    provenance, in which case what it contributed isn't known and subtracting a
+    guess could strip call sites the survivor genuinely owns — restore it from
+    the trash and adjust both records by hand instead.
+    """
+    uid = current_user_id()
+    try:
+        result = await snippets_svc.unmerge_snippet(uid, survivor_id, source_id)
+    except snippets_svc.UnmergeError as exc:
+        raise ValueError(str(exc)) from exc
+    if result is None:
+        raise ValueError(f"snippet {survivor_id} not found")
+    survivor, restored = result
+    return {
+        "survivor": snippets_svc.snippet_to_dict(survivor),
+        "restored": snippets_svc.snippet_to_dict(restored) if restored else None,
+    }
+
+
 async def find_duplicate_snippets(threshold: float = 0.0) -> dict:
     """Find snippets already recorded that look like duplicates of each other.
 
@@ -376,6 +416,10 @@ async def merge_snippets(target_id: int, source_ids: list[int]) -> dict:
     fields and as a "Merged from: #ids" line in the body), so a variant that got
     folded in leaves a trace outside the trash. It accumulates across merges.
 
+    Reversible: each entry also records what that source contributed, so
+    `unmerge_snippet(target_id, source_id)` can restore it and strip exactly
+    those locations back off — never the ones the target already had.
+
     Args:
         target_id: The snippet to keep (the canonical record).
         source_ids: Snippet ids to fold into the target and retire. Ids that
@@ -404,5 +448,6 @@ def register(mcp) -> None:
     for fn in (
         list_snippets, create_snippet, get_snippet, update_snippet,
         delete_snippet, merge_snippets, verify_snippet, find_duplicate_snippets,
+        unmerge_snippet,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/routes/snippets.py

@@ -205,6 +205,36 @@ async def update_snippet_route(snippet_id: int):
     return jsonify(out)
 
 
+@snippets_bp.route("/<int:snippet_id>/unmerge", methods=["POST"])
+@login_required
+async def unmerge_snippet_route(snippet_id: int):
+    """Pull one source back out of a merged survivor. Body: {"source_id": int}.
+
+    Also the repair for a half-undone merge: restoring a source from the trash by
+    hand leaves the survivor still claiming its call sites, and running this on an
+    already-restored source strips them."""
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    source_id = data.get("source_id")
+    if not isinstance(source_id, int):
+        return jsonify({"error": "source_id (int) is required"}), 400
+    try:
+        result = await snippets_svc.unmerge_snippet(uid, snippet_id, source_id)
+    except PermissionError as exc:
+        return jsonify({"error": str(exc)}), 403
+    except snippets_svc.UnmergeError as exc:
+        # 409, not 400: the request is well-formed, the record's state is what
+        # makes it impossible — and the message says which.
+        return jsonify({"error": str(exc)}), 409
+    if result is None:
+        return not_found("Snippet")
+    survivor, restored = result
+    return jsonify({
+        "survivor": snippets_svc.snippet_to_dict(survivor),
+        "restored": snippets_svc.snippet_to_dict(restored) if restored else None,
+    })
+
+
 @snippets_bp.route("/duplicates", methods=["GET"])
 @login_required
 async def duplicate_snippets_route():

src/scribe/services/snippets.py

@@ -134,23 +134,58 @@ def _render_location_block(locations: list[dict]) -> str | None:
     return f"**Locations:**\n{lines}"
 
 
-def _normalize_merged_from(ids: list[int] | None) -> list[int]:
-    """Merge provenance as a clean id list: ints only, no dups, order kept.
+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[int] = []
-    for raw in ids or []:
+    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(raw)
+            i = int(ident)
         except (TypeError, ValueError):
             continue
-        if i > 0 and i not in out:
-            out.append(i)
+        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,
@@ -188,8 +223,11 @@ def compose_body(
         # 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"#{i}" for i in merged)
+            "**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```"
@@ -859,15 +897,32 @@ def merge_snippet_fields(
     """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)."""
-    locations = list(target_fields.get("locations") or [])
+    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:
-        locations.extend(sfields.get("locations") or [])
+        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)
-    return _normalize_locations(locations), extra
+                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]):
@@ -917,15 +972,25 @@ async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
 
     tgt_fields = snippet_fields(target)
     parsed_sources = [(snippet_fields(s), s.tags) for s in sources]
-    locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_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 []) + [s.id for s in sources]
+        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.
@@ -962,3 +1027,112 @@ async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
 
     _embed_snippet(updated)
     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
+    _embed_snippet(updated)
+    return updated, restored

src/scribe/services/trash.py

@@ -181,6 +181,36 @@ async def restore(user_id: int, batch_id: str) -> int:
     return n
 
 
+async def restore_entity(user_id: int, entity_type: str, entity_id: int) -> int | None:
+    """Restore ONE trashed entity by id, by reviving the batch that took it.
+
+    The inverse of `delete(user_id, entity_type, entity_id)`, which returns a
+    batch id the caller usually doesn't keep. Callers that need to undo their own
+    soft-delete later — snippet un-merge (#2165) — know the entity id, not the
+    batch, and looking it up here keeps them from reaching into the column set.
+
+    Restores the whole batch on purpose, not just the row: the batch is the
+    entity plus its cascaded descendants, so reviving the parent alone would
+    leave them orphaned in the trash. That is the same thing the trash UI does.
+
+    Returns rows restored, or None if the entity isn't trashed / not owned.
+    """
+    model = next((m for m, label in _TYPE.items() if label == entity_type), None)
+    if model is None:
+        return None
+    async with async_session() as session:
+        batch = (await session.execute(
+            select(model.deleted_batch_id).where(
+                model.id == entity_id,
+                _owner_clause(model, user_id),
+                model.deleted_at.isnot(None),
+            )
+        )).scalars().first()
+    if not batch:
+        return None
+    return await restore(user_id, batch)
+
+
 async def purge(user_id: int, batch_id: str) -> int:
     """Hard-delete every row in the batch. Irreversible."""
     from sqlalchemy import delete as sql_delete

tests/test_mcp_tool_snippets.py

@@ -229,5 +229,5 @@ def test_register_attaches_all_tools():
     assert set(names) == {
         "list_snippets", "create_snippet", "get_snippet", "update_snippet",
         "delete_snippet", "merge_snippets", "verify_snippet",
-        "find_duplicate_snippets",
+        "find_duplicate_snippets", "unmerge_snippet",
     }

tests/test_routes_snippets.py

@@ -21,6 +21,7 @@ def test_snippet_handlers_callable():
         "list_snippets_route", "create_snippet_route", "get_snippet_route",
         "update_snippet_route", "delete_snippet_route", "merge_snippet_route",
         "verify_snippet_route", "duplicate_snippets_route",
+        "unmerge_snippet_route",
     ):
         assert callable(getattr(routes, name))
 
@@ -31,6 +32,7 @@ def test_service_functions_take_user_id():
     for fn_name in (
         "create_snippet", "list_snippets", "get_snippet", "update_snippet",
         "delete_snippet", "merge_snippets", "record_verification",
+        "unmerge_snippet",
     ):
         fn = getattr(svc, fn_name)
         assert callable(fn)
@@ -46,7 +48,8 @@ def test_agent_and_web_surfaces_stay_at_parity():
     from scribe.routes import snippets as routes
 
     # Every write verb the web surface offers, the agent surface offers too.
-    for verb in ("create", "get", "list", "update", "delete", "merge", "verify"):
+    for verb in ("create", "get", "list", "update", "delete", "merge", "verify",
+                 "unmerge"):
         assert callable(getattr(tools, f"{verb}_snippets", None) or
                         getattr(tools, f"{verb}_snippet", None)), f"MCP lacks {verb}"
 

tests/test_services_snippets.py

@@ -148,7 +148,9 @@ def test_merge_snippet_fields_unions_locations_and_tags():
             ["py", "snippet", "helper"])
     src2 = ({"language": "py", "locations": [{"repo": "a", "path": "a.py", "symbol": "f"}]},  # dup loc
             ["snippet"])
-    locs, extra = s.merge_snippet_fields(target_fields, ["py", "snippet", "core"], [src1, src2])
+    locs, extra, contributions = s.merge_snippet_fields(
+        target_fields, ["py", "snippet", "core"], [src1, src2]
+    )
     # target location first, then unique source locations; dup dropped.
     assert locs == [
         {"repo": "a", "path": "a.py", "symbol": "f"},
@@ -157,6 +159,14 @@ def test_merge_snippet_fields_unions_locations_and_tags():
     # extra tags unioned, language + "snippet" markers excluded.
     assert extra == ["core", "helper"]
 
+    # Attribution (#2165): only what each source ACTUALLY added. src2's location
+    # is one the target already had, so it contributed nothing — un-merging it
+    # must not strip a call site the survivor owns in its own right.
+    assert contributions[0]["locations"] == [{"repo": "b", "path": "b.py", "symbol": "g"}]
+    assert contributions[0]["tags"] == ["helper"]
+    assert contributions[1]["locations"] == []
+    assert contributions[1]["tags"] == []
+
 
 # --- the queryable mirror (notes.data, migration 0070) -----------------------
 
@@ -253,19 +263,44 @@ def test_data_and_body_round_trip_to_the_same_fields():
 
 def test_normalize_merged_from_keeps_history_order_and_drops_junk():
     """Order is history, not sorting — earlier merges stay first."""
-    assert s._normalize_merged_from([9, 3, 9, "4", None, 0, -2, "x"]) == [9, 3, 4]
+    assert s.merged_from_ids([9, 3, 9, "4", None, 0, -2, "x"]) == [9, 3, 4]
     assert s._normalize_merged_from(None) == []
 
 
+def test_normalize_merged_from_carries_per_source_attribution():
+    """The shape that makes un-merge exact (#2165): each entry holds what THAT
+    source contributed, so reversing one can't strip what the survivor owns."""
+    loc = {"repo": "a", "path": "a.py", "symbol": "f"}
+    out = s._normalize_merged_from([{"id": 7, "locations": [loc], "tags": ["x"]}])
+    assert out == [{"id": 7, "locations": [loc], "tags": ["x"]}]
+
+
+def test_a_bare_id_normalizes_to_an_entry_with_no_attribution():
+    """Not legacy tolerance: snippet_fields falls back to PARSING THE BODY when a
+    row has no `data`, and the body's provenance line can only carry ids. Such an
+    entry still shows history; un-merge refuses it rather than guessing."""
+    assert s._normalize_merged_from([5]) == [{"id": 5}]
+
+
 def test_compose_body_renders_merged_from_and_parse_reads_it_back():
     body = s.compose_body(code="x = 1", merged_from=[12, 13])
     assert "**Merged from:** #12, #13" in body
     got = s.parse_snippet_fields("n — u", body, None)
-    assert got["merged_from"] == [12, 13]
+    # Ids survive the body round-trip; attribution cannot, since the body only
+    # ever renders ids — which is exactly why `data` is the authority for it.
+    assert s.merged_from_ids(got["merged_from"]) == [12, 13]
     # The code fence still comes last — provenance is header metadata.
     assert body.rstrip().endswith("```")
 
 
+def test_compose_body_renders_ids_from_rich_entries():
+    body = s.compose_body(
+        code="x = 1",
+        merged_from=[{"id": 12, "locations": [{"repo": "", "path": "a.py", "symbol": ""}]}],
+    )
+    assert "**Merged from:** #12" in body
+
+
 def test_compose_body_omits_merged_from_when_there_is_none():
     """A snippet that was never merged says nothing about merging."""
     assert "Merged from" not in s.compose_body(code="x = 1")
@@ -273,7 +308,8 @@ def test_compose_body_omits_merged_from_when_there_is_none():
 
 
 def test_compose_data_mirrors_merged_from():
-    assert s.compose_data(name="f", merged_from=[3, 3, 2])["merged_from"] == [3, 2]
+    got = s.compose_data(name="f", merged_from=[3, 3, 2])["merged_from"]
+    assert s.merged_from_ids(got) == [3, 2]
     assert "merged_from" not in s.compose_data(name="f")
 
 
@@ -283,7 +319,7 @@ def test_snippet_fields_prefers_the_data_columns_merged_from():
     body = s.compose_body(code="x = 1", merged_from=[12])
     note = _Note(title="n — u", body=body, tags=["snippet"],
                  data=s.compose_data(name="n", merged_from=[12, 13]))
-    assert s.snippet_fields(note)["merged_from"] == [12, 13]
+    assert s.merged_from_ids(s.snippet_fields(note)["merged_from"]) == [12, 13]
 
 
 def test_snippet_to_dict_includes_parsed_fields():

tests/test_snippet_merge_provenance.py

@@ -51,7 +51,7 @@ async def test_merge_records_what_it_absorbed_in_body_and_mirror():
         _snippet(id=1), [_snippet(id=2), _snippet(id=3)], [2, 3],
     )
     assert "**Merged from:** #2, #3" in kwargs["body"]
-    assert kwargs["data"]["merged_from"] == [2, 3]
+    assert s.merged_from_ids(kwargs["data"]["merged_from"]) == [2, 3]
 
 
 @pytest.mark.asyncio
@@ -60,7 +60,7 @@ async def test_merge_accumulates_across_merges():
     replace the record of the first."""
     target = _snippet(id=1, data=s.compose_data(name="f", merged_from=[2]))
     kwargs = await _run_merge(target, [_snippet(id=3)], [3])
-    assert kwargs["data"]["merged_from"] == [2, 3]
+    assert s.merged_from_ids(kwargs["data"]["merged_from"]) == [2, 3]
     assert "**Merged from:** #2, #3" in kwargs["body"]
 
 
@@ -74,7 +74,7 @@ async def test_merge_does_not_record_a_skipped_cross_owner_source():
         [_snippet(id=2, owner=9), _snippet(id=3, owner=42)],
         [2, 3],
     )
-    assert kwargs["data"]["merged_from"] == [2]
+    assert s.merged_from_ids(kwargs["data"]["merged_from"]) == [2]
     assert "#3" not in kwargs["body"]
 
 
@@ -93,7 +93,7 @@ async def test_an_ordinary_edit_carries_provenance_forward():
         await s.update_snippet(7, 1, signature="f(ms) -> string")
 
     kwargs = mock_update.await_args.kwargs
-    assert kwargs["data"]["merged_from"] == [2, 3]
+    assert s.merged_from_ids(kwargs["data"]["merged_from"]) == [2, 3]
     assert "**Merged from:** #2, #3" in kwargs["body"]
 
 

tests/test_snippet_unmerge.py

@@ -0,0 +1,183 @@
+"""Tests for un-merge (#2165) — reversing one source out of a merged survivor.
+
+The design question this task said to settle first was how to subtract without
+stripping call sites the survivor legitimately owns. The answer is per-source
+attribution recorded AT MERGE TIME: each `merged_from` entry holds only what
+that source actually added. These tests pin that rule, and the refusal that
+guards the case where the attribution isn't there.
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import snippets as s
+
+
+def _loc(path, repo="r", symbol=""):
+    return {"repo": repo, "path": path, "symbol": symbol}
+
+
+def _survivor(locations, merged_from, tags=None, owner=7):
+    """A survivor note whose `data` carries locations + merge provenance."""
+    return SimpleNamespace(
+        id=1, user_id=owner, note_type="snippet", deleted_at=None,
+        title="f — does a thing", tags=tags or ["python", "snippet"],
+        body=s.compose_body(code="x = 1", language="python", locations=locations,
+                            merged_from=merged_from),
+        data=s.compose_data(name="f", when_to_use="does a thing", language="python",
+                            code="x = 1", locations=locations,
+                            merged_from=merged_from),
+    )
+
+
+async def _run_unmerge(survivor, *, source_alive=None, restore=1):
+    """Drive unmerge_snippet with the DB stubbed; return update_note's kwargs."""
+    async def fake_get(_uid, sid):
+        if sid == survivor.id:
+            return survivor
+        return source_alive
+
+    with (
+        patch.object(s, "get_snippet", AsyncMock(side_effect=fake_get)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)),
+        patch("scribe.services.trash.restore_entity", AsyncMock(return_value=restore)),
+        patch.object(s.notes_svc, "update_note",
+                     AsyncMock(return_value=survivor)) as upd,
+        patch.object(s, "_embed_snippet", MagicMock()),
+    ):
+        await s.unmerge_snippet(7, 1, 2)
+    return upd.await_args.kwargs
+
+
+# --- the subtraction ------------------------------------------------------
+
+
+async def test_unmerge_strips_only_what_the_source_contributed():
+    """The survivor's own location survives; the source's is removed."""
+    mine, theirs = _loc("mine.py"), _loc("theirs.py")
+    survivor = _survivor(
+        [mine, theirs],
+        [{"id": 2, "locations": [theirs], "tags": ["helper"]}],
+        tags=["python", "snippet", "helper", "core"],
+    )
+    kwargs = await _run_unmerge(survivor)
+    assert kwargs["data"]["locations"] == [mine]
+    assert "helper" not in kwargs["tags"]
+    assert "core" in kwargs["tags"]
+
+
+async def test_a_location_the_survivor_also_owned_is_never_stripped():
+    """The central hazard the task named. If a source brought a location the
+    survivor ALREADY had, merge attributes nothing to it — so reversing must
+    leave that call site in place."""
+    shared = _loc("shared.py")
+    survivor = _survivor([shared], [{"id": 2, "locations": [], "tags": ["t"]}])
+    kwargs = await _run_unmerge(survivor)
+    assert kwargs["data"]["locations"] == [shared]
+
+
+async def test_the_reversed_entry_leaves_the_provenance_list():
+    survivor = _survivor(
+        [_loc("a.py"), _loc("b.py")],
+        [{"id": 2, "locations": [_loc("b.py")]}, {"id": 3, "locations": []}],
+    )
+    kwargs = await _run_unmerge(survivor)
+    assert s.merged_from_ids(kwargs["data"]["merged_from"]) == [3]
+    assert "#2" not in kwargs["body"]
+    # The other source's history is untouched — un-merge reverses one thing.
+    assert "**Merged from:** #3" in kwargs["body"]
+
+
+async def test_unmerging_the_last_source_clears_the_provenance_line():
+    survivor = _survivor([_loc("a.py")], [{"id": 2, "locations": [], "tags": ["t"]}])
+    kwargs = await _run_unmerge(survivor)
+    assert "Merged from" not in kwargs["body"]
+
+
+# --- restoring the source -------------------------------------------------
+
+
+async def test_an_already_restored_source_is_repaired_not_refused():
+    """The scenario that motivated the feature: the operator restored the source
+    from the trash by hand, so both records claim its call sites and nothing ever
+    stripped the survivor's copy. Un-merge must fix that, not reject it."""
+    theirs = _loc("theirs.py")
+    survivor = _survivor([_loc("mine.py"), theirs],
+                         [{"id": 2, "locations": [theirs]}])
+    alive = SimpleNamespace(id=2, user_id=7, note_type="snippet", deleted_at=None,
+                            title="g — x", tags=["snippet"], body="", data=None)
+    with patch("scribe.services.trash.restore_entity", AsyncMock()) as revive:
+        kwargs = await _run_unmerge(survivor, source_alive=alive)
+    # Nothing to revive — it's already alive — but the subtraction still happens.
+    revive.assert_not_called()
+    assert kwargs["data"]["locations"] == [_loc("mine.py")]
+
+
+async def test_a_purged_source_is_refused_and_the_survivor_is_untouched():
+    """If the source can't come back, stripping the survivor would lose the
+    locations entirely — no record would claim them."""
+    survivor = _survivor([_loc("a.py")], [{"id": 2, "locations": [_loc("a.py")]}])
+
+    async def fake_get(_uid, sid):
+        return survivor if sid == 1 else None
+
+    with (
+        patch.object(s, "get_snippet", AsyncMock(side_effect=fake_get)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)),
+        patch("scribe.services.trash.restore_entity", AsyncMock(return_value=None)),
+        patch.object(s.notes_svc, "update_note", AsyncMock()) as upd,
+        patch.object(s, "_embed_snippet", MagicMock()),
+    ):
+        with pytest.raises(s.UnmergeError, match="purged"):
+            await s.unmerge_snippet(7, 1, 2)
+    upd.assert_not_called()
+
+
+# --- refusals -------------------------------------------------------------
+
+
+async def test_an_entry_without_attribution_is_refused_not_guessed():
+    """Bare-id provenance comes from parsing the body, which can only hold ids.
+    Subtracting a guess could strip call sites the survivor owns — so refuse and
+    say what to do instead."""
+    survivor = _survivor([_loc("a.py")], [2])
+
+    async def fake_get(_uid, sid):
+        return survivor if sid == 1 else None
+
+    with (
+        patch.object(s, "get_snippet", AsyncMock(side_effect=fake_get)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)),
+        patch.object(s.notes_svc, "update_note", AsyncMock()) as upd,
+        patch.object(s, "_embed_snippet", MagicMock()),
+    ):
+        with pytest.raises(s.UnmergeError, match="provenance"):
+            await s.unmerge_snippet(7, 1, 2)
+    upd.assert_not_called()
+
+
+async def test_unmerging_something_never_absorbed_is_refused():
+    survivor = _survivor([_loc("a.py")], [{"id": 99, "locations": []}])
+    with (
+        patch.object(s, "get_snippet", AsyncMock(return_value=survivor)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)),
+    ):
+        with pytest.raises(s.UnmergeError, match="no record of absorbing"):
+            await s.unmerge_snippet(7, 1, 2)
+
+
+async def test_unmerge_requires_write_access():
+    """Same rule as merge: a read-only share can see the record, not rearrange it."""
+    survivor = _survivor([_loc("a.py")], [{"id": 2, "locations": []}])
+    with (
+        patch.object(s, "get_snippet", AsyncMock(return_value=survivor)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)),
+    ):
+        with pytest.raises(PermissionError):
+            await s.unmerge_snippet(7, 1, 2)
+
+
+async def test_unmerge_on_a_missing_survivor_returns_none():
+    with patch.object(s, "get_snippet", AsyncMock(return_value=None)):
+        assert await s.unmerge_snippet(7, 1, 2) is None