diff --git a/alembic/versions/0071_note_usage_events.py b/alembic/versions/0071_note_usage_events.py
new file mode 100644
index 0000000..c9ac3cb
--- /dev/null
+++ b/alembic/versions/0071_note_usage_events.py
@@ -0,0 +1,72 @@
+"""add note_usage_events — did anyone actually open what we surfaced?
+
+Revision ID: 0071
+Revises: 0070
+Create Date: 2026-07-28
+
+`retrieval_logs` records what the ranker returned and with what scores, which is
+the right substrate for tuning a similarity threshold. It cannot answer the
+different question the snippet corpus needs: was a surfaced snippet ever pulled
+in full? A snippet nobody opens still competes for the injection budget on every
+turn, so the surfaced:pulled ratio is what makes dead weight visible.
+
+Two reasons this is its own table rather than columns on `notes` or rows in
+`retrieval_logs`:
+
+  - Counters on `notes` would answer "how many" but not "when, from where, and
+    by which arm" — and the place arm vs semantic arm comparison is precisely
+    what was missing (the write-path place arm surfaced snippets while leaving
+    no trace anywhere).
+  - Folding un-scored surfacing into `retrieval_logs` would corrupt the score
+    distribution that table exists to capture. Location hits have no score.
+
+Grain is one row per note per event, which is what the per-snippet readout needs
+and what `retrieval_logs.result_ids` (a JSONB array, one row per *call*) cannot
+be indexed at.
+
+FK-free on note_id and user_id, matching retrieval_logs and app_logs: telemetry
+should outlive what it describes. Deleting a note must not erase the evidence
+that it was surfaced forty times and opened none.
+
+Downgrade drops the table outright. The data is purely observational — nothing
+reads it for correctness, so losing it costs history and no behavior.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0071"
+down_revision = "0070"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "note_usage_events",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("note_id", sa.Integer(), nullable=False),
+        sa.Column("event", sa.Text(), nullable=False),
+        sa.Column("source", sa.Text(), nullable=False),
+    )
+    # Every readout is "these note ids, split by event", so the composite is the
+    # one that actually gets used; the others serve pruning and per-user views.
+    op.create_index(
+        "ix_note_usage_note_event", "note_usage_events", ["note_id", "event"]
+    )
+    op.create_index("ix_note_usage_created_at", "note_usage_events", ["created_at"])
+    op.create_index("ix_note_usage_user_id", "note_usage_events", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_note_usage_user_id", table_name="note_usage_events")
+    op.drop_index("ix_note_usage_created_at", table_name="note_usage_events")
+    op.drop_index("ix_note_usage_note_event", table_name="note_usage_events")
+    op.drop_table("note_usage_events")
diff --git a/frontend/src/api/snippets.ts b/frontend/src/api/snippets.ts
index fa7ef03..e912d4c 100644
--- a/frontend/src/api/snippets.ts
+++ b/frontend/src/api/snippets.ts
@@ -46,6 +46,16 @@ export interface Snippet {
   owner?: string | null;
 }
 
+/** How often a record was put in front of an agent versus actually opened.
+ *  A high `surfaced_count` with `pull_count: 0` is dead weight — it occupies a
+ *  slot in every future auto-inject menu while never being used. */
+export interface SnippetUsage {
+  surfaced_count: number;
+  pull_count: number;
+  last_surfaced_at: string | null;
+  last_pulled_at: string | null;
+}
+
 /** Lightweight list item from the knowledge preview feed. Note: the `snippet`
  *  field here is a truncated *body preview* (the knowledge feed's naming), not
  *  the parsed fields above. */
@@ -61,6 +71,8 @@ export interface SnippetListItem {
    *  them, not one of your own. Absent means it's yours. */
   shared?: boolean;
   owner?: string | null;
+  /** Always present from the backend, zero-filled for records with no events. */
+  usage?: SnippetUsage;
 }
 
 /** Create/update payload — discrete fields the backend serializes into the
diff --git a/frontend/src/views/SnippetListView.vue b/frontend/src/views/SnippetListView.vue
index 8fbed78..2af502f 100644
--- a/frontend/src/views/SnippetListView.vue
+++ b/frontend/src/views/SnippetListView.vue
@@ -126,6 +126,40 @@ function splitTitle(title: string): { name: string; when: string } {
 function languageOf(tags: string[]): string {
   return tags.find((t) => t && t !== "snippet") ?? "";
 }
+
+/** A snippet that has been offered repeatedly and never opened. The threshold
+ *  is 3 rather than 1 because one or two surfacings is noise — the record may
+ *  simply not have come up in a relevant context yet. */
+function isDeadWeight(s: SnippetListItem): boolean {
+  const u = s.usage;
+  return !!u && u.pull_count === 0 && u.surfaced_count >= 3;
+}
+
+/** Short badge text, or "" to render nothing. A record nobody has surfaced yet
+ *  gets no badge at all: "0 / 0" would read as a verdict when it's an absence
+ *  of evidence. */
+function usageBadge(s: SnippetListItem): string {
+  const u = s.usage;
+  if (!u || u.surfaced_count === 0) return "";
+  return `${u.pull_count}/${u.surfaced_count} used`;
+}
+
+function usageTitle(s: SnippetListItem): string {
+  const u = s.usage;
+  if (!u) return "";
+  const last = u.last_pulled_at
+    ? `Last opened ${new Date(u.last_pulled_at).toLocaleDateString()}.`
+    : "Never opened.";
+  const verdict = isDeadWeight(s)
+    ? " Offered repeatedly without ever being opened — consider rewriting its" +
+      " “when to reach for it” so it says when, or deleting it. It takes a slot" +
+      " in every future auto-inject menu."
+    : "";
+  return (
+    `Surfaced to an agent ${u.surfaced_count}×, opened in full ` +
+    `${u.pull_count}×. ${last}${verdict}`
+  );
+}
 </script>
 
 <template>
@@ -261,6 +295,14 @@ function languageOf(tags: string[]): string {
         </p>
         <div class="card-footer">
           <span class="meta-date">Updated {{ new Date(s.updated_at).toLocaleDateString() }}</span>
+          <span
+            v-if="usageBadge(s)"
+            class="usage-tag"
+            :class="{ 'usage-dead': isDeadWeight(s) }"
+            :title="usageTitle(s)"
+          >
+            {{ usageBadge(s) }}
+          </span>
           <span v-if="s.shared" class="shared-tag" :title="`Shared by ${s.owner ?? 'another user'} — a suggestion, not your own record`">
             by {{ s.owner ?? "another user" }}
           </span>
@@ -576,6 +618,23 @@ function languageOf(tags: string[]): string {
   color: var(--color-text-muted);
 }
 
+.usage-tag {
+  font-size: 0.7rem;
+  padding: 0.1rem 0.4rem;
+  border-radius: 4px;
+  white-space: nowrap;
+  font-variant-numeric: tabular-nums;
+  background: color-mix(in srgb, var(--color-text-muted) 15%, transparent);
+  color: var(--color-text-muted);
+}
+
+/* Dead weight is a nudge, not an error — it warns in the warning colour rather
+   than the danger one, because the record isn't broken, just unearned. */
+.usage-tag.usage-dead {
+  background: color-mix(in srgb, var(--color-warning, #b45309) 18%, transparent);
+  color: var(--color-warning, #b45309);
+}
+
 /* Header + select-mode */
 .header-actions {
   display: flex;
diff --git a/src/scribe/mcp/tools/notes.py b/src/scribe/mcp/tools/notes.py
index 58f696a..c6dc038 100644
--- a/src/scribe/mcp/tools/notes.py
+++ b/src/scribe/mcp/tools/notes.py
@@ -19,6 +19,7 @@ from scribe.services import dedup as dedup_svc
 from scribe.services import notes as notes_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
+from scribe.services.note_usage import record_pulled
 
 
 async def list_notes(
@@ -68,6 +69,11 @@ async def get_note(note_id: int) -> dict:
         raise ValueError(f"note {note_id} not found")
     out = note.to_dict()
     out.update(await access_svc.describe_provenance(uid, note))
+    # Records the pull for ANY note kind, not just snippets: the auto-inject
+    # menu surfaces notes, tasks and processes too, so restricting this to
+    # snippets would leave those permanently at zero pulls and make them look
+    # like dead weight next to snippets that merely had a counter (#2085).
+    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_note")
     return out
 
 
diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py
index f0ec970..8824c0b 100644
--- a/src/scribe/mcp/tools/snippets.py
+++ b/src/scribe/mcp/tools/snippets.py
@@ -14,6 +14,7 @@ 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 snippets as snippets_svc
+from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
 from scribe.services import systems as systems_svc
 
 
@@ -48,8 +49,16 @@ async def list_snippets(
     snippet that lives in repo A and, separately, at path B in another repo is
     not returned for repo=A + path=B.
 
-    Returns {"snippets": [{id, title, tags, preview}], "total": int}. The title
-    reads "name — when to reach for it"; open one in full with get_snippet(id).
+    Returns {"snippets": [{id, title, tags, preview, usage}], "total": int}. The
+    title reads "name — when to reach for it"; open one in full with
+    get_snippet(id).
+
+    `usage` is {surfaced_count, pull_count, last_surfaced_at, last_pulled_at}:
+    how often the entry has been put in front of an agent versus actually
+    opened. Treat a high surfaced_count with a zero pull_count as a prompt to
+    fix the record — usually its "when to reach for it" doesn't say when — or to
+    delete it. Such an entry is not harmless: it takes a slot in every future
+    auto-inject menu and crowds out something useful.
 
     An entry marked `shared: true` with an `owner` belongs to someone else — one
     person's suggestion, not settled practice here. Weigh it on its merits and
@@ -63,10 +72,11 @@ async def list_snippets(
         project_id=project_id or None,
         repo=repo, path=path, symbol=symbol,
     )
-    return {
-        "snippets": await access_svc.label_shared_items(uid, items),
-        "total": total,
-    }
+    labeled = await access_svc.label_shared_items(uid, items)
+    usage = await usage_for_notes([int(it["id"]) for it in labeled])
+    for it in labeled:
+        it["usage"] = usage.get(int(it["id"]), empty_usage())
+    return {"snippets": labeled, "total": total}
 
 
 async def create_snippet(
@@ -166,6 +176,11 @@ async def get_snippet(snippet_id: int) -> dict:
         raise ValueError(f"snippet {snippet_id} not found")
     data = snippets_svc.snippet_to_dict(note)
     data.update(await access_svc.describe_provenance(uid, note))
+    # A "pull" is an explicit open, so it's recorded HERE rather than in
+    # snippets_svc.get_snippet — the service is also reached by update/merge
+    # paths, and counting those would inflate exactly the number that is
+    # supposed to mean "someone chose to look at this" (#2085).
+    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_snippet")
     return data
 
 
diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py
index 2d15043..2297cdd 100644
--- a/src/scribe/models/__init__.py
+++ b/src/scribe/models/__init__.py
@@ -27,6 +27,7 @@ from scribe.models.password_reset import PasswordResetToken  # noqa: E402, F401
 from scribe.models.invitation import InvitationToken  # noqa: E402, F401
 from scribe.models.embedding import NoteEmbedding  # noqa: E402, F401
 from scribe.models.retrieval_log import RetrievalLog  # noqa: E402, F401
+from scribe.models.note_usage import NoteUsageEvent  # noqa: E402, F401
 from scribe.models.project import Project  # noqa: E402, F401
 from scribe.models.milestone import Milestone  # noqa: E402, F401
 from scribe.models.task_log import TaskLog  # noqa: E402, F401
diff --git a/src/scribe/models/note_usage.py b/src/scribe/models/note_usage.py
new file mode 100644
index 0000000..cb3671c
--- /dev/null
+++ b/src/scribe/models/note_usage.py
@@ -0,0 +1,76 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, Index, Integer, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+
+SURFACED = "surfaced"
+PULLED = "pulled"
+
+
+class NoteUsageEvent(Base):
+    """One row per time a note was SURFACED to the agent, or PULLED in full.
+
+    Answers the question RetrievalLog cannot: not "what did the ranker return
+    and with what scores" but "did anyone ever actually open this?" A snippet
+    nobody opens is not neutral — it competes for the injection budget on every
+    future turn and dilutes the menu — so the surfaced:pulled ratio is what
+    makes dead weight visible and prunable.
+
+    WHY A SEPARATE TABLE FROM RetrievalLog. RetrievalLog is one row per *call*,
+    keyed on the score distribution it exists to capture; folding un-scored
+    events into it would corrupt exactly the distribution threshold tuning reads
+    (see the KNOWN GAP note this closes in plugin_context.build_write_path_hint).
+    This table is one row per *note per event*, which is the grain the usage
+    readout needs and the grain RetrievalLog's JSONB `result_ids` can't be
+    indexed at. The two are complements: RetrievalLog tunes the threshold, this
+    tunes the corpus.
+
+    WHY NOT IN-SESSION CORRELATION. The original framing was "correlate
+    result_ids against a later get_note in the same session". There is no
+    session identity server-side — the MCP endpoint is stateless and hooks send
+    no session id — and introducing one would mean threading an opaque,
+    client-supplied token through every read path for a signal that does not
+    need it. Two independent counters answer the question without it: a note
+    surfaced 40 times and never pulled is dead weight regardless of which
+    sessions those events fell in.
+
+    Deliberately FK-free on user_id and note_id (mirrors RetrievalLog/AppLog):
+    telemetry outlives the row it describes, and deleting a note should not
+    erase the evidence that it was surfaced 40 times and never once opened.
+    """
+
+    __tablename__ = "note_usage_events"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    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' |
+    # 'write_path_semantic' | 'mcp_get_snippet' | 'mcp_get_note' | 'rest_note'.
+    # Kept granular so the place arm and the semantic arm can be compared —
+    # that comparison is the whole reason the place arm needed logging at all.
+    source: Mapped[str] = mapped_column(Text, nullable=False)
+
+    __table_args__ = (
+        # The readout is always "these note ids, split by event" — a covering
+        # composite beats separate single-column indexes for it.
+        Index("ix_note_usage_note_event", "note_id", "event"),
+        Index("ix_note_usage_created_at", "created_at"),
+        Index("ix_note_usage_user_id", "user_id"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "user_id": self.user_id,
+            "note_id": self.note_id,
+            "event": self.event,
+            "source": self.source,
+        }
diff --git a/src/scribe/routes/snippets.py b/src/scribe/routes/snippets.py
index a7f5cc0..426cfb4 100644
--- a/src/scribe/routes/snippets.py
+++ b/src/scribe/routes/snippets.py
@@ -19,6 +19,7 @@ from scribe.routes.utils import not_found, parse_pagination
 from scribe.services import dedup as dedup_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services import systems as systems_svc
+from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
 from scribe.services.access import (
     can_write_note,
     describe_provenance,
@@ -69,6 +70,12 @@ async def list_snippets_route():
     # Mark rows owned by someone else so the UI can show whose they are — an
     # unmarked row in your own list reads as one you recorded and vetted.
     items = await label_shared_items(uid, items)
+    # One aggregate for the whole page — a per-row lookup here would be N+1 by
+    # construction. Every row gets the key, zero-filled, so the UI renders
+    # "never pulled" rather than having to treat a missing field as a state.
+    usage = await usage_for_notes([int(it["id"]) for it in items])
+    for it in items:
+        it["usage"] = usage.get(int(it["id"]), empty_usage())
     return jsonify({"snippets": items, "total": total})
 
 
@@ -148,6 +155,13 @@ async def get_snippet_route(snippet_id: int):
         for s in await systems_svc.list_record_systems(note.user_id, snippet_id)
     ]
     data.update(await describe_provenance(uid, note))
+    data["usage"] = (await usage_for_notes([snippet_id])).get(
+        snippet_id, empty_usage()
+    )
+    # Opening the detail view IS a pull — the operator chose to look. Tagged
+    # apart from the MCP sources so "the agent reused it" and "a human read it"
+    # stay distinguishable; they mean different things for pruning (#2085).
+    record_pulled(user_id=uid, note_id=snippet_id, source="rest_snippet")
     return jsonify(data)
 
 
diff --git a/src/scribe/services/note_usage.py b/src/scribe/services/note_usage.py
new file mode 100644
index 0000000..a268020
--- /dev/null
+++ b/src/scribe/services/note_usage.py
@@ -0,0 +1,155 @@
+"""Note usage telemetry — was a surfaced note ever actually pulled?
+
+Two event streams, deliberately independent:
+
+  - SURFACED: we put this note's title in front of the agent (auto-inject, or
+    either arm of the write-path prior-art trigger).
+  - PULLED: someone then opened it in full (get_snippet / get_note / the REST
+    detail route).
+
+The ratio between them is the signal. A snippet surfaced forty times and never
+pulled is not neutral — it occupies the injection budget on every future turn
+and dilutes the menu — so this is what makes dead weight visible and prunable.
+
+Design notes (mirrors retrieval_telemetry, for the same reasons):
+  - Writes are fire-and-forget. `record_surfaced` / `record_pulled` extract
+    plain ints synchronously and schedule the insert as a background task, so
+    telemetry never adds latency to — or can break — the surface it observes.
+  - Every failure path is swallowed. Losing a usage row costs a data point;
+    raising would cost the operator their retrieval.
+  - Reads (`usage_for_notes`) are NOT fire-and-forget — a readout the caller
+    awaits, aggregated in one round-trip for a whole page of snippets rather
+    than per row.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import func, select
+
+from scribe.models import async_session
+from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
+
+logger = logging.getLogger(__name__)
+
+
+async def _insert_events(rows: list[dict]) -> None:
+    """Persist usage rows. Best-effort: all errors are swallowed."""
+    try:
+        async with async_session() as session:
+            session.add_all([NoteUsageEvent(**row) for row in rows])
+            await session.commit()
+    except Exception:
+        logger.debug("note usage telemetry write skipped", exc_info=True)
+
+
+def _schedule(rows: list[dict]) -> None:
+    if not rows:
+        return
+    try:
+        asyncio.get_running_loop().create_task(_insert_events(rows))
+    except RuntimeError:
+        # No running loop (sync context outside the app) — skip rather than
+        # block. Every app path runs on the loop.
+        logger.debug("note usage telemetry skipped — no running event loop")
+
+
+def record_surfaced(
+    *, user_id: int | None, note_ids: list[int] | set[int], source: str
+) -> None:
+    """Fire-and-forget: record that these notes were shown to the agent.
+
+    Takes the whole menu at once — one insert per surfacing event, not per note
+    — because a menu is a single decision and its rows should land together.
+    """
+    try:
+        rows = [
+            {
+                "user_id": user_id,
+                "note_id": int(nid),
+                "event": SURFACED,
+                "source": source,
+            }
+            for nid in note_ids
+        ]
+    except Exception:
+        logger.debug("note usage payload build failed", exc_info=True)
+        return
+    _schedule(rows)
+
+
+def record_pulled(*, user_id: int | None, note_id: int, source: str) -> None:
+    """Fire-and-forget: record that a note was opened in full."""
+    try:
+        rows = [
+            {
+                "user_id": user_id,
+                "note_id": int(note_id),
+                "event": PULLED,
+                "source": source,
+            }
+        ]
+    except Exception:
+        logger.debug("note usage payload build failed", exc_info=True)
+        return
+    _schedule(rows)
+
+
+def empty_usage() -> dict:
+    """The zero readout — what a note with no recorded events looks like.
+
+    Callers render this shape unconditionally, so a note predating the table
+    reads as "never surfaced, never pulled" rather than as a missing key.
+    """
+    return {
+        "surfaced_count": 0,
+        "pull_count": 0,
+        "last_surfaced_at": None,
+        "last_pulled_at": None,
+    }
+
+
+async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
+    """Aggregate usage for a set of notes: {note_id: {counts + timestamps}}.
+
+    One GROUP BY for the whole page rather than a query per row — this feeds a
+    list view, so the per-row shape would be N+1 by construction. Notes with no
+    events are returned with `empty_usage()` so the caller never has to
+    distinguish "no events" from "not in the result".
+    """
+    ids = [int(n) for n in note_ids]
+    out: dict[int, dict] = {nid: empty_usage() for nid in ids}
+    if not ids:
+        return out
+
+    try:
+        async with async_session() as session:
+            rows = (
+                await session.execute(
+                    select(
+                        NoteUsageEvent.note_id,
+                        NoteUsageEvent.event,
+                        func.count().label("n"),
+                        func.max(NoteUsageEvent.created_at).label("last_at"),
+                    )
+                    .where(NoteUsageEvent.note_id.in_(ids))
+                    .group_by(NoteUsageEvent.note_id, NoteUsageEvent.event)
+                )
+            ).all()
+    except Exception:
+        # A telemetry readout must not be able to break the list it decorates.
+        logger.debug("note usage readout failed", exc_info=True)
+        return out
+
+    for note_id, event, n, last_at in rows:
+        slot = out.get(int(note_id))
+        if slot is None:
+            continue
+        if event == SURFACED:
+            slot["surfaced_count"] = int(n)
+            slot["last_surfaced_at"] = last_at.isoformat() if last_at else None
+        elif event == PULLED:
+            slot["pull_count"] = int(n)
+            slot["last_pulled_at"] = last_at.isoformat() if last_at else None
+    return out
diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index 3924337..c5d878f 100644
--- a/src/scribe/services/plugin_context.py
+++ b/src/scribe/services/plugin_context.py
@@ -29,6 +29,7 @@ from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services.access import label_shared_items, owner_names_for
 from scribe.services.embeddings import semantic_search_notes
+from scribe.services.note_usage import record_surfaced
 from scribe.services.retrieval_telemetry import record_retrieval
 from scribe.services.settings import get_setting
 
@@ -271,6 +272,12 @@ async def build_autoinject_hint(
             line += f" — shared by {who}, treat as a suggestion"
         lines.append(line)
 
+    # Records what SURVIVED the margin gate, not what the ranker returned — the
+    # menu the agent actually saw. retrieval_logs already holds the full
+    # candidate set for threshold tuning; conflating the two would make
+    # "surfaced" mean two different things depending on the surface (#2085).
+    record_surfaced(user_id=user_id, note_ids=note_ids, source="auto_inject")
+
     return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
 
 
@@ -341,12 +348,12 @@ async def build_write_path_hint(
     arm is logged to retrieval_logs as source='write_path' — its own source, so
     its precision is tunable separately from auto-inject's.
 
-    KNOWN GAP: only the semantic arm is logged, matching auto-inject's convention
-    of recording the candidate set for threshold tuning. Location hits carry no
-    score, so folding them in would corrupt the score distribution the log exists
-    to capture — but it does mean a snippet surfaced BY PLACE leaves no trace. The
-    usage signal (#2085) correlates retrieval_logs against later pulls, so it will
-    need a home for un-scored surfacing before it can measure this arm.
+    Location hits still carry no score and so stay out of retrieval_logs, whose
+    score distribution they would corrupt. What closed the gap (#2085) is that
+    un-scored surfacing now has its own home: BOTH arms emit note_usage_events,
+    tagged 'write_path_place' vs 'write_path_semantic', so the place arm is
+    finally measurable — and the two arms' pull-through rates are comparable,
+    which is the number that says whether place really does beat meaning here.
     """
     cfg = await get_writepath_config(user_id)
     empty = {"context": "", "note_ids": [], "config": cfg}
@@ -439,6 +446,18 @@ async def build_write_path_hint(
             owner = owners.get(int(owner_id)) or "another user"
         lines.append(_prior_art_line(item, marker, owner))
 
+    # Split by arm, which is the whole reason this table exists. The place arm
+    # carries no score and so has no home in retrieval_logs; before #2085 a
+    # snippet surfaced BY PLACE left no trace anywhere, making the arm that
+    # fires on the strongest possible claim ("there is already a canonical
+    # helper in this exact file") the one arm nobody could measure.
+    by_arm: dict[str, list[int]] = {}
+    for marker, item in menu:
+        arm = "write_path_place" if marker in ("here", "nearby") else "write_path_semantic"
+        by_arm.setdefault(arm, []).append(int(item["id"]))
+    for arm, ids in by_arm.items():
+        record_surfaced(user_id=user_id, note_ids=ids, source=arm)
+
     return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
 
 
diff --git a/tests/test_note_usage.py b/tests/test_note_usage.py
new file mode 100644
index 0000000..b70292c
--- /dev/null
+++ b/tests/test_note_usage.py
@@ -0,0 +1,185 @@
+"""Tests for the usage signal (#2085) — was a surfaced record ever pulled?
+
+Covers the payload shaping and the two contracts that make this safe to leave in
+the hot path: telemetry never raises, and telemetry never blocks. Plus the thing
+the feature exists for — that the write-path PLACE arm is now recorded, since
+before this it surfaced snippets while leaving no trace anywhere.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import note_usage
+from scribe.services.note_usage import (
+    empty_usage,
+    record_pulled,
+    record_surfaced,
+    usage_for_notes,
+)
+
+
+def _note(nid, title, user_id=1):
+    n = MagicMock()
+    n.id, n.title, n.user_id = nid, title, user_id
+    n.note_type = "snippet"
+    return n
+
+
+# --- recording ------------------------------------------------------------
+
+
+async def test_record_surfaced_writes_one_row_per_note():
+    with patch.object(note_usage, "_schedule") as sched:
+        record_surfaced(user_id=7, note_ids=[11, 12], source="auto_inject")
+    rows = sched.call_args[0][0]
+    assert [r["note_id"] for r in rows] == [11, 12]
+    assert {r["event"] for r in rows} == {"surfaced"}
+    assert {r["source"] for r in rows} == {"auto_inject"}
+    assert {r["user_id"] for r in rows} == {7}
+
+
+async def test_record_pulled_writes_a_single_row():
+    with patch.object(note_usage, "_schedule") as sched:
+        record_pulled(user_id=7, note_id=11, source="mcp_get_snippet")
+    rows = sched.call_args[0][0]
+    assert rows == [
+        {"user_id": 7, "note_id": 11, "event": "pulled", "source": "mcp_get_snippet"}
+    ]
+
+
+async def test_empty_menu_schedules_nothing():
+    """No notes surfaced is not an event — it must not cost a write."""
+    with patch.object(note_usage, "_insert_events") as ins:
+        record_surfaced(user_id=1, note_ids=[], source="auto_inject")
+    ins.assert_not_called()
+
+
+async def test_recording_never_raises_on_bad_input():
+    """Telemetry sits in the hot path of every retrieval. A malformed id must
+    cost a data point, never the operator's request."""
+    with patch.object(note_usage, "_schedule"):
+        record_surfaced(user_id=1, note_ids=["not-an-int"], source="auto_inject")
+        record_pulled(user_id=1, note_id=None, source="mcp_get_note")
+
+
+async def test_recording_without_an_event_loop_is_skipped_not_raised():
+    """Called from a sync context outside the app (a script, a test helper),
+    there is no loop to schedule on. Skip rather than blow up."""
+    with patch.object(
+        note_usage.asyncio, "get_running_loop", side_effect=RuntimeError
+    ):
+        record_pulled(user_id=1, note_id=5, source="mcp_get_note")
+
+
+# --- readout --------------------------------------------------------------
+
+
+async def test_usage_for_notes_zero_fills_every_requested_id():
+    """The caller renders this shape unconditionally, so a note with no events
+    must come back as zeroes, not as a missing key."""
+    session = MagicMock()
+    session.execute = AsyncMock(return_value=MagicMock(all=MagicMock(return_value=[])))
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=session)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    with patch.object(note_usage, "async_session", return_value=ctx):
+        out = await usage_for_notes([3, 4])
+    assert out == {3: empty_usage(), 4: empty_usage()}
+
+
+async def test_usage_for_notes_splits_counts_by_event():
+    from datetime import datetime, timezone
+
+    ts = datetime(2026, 7, 28, tzinfo=timezone.utc)
+    rows = [(3, "surfaced", 9, ts), (3, "pulled", 2, ts)]
+    session = MagicMock()
+    session.execute = AsyncMock(
+        return_value=MagicMock(all=MagicMock(return_value=rows))
+    )
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=session)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    with patch.object(note_usage, "async_session", return_value=ctx):
+        out = await usage_for_notes([3])
+    assert out[3]["surfaced_count"] == 9
+    assert out[3]["pull_count"] == 2
+    assert out[3]["last_pulled_at"] == ts.isoformat()
+
+
+async def test_usage_readout_failure_degrades_to_zeroes():
+    """A telemetry readout must not be able to break the list it decorates."""
+    with patch.object(note_usage, "async_session", side_effect=RuntimeError("boom")):
+        out = await usage_for_notes([3])
+    assert out == {3: empty_usage()}
+
+
+async def test_no_ids_short_circuits_without_a_query():
+    with patch.object(note_usage, "async_session") as sess:
+        assert await usage_for_notes([]) == {}
+    sess.assert_not_called()
+
+
+# --- the gap this closes --------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "marker, expected_source",
+    [("here", "write_path_place"), ("nearby", "write_path_place")],
+)
+async def test_write_path_place_arm_is_recorded(marker, expected_source):
+    """The place arm carries no score, so it has no home in retrieval_logs — it
+    surfaced snippets while leaving no trace anywhere. That was the blocker
+    #2082 recorded against this task; this is the assertion that it's closed."""
+    from scribe.services import plugin_context
+
+    here = [{"id": 42, "title": "helper", "user_id": 1, "note_type": "snippet"}]
+    with (
+        patch.object(
+            plugin_context,
+            "get_writepath_config",
+            AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3}),
+        ),
+        patch.object(
+            plugin_context.snippets_svc,
+            "list_snippets",
+            AsyncMock(side_effect=[(here, 1), ([], 0)]),
+        ),
+        patch.object(
+            plugin_context, "semantic_search_notes", AsyncMock(return_value=[])
+        ),
+        patch.object(plugin_context, "owner_names_for", AsyncMock(return_value={})),
+        patch.object(plugin_context, "record_retrieval"),
+        patch.object(plugin_context, "record_surfaced") as surfaced,
+    ):
+        out = await plugin_context.build_write_path_hint(
+            1, "src/a.py", code="def f(): pass"
+        )
+
+    assert out["note_ids"] == [42]
+    sources = {c.kwargs["source"] for c in surfaced.call_args_list}
+    assert expected_source in sources
+
+
+async def test_auto_inject_records_what_survived_the_margin_gate():
+    """Not what the ranker returned — retrieval_logs already holds that. These
+    two numbers must not silently mean different things per surface."""
+    from scribe.services import plugin_context
+
+    hits = [(0.90, _note(1, "kept")), (0.40, _note(2, "cut by the margin gate"))]
+    with (
+        patch.object(
+            plugin_context,
+            "get_autoinject_config",
+            AsyncMock(return_value={"enabled": True, "threshold": 0.3, "top_k": 5}),
+        ),
+        patch.object(
+            plugin_context, "semantic_search_notes", AsyncMock(return_value=hits)
+        ),
+        patch.object(plugin_context, "owner_names_for", AsyncMock(return_value={})),
+        patch.object(plugin_context, "record_retrieval"),
+        patch.object(plugin_context, "record_surfaced") as surfaced,
+    ):
+        await plugin_context.build_autoinject_hint(1, "a query")
+
+    assert surfaced.call_args.kwargs["note_ids"] == [1]
+    assert surfaced.call_args.kwargs["source"] == "auto_inject"