feat(snippets): near-duplicate finder — surface the sets worth merging

#231's premise was unifying reusable things already scattered as one-offs.
The create gate PREVENTS a new duplicate and merge_snippets CURES one you
point it at, but nothing FOUND the duplicates already in the record —
someone had to notice them by hand, which is the exact failure the Drafter
exists to remove.

One indexed self-join over note_embeddings, not an N² Python scan:
pgvector's cosine distance is the same operator semantic search uses, so a
similarity floor is a distance ceiling and the work stays in Postgres.
`left.note_id < right.note_id` yields each unordered pair once and drops
the self-pair that would otherwise dominate the ranking.

Pairs are collapsed into merge SETS by connected components. Transitive on
purpose: A~B plus B~C puts all three together even when A and C don't
directly clear the bar, which is what merge actually does (it folds every
source into one survivor). The cost is that a chain of mild resemblances
can rope in a member that isn't really alike — so the UI presents a set as
a proposal, shows the members, and never merges without a confirm.

Two scope decisions worth naming:

- OWN snippets only. merge_snippets requires one owner across the set, so
  surfacing someone else's would propose a merge that cannot be performed.
  The report is bounded by what the operator can act on, not what they can
  see.

- Threshold defaults to 0.82, LOOSER than the write gate's 0.90, and is a
  setting rather than a constant (rule #25). The gate blocks a create and
  has to be unforgiving of noise; this only suggests a merge under review,
  so it must reach further or it would never surface the pairs the gate
  already let through — which are precisely the ones that accumulated.

Fixes a real bug in the merge flow while wiring the UI: selectedList
filtered the selection against the CURRENT PAGE, and doMerge derives its
source ids from that list. A corpus-wide suggested group with off-page
members would have rendered incomplete and silently merged only the
visible subset. A group under review is now the authority for that list.

Refs #2088

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

src/scribe/services/dedup.py

@@ -26,11 +26,17 @@ import logging
 from dataclasses import dataclass
 
 from sqlalchemy import func, select
+from sqlalchemy.orm import aliased
 
 from scribe.models import async_session
+from scribe.models.embedding import NoteEmbedding
 from scribe.models.note import Note
 from scribe.models.rulebook import Rule
 from scribe.services import embeddings as embeddings_svc
+# Imported rather than redeclared: no service imports this module (the create
+# gate is called from the routes/tools layer), so there is no cycle to dodge,
+# and a second copy of the constant is a thing to drift.
+from scribe.services.snippets import SNIPPET_NOTE_TYPE
 
 logger = logging.getLogger(__name__)
 
@@ -139,6 +145,174 @@ async def find_duplicate_note(
     return None
 
 
+# --- corpus-wide near-duplicate report (#2088) -------------------------------
+# The gate above PREVENTS a new duplicate; merge_snippets CURES one you point it
+# at. Neither FINDS the duplicates already sitting in the record — someone had to
+# notice them by hand, which is the exact failure the Drafter exists to remove.
+#
+# WHY OWN SNIPPETS ONLY. merge_snippets requires every record to share the
+# target's owner (cross-owner merge is out of scope), so a report that surfaced
+# someone else's snippet would propose a merge that cannot be performed. The
+# scope here is set by what the operator can actually act on, not by what they
+# can see.
+#
+# WHY A LOWER THRESHOLD THAN THE GATE. The gate BLOCKS a write at 0.90 and has to
+# be unforgiving of noise. This report only makes a suggestion the operator
+# reviews, so it can afford to be looser and catch the pairs the gate lets
+# through — which are precisely the ones that accumulated. It is a setting rather
+# than a constant (rule #25) because the right value depends on how uniform a
+# corpus is, and nobody can guess that from here.
+
+DUPLICATE_THRESHOLD_KEY = "kb_duplicate_threshold"
+DUPLICATE_DEFAULT_THRESHOLD = 0.82
+# Hard cap on returned pairs. A pathologically uniform corpus is O(n²) pairs, and
+# a report nobody can read is not a report.
+_MAX_DUPLICATE_PAIRS = 200
+
+
+async def get_duplicate_threshold(user_id: int) -> float:
+    """The user's near-duplicate similarity floor, clamped to [0, 1]."""
+    from scribe.services.settings import get_setting
+
+    try:
+        value = float(await get_setting(
+            user_id, DUPLICATE_THRESHOLD_KEY, str(DUPLICATE_DEFAULT_THRESHOLD)
+        ))
+    except (TypeError, ValueError):
+        value = DUPLICATE_DEFAULT_THRESHOLD
+    return min(1.0, max(0.0, value))
+
+
+def group_pairs(pairs: list[tuple[int, int, float]]) -> list[list[int]]:
+    """Collapse similar-pairs into candidate merge SETS (connected components).
+
+    Pure and synchronous so the grouping rule is testable without a database.
+
+    Transitive on purpose: if A~B and B~C, all three land in one set even when
+    A and C fall below the threshold. That matches what merge does — it folds
+    every source into one survivor — and it avoids handing the operator three
+    overlapping pairs to reconcile by hand, which is the chore being removed.
+    The cost is that a chain of mild resemblances can rope in a pair that isn't
+    really alike; the operator sees the members and picks, so a set is a
+    proposal, never an action.
+    """
+    parent: dict[int, int] = {}
+
+    def find(x: int) -> int:
+        parent.setdefault(x, x)
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a: int, b: int) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[rb] = ra
+
+    for left, right, _score in pairs:
+        union(left, right)
+
+    groups: dict[int, list[int]] = {}
+    for node in parent:
+        groups.setdefault(find(node), []).append(node)
+    # Biggest clusters first — the most tangled thing is the most worth fixing.
+    # Ids ascending within a set so the output is stable across runs.
+    return sorted((sorted(g) for g in groups.values() if len(g) > 1),
+                  key=lambda g: (-len(g), g[0]))
+
+
+async def find_duplicate_snippets(
+    user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
+) -> dict:
+    """Near-duplicate snippets already in the record, grouped into merge sets.
+
+    One indexed self-join over `note_embeddings` rather than an N² Python scan:
+    pgvector's cosine distance is the same operator semantic search uses, so a
+    similarity floor is a distance ceiling and the work stays in Postgres.
+
+    Returns {"groups": [{"note_ids": [...], "snippets": [...], "top_score": f}],
+    "pairs": [...], "threshold": f}. Fail-open (an empty report) like the rest of
+    this module — a suggestion feature must not be able to break the page it
+    decorates.
+    """
+    floor = await get_duplicate_threshold(user_id) if threshold is None else threshold
+    floor = min(1.0, max(0.0, floor))
+    max_distance = min(2.0, max(0.0, 1.0 - floor))
+
+    left = aliased(NoteEmbedding, name="left_emb")
+    right = aliased(NoteEmbedding, name="right_emb")
+    left_note = aliased(Note, name="left_note")
+    right_note = aliased(Note, name="right_note")
+    distance = left.embedding.cosine_distance(right.embedding)
+
+    pairs: list[tuple[int, int, float]] = []
+    try:
+        async with async_session() as session:
+            stmt = (
+                select(left.note_id, right.note_id, distance.label("distance"))
+                .select_from(left)
+                # `<` not `!=`: each unordered pair exactly once, and it drops
+                # the self-pair (distance 0) that would otherwise dominate.
+                .join(right, left.note_id < right.note_id)
+                .join(left_note, left_note.id == left.note_id)
+                .join(right_note, right_note.id == right.note_id)
+                .where(
+                    left_note.note_type == SNIPPET_NOTE_TYPE,
+                    right_note.note_type == SNIPPET_NOTE_TYPE,
+                    left_note.deleted_at.is_(None),
+                    right_note.deleted_at.is_(None),
+                    # Owner-scoped on both sides — see the note above on why the
+                    # report is bounded by what merge can actually act on.
+                    left_note.user_id == user_id,
+                    right_note.user_id == user_id,
+                    distance <= max_distance,
+                )
+                .order_by(distance.asc())
+                .limit(max(1, limit))
+            )
+            rows = list((await session.execute(stmt)).all())
+        pairs = [(int(a), int(b), round(1.0 - float(d), 4)) for a, b, d in rows]
+    except Exception:
+        logger.warning("Near-duplicate snippet scan failed", exc_info=True)
+        return {"groups": [], "pairs": [], "threshold": floor}
+
+    if not pairs:
+        return {"groups": [], "pairs": [], "threshold": floor}
+
+    best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
+    grouped = group_pairs(pairs)
+
+    # Titles for presentation. One fetch for every id in the report.
+    ids = sorted({n for g in grouped for n in g})
+    titles: dict[int, str] = {}
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                select(Note.id, Note.title).where(Note.id.in_(ids))
+            )).all()
+            titles = {int(i): t for i, t in rows}
+    except Exception:
+        logger.debug("duplicate report titles unavailable", exc_info=True)
+
+    groups = []
+    for members in grouped:
+        scores = [
+            s for (a, b), s in best.items() if a in members and b in members
+        ]
+        groups.append({
+            "note_ids": members,
+            "snippets": [
+                {"id": nid, "title": titles.get(nid, "")} for nid in members
+            ],
+            # The strongest resemblance in the set — how confident the suggestion
+            # is, and what the list sorts on.
+            "top_score": max(scores) if scores else floor,
+        })
+    groups.sort(key=lambda g: (-g["top_score"], g["note_ids"][0]))
+    return {"groups": groups, "pairs": pairs, "threshold": floor}
+
+
 async def find_duplicate_rule(
     title: str,
     topic_id: int | None = None,