frontend/src/views/ProjectView.vue

@@ -956,13 +956,25 @@ async function confirmDelete() {
 .stat-notes     { background: color-mix(in srgb, var(--color-primary) 8%, transparent); color: var(--color-primary); border-color: color-mix(in srgb, var(--color-primary) 22%, transparent); }
 
 /* ── Two-column body ─────────────────────────────────────────── */
+/* `minmax(0, 1fr)`, not `1fr`. A bare `1fr` track has an AUTO minimum, so it
+   cannot shrink below its content — one wide descendant anywhere in the
+   content column widens the whole column past the grid, and everything inside
+   it then overflows the page and gets cut by `.project-view`'s
+   `overflow-x: clip`.
+   This is the same property the header nav relies on and wants (neither side
+   squeezed under its content); here it is exactly wrong, because the column
+   holds a kanban whose own tracks push outward. `min-width: 0` on the item is
+   the twin half — a grid item's default `min-width: auto` refuses to shrink
+   even when its track will. */
 .project-body {
   display: grid;
-  grid-template-columns: 248px 1fr;
+  grid-template-columns: 248px minmax(0, 1fr);
   gap: 1.25rem;
   align-items: start;
 }
 
+.content-area { min-width: 0; }
+
 /* ── Edit panel ──────────────────────────────────────────────── */
 .edit-panel {
   background: var(--color-bg-card);
@@ -1190,7 +1202,10 @@ async function confirmDelete() {
 /* ── Kanban ──────────────────────────────────────────────────── */
 .kanban {
   display: grid;
-  grid-template-columns: repeat(3, 1fr);
+  /* Same reason as .project-body: three auto-minimum tracks add up to more
+     than the column when a card title or a column header won't compress, and
+     the excess pushes the whole milestone card wider than the page. */
+  grid-template-columns: repeat(3, minmax(0, 1fr));
   gap: 0.75rem;
   align-items: start;
   padding: 0.75rem;

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.23",
+  "version": "0.1.24",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/hooks/scribe_static_context.md

@@ -51,5 +51,17 @@ for the operator's work, and as your own working memory across sessions.
   `/compact` (name what you logged). You can't run it yourself — surface the
   recommendation and let them decide. Suggest it at seams, not every turn.
 
+**If two Scribe instruction surfaces disagree** — this file, the MCP server's
+tool instructions, the `using-scribe` skill — **follow the one that assumes
+least about its own delivery.** This file is the floor: it ships with the
+plugin and needs no API key and no network, so it still applies in exactly the
+session where the others never arrived. The others may elaborate on what is
+written here; they must not contradict it. Weigh a disagreement by which way it
+fails, not by which surface said more: doing something a push would also have
+covered costs one redundant call, while skipping it because you expected a push
+that never came means working without the operator's rules and not knowing.
+A contradiction between surfaces is a defect in the product — say so, so it
+gets recorded and fixed rather than silently arbitrated again next session.
+
 If the Scribe tools are unavailable, say so rather than silently falling back
 to local notes.

src/scribe/mcp/server.py

@@ -174,19 +174,30 @@ written as a rule. A repeatable procedure is a PROCESS. Reusable code is a
 SNIPPET. Reach for a rule when the thing genuinely is a standing instruction
 about how to work, and nothing else can hold it.
 
+Rules are pull-only: CALL list_always_on_rules() YOURSELF at the start of a
+session, before substantive work. Do not wait to be handed them. The plugin's
+SessionStart hook also pushes the always-on rules + active-project context, but
+that is a delivery optimisation, not the bridge — it can be absent (instance
+unreachable, key not configured, or the hook itself silently inert, which is
+what #2198 was) and its absence does not announce itself. When the push did
+arrive, the pull costs one redundant call; when it didn't, skipping the pull
+means working without the operator's binding rules and having no signal that
+you are. Those two costs are not comparable, so pull.
+
 One thing NOT to do: don't bridge Scribe into a session by writing to the
-host's native memory. Rules are pull-only, so a fresh session won't reach for
+host's native memory. That explicit pull IS the bridge — it is written into
-them unless its always-loaded context says to — but the bridge for that is the
+every instruction surface the session already loads, so it needs no pointer on
-Scribe plugin's SessionStart hook, which pushes the always-on rules +
+disk. So do NOT create or refresh a "rules live in Scribe" pointer in
-active-project context into each session directly. So do NOT create or refresh
+CLAUDE.md / AGENTS.md / ~/.claude memory, and do NOT keep rules, recall, or
-a "rules live in Scribe" pointer in CLAUDE.md / AGENTS.md / ~/.claude memory,
+plans in those stores in parallel with Scribe — Scribe holds the single copy,
-and do NOT keep rules, recall, or plans in those stores in parallel with Scribe
+and a pointer Scribe cannot update is one that goes stale without anyone
-— Scribe holds the single copy. Native auto-memory stays for facts about the
+noticing. Native auto-memory stays for facts about the user; CLAUDE.md for
-user; CLAUDE.md for codebase onboarding. Never make Scribe's correctness depend
+codebase onboarding. Never make Scribe's correctness depend on the operator
-on the operator disabling a native function (e.g. autoMemoryEnabled): the
+disabling a native function (e.g. autoMemoryEnabled): the plugin must work with
-plugin must work with auto-memory at its default. If the plugin is ever removed
+auto-memory at its default. If the plugin is ever removed the session loses both
-the session loses this push and rebuilds context over time — an acceptable cost,
+the push and these instructions, and rebuilds context over time — an acceptable
-and far better than a silent settings change the operator may not know about.
+cost, and far better than a silent settings change the operator may not know
+about.
 
 When you are working on a specific project, call enter_project(project_id)
 ONCE at session start (or whenever the active project changes). It returns the
@@ -225,7 +236,11 @@ Scribe stores reusable Processes — saved prompts/workflows (note_type
 X process" or otherwise references a saved process, call list_processes() /
 get_process(name) and follow the returned prompt verbatim, including any
 "clarify first" steps it contains. Author a new one with create_process(title,
-body); edit with update_process.
+body); edit with update_process; retire one with delete_process (recoverable —
+it goes to the trash like anything else). A near-duplicate is refused at create
+time, because every Process becomes a skill file that auto-surfaces on the
+operator's machine: two near-identical procedures don't merely bloat the record,
+they compete to be followed.
 
 Scribe also stores Snippets — reusable functions/components recorded once for
 recall (note_type "snippet"): a name, language, signature, canonical location
@@ -267,6 +282,20 @@ operator. "Works for one user" is not done.
 # Tools a read-only API key may call. Anything not listed is treated as a
 # write for read keys (default-deny), so a newly-added tool is locked down
 # until explicitly classified here.
+#
+# The list stays EXPLICIT rather than being derived from the name. A read key is
+# what you hand to something you don't fully trust — a dashboard, a CI job, a
+# shared integration — and a boundary inferred from a naming convention grants
+# access to whatever a future author happens to call `get_*`. Enumerating it is
+# the point; staleness is the cost, and test_mcp_auth covers that (a read-shaped
+# tool must appear here or in _DELIBERATELY_WRITE_SCOPED below, so adding one
+# forces a decision instead of silently denying it).
+#
+# Membership means "reads the operator's data and mutates none of it". Several
+# getters record a retrieval event via record_pulled; that is telemetry about
+# the read itself, not a change to what was read, and it must keep working for a
+# read key or the corpus's surfaced:pulled ratio silently under-counts whichever
+# consumers hold one.
 _READ_ONLY_TOOLS = frozenset({
     "get_note", "get_project", "get_rule", "get_rulebook",
     "get_task", "get_milestone", "get_recent", "enter_project",
@@ -277,8 +306,30 @@ _READ_ONLY_TOOLS = frozenset({
     # Reports on the snippet corpus. Reads only — the merge it suggests is a
     # separate, explicitly-called write.
     "find_duplicate_snippets",
+    # 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).
+    "get_snippet", "list_snippets",
+    "get_process", "list_processes",
+    # Design systems: read, resolve (inheritance + mode), render, and compare
+    # against recorded snippets. All four compute from stored records and write
+    # nothing — the drift report is a report, and applying it is a separate
+    # explicit call.
+    "get_design_system", "list_design_systems", "resolve_design_system",
+    "get_design_system_stylesheet", "list_design_tokens",
+    "check_snippets_against_design_system", "list_starter_role_groups",
+    # Which repos map to which project. Read-only by nature; bind_repo /
+    # unbind_repo are the writes.
+    "list_repo_bindings",
 })
 
+# Read-SHAPED tools that must NOT be reachable with a read key — a getter that
+# creates on miss, a list that has a side effect. Empty today, and deliberately
+# kept as a declared escape hatch rather than left implicit: without it, the
+# completeness test would push a future `get_or_create_*` into the allow-list
+# above, which is exactly the wrong way to make a test pass.
+_DELIBERATELY_WRITE_SCOPED: frozenset[str] = frozenset()
+
 
 async def _buffer_request_body(receive):
     """Drain the ASGI request body and return (body_bytes, replay_receive).

src/scribe/mcp/tools/processes.py

@@ -8,8 +8,11 @@ from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
 from scribe.services import access as access_svc
+from scribe.services import dedup as dedup_svc
 from scribe.services import knowledge as knowledge_svc
 from scribe.services import notes as notes_svc
+from scribe.services import trash as trash_svc
+from scribe.services.note_usage import record_pulled
 
 
 async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
@@ -41,17 +44,38 @@ async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
     return {"processes": procs, "total": total}
 
 
-async def create_process(title: str, body: str, tags: list[str] | None = None) -> dict:
+async def create_process(
+    title: str, body: str, tags: list[str] | None = None, force: bool = False,
+) -> dict:
     """Create a stored process (a reusable saved prompt).
 
     Args:
         title: Process name, e.g. "Drift Audit" (required).
         body: The full prompt to run later (markdown). Required.
         tags: Plain-string tags, no # prefix.
+        force: Bypass the near-duplicate gate. By default, if a title- or
+            meaning-similar process already exists, creation is BLOCKED and the
+            existing one's id is returned so you update it instead. Set true
+            only for a genuinely distinct procedure.
+
+    Returns the created process, OR — when a near-duplicate is found and force
+    is false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing
+    created).
+
+    The gate matters more here than for other kinds: every process becomes a
+    skill file that auto-surfaces on the operator's machine, so two near-identical
+    procedures don't merely bloat the corpus — they compete to be followed, and
+    which one wins is decided by a slug.
     """
     if not (title or "").strip() or not (body or "").strip():
         raise ValueError("create_process requires a non-empty title and body")
     uid = current_user_id()
+    if not force:
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body, is_task=False, note_type="process",
+        )
+        if dup is not None:
+            return dedup_svc.duplicate_response(dup, "process")
     note = await notes_svc.create_note(
         uid, title=title.strip(), body=body, note_type="process", tags=tags,
     )
@@ -82,6 +106,12 @@ async def get_process(name_or_id: str) -> dict:
     if candidates:
         out["other_matches"] = candidates
     out.update(await access_svc.describe_provenance(uid, note))
+    # A process is embedded like any other note, so auto-inject can surface one —
+    # and its menu header names THIS tool as the way to open that kind. Without
+    # this, the getter the product points at is the one getter that records
+    # nothing, and every process sits permanently at zero pulls looking like dead
+    # weight beside kinds that merely had a counter (#2476, the repeat of #2245).
+    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_process")
     return out
 
 
@@ -119,6 +149,44 @@ async def update_process(process_id: int, title: str = "", body: str = "",
     return out
 
 
+async def delete_process(process_id: int) -> dict:
+    """Retire a stored process — it moves to the trash and is recoverable.
+
+    Reach for this when a procedure is wrong, superseded, or was never worth
+    keeping. A stored process is installed as a skill file on the operator's
+    machine and auto-surfaces there, so a bad one is followed rather than merely
+    ignored — it costs more than a missing one.
+
+    Deletion was always possible through `delete_note` (a process is a note, and
+    the trash is kind-agnostic), but nothing said so, and a kind whose own tools
+    offer create/read/update reads as one you cannot retire (#2250).
+    """
+    uid = current_user_id()
+    loaded = await notes_svc.get_note_for_user(uid, process_id)
+    note = loaded[0] if loaded else None
+    # Check the KIND before deleting: this tool is reached for by name, and
+    # letting it trash an ordinary note because the id happened to resolve would
+    # be a destructive action taken on a mistyped argument.
+    if note is None or note.note_type != "process" or note.deleted_at is not None:
+        raise ValueError(f"process {process_id} not found")
+    batch = await trash_svc.delete(uid, "note", process_id)
+    if batch is None:
+        raise ValueError(f"process {process_id} not found")
+    return {
+        "deleted_batch_id": batch,
+        "message": (
+            f"Process {process_id} moved to trash. Restore with restore('{batch}'). "
+            f"Its skill stub disappears on the operator's next process sync."
+        ),
+    }
+
+
 def register(mcp) -> None:
-    for fn in (list_processes, create_process, get_process, update_process):
+    for fn in (
+        list_processes,
+        create_process,
+        get_process,
+        update_process,
+        delete_process,
+    ):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/mcp/tools/snippets.py

@@ -127,12 +127,19 @@ async def create_snippet(
         force: Bypass the near-duplicate gate (see below).
 
     Returns the created snippet (including a parsed `snippet` field), OR — when a
-    near-duplicate snippet already exists and force is false — {"duplicate": true,
+    duplicate already exists and force is false — {"duplicate": true,
     "existing_id": ..., "message": ...} and nothing is created. When that happens
     and it really is the same reusable thing found in another place, prefer
     merge_snippets(existing_id, [new...]) — or record then merge — to unify them
     into ONE canonical record (which then carries every call site as a location),
     rather than forcing a second copy with force=true.
+
+    WHAT THE GATE MATCHES ON. Exact identity first — an existing snippet at the
+    same repo · path · symbol, or holding byte-identical code. Those are certain,
+    and force is almost never the right answer to them. Only then a semantic
+    check, held to a high bar so that VARIANTS of one component are not refused:
+    `.btn-primary` and `.btn-secondary` read alike and are two different things,
+    so record both (#2518).
     """
     if not (name or "").strip() or not (code or "").strip():
         raise ValueError("create_snippet requires a non-empty name and code")
@@ -148,6 +155,10 @@ async def create_snippet(
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
             is_task=False, note_type=snippets_svc.SNIPPET_NOTE_TYPE,
+            # The artefact itself, not just its description — the gate compares
+            # location and code before it compares prose (#2518).
+            code=code,
+            locations=snippets_svc.resolve_locations(repo, path, symbol, locations),
         )
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "snippet")

src/scribe/models/note_usage.py

@@ -51,10 +51,23 @@ class NoteUsageEvent(Base):
     note_id: Mapped[int] = mapped_column(Integer, nullable=False)
     # 'surfaced' | 'pulled'
     event: Mapped[str] = mapped_column(Text, nullable=False)
-    # Which surface produced it: 'auto_inject' | 'write_path_place' |
+    # Which surface produced it. Kept granular so the place arm and the
-    # 'write_path_semantic' | 'mcp_get_snippet' | 'mcp_get_note' | 'rest_note'.
+    # semantic arm can be compared — that comparison is the whole reason the
-    # Kept granular so the place arm and the semantic arm can be compared —
+    # place arm needed logging at all.
-    # that comparison is the whole reason the place arm needed logging at all.
+    #
+    # A CONVENTION, not a fixed vocabulary: `mcp_<tool>` for an agent call,
+    # `rest_<kind>` for a human opening a detail view, and a bare name for a
+    # hook or background surface ('auto_inject', 'write_path_place',
+    # 'write_path_semantic'). This comment deliberately no longer lists the
+    # members — the previous list had gone stale, naming 'rest_note' that
+    # nothing wrote while omitting sources that existed, and a half-true
+    # enumeration reads as authoritative in exactly the way that misleads
+    # (#2476). `grep -rn record_pulled\\\|record_surfaced src/` is the
+    # authoritative list, and unlike a comment it cannot drift.
+    #
+    # The mcp_/rest_ split is load-bearing. "Is this dead weight?" is served by
+    # any pull; "was that injected line useful?" is served by AGENT pulls only,
+    # so never aggregate across the prefix without saying why (#1038, #2085).
     source: Mapped[str] = mapped_column(Text, nullable=False)
 
     __table_args__ = (

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.note_usage import record_pulled
 from scribe.services.note_versions import list_versions, get_version
 
 logger = logging.getLogger(__name__)
@@ -178,6 +179,13 @@ async def get_note_route(note_id: int):
     note, permission = result
     data = note.to_dict()
     data["permission"] = permission
+    # Opening the detail view IS a pull — the operator chose to look. Tagged by
+    # SURFACE, not by the record's kind, matching rest_snippet: the kind is a
+    # join away, but which surface asked is not recoverable after the fact.
+    # Keeping rest_* apart from mcp_* is load-bearing, not tidiness — "was that
+    # injected line useful?" is answered by agent pulls alone, and a human
+    # clicking a link would inflate exactly the number #1038 and #2085 gate on.
+    record_pulled(user_id=uid, note_id=note_id, source="rest_note")
     return jsonify(data)
 
 

src/scribe/routes/snippets.py

@@ -112,6 +112,15 @@ async def create_snippet_route():
             project_id=project_id,
             is_task=False,
             note_type=snippets_svc.SNIPPET_NOTE_TYPE,
+            # Matched on the artefact — location and code — before prose, the
+            # same way the MCP create path does (#2518). Both surfaces must
+            # apply the identical gate or the web UI becomes the way to record
+            # a duplicate the agent would have been stopped from writing.
+            code=data.get("code", ""),
+            locations=snippets_svc.resolve_locations(
+                data.get("repo", ""), data.get("path", ""), data.get("symbol", ""),
+                data.get("locations"),
+            ),
         )
         if dup is not None:
             return jsonify(dedup_svc.duplicate_response(dup, "snippet")), 409

src/scribe/routes/tasks.py

@@ -13,6 +13,7 @@ from scribe.services.notes import (
     list_notes,
     update_note,
 )
+from scribe.services.note_usage import record_pulled
 from scribe.services.planning import start_planning as svc_start_planning
 from scribe.services.recurrence import calculate_next_due, validate_recurrence_rule
 
@@ -186,6 +187,9 @@ async def get_task_route(task_id: int):
         parent = await get_note_for_user(uid, task.parent_id)
         data["parent_title"] = parent[0].title if parent else None
     data["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)]
+    # Opening the detail view IS a pull — see the note beside rest_note in
+    # routes/notes.py for why the rest_* and mcp_* prefixes stay separable.
+    record_pulled(user_id=uid, note_id=task_id, source="rest_task")
     return jsonify(data)
 
 

src/scribe/services/dedup.py

@@ -48,6 +48,27 @@ _MIN_BODY_FOR_SEMANTIC = 200
 # 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
+
 
 @dataclass
 class DuplicateMatch:
@@ -58,25 +79,124 @@ class DuplicateMatch:
     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' or 'task' (drives the
+    when the gate finds a near-duplicate. `kind` is 'note', 'task' or 'snippet'
-    update_<kind> hint)."""
+    (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": 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."
-        ),
     }
 
 
+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,
@@ -84,11 +204,19 @@ async def find_duplicate_note(
     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 semantic when the body
+    kind, or None. Title match first (cheap, exact), then — for snippets — the
-    is long enough to be meaningful. Never raises — embedder failure degrades to
+    structural signals, then semantic when the body is long enough to be
-    title-only (callers should still be able to create)."""
+    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) ---
@@ -118,7 +246,17 @@ async def find_duplicate_note(
             logger.debug("dedup title check skipped — query failed", exc_info=True)
             return None
 
-    # --- Signal 2: semantic similarity (only with a substantial body) ---
+    # --- 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 = f"{title}\n{body}".strip()
         # Scope the semantic check the same way as the title check: a record in
@@ -129,7 +267,9 @@ async def find_duplicate_note(
         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=_SEMANTIC_THRESHOLD,
+            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
@@ -222,6 +362,52 @@ def group_pairs(pairs: list[tuple[int, int, float]]) -> list[list[int]]:
                   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
+
+
 async def find_duplicate_snippets(
     user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
 ) -> dict:
@@ -280,21 +466,32 @@ async def find_duplicate_snippets(
     if not pairs:
         return {"groups": [], "pairs": [], "threshold": floor}
 
-    best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
+    # Titles + the structural fields, for presentation AND for the sibling
-    grouped = group_pairs(pairs)
+    # filter below. One fetch covers every id the scan proposed.
-
+    scanned = sorted({n for pair in pairs for n in pair[:2]})
-    # 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] = {}
+    records: dict[int, dict] = {}
     try:
         async with async_session() as session:
             rows = (await session.execute(
-                select(Note.id, Note.title).where(Note.id.in_(ids))
+                select(Note.id, Note.title, Note.data).where(Note.id.in_(scanned))
             )).all()
-            titles = {int(i): t for i, t in rows}
+            titles = {int(i): t for i, t, _ in rows}
+            records = {int(i): (d or {}) for i, _, d in rows}
     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}
+
+    best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
+    grouped = group_pairs(pairs)
+
     groups = []
     for members in grouped:
         scores = [

src/scribe/services/snippets.py

@@ -96,6 +96,27 @@ def _normalize_locations(locations: list[dict] | None) -> list[dict]:
     return out
 
 
+def resolve_locations(
+    repo: str = "", path: str = "", symbol: str = "",
+    locations: list[dict] | None = None,
+) -> list[dict]:
+    """The location list a caller meant, from either calling convention.
+
+    `locations` is the general form (one entry per call site); repo/path/symbol
+    are the single-location shorthand and apply only when `locations` was not
+    given — passing both is not a merge, it is the caller having decided.
+
+    Extracted because compose_body, create_snippet and the dedup gate must all
+    read the shorthand the SAME way. They each had their own copy of the
+    `if locations is None` fallback, which is fine until one of them gains a
+    rule the others don't — and the gate (#2518) is the one where a disagreement
+    would mean comparing a location the record won't actually be stored with.
+    """
+    if locations is None:
+        locations = [{"repo": repo, "path": path, "symbol": symbol}]
+    return _normalize_locations(locations)
+
+
 def _location_str(loc: dict) -> str:
     """`repo` · `path` · `symbol` — only the non-empty parts."""
     parts = [(loc.get(k) or "").strip() for k in ("repo", "path", "symbol")]
@@ -186,9 +207,7 @@ def compose_body(
     a back-compat shorthand for one location and are used only when ``locations``
     is not given.
     """
-    if locations is None:
+    locs = resolve_locations(repo, path, symbol, locations)
-        locations = [{"repo": repo, "path": path, "symbol": symbol}]
-    locs = _normalize_locations(locations)
 
     header: list[str] = []
     if (when_to_use or "").strip():
@@ -601,8 +620,7 @@ async def create_snippet(
     """Create a snippet note (embedded on create for immediate recall). Returns
     the created Note. Pass ``locations`` for the multi-location case; the single
     ``repo``/``path``/``symbol`` are the one-location shorthand."""
-    if locations is None:
+    locations = resolve_locations(repo, path, symbol, locations)
-        locations = [{"repo": repo, "path": path, "symbol": symbol}]
     note = await notes_svc.create_note(
         user_id,
         title=compose_title(name, when_to_use),

tests/test_instruction_surfaces_agree.py

@@ -0,0 +1,104 @@
+"""The instruction surfaces must agree that the agent pulls the rules itself.
+
+WHY THIS EXISTS
+
+Rule #119 makes the instruction surfaces the SPECIFICATION for product
+behaviour — there is no other place the "load the operator's rules" obligation
+is written down, and no code path enforces it. So a surface that states it
+differently isn't a documentation slip; it is the product behaving differently.
+
+That happened (#2497). `_INSTRUCTIONS` said the SessionStart hook "is the
+bridge" for getting rules into a session, while the `using-scribe` skill said to
+pull them yourself and treat any push as a bonus. An agent weighting the first
+would reasonably skip the pull.
+
+#2198 is the case where that is wrong: every plugin hook was silently inert for
+an extended period, and nothing announced it. An agent trusting the push would
+have run with no binding rules and no signal — while those rules govern branch,
+commit, push and other hard-to-reverse actions.
+
+The asymmetry is the whole argument, and it is what these tests pin: pulling
+when a push also arrived costs one redundant call; not pulling when the push
+never came costs the operator's rules entirely.
+
+WHAT THIS DOES NOT DO
+
+It cannot tell whether two surfaces contradict each other in prose generally —
+that needs a reader. It pins the one instruction whose absence is known to be
+load-bearing, and the specific shape the #2497 defect took: naming the push
+without also stating the pull.
+"""
+from __future__ import annotations
+
+import pathlib
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+# The pull instruction, however a surface phrases the surrounding prose.
+PULL = "list_always_on_rules"
+
+# Surfaces a session loads before substantive work. Hand-written because
+# "is this a session-start surface?" is an editorial fact, not a derivable one —
+# but each entry is asserted to EXIST, so a move or rename fails loudly here
+# instead of quietly dropping that surface from the check.
+SESSION_START_SURFACES = (
+    ROOT / "src" / "scribe" / "mcp" / "server.py",
+    ROOT / "plugin" / "hooks" / "scribe_static_context.md",
+    ROOT / "plugin" / "skills" / "using-scribe" / "SKILL.md",
+)
+
+
+def _all_surfaces() -> list[tuple[str, str]]:
+    """(label, text) for every file a SESSION loads as instructions.
+
+    Deliberately not every markdown file under plugin/: `README.md` describes
+    the push channel to the operator installing the plugin, and telling a human
+    what the hook does is not the same act as telling an agent it need not pull.
+    The boundary is "does a session read this", which is skills (loaded by
+    description match), the hook-injected static context, and the MCP server's
+    own instructions.
+    """
+    found = [(str(p.relative_to(ROOT)), p.read_text())
+             for p in (ROOT / "plugin" / "skills").rglob("SKILL.md")]
+    found += [(str(p.relative_to(ROOT)), p.read_text())
+              for p in (ROOT / "plugin" / "hooks").glob("*.md")]
+    server = ROOT / "src" / "scribe" / "mcp" / "server.py"
+    found.append((str(server.relative_to(ROOT)), server.read_text()))
+    return found
+
+
+def test_every_session_start_surface_states_the_pull():
+    missing = []
+    for path in SESSION_START_SURFACES:
+        assert path.exists(), (
+            f"{path.relative_to(ROOT)} is gone — it was one of the surfaces "
+            f"carrying the load-the-rules instruction. If it moved, update "
+            f"SESSION_START_SURFACES; if it was retired, check the instruction "
+            f"still lives somewhere a fresh session reads."
+        )
+        if PULL not in path.read_text():
+            missing.append(str(path.relative_to(ROOT)))
+    assert not missing, (
+        f"these surfaces no longer tell the agent to call {PULL}(): {missing}. "
+        f"The rules are pull-only and the push is best-effort, so a surface "
+        f"that omits this leaves a session bound by nothing (#2198, #2497)."
+    )
+
+
+def test_no_surface_names_the_push_without_stating_the_pull():
+    """The exact shape #2497 took.
+
+    Mentioning the SessionStart hook is fine and often useful. Mentioning it
+    *instead of* the pull is the defect: it reads as "this is handled", and the
+    surface that says so is the one an agent has least reason to doubt.
+    """
+    offenders = [
+        label for label, text in _all_surfaces()
+        if "SessionStart" in text and PULL not in text
+    ]
+    assert not offenders, (
+        f"these surfaces describe the SessionStart push but never state the "
+        f"explicit pull: {offenders}. The push is a delivery optimisation, not "
+        f"the bridge — it can be absent without saying so. Name it if it helps, "
+        f"but say to call {PULL}() regardless."
+    )

tests/test_mcp_auth.py

@@ -92,3 +92,68 @@ def test_body_calls_write_tool_classifies_correctly():
         json.dumps({"method": "tools/list"}).encode()
     ) is False
     assert _body_calls_write_tool(b"not json") is False
+
+
+def test_every_read_shaped_tool_is_explicitly_classified():
+    """A read-shaped tool must be classified, not left to default-deny.
+
+    `_READ_ONLY_TOOLS` is hand-maintained, and default-deny means a getter
+    omitted from it fails CLOSED — safe, but silent. That is how a read key
+    ended up able to `get_note` and not `get_snippet`, both pure reads of the
+    same table, while design systems were unreachable entirely (#2496). The
+    `find_duplicate_snippets` entry was the tell: someone classified the report
+    and missed the getters beside it.
+
+    This is the same shape as #2476 (record_pulled on three of four getters) —
+    a hand-written enumeration that missed the members added after it. The fix
+    there and here is the same: derive the CANDIDATES, keep the DECISION
+    explicit. Deriving the decision itself would be worse than a stale list —
+    it would make a security boundary follow a naming convention, so any future
+    `get_*` grants itself access.
+
+    So: every tool whose name reads like a read must appear in one of the two
+    sets. Adding a getter then forces a choice at review time.
+    """
+    import ast
+    import pathlib
+
+    from scribe.mcp.server import _DELIBERATELY_WRITE_SCOPED, _READ_ONLY_TOOLS
+
+    tools_dir = (pathlib.Path(__file__).resolve().parents[1]
+                 / "src" / "scribe" / "mcp" / "tools")
+    read_shaped = {
+        node.name
+        for path in tools_dir.glob("*.py") if path.name != "__init__.py"
+        for node in ast.parse(path.read_text()).body
+        if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
+        and node.name.startswith(("get_", "list_", "search", "resolve_",
+                                  "check_", "find_"))
+    }
+    assert read_shaped, "found no read-shaped tools — the tools package moved"
+
+    unclassified = sorted(read_shaped - _READ_ONLY_TOOLS
+                          - _DELIBERATELY_WRITE_SCOPED)
+    assert not unclassified, (
+        f"these read-shaped tools are classified by neither set: {unclassified}. "
+        f"They currently fail closed for read-only keys, silently. Add each to "
+        f"_READ_ONLY_TOOLS if it mutates nothing, or to "
+        f"_DELIBERATELY_WRITE_SCOPED with a comment saying what it writes."
+    )
+
+    # The reverse: a name in either set that no longer exists is a rename or a
+    # deletion, and a stale grant is worth surfacing even though it grants
+    # access to nothing. `enter_project` is the one read tool without a read
+    # prefix, so it is checked against the full tool set, not `read_shaped`.
+    all_tools = {
+        node.name
+        for path in tools_dir.glob("*.py") if path.name != "__init__.py"
+        for node in ast.parse(path.read_text()).body
+        if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
+        and not node.name.startswith("_") and node.name != "register"
+    }
+    phantom = sorted((_READ_ONLY_TOOLS | _DELIBERATELY_WRITE_SCOPED) - all_tools)
+    assert not phantom, (
+        f"these names are classified but are not tools: {phantom}. They were "
+        f"renamed or removed — drop them, and check whatever replaced them got "
+        f"classified."
+    )

tests/test_mcp_tool_processes.py

@@ -37,7 +37,9 @@ async def test_create_process_requires_title_and_body():
 @pytest.mark.asyncio
 async def test_create_process_sets_note_type():
     created = _fake_note()
-    with patch("scribe.services.notes.create_note",
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=None)), \
+         patch("scribe.services.notes.create_note",
                AsyncMock(return_value=created)) as mock_create:
         from scribe.mcp.tools.processes import create_process
         out = await create_process(title="Drift Audit", body="the prompt", tags=["audit"])
@@ -47,6 +49,39 @@ async def test_create_process_sets_note_type():
     assert mock_create.await_args.kwargs["title"] == "Drift Audit"
 
 
+@pytest.mark.asyncio
+async def test_create_process_blocks_a_near_duplicate():
+    """The gate matters more for processes than for other kinds: each one becomes
+    a skill file that auto-surfaces, so two near-identical procedures don't just
+    bloat the corpus — they compete to be followed (#2250)."""
+    from scribe.services.dedup import DuplicateMatch
+
+    # The real dataclass, not a MagicMock: a mock answers every attribute, so it
+    # would pass whatever field names this test happened to guess and prove
+    # nothing about the payload the tool actually returns.
+    match = DuplicateMatch(id=42, title="Drift Audit", similarity=0.94, reason="semantic")
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=match)), \
+         patch("scribe.services.notes.create_note", AsyncMock()) as mock_create:
+        from scribe.mcp.tools.processes import create_process
+        out = await create_process(title="Drift Audit", body="the prompt")
+    assert out["duplicate"] is True
+    assert out["existing_id"] == 42
+    mock_create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_create_process_force_bypasses_the_gate():
+    created = _fake_note()
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock()) as find_mock, \
+         patch("scribe.services.notes.create_note",
+               AsyncMock(return_value=created)):
+        from scribe.mcp.tools.processes import create_process
+        await create_process(title="Drift Audit", body="the prompt", force=True)
+    find_mock.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_get_process_returns_body_and_candidates():
     note = _fake_note(id=7)
@@ -117,7 +152,42 @@ async def test_update_process_refuses_a_read_only_share_with_the_reason():
     mock_update.assert_not_awaited()
 
 
-def test_register_attaches_four_tools():
+@pytest.mark.asyncio
+async def test_delete_process_trashes_it_recoverably():
+    proc = _fake_note(id=4)
+    proc.deleted_at = None
+    with patch("scribe.services.notes.get_note_for_user",
+               AsyncMock(return_value=(proc, "owner"))), \
+         patch("scribe.mcp.tools.processes.trash_svc.delete",
+               AsyncMock(return_value="batch-1")) as mock_delete:
+        from scribe.mcp.tools.processes import delete_process
+        out = await delete_process(process_id=4)
+    assert out["deleted_batch_id"] == "batch-1"
+    # Through the trash, not a hard delete — restorable like every other kind.
+    assert mock_delete.await_args.args[1] == "note"
+
+
+@pytest.mark.asyncio
+async def test_delete_process_refuses_a_plain_note():
+    """This tool is reached for by name. Letting it trash an ordinary note
+    because the id happened to resolve would be a destructive action taken on a
+    mistyped argument."""
+    plain = _fake_note(id=3, note_type="note")
+    plain.deleted_at = None
+    with patch("scribe.services.notes.get_note_for_user",
+               AsyncMock(return_value=(plain, "owner"))), \
+         patch("scribe.mcp.tools.processes.trash_svc.delete", AsyncMock()) as mock_delete:
+        from scribe.mcp.tools.processes import delete_process
+        with pytest.raises(ValueError):
+            await delete_process(process_id=3)
+    mock_delete.assert_not_awaited()
+
+
+def test_register_attaches_every_tool_in_the_module():
+    """Derived from the module rather than listed: a tool written but never
+    registered is invisible to an agent, and nothing else would notice."""
+    import inspect
+
     from scribe.mcp.tools import processes
     names: list[str] = []
 
@@ -129,6 +199,8 @@ def test_register_attaches_four_tools():
             return deco
 
     processes.register(FakeMcp())
-    assert set(names) == {
+    public = {
-        "list_processes", "create_process", "get_process", "update_process",
+        name for name, obj in vars(processes).items()
+        if inspect.iscoroutinefunction(obj) and not name.startswith("_")
     }
+    assert set(names) == public

tests/test_pull_telemetry.py

@@ -0,0 +1,203 @@
+"""Every surface that opens ONE note-backed record must record the pull.
+
+Covers both halves: the MCP getters an agent calls, and the REST detail views a
+human opens. They are one ledger with two prefixes (`mcp_*` / `rest_*`), and a
+gap on either side makes the same records look untouched.
+
+WHY THIS EXISTS
+
+`note_usage_events` answers "did anyone ever actually open this?" — the
+surfaced:pulled ratio is what makes dead weight visible and prunable. A getter
+that opens a record without recording it leaves that kind permanently at zero
+pulls, so it looks like dead weight beside kinds that merely had a counter.
+
+That has now happened twice:
+
+  #2245  `get_task` recorded nothing while auto-inject surfaced mostly tasks.
+         Fixed by adding the call to notes, tasks and snippets.
+  #2476  `get_process` recorded nothing — and the auto-inject menu header names
+         `get_process` as the way to open that kind. Processes were embedded
+         when #2245 was fixed; the fix enumerated the kinds someone thought of
+         rather than the kinds that exist. The same issue found the REST half:
+         `rest_snippet` was recorded, note and task detail were not, and the
+         model's own comment named a `'rest_note'` source nothing wrote.
+
+A missing call is the shape no behavioural test catches: it changes no return
+value (#2278, shape 4). Source inspection is the only thing that sees it.
+
+WHAT MAKES THIS DERIVED RATHER THAN A LIST
+
+The getters are not enumerated here. They are discovered from the tool modules
+by AST, and the ones that must record are identified by the loader they call —
+so a `get_<newkind>` added tomorrow is covered the moment it loads a note the
+way every other getter does.
+
+The loader names ARE a list, and that is the residual weakness. The second test
+pins them against a RENAME — the failure mode that would silently empty the
+candidate set and let this pass while checking nothing.
+
+It does not discover NEW loaders, and an earlier draft that tried to failed for
+the wrong reason: `create_note` and `update_note` also return a `Note`, so an
+annotation scan finds writers, not readers. Distinguishing them needs more than
+a type, so the honest position is a pinned list plus a non-empty assertion,
+and this paragraph saying so.
+"""
+from __future__ import annotations
+
+import ast
+import inspect
+import pathlib
+import pkgutil
+
+# Loaders that return ONE note-backed record in full. A getter calling any of
+# these is opening a record, which is the act `pulled` describes.
+#
+# `list_notes` is deliberately absent: `get_milestone` calls it to list a
+# milestone's steps, and that is a LIST — the milestone itself is not a note,
+# and its steps are surfaced rather than opened.
+SINGLE_NOTE_LOADERS = (
+    "get_note_for_user",
+    "resolve_process",
+    "get_snippet",
+)
+
+_SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+TOOLS_DIR = _SRC / "mcp" / "tools"
+ROUTES_DIR = _SRC / "routes"
+
+
+def _getters():
+    """(module name, function name, source) for every `get_*` MCP tool."""
+    for mod in pkgutil.iter_modules([str(TOOLS_DIR)]):
+        path = TOOLS_DIR / f"{mod.name}.py"
+        source = path.read_text()
+        for node in ast.parse(source).body:
+            if isinstance(node, ast.AsyncFunctionDef) and node.name.startswith("get_"):
+                yield mod.name, node.name, ast.get_source_segment(source, node) or ""
+
+
+def test_every_single_record_getter_records_a_pull():
+    missing = []
+    checked = []
+    for module, name, body in _getters():
+        if not any(loader in body for loader in SINGLE_NOTE_LOADERS):
+            continue
+        checked.append(f"{module}.{name}")
+        if "record_pulled" not in body:
+            missing.append(f"{module}.{name}")
+
+    # If this ever drops to zero the test has stopped testing anything — a
+    # renamed loader would silently empty the candidate set and pass.
+    assert checked, "found no note-backed getters; the loader names must have moved"
+    assert not missing, (
+        f"these getters open a record without recording the pull: {missing}. "
+        f"Add record_pulled(user_id=…, note_id=…, source='mcp_<tool>') before "
+        f"returning — see mcp/tools/notes.py:get_note."
+    )
+
+
+def _detail_routes():
+    """(module, handler, expanded source) for every bare-id GET route.
+
+    A route registered at exactly `/<int:x>` for GET is the DETAIL view of one
+    record — that shape is what distinguishes it from a list, a sub-resource
+    (`/<int:x>/versions`) or a write. Nothing else about the handler has to be
+    guessed.
+
+    The source is expanded one level through module-private helpers, because
+    `get_snippet_route` loads via `_load_snippet` rather than calling the loader
+    itself. Without the expansion the snippet route — the one that already got
+    this right — would drop out of the check.
+    """
+    import re
+    bare_id = re.compile(r"^/<int:\w+>$")
+    for path in sorted(ROUTES_DIR.glob("*.py")):
+        source = path.read_text()
+        tree = ast.parse(source)
+        helpers = {
+            node.name: ast.get_source_segment(source, node) or ""
+            for node in tree.body
+            if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
+            and node.name.startswith("_")
+        }
+        for node in tree.body:
+            if not isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)):
+                continue
+            for dec in node.decorator_list:
+                if not isinstance(dec, ast.Call):
+                    continue
+                route = next((a.value for a in dec.args
+                              if isinstance(a, ast.Constant)), None)
+                if not isinstance(route, str) or not bare_id.match(route):
+                    continue
+                methods = [
+                    e.value for kw in dec.keywords if kw.arg == "methods"
+                    and isinstance(kw.value, ast.List)
+                    for e in kw.value.elts if isinstance(e, ast.Constant)
+                ]
+                if "GET" not in methods:
+                    continue
+                body = ast.get_source_segment(source, node) or ""
+                expanded = body + "".join(
+                    src for name, src in helpers.items() if name in body
+                )
+                yield path.name, node.name, expanded
+
+
+def test_every_rest_detail_view_records_a_pull():
+    """The human half of the same ledger.
+
+    `rest_snippet` was recorded; note and task detail recorded nothing, so the
+    UI's most direct evidence of interest — someone opened the record — existed
+    for one kind out of three. The model's own comment listed `'rest_note'` as
+    a source, which means the design intended it and the implementation stopped
+    at snippets.
+
+    Same derivation as the MCP test above, over the other surface: the routes
+    are discovered, and the ones that must record are identified by the loader
+    they reach. A `/<int:x>` GET added for a fourth note-backed kind is covered
+    the day it is written.
+    """
+    missing = []
+    checked = []
+    for module, handler, body in _detail_routes():
+        if not any(loader in body for loader in SINGLE_NOTE_LOADERS):
+            continue  # not note-backed — groups and projects land here
+        checked.append(f"{module}:{handler}")
+        if "record_pulled" not in body:
+            missing.append(f"{module}:{handler}")
+
+    assert checked, (
+        "found no note-backed detail routes; the route shape or the loader "
+        "names must have moved"
+    )
+    assert not missing, (
+        f"these detail views open a record without recording the pull: "
+        f"{missing}. Add record_pulled(user_id=…, note_id=…, source='rest_<kind>') "
+        f"before returning — see routes/snippets.py:get_snippet_route."
+    )
+
+
+def test_every_named_loader_still_exists():
+    """Pins the hand-written list against a rename.
+
+    A renamed loader is the failure that matters: the candidate set above would
+    quietly empty and the first test would pass while checking nothing. The
+    `assert checked` there catches it too; this says WHICH name moved, which is
+    the difference between a five-minute fix and a puzzle.
+    """
+    from scribe.services import notes as notes_svc
+    from scribe.services import snippets as snippets_svc
+
+    available = {
+        name
+        for svc in (notes_svc, snippets_svc)
+        for name, obj in vars(svc).items()
+        if inspect.iscoroutinefunction(obj)
+    }
+    gone = [name for name in SINGLE_NOTE_LOADERS if name not in available]
+    assert not gone, (
+        f"SINGLE_NOTE_LOADERS names {gone} that no longer exist — they were "
+        f"renamed or moved. Update the list, or the pull check silently stops "
+        f"covering whatever used them."
+    )

tests/test_services_dedup.py

@@ -110,3 +110,175 @@ def test_duplicate_response_shape():
     assert r["match"] == "title"
     assert "force=true" in r["message"]
     assert "update_task" in r["message"]
+
+
+# --- snippet structural identity (#2518) -------------------------------------
+#
+# The gate used to compare a snippet's rendered DOCUMENT, which is mostly prose
+# about the code. Measured on the button corpus, that failed in both directions
+# at once: two deliberately-parallel variants were refused at 0.92, while a
+# verbatim re-record of one snippet under a different name scored below 0.90 and
+# was created. These tests pin the structural signals that replaced it.
+
+
+def _session_sequence(results):
+    """A mocked async_session() whose successive execute() calls yield `results`.
+
+    The single-result helper above can't express this: the structural check runs
+    a location query and then a code query, and the whole point is that they
+    answer differently.
+    """
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+    wrapped = []
+    for note in results:
+        r = MagicMock()
+        r.scalars.return_value.first.return_value = note
+        wrapped.append(r)
+    s.execute = AsyncMock(side_effect=wrapped)
+    return s
+
+
+@pytest.mark.asyncio
+async def test_same_location_is_a_duplicate_however_it_is_described():
+    """The measured false NEGATIVE: identical code at an identical
+    repo·path·symbol was created because the prose around it differed."""
+    existing = _fake_note(id=30, title=".btn-primary — a page's main action",
+                          note_type="snippet")
+    sem = AsyncMock()
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_sequence([None, existing])), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "primaryButton — something else entirely", body="x" * 400,
+            project_id=2, is_task=False, note_type="snippet",
+            code=".btn-primary { color: red; }",
+            locations=[{"repo": "Scribe", "path": "a/b.css", "symbol": ".btn-primary"}],
+        )
+    assert dup is not None
+    assert dup.reason == "location"
+    assert dup.similarity == 1.0
+    # Structural identity is certain, so it must not be diluted by asking the
+    # embedder for a second opinion.
+    sem.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_identical_code_is_a_duplicate_at_a_different_location():
+    existing = _fake_note(id=31, title="group_pairs", note_type="snippet")
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_sequence([None, None, existing])), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes",
+               AsyncMock(return_value=[])):
+        dup = await find_duplicate_note(
+            7, "unionFind", body="x" * 400, project_id=2, is_task=False,
+            note_type="snippet", code="def f():\n    return 1",
+            locations=[{"repo": "Scribe", "path": "z.py", "symbol": "f"}],
+        )
+    assert dup is not None
+    assert dup.reason == "code"
+
+
+@pytest.mark.asyncio
+async def test_a_location_without_a_symbol_is_not_an_identity():
+    """A path alone is a DIRECTORY of artefacts. Matching on it would refuse
+    every second snippet recorded from one file — which is exactly the corpus
+    the button recipes form."""
+    session = _session_sequence([None])
+    sem = AsyncMock(return_value=[])
+    with patch("scribe.services.dedup.async_session", return_value=session), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "Some recipe", body="x" * 400, project_id=2, is_task=False,
+            note_type="snippet", code="",
+            locations=[{"repo": "Scribe", "path": "a/b.css", "symbol": ""}],
+        )
+    assert dup is None
+    # Exactly one query — the title check. With no symbol and no code there is
+    # nothing to match structurally, and the sequence above would raise
+    # StopIteration if a second query were issued.
+    assert session.execute.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_snippets_use_the_raised_semantic_threshold():
+    """Variants of one component legitimately reach 0.92. The semantic arm has
+    to sit above that band or it refuses the corpus it exists to protect."""
+    from scribe.services.dedup import (
+        _SEMANTIC_THRESHOLD,
+        _SNIPPET_SEMANTIC_THRESHOLD,
+    )
+    sem = AsyncMock(return_value=[])
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_sequence([None, None, None])), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        await find_duplicate_note(
+            7, "A recipe", body="x" * 400, project_id=2, is_task=False,
+            note_type="snippet", code="x",
+            locations=[{"repo": "R", "path": "p", "symbol": "s"}],
+        )
+    assert sem.await_args.kwargs["threshold"] == _SNIPPET_SEMANTIC_THRESHOLD
+    assert _SNIPPET_SEMANTIC_THRESHOLD > 0.92, (
+        "the observed sibling band tops out at 0.92 (.btn-danger vs "
+        ".btn-danger-outline); a threshold at or below it blocks legitimate "
+        "variants again"
+    )
+    assert _SNIPPET_SEMANTIC_THRESHOLD > _SEMANTIC_THRESHOLD
+
+
+def test_sibling_variants_are_not_reported_as_merge_candidates():
+    """The measured false POSITIVE: eight button recipes, every direct pair over
+    the floor, proposed as ONE merge set."""
+    from scribe.services.dedup import _drop_sibling_pairs
+
+    records = {
+        1: {"locations": [{"repo": "S", "path": "c.css", "symbol": ".btn-primary"}],
+            "code_sha": "aaa"},
+        2: {"locations": [{"repo": "S", "path": "c.css", "symbol": ".btn-secondary"}],
+            "code_sha": "bbb"},
+    }
+    assert _drop_sibling_pairs([(1, 2, 0.87)], records) == []
+
+
+def test_identical_code_still_reports_even_with_different_symbols():
+    """The filter keys on "the author named these apart", but a shared code
+    fingerprint overrides that — the same code under two names IS the
+    copy-paste the report exists to surface."""
+    from scribe.services.dedup import _drop_sibling_pairs
+
+    records = {
+        1: {"locations": [{"repo": "S", "path": "a.py", "symbol": "debounce"}],
+            "code_sha": "same"},
+        2: {"locations": [{"repo": "S", "path": "b.py", "symbol": "useDebounced"}],
+            "code_sha": "same"},
+    }
+    assert _drop_sibling_pairs([(1, 2, 0.9)], records) == [(1, 2, 0.9)]
+
+
+def test_unnamed_snippets_still_report():
+    """A snippet with no recorded symbol made no identity claim, so the filter
+    must not protect it — re-recording without a location is a common way to
+    duplicate."""
+    from scribe.services.dedup import _drop_sibling_pairs
+
+    records = {1: {"code_sha": "aaa"}, 2: {"code_sha": "bbb"}}
+    assert _drop_sibling_pairs([(1, 2, 0.9)], records) == [(1, 2, 0.9)]
+
+
+def test_duplicate_response_names_what_matched_for_structural_hits():
+    """A structural hit is certain, so the message must not hedge with
+    "similar" — and it points at merge, which is what two records of one
+    artefact actually need."""
+    r = duplicate_response(
+        DuplicateMatch(id=9, title=".btn-primary", similarity=1.0, reason="location"),
+        "snippet",
+    )
+    assert "repo · path · symbol" in r["message"]
+    assert "merge_snippets" in r["message"]
+    assert "similar" not in r["message"]
+
+    r = duplicate_response(
+        DuplicateMatch(id=9, title="x", similarity=1.0, reason="code"), "snippet",
+    )
+    assert "identical code" in r["message"]