feat(dedup): the duplicate report reaches notes and tasks, with per-kind cures

Step 5 of #278, folding in #2534. The operator's no-gate decision for the web
UI (#2482 — "an llm attached to this surface is the corrections system") has a
precondition nobody had built: the corrector has to be able to SEE what needs
correcting. find_duplicate_snippets had no equivalent for notes or tasks, so a
duplicate note was only ever noticed by accident.

find_duplicate_records(kind="snippet"|"note"|"task") — the same indexed
self-join, parameterised. Tasks are notes with a status, not a note_type, so
the kind split is a status predicate; mixing them would propose folding a
to-do into a write-up. find_duplicate_snippets stays as a wrapper because both
surfaces and SnippetListView consume it by name.

What differs by kind is the CURE, and the report says so in a `suggestion`
field rather than leaving the caller to guess:

  snippet  merge — lossless, the survivor keeps every call site
  note     NEVER merge. A correction pair → supersedes on the newer; state
           smeared across dated records → extract to the System's reference
           note; genuinely parallel → leave alone. Choosing needs the records
           READ, which is the agent's job — so non-snippet groups carry
           `members` with dates and any `existing_supersessions` already
           declared inside the group. A pair someone ruled on is not an open
           question.
  task     usually the same work opened twice — keep the one with the history,
           cancel the other with a pointer.

The snippet sibling filter stays snippet-only: it keys on symbol/code_sha,
which other kinds don't carry — and for them a look-alike is a finding.

Surfaces: MCP find_duplicate_records (classified into _READ_ONLY_TOOLS — the
completeness test would have caught the omission), REST /api/notes/duplicates,
and a KnowledgeView panel mirroring SnippetListView's — links only, no merge
button, because for notes the report proposes and the correction is a read-
and-decide act. The panel follows the type filter and clears when it changes,
so a note report can't linger under a task view.

Correcting the task's own premise: it claimed the snippet report had "no view
consuming it" — stale; SnippetListView has consumed it since it shipped. The
UI gap was only ever notes/tasks.

Answers the question carried from #2482: yes, the update routes on BOTH
surfaces can turn a record into a duplicate — the gate is create-time by
design. This report is the mechanism that catches it after the fact, which is
the model the operator chose.

Refs #278, #2547

src/scribe/mcp/server.py

@@ -332,9 +332,9 @@ _READ_ONLY_TOOLS = frozenset({
     "list_rules", "list_tags", "list_tasks", "list_topics", "list_trash",
     "list_always_on_rules", "search",
     "get_system", "list_systems", "list_system_records",
-    # Reports on the snippet corpus. Reads only — the merge it suggests is a
-    # separate, explicitly-called write.
-    "find_duplicate_snippets",
+    # Reports on the corpus. Reads only — the merge or supersession each
+    # suggests is a separate, explicitly-called write.
+    "find_duplicate_snippets", "find_duplicate_records",
     # Snippets and processes are notes with a kind. A key that may read a note
     # but not a snippet inverts the sensitivity ordering: it exposes the
     # free-text records and withholds the structured ones (#2496).

src/scribe/mcp/tools/notes.py

@@ -231,6 +231,38 @@ async def update_note(
     return data
 
 
+async def find_duplicate_records(kind: str = "note", threshold: float = 0.0) -> dict:
+    """Notes or tasks already recorded that closely resemble each other.
+
+    The create gate PREVENTS a duplicate arriving through an agent; the web UI
+    deliberately has no gate (a human mid-thought must not be blocked by a 409),
+    which makes YOU the corrections system — and a corrector has to be able to
+    SEE what needs correcting. This is the finder. Run it when tidying a
+    project's records, or when you suspect the same ground was covered twice.
+
+    Args:
+        kind: "note" (documents) or "task". Snippets have their own report,
+            find_duplicate_snippets, whose groups propose a lossless merge.
+        threshold: Similarity floor, 0-1. 0 uses the configured setting.
+
+    READ THE `suggestion` FIELD BEFORE ACTING, because the right fix differs by
+    what the records ARE, not how alike they score. For notes: a correction
+    pair → declare `supersedes` on the newer; state smeared across dated
+    records → extract it into the System's reference note and leave these as
+    history; genuinely parallel records → leave them alone. NEVER merge notes.
+    Each group carries `members` with dates and any `existing_supersessions`
+    already declared inside it — a pair someone has ruled on is not an open
+    question.
+    """
+    uid = current_user_id()
+    if kind not in ("note", "task"):
+        raise ValueError('kind must be "note" or "task" — snippets have their '
+                         "own report, find_duplicate_snippets")
+    return await dedup_svc.find_duplicate_records(
+        uid, kind=kind, threshold=threshold if threshold > 0 else None,
+    )
+
+
 async def delete_note(note_id: int) -> dict:
     """Move a Scribe note to the trash (recoverable). Restore via restore(batch_id)."""
     uid = current_user_id()
@@ -247,6 +279,7 @@ def register(mcp) -> None:
         get_note,
         create_note,
         update_note,
+        find_duplicate_records,
         delete_note,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/routes/notes.py

@@ -22,6 +22,7 @@ from scribe.services.notes import (
     update_note,
 )
 from scribe.services.note_drafts import upsert_draft, get_draft, delete_draft
+from scribe.services import dedup as dedup_svc
 from scribe.services import supersession as supersession_svc
 from scribe.services.note_usage import record_pulled
 
@@ -199,6 +200,30 @@ async def resolve_title_route():
     return jsonify(note.to_dict())
 
 
+@notes_bp.route("/duplicates", methods=["GET"])
+@login_required
+async def find_duplicate_records_route():
+    """Near-duplicate notes or tasks, grouped, with the per-kind suggestion.
+
+    ?kind=note|task, ?threshold= (0 = configured setting). Registered above the
+    `/<int:note_id>` routes on purpose — the literal path first, same reasoning
+    as /api/snippets/duplicates. Mirrors the MCP tool (#33): the web UI has no
+    dedup gate by design, so this report is the UI's only window onto what that
+    decision admits.
+    """
+    uid = get_current_user_id()
+    kind = request.args.get("kind", "note")
+    if kind not in ("note", "task"):
+        return jsonify({"error": 'kind must be "note" or "task"'}), 400
+    try:
+        threshold = float(request.args.get("threshold", 0) or 0)
+    except (TypeError, ValueError):
+        threshold = 0.0
+    return jsonify(await dedup_svc.find_duplicate_records(
+        uid, kind=kind, threshold=threshold if threshold > 0 else None,
+    ))
+
+
 @notes_bp.route("/<int:note_id>", methods=["GET"])
 @login_required
 async def get_note_route(note_id: int):

src/scribe/services/dedup.py

@@ -420,20 +420,79 @@ def _drop_sibling_pairs(
     return kept
 
 
-async def find_duplicate_snippets(
-    user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
+# 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 snippets already in the record, grouped into merge sets.
+    """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.
 
-    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.
+    `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) 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))
@@ -456,8 +515,8 @@ async def find_duplicate_snippets(
                 .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,
+                    *_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
@@ -472,54 +531,128 @@ async def find_duplicate_snippets(
             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}
+        logger.warning("Near-duplicate %s scan failed", kind, exc_info=True)
+        return _empty_report(kind, floor)
 
     if not pairs:
-        return {"groups": [], "pairs": [], "threshold": floor}
+        return _empty_report(kind, floor)
 
-    # Titles + the structural fields, for presentation AND for the sibling
-    # filter below. One fetch covers every id the scan proposed.
+    # 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).where(Note.id.in_(scanned))
+                select(
+                    Note.id, Note.title, Note.data,
+                    Note.created_at, Note.updated_at, Note.task_kind,
+                ).where(Note.id.in_(scanned))
             )).all()
-            titles = {int(i): t for i, t, _ in rows}
-            records = {int(i): (d or {}) for i, _, d in rows}
+            for i, t, d, created, updated, task_kind in rows:
+                titles[int(i)] = t
+                records[int(i)] = d or {}
+                meta[int(i)] = {
+                    "created_at": created.isoformat() if created else None,
+                    "updated_at": updated.isoformat() if updated else None,
+                    "task_kind": task_kind,
+                }
     except Exception:
         logger.debug("duplicate report titles unavailable", exc_info=True)
 
-    # Fails OPEN, and the direction matters: with `records` empty the filter
-    # below 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", which is the wrong lie.
-    pairs = _drop_sibling_pairs(pairs, records)
-    if not pairs:
-        return {"groups": [], "pairs": [], "threshold": floor}
+    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
         ]
-        groups.append({
+        group: dict = {
             "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.
+            # 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}
+    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(