feat(snippets): usage signal — was a surfaced record ever actually pulled?

retrieval_logs answers "what did the ranker return, at what scores" — the
right substrate for tuning a threshold. It cannot answer the question the
snippet corpus actually needs: did anyone open this? A snippet nobody
opens is not neutral. It takes a slot in every future auto-inject menu and
crowds out something useful.

Adds note_usage_events (migration 0071): one row per note per event,
either 'surfaced' (we put its title in front of an agent) or 'pulled'
(someone opened it in full), tagged with which surface produced it.

Closes the gap #2082 recorded against this work. The write-path PLACE arm
carries no score, so it has no home in retrieval_logs — folding it in
would corrupt the score distribution that table exists to capture. The
result was that the arm firing on the STRONGEST claim ("there is already a
canonical helper in this exact file") was the one arm nobody could
measure. Both arms now emit usage events under distinct sources, so their
pull-through rates are finally comparable.

Deliberate departures from the task as written:

- 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 the
  hooks send no session id — and adding one would mean threading an
  opaque client-supplied token through every read path. Two independent
  counters answer the question without it: surfaced 40×, pulled 0 is dead
  weight regardless of how those events distribute across sessions.

- Pulls record at the ENTRY POINTS (MCP tools, REST detail route), not in
  snippets_svc.get_snippet, which update and merge also reach. Counting
  those would inflate precisely the number meant to say "someone chose to
  look at this."

- get_note records for every note kind, not just snippets. The auto-inject
  menu surfaces tasks and processes too; scoping this to snippets would
  pin those at zero pulls forever and make them read as dead weight next
  to snippets that merely had a counter.

Surfaced in the Snippets list as an "N/M used" badge, warning-toned once a
record has been offered 3+ times and never opened, with the tooltip saying
what to do about it (usually: its "when to reach for it" doesn't say
when). No badge at all below one surfacing — "0/0" reads as a verdict when
it's an absence of evidence. Also returned from MCP list_snippets so the
agent can see dead weight without opening the UI.

Telemetry keeps the retrieval_telemetry contract throughout: writes are
fire-and-forget, reads degrade to zeroes, and no path can raise into the
surface it observes.

Refs #2085

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

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")

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

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;

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
 
 

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
 
 

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

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,
+        }

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)
 
 

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

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}
 
 

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"