FabledScribe/src/scribe/services/dedup.py

"""Write-time near-duplicate detection — the update-over-create gate.

Goal: stop a second near-identical row from being created when an existing one
should be UPDATED instead. Duplicates bloat the store and, worse, get surfaced
by semantic search (RAG) later as competing/stale copies that then have to be
reconciled. This is the enforcement half of the instruction-level "prefer
updating over creating" reflex.

OPT-IN by design: the interactive create paths (MCP create tools + REST create
routes) run this gate; internal/programmatic creates do NOT (e.g. a recurring
task spawning its next instance, or a bulk import — those legitimately repeat a
title and must not be blocked). Callers that want the gate call find_duplicate_*
themselves and act on a hit; nothing here mutates.

Two signals, both scoped to the same owner + project + kind:
  1. Normalized-title exact match — cheap, always checked.
  2. Semantic similarity (cosine ≥ _SEMANTIC_THRESHOLD) — only when the incoming
     body is substantial. Short/title-only embeddings sit in a tight neighborhood
     and false-positive (the pre-pivot lesson: "Lore: Shell 0" vs
     "Lore: Reinitialization 0" matched at 0.91 with no body), so we gate it on
     a minimum body length.
"""
from __future__ import annotations

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.models.base import iso
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__)

# Run the semantic check only when the incoming body has at least this many
# characters — below it, embeddings are dominated by the title and false-positive.
_MIN_BODY_FOR_SEMANTIC = 200
# Cosine threshold for "this is the same thing, reworded." Deliberately high to
# keep false positives rare (a hard block with a force-override is unforgiving of
# noise). Matches the 0.90 the pre-pivot dedup settled on.
_SEMANTIC_THRESHOLD = 0.90

# SNIPPETS ARE MEASURED DIFFERENTLY, and #2518 is why. A snippet's embedded
# document is mostly PROSE ABOUT the code — name, when-to-reach-for-it,
# signature, the comments explaining the choice — with the artefact itself a
# minority of the text. Two measurements on the same corpus:
#
#   .btn-danger vs .btn-danger-outline   0.92   siblings, blocked (false positive)
#   .btn-primary re-recorded verbatim
#   under a different name              <0.90   a literal copy, ALLOWED THROUGH
#
# The second is the one that settles it. Identical code at an identical
# repo·path·symbol sailed past the gate because the description differed, while
# two deliberately-parallel variants were refused because theirs did not. The
# arm is not mis-tuned; it is reading the wrong field, and no threshold fixes
# that — lowering it blocks more siblings, raising it allows more copies.
#
# So: STRUCTURE decides, and the semantic arm becomes a backstop set above the
# band where legitimate variants live (0.92 observed). It still catches a
# genuine reword that shares neither location nor code, which is the case the
# structural signals cannot see.
_SNIPPET_SEMANTIC_THRESHOLD = 0.96

# The gate queries per CHUNK of the candidate (#280) — this caps how many
# searches one save may cost. Eight chunks ≈ five thousand words of candidate;
# a duplicate hiding past that is the duplicate report's job to find, not a
# reason to stall the write path.
_GATE_MAX_CHUNKS = 8


@dataclass
class DuplicateMatch:
    """An existing record judged a near-duplicate of an incoming create."""
    id: int
    title: str
    similarity: float  # 1.0 for an exact normalized-title match
    reason: str        # "title" | "semantic"


# How each signal describes itself when it blocks a write. The structural ones
# are CERTAIN, so they say what was matched instead of hedging with "similar" —
# and they point at merge, not update, because two records of one artefact is
# what merge exists to fold back together.
_REASON_PHRASING = {
    "location": (
        "is already recorded at that exact repo · path · symbol",
        "Update it (update_{kind}), or if you meant to record a second call "
        "site, use merge_snippets so one record carries both locations.",
    ),
    "code": (
        "already holds identical code",
        "Update it (update_{kind}) rather than keeping two copies that must "
        "then be kept in step by hand.",
    ),
}


def duplicate_response(dup: "DuplicateMatch", kind: str) -> dict:
    """Standard 'blocked — update instead' payload returned by a create tool
    when the gate finds a near-duplicate. `kind` is 'note', 'task' or 'snippet'
    (drives the update_<kind> hint)."""
    phrasing = _REASON_PHRASING.get(dup.reason)
    if phrasing:
        claim, advice = phrasing
        message = (
            f'An existing {kind} (id {dup.id}: "{dup.title}") {claim}. '
            f"{advice.format(kind=kind)} If this really is a distinct {kind}, "
            f"retry with force=true."
        )
    else:
        message = (
            f'A {dup.reason}-similar {kind} already exists (id {dup.id}: '
            f'"{dup.title}"). Prefer UPDATING it (update_{kind}) over creating a '
            f"near-duplicate. If this really is a distinct {kind}, retry with "
            f"force=true."
        )
    return {
        "duplicate": True,
        "existing_id": dup.id,
        "existing_title": dup.title,
        "similarity": dup.similarity,
        "match": dup.reason,
        "message": message,
    }


async def _find_snippet_by_structure(
    user_id: int,
    code: str,
    locations: list[dict] | None,
    project_id: int | None,
) -> DuplicateMatch | None:
    """Exact-identity duplicate of an incoming snippet, or None.

    Two signals, both index-served off `notes.data` and both CERTAIN rather than
    probabilistic — which is what the semantic arm could not be (#2518):

      location  the same named thing in the same file. Requires BOTH path and
                symbol: a path alone is a directory of many artefacts, and
                matching on it would refuse every second snippet from one file.
      code      byte-identical code, wherever it lives. Uses the same
                fingerprint the drift check uses, so "identical" means the same
                thing in both places.

    Fail-open like the rest of this module: a failed lookup lets the write
    through rather than blocking on an infrastructure problem.
    """
    from scribe.services.knowledge import location_jsonpath
    from scribe.services.snippets import code_sha

    identifying = [
        loc for loc in (locations or [])
        if (loc.get("path") or "").strip() and (loc.get("symbol") or "").strip()
    ]
    if not identifying and not (code or "").strip():
        return None          # nothing to match on — don't open a session for it

    def _scoped(stmt):
        stmt = stmt.where(
            Note.user_id == user_id,
            Note.deleted_at.is_(None),
            Note.note_type == SNIPPET_NOTE_TYPE,
        )
        # Same scoping rule as the title and semantic arms: a project's records
        # compare only within that project, orphans only to orphans.
        return (stmt.where(Note.project_id == project_id) if project_id is not None
                else stmt.where(Note.project_id.is_(None)))

    try:
        async with async_session() as session:
            for loc in identifying:
                parts = {
                    "path": (loc["path"]).strip(),
                    "symbol": (loc["symbol"]).strip(),
                }
                repo = (loc.get("repo") or "").strip()
                if repo:
                    parts["repo"] = repo
                stmt = _scoped(select(Note)).where(
                    Note.data.path_exists(location_jsonpath(parts))
                )
                hit = (await session.execute(stmt.limit(1))).scalars().first()
                if hit is not None:
                    return DuplicateMatch(hit.id, hit.title, 1.0, "location")

            if (code or "").strip():
                stmt = _scoped(select(Note)).where(
                    Note.data["code_sha"].astext == code_sha(code)
                )
                hit = (await session.execute(stmt.limit(1))).scalars().first()
                if hit is not None:
                    return DuplicateMatch(hit.id, hit.title, 1.0, "code")
    except Exception:
        logger.debug("snippet structural dedup skipped — query failed", exc_info=True)
    return None


async def find_duplicate_note(
    user_id: int,
    title: str,
    body: str = "",
    project_id: int | None = None,
    is_task: bool | None = None,
    note_type: str = "note",
    code: str = "",
    locations: list[dict] | None = None,
) -> DuplicateMatch | None:
    """Best near-duplicate of (title, body) within the same owner + project +
    kind, or None. Title match first (cheap, exact), then — for snippets — the
    structural signals, then semantic when the body is long enough to be
    meaningful. Never raises — embedder failure degrades to title-only (callers
    should still be able to create).

    `code` and `locations` are the snippet's structured fields. They are ignored
    for every other kind, and passing them is what lets the gate compare
    ARTEFACTS rather than descriptions of artefacts (#2518).
    """
    norm = " ".join((title or "").split()).lower()

    # --- Signal 1: normalized-title exact match (same scope) ---
    # Fail-open: a dedup-check failure (DB down, etc.) must never block a
    # legitimate create — degrade to "no duplicate found" and let it through.
    if norm:
        try:
            async with async_session() as session:
                stmt = select(Note).where(
                    Note.user_id == user_id,
                    Note.deleted_at.is_(None),
                    Note.note_type == note_type,
                    func.lower(func.trim(Note.title)) == norm,
                )
                if project_id is not None:
                    stmt = stmt.where(Note.project_id == project_id)
                else:
                    stmt = stmt.where(Note.project_id.is_(None))
                if is_task is True:
                    stmt = stmt.where(Note.status.isnot(None))
                elif is_task is False:
                    stmt = stmt.where(Note.status.is_(None))
                existing = (await session.execute(stmt.limit(1))).scalars().first()
                if existing is not None:
                    return DuplicateMatch(existing.id, existing.title, 1.0, "title")
        except Exception:
            logger.debug("dedup title check skipped — query failed", exc_info=True)
            return None

    # --- Signal 2: structural identity (snippets only) ---
    # Ahead of the semantic arm because it is exact: when it fires there is
    # nothing to weigh, and its verdict is the one worth showing.
    if note_type == SNIPPET_NOTE_TYPE:
        structural = await _find_snippet_by_structure(
            user_id, code, locations, project_id
        )
        if structural is not None:
            return structural

    # --- Signal 3: semantic similarity (only with a substantial body) ---
    if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
        # Query with the SAME chunker the corpus was embedded with (#280). This
        # was the copy that mattered most and was easiest to miss: these are
        # QUERY documents, compared against embedded ones — shaped differently
        # from the corpus, the gate degrades silently. Chunking also makes the
        # gate see what the whole-document query diluted: a long candidate that
        # duplicates an existing record IN ONE SECTION now matches on that
        # section. Capped so one pathological paste can't turn a save into
        # dozens of searches — a duplicate past the cap is the duplicate
        # report's job, not the gate's.
        for query in embeddings_svc.chunk_document(title, body)[:_GATE_MAX_CHUNKS]:
            # Scope the semantic check the same way as the title check: a record
            # in project P compares only to P; a project-less (orphan) record
            # compares only to other orphans (orphan_only), NOT across every
            # project — without this, semantic_search_notes applies no project
            # filter when project_id is None and would match an orphan note
            # against any project's notes.
            hits = await embeddings_svc.semantic_search_notes(
                user_id, query, project_id=project_id, is_task=is_task,
                orphan_only=(project_id is None),
                limit=3,
                threshold=(_SNIPPET_SEMANTIC_THRESHOLD
                           if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
                # Owner-only, deliberately: this gate BLOCKS a create and tells
                # the caller to update the match instead. Matching someone
                # else's record would refuse their write and point them at
                # something they may not be able to edit.
                scope="own",
                # NOT demoted by supersession (#278). A superseded record is
                # still a duplicate of what you are about to write — the claim
                # is that it is no longer CURRENT, not that it is gone. Demoting
                # it here would let the same note be recorded a second time, and
                # the second copy would be the one nothing warns about.
                demote_superseded=False,
            )
            for score, note in hits:
                # semantic_search_notes doesn't filter note_type — enforce it
                # here so a note doesn't shadow a task of the same wording, etc.
                if note.note_type == note_type:
                    return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")

    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 THE FLOOR IS PER-KIND. Chunked embeddings (#280) changed what a pair
# score MEANS for multi-chunk records: a note-pair's similarity is its closest
# chunk pair, so any family of related long records — a dev-log run, a research
# fan-out — clears a floor that whole-document vectors used to dilute below it.
# Measured on the live corpus (2026-08-09): at 0.82 the note/task reports
# saturate the pair cap with related-but-distinct families, while 0.93+ returns
# the genuinely-alike records. Snippets are single-chunk (short by nature), so
# their similarity scale never shifted and they keep the old floor.
#
# Snippets sit BELOW the 0.90 write gate — the report catches what the gate
# lets through. Notes/tasks sit ABOVE it, and that is not a contradiction: the
# gate compares a new record against best-matching chunks too, but it blocks a
# WRITE and must stay forgiving, while the report proposes a REVIEW and at
# chunk grain 0.90 would still drown it in families. Each 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_KEYS = {
    "snippet": "kb_duplicate_threshold_snippet",
    "note": "kb_duplicate_threshold_note",
    "task": "kb_duplicate_threshold_task",
}
DUPLICATE_DEFAULT_THRESHOLDS = {"snippet": 0.82, "note": 0.93, "task": 0.93}
# 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, kind: str = "snippet") -> float:
    """The user's near-duplicate similarity floor for `kind`, clamped to [0, 1]."""
    from scribe.services.settings import get_setting

    default = DUPLICATE_DEFAULT_THRESHOLDS[kind]
    try:
        value = float(await get_setting(
            user_id, DUPLICATE_THRESHOLD_KEYS[kind], str(default)
        ))
    except (TypeError, ValueError):
        value = default
    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]))


def _symbols(data: dict | None) -> set[str]:
    """Every symbol a snippet claims, from its indexed location mirror."""
    return {
        (loc.get("symbol") or "").strip()
        for loc in (data or {}).get("locations") or []
        if (loc.get("symbol") or "").strip()
    }


def _drop_sibling_pairs(
    pairs: list[tuple[int, int, float]], records: dict[int, dict]
) -> list[tuple[int, int, float]]:
    """Remove pairs that are VARIANTS of one thing rather than copies of it.

    Two snippets that both name a symbol, name DIFFERENT symbols, and hold
    different code are two artefacts. The author asserted that by naming them
    apart, and merging them would destroy a distinction someone made on purpose.

    Without this, a design system's button family reports as a single merge set:
    eight recipes, every direct pair over the floor, top score 0.92 (#2518).
    They resemble each other because variants of one component are SUPPOSED to —
    same selector prefix, same token families, deliberately parallel prose. The
    similarity is read correctly; it just does not mean "duplicate".

    THE COST, stated plainly: a helper genuinely recorded twice under two names
    — `debounce` and `useDebouncedRef` — is no longer reported. That is real
    recall lost. It is the better trade because the report is a merge PROPOSAL:
    a missed pair costs a duplicate nobody was going to notice anyway, while a
    wrong set invites an operator to collapse a component family in one click.
    Same-symbol and no-symbol duplicates, which is how re-recording usually
    looks, still report.
    """
    kept = []
    for left, right, score in pairs:
        left_data, right_data = records.get(left) or {}, records.get(right) or {}
        left_syms, right_syms = _symbols(left_data), _symbols(right_data)
        shas = (left_data.get("code_sha"), right_data.get("code_sha"))
        identical_code = shas[0] is not None and shas[0] == shas[1]
        # Both named, no name in common, and the code differs → siblings.
        if (left_syms and right_syms and not (left_syms & right_syms)
                and not identical_code):
            continue
        kept.append((left, right, score))
    return kept


# What a duplicate group should become, per kind. Snippets merge losslessly —
# one helper, every call site folded into the survivor. Notes and tasks do NOT
# merge: folding two records destroys what each actually said, which is why
# consolidated_at was dropped rather than built (#2483). The report can detect
# the cluster and name the options; deciding which applies is the reader's job,
# because it turns on what the records SAY, not how alike they score (#2547).
_KIND_SUGGESTION = {
    "snippet": (
        "Same reusable thing recorded more than once → merge_snippets(target, "
        "others); the survivor keeps every call site. Deliberately-parallel "
        "variants (a component family) are siblings — leave them."
    ),
    "note": (
        "Read before acting — records this alike are one of three things. A "
        "correction pair (one re-measures or reverses the other): declare "
        "supersedes on the newer, both survive, the older is demoted and "
        "labelled. State smeared across dated records: extract it into the "
        "System's reference note (updated in place) and leave these as "
        "history. Genuinely parallel records: leave them alone. Do NOT merge "
        "notes — folding them destroys what each said."
    ),
    "task": (
        "Two tasks this alike usually mean the same work opened twice: keep "
        "the one with the real history, fold anything unique into its body or "
        "a work-log, and cancel the other with a pointer. If one CORRECTS the "
        "other's conclusions, supersession also works for tasks."
    ),
}

# kind → the Note-model predicate for BOTH sides of the self-join. Tasks are
# notes with a status, not a note_type of their own — the same split every
# list surface makes.
_REPORT_KINDS = ("snippet", "note", "task")


def _kind_clauses(kind: str, note_alias):
    """The WHERE terms that make an aliased Note row one `kind` of record."""
    if kind == "snippet":
        return (note_alias.note_type == SNIPPET_NOTE_TYPE,)
    if kind == "task":
        return (note_alias.note_type == "note", note_alias.status.isnot(None))
    # kind == "note": documents only — a task is a note with a status, and
    # mixing them would propose folding a to-do into a write-up.
    return (note_alias.note_type == "note", note_alias.status.is_(None))


async def find_duplicate_records(
    user_id: int,
    *,
    kind: str = "snippet",
    threshold: float | None = None,
    limit: int = _MAX_DUPLICATE_PAIRS,
) -> dict:
    """Near-duplicate records of one kind, grouped into candidate 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.

    `kind` is "snippet", "note", or "task". The query is the same; what differs
    is the group payload and the SUGGESTION attached to it — merge is only ever
    proposed for snippets (see _KIND_SUGGESTION). Non-snippet groups also carry
    `members` with dates and any supersession claims that already exist inside
    the group, because "is this a correction pair or a chronicle cluster?" is
    answered by reading, and dates plus existing claims are the evidence.

    Returns {"groups": [...], "pairs": [...], "threshold": f, "suggestion": s}.
    Snippet groups keep their historical shape (`snippets` key) so the existing
    UI and MCP consumers don't churn. Fail-open (an empty report) like the rest
    of this module — a suggestion feature must not break the page it decorates.
    """
    if kind not in _REPORT_KINDS:
        raise ValueError(f"kind must be one of {_REPORT_KINDS}, not {kind!r}")
    floor = (
        await get_duplicate_threshold(user_id, kind)
        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)
    # Chunk grain (#280): a note-pair's similarity is its closest CHUNK pair —
    # two records duplicate each other where their most similar sections do,
    # which is the honest definition when one section of a long note restates
    # another record. GROUP BY collapses the chunk cross-product to one row
    # per note pair.
    best = func.min(distance)

    pairs: list[tuple[int, int, float]] = []
    try:
        async with async_session() as session:
            stmt = (
                select(left.note_id, right.note_id, best.label("distance"))
                .select_from(left)
                # `<` not `!=`: each unordered pair exactly once, and it drops
                # the self-pairs (including cross-chunk self-pairs, which would
                # otherwise flag every multi-chunk note against itself).
                .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(
                    *_kind_clauses(kind, left_note),
                    *_kind_clauses(kind, right_note),
                    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,
                )
                .group_by(left.note_id, right.note_id)
                .having(best <= max_distance)
                .order_by(best.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 %s scan failed", kind, exc_info=True)
        return _empty_report(kind, floor)

    if not pairs:
        return _empty_report(kind, floor)

    # Titles + structural fields + dates, one fetch for every id the scan
    # proposed. Dates matter for non-snippet groups: "is this a correction pair
    # or a chronicle cluster?" is partly answered by when each was written.
    scanned = sorted({n for pair in pairs for n in pair[:2]})
    titles: dict[int, str] = {}
    records: dict[int, dict] = {}
    meta: dict[int, dict] = {}
    try:
        async with async_session() as session:
            rows = (await session.execute(
                select(
                    Note.id, Note.title, Note.data,
                    Note.created_at, Note.updated_at, Note.task_kind,
                ).where(Note.id.in_(scanned))
            )).all()
            for i, t, d, created, updated, task_kind in rows:
                titles[int(i)] = t
                records[int(i)] = d or {}
                meta[int(i)] = {
                    "created_at": iso(created),
                    "updated_at": iso(updated),
                    "task_kind": task_kind,
                }
    except Exception:
        logger.debug("duplicate report titles unavailable", exc_info=True)

    if kind == "snippet":
        # Fails OPEN, and the direction matters: with `records` empty the
        # filter keeps every pair, so a lookup failure degrades to the
        # unfiltered report rather than to an empty one. A report that silently
        # returns nothing reads as "your corpus is clean", the wrong lie.
        # Snippet-only: the filter keys on symbol/code_sha, which other kinds
        # don't carry — and for them a look-alike is a finding, not a sibling.
        pairs = _drop_sibling_pairs(pairs, records)
        if not pairs:
            return _empty_report(kind, floor)

    best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
    grouped = group_pairs(pairs)

    # Supersession claims already declared WITHIN a group. A pair someone has
    # already ruled on must not be re-proposed as an open question — and for
    # the reader, an existing claim is the strongest evidence the group is a
    # correction chain rather than a chronicle cluster.
    claims: set[tuple[int, int]] = set()
    if kind != "snippet" and grouped:
        try:
            from scribe.models.note_supersession import NoteSupersession
            all_ids = sorted({n for g in grouped for n in g})
            async with async_session() as session:
                rows = (await session.execute(
                    select(
                        NoteSupersession.superseder_id,
                        NoteSupersession.superseded_id,
                    ).where(
                        NoteSupersession.superseder_id.in_(all_ids),
                        NoteSupersession.superseded_id.in_(all_ids),
                    )
                )).all()
            claims = {(int(a), int(b)) for a, b in rows}
        except Exception:
            logger.debug("supersession lookup for report failed", 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
        ]
        group: dict = {
            "note_ids": 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,
        }
        if kind == "snippet":
            # Historical shape — the existing UI and MCP consumers read
            # `snippets`, and churning them buys nothing.
            group["snippets"] = [
                {"id": nid, "title": titles.get(nid, "")} for nid in members
            ]
        else:
            group["members"] = [
                {"id": nid, "title": titles.get(nid, ""), **meta.get(nid, {})}
                for nid in members
            ]
            in_group = [
                {"superseder_id": a, "superseded_id": b}
                for (a, b) in sorted(claims)
                if a in members and b in members
            ]
            if in_group:
                group["existing_supersessions"] = in_group
        groups.append(group)
    groups.sort(key=lambda g: (-g["top_score"], g["note_ids"][0]))
    return {
        "groups": groups, "pairs": pairs, "threshold": floor,
        "suggestion": _KIND_SUGGESTION[kind],
    }


def _empty_report(kind: str, floor: float) -> dict:
    return {
        "groups": [], "pairs": [], "threshold": floor,
        "suggestion": _KIND_SUGGESTION[kind],
    }


async def find_duplicate_snippets(
    user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
) -> dict:
    """The snippet report — find_duplicate_records(kind="snippet"), kept under
    its established name because both surfaces and the SnippetListView consume
    it. New kinds go through the general function."""
    return await find_duplicate_records(
        user_id, kind="snippet", threshold=threshold, limit=limit
    )


async def find_duplicate_rule(
    title: str,
    topic_id: int | None = None,
    project_id: int | None = None,
) -> DuplicateMatch | None:
    """Title-based near-duplicate of a rule, scoped to the same topic (a rulebook
    rule) or the same project (a project rule). Rules aren't a semantic-retrieval
    surface, so a normalized-title match is the right (and only) signal. Fail-open
    like find_duplicate_note."""
    norm = " ".join((title or "").split()).lower()
    if not norm or (topic_id is None and project_id is None):
        return None
    try:
        async with async_session() as session:
            stmt = select(Rule).where(
                Rule.deleted_at.is_(None),
                func.lower(func.trim(Rule.title)) == norm,
            )
            if topic_id is not None:
                stmt = stmt.where(Rule.topic_id == topic_id)
            else:
                stmt = stmt.where(Rule.project_id == project_id)
            existing = (await session.execute(stmt.limit(1))).scalars().first()
            if existing is not None:
                return DuplicateMatch(existing.id, existing.title, 1.0, "title")
    except Exception:
        logger.debug("dedup rule title check skipped — query failed", exc_info=True)
    return None