frontend/src/views/KnowledgeView.vue

@@ -46,6 +46,50 @@ const sortMode = ref<"modified" | "created" | "alpha" | "type">("modified");
 const searchQuery = ref("");
 let searchDebounce: ReturnType<typeof setTimeout> | null = null;
 
+// ─── Near-duplicate report ────────────────────────────────────────────────────
+// On demand, never automatic — a corpus-wide pairwise scan, and most visits to
+// this page aren't a tidy-up (same reasoning as SnippetListView's report).
+
+interface DupMember {
+  id: number; title: string;
+  created_at?: string | null; updated_at?: string | null;
+  task_kind?: string | null;
+}
+interface DupGroup {
+  note_ids: number[];
+  members: DupMember[];
+  top_score: number;
+  existing_supersessions?: { superseder_id: number; superseded_id: number }[];
+}
+
+const dupGroups = ref<DupGroup[]>([]);
+const dupSuggestion = ref("");
+const dupLoading = ref(false);
+const dupChecked = ref(false);
+// The report follows the type filter: viewing tasks checks tasks. Anything
+// else (all / plan / process) checks notes — the kind with the most to find.
+const dupKind = computed(() => (activeType.value === "task" ? "task" : "note"));
+
+async function loadDuplicates() {
+  dupLoading.value = true;
+  try {
+    const data = await apiGet<{ groups: DupGroup[]; suggestion: string }>(
+      `/api/notes/duplicates?kind=${dupKind.value}`
+    );
+    dupGroups.value = data.groups;
+    dupSuggestion.value = data.suggestion;
+    dupChecked.value = true;
+  } catch {
+    dupChecked.value = false;
+  } finally {
+    dupLoading.value = false;
+  }
+}
+
+// A stale report is worse than none: switching the type filter changes which
+// kind the button checks, so the old kind's groups must not linger under it.
+watch(dupKind, () => { dupChecked.value = false; dupGroups.value = []; });
+
 // ─── Type counts ──────────────────────────────────────────────────────────────
 
 interface KnowledgeCounts { note: number; task: number; plan: number; process: number; total: number }
@@ -385,6 +429,51 @@ onUnmounted(() => {
             <Share2 :size="16" />
             Graph
           </button>
+          <button
+            class="btn-ghost btn-compact"
+            :disabled="dupLoading"
+            title="Find notes or tasks already recorded that closely resemble each other — the report proposes, it never changes anything"
+            @click="loadDuplicates"
+          >
+            {{ dupLoading ? "Checking…" : "Find duplicates" }}
+          </button>
+        </div>
+
+        <!-- Near-duplicate report. A proposal surface only: unlike snippets
+             (which merge losslessly), notes are never merged — the right fix is
+             supersession, extraction into a reference note, or leaving parallel
+             records alone, and choosing needs the records READ. That reading is
+             the assistant's job; this panel shows the human what exists. -->
+        <div v-if="dupChecked && !dupLoading" class="dup-panel">
+          <p v-if="!dupGroups.length" class="dup-empty">
+            No near-duplicate {{ dupKind }}s found — nothing recorded resembles
+            anything else closely enough to flag.
+          </p>
+          <template v-else>
+            <p class="dup-head">
+              {{ dupGroups.length }} possible duplicate
+              {{ dupGroups.length > 1 ? "sets" : "set" }} among your
+              {{ dupKind }}s. {{ dupSuggestion }}
+            </p>
+            <div v-for="(g, i) in dupGroups" :key="i" class="dup-group">
+              <div class="dup-members">
+                <router-link
+                  v-for="m in g.members"
+                  :key="m.id"
+                  class="dup-member"
+                  :to="dupKind === 'task' ? `/tasks/${m.id}` : `/notes/${m.id}`"
+                >
+                  #{{ m.id }} {{ m.title }}
+                </router-link>
+              </div>
+              <span class="dup-score">{{ Math.round(g.top_score * 100) }}% alike</span>
+              <span
+                v-if="g.existing_supersessions?.length"
+                class="dup-claimed"
+                title="A supersession has already been declared inside this set — it is not an open question"
+              >already ruled on</span>
+            </div>
+          </template>
         </div>
 
         <!-- Loading / empty -->
@@ -952,4 +1041,64 @@ onUnmounted(() => {
   height: 100%;
 }
 
+
+/* ── Near-duplicate report ──────────────────────────────────────────────────
+   Mirrors SnippetListView's panel so the two reports read as one feature.
+   Scoped styles can't be shared across SFCs; if a third view ever grows this
+   panel, promote the family to components.css and record it (#2464's rule:
+   two-or-more is when a recipe earns the shared sheet). */
+.dup-panel {
+  margin-bottom: 1.25rem;
+  padding: 0.85rem 1rem;
+  border: 1px solid var(--color-border);
+  border-radius: 8px;
+  background: var(--color-surface-alt);
+}
+.dup-empty,
+.dup-head {
+  margin: 0 0 0.5rem;
+  font-size: 0.85rem;
+  color: var(--color-text-muted);
+}
+.dup-empty { margin-bottom: 0; }
+.dup-group {
+  display: flex;
+  align-items: center;
+  gap: 0.75rem;
+  flex-wrap: wrap;
+  padding: 0.5rem 0;
+  border-top: 1px solid var(--color-border);
+}
+.dup-members {
+  display: flex;
+  gap: 0.4rem;
+  flex-wrap: wrap;
+  flex: 1 1 20rem;
+  min-width: 0;
+}
+.dup-member {
+  font-size: 0.8rem;
+  padding: 0.1rem 0.45rem;
+  border-radius: 4px;
+  background: color-mix(in srgb, var(--color-text-muted) 12%, transparent);
+  color: var(--color-text);
+  text-decoration: none;
+  overflow-wrap: anywhere;
+}
+.dup-member:hover { background: var(--color-hover); }
+.dup-score {
+  font-size: 0.75rem;
+  color: var(--color-text-muted);
+  font-variant-numeric: tabular-nums;
+  white-space: nowrap;
+}
+/* A set someone already ruled on — quiet, not celebratory: it means "skip". */
+.dup-claimed {
+  font-size: 0.72rem;
+  color: var(--color-text-muted);
+  border: 1px solid var(--color-border);
+  border-radius: 4px;
+  padding: 0.05rem 0.4rem;
+  white-space: nowrap;
+}
 </style>

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
+    # Reports on the corpus. Reads only — the merge or supersession each
-    # separate, explicitly-called write.
+    # suggests is a separate, explicitly-called write.
-    "find_duplicate_snippets",
+    "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(
+# What a duplicate group should become, per kind. Snippets merge losslessly —
-    user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
+# 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}],
+    `kind` is "snippet", "note", or "task". The query is the same; what differs
-    "pairs": [...], "threshold": f}. Fail-open (an empty report) like the rest of
+    is the group payload and the SUGGESTION attached to it — merge is only ever
-    this module — a suggestion feature must not be able to break the page it
+    proposed for snippets (see _KIND_SUGGESTION). Non-snippet groups also carry
-    decorates.
+    `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,
+                    *_kind_clauses(kind, left_note),
-                    right_note.note_type == SNIPPET_NOTE_TYPE,
+                    *_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)
+        logger.warning("Near-duplicate %s scan failed", kind, exc_info=True)
-        return {"groups": [], "pairs": [], "threshold": floor}
+        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
+    # Titles + structural fields + dates, one fetch for every id the scan
-    # filter below. One fetch covers every id the scan proposed.
+    # 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}
+            for i, t, d, created, updated, task_kind in rows:
-            records = {int(i): (d or {}) for i, _, d 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
+    if kind == "snippet":
-    # below keeps every pair, so a lookup failure degrades to the unfiltered
+        # Fails OPEN, and the direction matters: with `records` empty the
-    # report rather than to an empty one. A report that silently returns
+        # filter keeps every pair, so a lookup failure degrades to the
-    # nothing reads as "your corpus is clean", which is the wrong lie.
+        # 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 {"groups": [], "pairs": [], "threshold": floor}
+            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": [
+            # The strongest resemblance in the set — how confident the
-                {"id": nid, "title": titles.get(nid, "")} for nid in members
+            # 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(

tests/test_services_dedup.py

@@ -282,3 +282,46 @@ def test_duplicate_response_names_what_matched_for_structural_hits():
         DuplicateMatch(id=9, title="x", similarity=1.0, reason="code"), "snippet",
     )
     assert "identical code" in r["message"]
+
+
+# --- the generalised report (#2547) ------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_report_refuses_an_unknown_kind():
+    """A typo'd kind must fail loudly, not scan snippets by default — the
+    caller asked a question about a kind that doesn't exist, and answering a
+    different question instead is how wrong conclusions get confident."""
+    from scribe.services.dedup import find_duplicate_records
+
+    with pytest.raises(ValueError, match="kind must be one of"):
+        await find_duplicate_records(7, kind="rule")
+
+
+def test_kind_clauses_split_notes_from_tasks_on_status():
+    """Tasks are notes with a status, not a note_type of their own. A report
+    that mixed them would propose folding a to-do into a write-up."""
+    from scribe.models.note import Note
+    from scribe.services.dedup import _kind_clauses
+
+    note_sql = " AND ".join(str(c) for c in _kind_clauses("note", Note))
+    task_sql = " AND ".join(str(c) for c in _kind_clauses("task", Note))
+    snip_sql = " AND ".join(str(c) for c in _kind_clauses("snippet", Note))
+
+    assert "status IS NULL" in note_sql
+    assert "status IS NOT NULL" in task_sql
+    assert "note_type" in snip_sql and "status" not in snip_sql
+
+
+def test_every_kind_has_a_suggestion_and_none_proposes_merging_notes():
+    """The suggestion is the report's point: what to DO differs by what the
+    records are, and 'merge' is only ever the answer for snippets — folding two
+    notes destroys what each said, which is why consolidated_at was dropped
+    rather than built (#2483)."""
+    from scribe.services.dedup import _KIND_SUGGESTION, _REPORT_KINDS
+
+    for kind in _REPORT_KINDS:
+        assert _KIND_SUGGESTION.get(kind), f"no suggestion for {kind}"
+    assert "merge" in _KIND_SUGGESTION["snippet"]
+    assert "NOT merge" in _KIND_SUGGESTION["note"]
+    assert "supersedes" in _KIND_SUGGESTION["note"]

tests/test_snippet_duplicates.py

@@ -96,4 +96,9 @@ async def test_a_failed_scan_returns_an_empty_report_not_an_error():
         patch.object(dedup_svc, "async_session", side_effect=RuntimeError("boom")),
     ):
         out = await dedup_svc.find_duplicate_snippets(1)
-    assert out == {"groups": [], "pairs": [], "threshold": 0.8}
+    # The empty report still carries its per-kind `suggestion` (#2547) — a
+    # failed scan must be indistinguishable in SHAPE from a clean one, or every
+    # consumer needs a second code path for the degraded case.
+    assert out["groups"] == [] and out["pairs"] == []
+    assert out["threshold"] == 0.8
+    assert "merge" in out["suggestion"]