Merge pull request 'Edit-time record-sync nudge — the sync class (#2708)' (#112) from dev into main

frontend/src/views/SettingsView.vue

@@ -1316,7 +1316,10 @@ function formatUserDate(iso: string): string {
             Checks the file Claude is about to write or edit against your recorded
             snippets — what's already kept at that path, and what resembles the code
             being written — so a helper you already have is offered before it's
-            rewritten. Uses the ceiling above with its own threshold below, and never
+            rewritten. When the file being edited is itself a recorded snippet's
+            location, the hint instead asks Claude to update or re-verify that
+            record as part of the edit — how records stay current without any forge
+            connection. Uses the ceiling above with its own threshold below, and never
             blocks the edit. Off = prior art surfaces only on your own prompts.
           </p>
         </div>

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.30",
+  "version": "0.1.31",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/README.md

@@ -50,6 +50,9 @@ On install you'll be asked for:
   (`hooks/scribe_prior_art.sh`) → `GET /api/plugin/prior-art`. Returns
   `additionalContext` with **no** permission decision, so it can inform the write
   but never stop it; silent when nothing is recorded, which is most of the time.
+  Two framings: a REUSE menu (similar/nearby records), and a SYNC nudge when a
+  snippet records the exact file being edited — "updating the record is part of
+  the edit" — each with its own once-per-session dedup.
   Toggle in **Settings → Knowledge auto-inject**.
 - `skills/` → the universal process-skills, surfaced by description match.
 - `hooks/scribe_sync_processes.sh` (a 2nd SessionStart hook) + the `/scribe:sync`

plugin/hooks/scribe_prior_art.sh

@@ -8,6 +8,12 @@
 # that path or in its directory, plus snippets resembling the code about to be
 # written. Titles + ids only, never bodies.
 #
+# The answer comes in two framings (#2708). A snippet recorded AT the exact
+# file being edited is the SYNC class — "you are editing the recorded file;
+# updating the record is part of the edit" — which is how records stay current
+# on an instance with no forge connection (decision #2707). Everything else is
+# the REUSE menu. The two dedup separately (see the state files below).
+#
 # NEVER BLOCKS. It returns `additionalContext` with no `permissionDecision`, so
 # the write proceeds untouched and Claude sees the note beside the tool result.
 # Any failure — unconfigured, unreachable, malformed — exits 0 in silence. A
@@ -194,31 +200,51 @@ fi
 # surface shows a given snippet at most once per session, but they don't silence
 # each other: a title that flew past in a prompt menu twenty turns ago is
 # exactly what should reappear at the moment the duplicate is being written.
+#
+# TWO channels, not one (#2708). The server answers in two classes — REUSE
+# ("something similar/nearby is recorded") and SYNC ("a snippet records the
+# exact file being edited — updating the record is part of the edit"). They
+# dedup separately: a reuse hint shown early in the session must not suppress
+# the sync nudge when the recorded file itself is edited later.
 state_dir="${TMPDIR:-/tmp}/scribe-priorart"
 mkdir -p "$state_dir" 2>/dev/null || true
 idfile=""
+syncfile=""
 exclude_q=""
+sync_exclude_q=""
 if [ -n "$session_id" ]; then
   safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
   idfile="$state_dir/${safe_sid}.ids"
+  syncfile="$state_dir/${safe_sid}.sync.ids"
   if [ -f "$idfile" ]; then
     seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
     [ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
   fi
+  if [ -f "$syncfile" ]; then
+    sync_seen=$(tr '\n' ',' < "$syncfile" 2>/dev/null | sed 's/,$//')
+    [ -n "$sync_seen" ] && sync_exclude_q="&exclude_sync_ids=${sync_seen}"
+  fi
 fi
 
 # `|| true`, not `|| exit 0`: an unreachable instance must not discard a local
 # finding that needed no instance to produce.
 body=$(curl -fsS --max-time 5 \
   -H "Authorization: Bearer ${token}" \
-  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}" 2>/dev/null) || body=""
+  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}" 2>/dev/null) || body=""
 
 context=""
 if [ -n "$body" ]; then
   context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || context=""
-  # Remember what was surfaced so it isn't shown again this session.
-  if [ -n "$idfile" ] && [ -n "$context" ]; then
-    printf '%s' "$body" | jq -r '.note_ids[]? // empty' 2>/dev/null >> "$idfile" || true
+  # Remember what was surfaced so it isn't shown again this session — each
+  # class into its own channel: sync ids (snippets recording the edited file)
+  # to the sync file, everything else to the reuse file.
+  if [ -n "$context" ]; then
+    if [ -n "$idfile" ]; then
+      printf '%s' "$body" | jq -r '((.note_ids // []) - (.sync_note_ids // []))[]?' 2>/dev/null >> "$idfile" || true
+    fi
+    if [ -n "$syncfile" ]; then
+      printf '%s' "$body" | jq -r '(.sync_note_ids // [])[]?' 2>/dev/null >> "$syncfile" || true
+    fi
   fi
 fi
 

plugin/skills/reusing-code/SKILL.md

@@ -41,6 +41,14 @@ through recall/auto-inject; this skill is the active reflex around that.
   it before you go any further. Either it's the helper you were about to
   duplicate — reuse it and drop yours — or it isn't, and the record needs the new
   location adding. Both are cheaper now than after the duplicate settles in.
+- **A `[records this file]` hint is a duty, not a menu.** When the hint says a
+  snippet records the very file you're editing, the record's freshness is now
+  YOUR edit's responsibility: if the edit changes the recorded shape,
+  `update_snippet(id, code=…)` with the new form as part of the same task; if
+  it doesn't, `verify_snippet(id, status="ok", commit_sha=…)` costs one call
+  and re-stamps the record as checked. Scribe never reads the repo — this
+  moment, in the session that has the context, is the only place the record
+  gets kept true.
 
 ## The first time a shape is built — record it
 

src/scribe/routes/plugin.py

@@ -120,7 +120,13 @@ async def write_path_prior_art():
                              is NOT used as the location `repo` filter — see the
                              service docstring for why those two differ.
         project_id (opt)   — explicit project scope override (ad-hoc/testing).
-        exclude_ids (opt)  — comma-separated ids already surfaced this session.
+        exclude_ids (opt)  — comma-separated REUSE-class ids already surfaced
+                             this session.
+        exclude_sync_ids (opt) — comma-separated SYNC-class ids (snippets
+                             recorded AT the edited file, #2708) already
+                             surfaced. A separate channel on purpose: a reuse
+                             hint shown early must not suppress the record-sync
+                             nudge when the recorded file is edited later.
     """
     path = (request.args.get("path") or "").strip()
     code = request.args.get("code") or ""
@@ -139,9 +145,14 @@ async def write_path_prior_art():
         int(p) for p in (request.args.get("exclude_ids") or "").split(",")
         if p.strip().isdigit()
     ]
+    exclude_sync_ids = [
+        int(p) for p in (request.args.get("exclude_sync_ids") or "").split(",")
+        if p.strip().isdigit()
+    ]
 
     result = await plugin_ctx_svc.build_write_path_hint(
-        g.user.id, path, code=code, project_id=project_id, exclude_ids=exclude_ids
+        g.user.id, path, code=code, project_id=project_id,
+        exclude_ids=exclude_ids, exclude_sync_ids=exclude_sync_ids,
     )
     return jsonify(result)
 

src/scribe/services/plugin_context.py

@@ -702,6 +702,7 @@ async def build_write_path_hint(
     code: str = "",
     project_id: int = 0,
     exclude_ids: list[int] | None = None,
+    exclude_sync_ids: list[int] | None = None,
 ) -> dict:
     """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
 
@@ -709,11 +710,24 @@ async def build_write_path_hint(
     convention snippet locations are recorded in. `code` is what's about to be
     written, used only as the semantic query.
 
-    Carries auto-inject's anti-bloat gates (margin, session dedup via
-    `exclude_ids`, titles-never-bodies) plus the shared top-k cap across BOTH
-    arms — so a file with a lot of recorded history can't turn one edit into a
-    wall of text. Two gates are its OWN, because code is not prose: a stricter
-    similarity threshold, and a minimum-substance floor on `code` below which the
+    A hit recorded AT this exact path is not a reuse suggestion — it IS the
+    record of the file being changed, so it renders as the SYNC class (#2708):
+    "this snippet records the file you're editing; if the edit changes the
+    recorded shape, updating the record is part of the edit." That is the
+    operator's chosen alternative to server-side drift flagging (decision
+    #2707): the record gets corrected in the session that has the context,
+    at the moment of change. Nearby and semantic hits stay the REUSE menu.
+
+    The two classes dedup on SEPARATE channels — `exclude_ids` (reuse) and
+    `exclude_sync_ids` (sync) — because they answer different questions: a
+    title shown as "consider reusing this" twenty turns ago must not silence
+    "you are editing the recorded file right now" (#2708).
+
+    Carries auto-inject's anti-bloat gates (margin, session dedup,
+    titles-never-bodies) plus the shared top-k cap across ALL arms — so a file
+    with a lot of recorded history can't turn one edit into a wall of text. Two
+    gates are its OWN, because code is not prose: a stricter similarity
+    threshold, and a minimum-substance floor on `code` below which the
     semantic arm doesn't run at all (#2223 — see WRITEPATH_DEFAULT_THRESHOLD and
     WRITEPATH_MIN_CODE_CHARS). Returns empty context when disabled, when there's
     no path, or when nothing is recorded — which is the common case, and the point.
@@ -724,28 +738,34 @@ async def build_write_path_hint(
     free-text label the operator typed ("Scribe"), not a remote URL, and matching
     one against the other would silently return nothing.
 
-    Returns {"context": str, "note_ids": list[int], "config": dict}. The semantic
-    arm is logged to retrieval_logs as source='write_path' — its own source, so
-    its precision is tunable separately from auto-inject's.
+    Returns {"context": str, "note_ids": list[int], "sync_note_ids": list[int],
+    "config": dict} — `sync_note_ids` is the subset of `note_ids` shown as the
+    sync class, so the hook can feed each dedup channel its own ids. The
+    semantic arm is logged to retrieval_logs as source='write_path' — its own
+    source, so its precision is tunable separately from auto-inject's.
 
     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.
+    un-scored surfacing now has its own home: every arm emits note_usage_events,
+    tagged 'write_path_sync' vs 'write_path_place' vs 'write_path_semantic', so
+    each claim's pull-through rate is measurable on its own.
     """
     cfg = await get_writepath_config(user_id)
-    empty = {"context": "", "note_ids": [], "config": cfg}
+    empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg}
     path = (path or "").strip()
     if not cfg["enabled"] or not path:
         return empty
 
     top_k = cfg["top_k"]
     excluded = set(exclude_ids or [])
+    sync_excluded = set(exclude_sync_ids or [])
     scope_project = project_id or None
 
-    # --- arm 1: by place ---
+    # --- the sync class, and arm 1 by place ---
+    # `path` matches exact-or-under (see knowledge.py), and nothing sits under
+    # a FILE path — so the file query returns precisely the snippets recorded
+    # AT this path: the sync class. The directory query is the reuse-shaped
+    # "nearby" arm, unchanged.
     here: list[dict] = []
     nearby: list[dict] = []
     try:
@@ -760,19 +780,29 @@ async def build_write_path_hint(
     except Exception:
         logger.warning("Write-path location lookup failed", exc_info=True)
 
+    # Sync hits dedup ONLY against their own channel — reuse-`excluded` ids
+    # stay eligible here, which is the whole point of the split. Either way
+    # they join `seen`, so the reuse arms (where the directory query would
+    # surface them again) never re-list a record the sync block owns.
     seen: set[int] = set(excluded)
+    synced: list[dict] = []
+    for item in here:
+        nid = int(item["id"])
+        seen.add(nid)
+        if nid not in sync_excluded and len(synced) < top_k:
+            synced.append(item)
+
     placed: list[tuple[str, dict]] = []
-    for marker, items in (("here", here), ("nearby", nearby)):
-        for item in items:
-            nid = int(item["id"])
-            if nid in seen:
-                continue
-            seen.add(nid)
-            placed.append((marker, item))
+    for item in nearby:
+        nid = int(item["id"])
+        if nid in seen:
+            continue
+        seen.add(nid)
+        placed.append(("nearby", item))
 
     # --- arm 2: by meaning ---
     scored: list[tuple[str, dict]] = []
-    remaining = top_k - len(placed)
+    remaining = top_k - len(synced) - len(placed)
     query = (code or "").strip()
     # Drop payloads too small to carry meaning before spending an embedding on
     # them — a one-line Edit is not a helper being rewritten, and its embedding
@@ -848,31 +878,54 @@ async def build_write_path_hint(
                     },
                 ))
 
-    menu = (placed + scored)[:top_k]
-    if not menu:
+    menu = (placed + scored)[:max(0, top_k - len(synced))]
+    if not synced and not menu:
         return empty
 
     owners = await owner_names_for({
-        int(it["user_id"]) for _m, it in menu
+        int(it["user_id"]) for it in synced + [it for _m, it in menu]
         if it.get("user_id") is not None and int(it["user_id"]) != user_id
     })
 
+    def _owner_of(item: dict) -> str | None:
+        owner_id = item.get("user_id")
+        if owner_id is None or int(owner_id) == user_id:
+            return None
+        return owners.get(int(owner_id)) or "another user"
+
     target_lang = _language_for_path(path)
     rendered: list[tuple[dict, str, str | None, str]] = []
     for marker, item in menu:
-        owner_id = item.get("user_id")
-        owner = None
-        if owner_id is not None and int(owner_id) != user_id:
-            owner = owners.get(int(owner_id)) or "another user"
-        rendered.append((item, marker, owner, _foreign_language(item, target_lang)))
+        rendered.append((item, marker, _owner_of(item), _foreign_language(item, target_lang)))
 
-    lines = [
-        f"> Prior art already recorded in Scribe for `{path}` — open one with "
-        "`get_snippet(id)` for a snippet, `get_task(id)` for an issue, "
-        "`get_note(id)` otherwise. Reuse a snippet rather than writing a fresh "
-        "one-off; read an issue before repeating what it records "
-        "(titles only; shown once per session):",
-    ]
+    lines: list[str] = []
+    sync_note_ids: list[int] = []
+    if synced:
+        # The sync framing (#2708). Deliberately imperative about the record —
+        # the reuse claim ("start from this shape") is still implied by the
+        # title being right there, but the load-bearing sentence is the one no
+        # other surface says: keeping the record true is part of THIS edit.
+        lines.append(
+            f"> Recorded in Scribe AT `{path}` — the snippet(s) below record "
+            "the file this edit is changing. Reuse/extend the recorded shape "
+            "rather than writing a parallel one; and if this edit changes what "
+            "a record captures, updating it is part of the edit: "
+            "`update_snippet(id, code=…)` with the new shape, or "
+            "`verify_snippet(id, status=\"ok\", commit_sha=…)` after confirming "
+            "it still holds. Open with `get_snippet(id)` (shown once per session):"
+        )
+        for item in synced:
+            sync_note_ids.append(int(item["id"]))
+            lines.append(_prior_art_line(item, "records this file", _owner_of(item)))
+
+    if menu:
+        lines.append(
+            f"> Prior art already recorded in Scribe for `{path}` — open one with "
+            "`get_snippet(id)` for a snippet, `get_task(id)` for an issue, "
+            "`get_note(id)` otherwise. Reuse a snippet rather than writing a fresh "
+            "one-off; read an issue before repeating what it records "
+            "(titles only; shown once per session):"
+        )
     # Say what a language tag MEANS, and only when one is actually on the menu.
     # Without this the reader has to infer why "· python" is attached to a hit on
     # a .ts file, and the two ways of guessing wrong are both bad: dismiss it as
@@ -885,7 +938,7 @@ async def build_write_path_hint(
             "shape of a solution to adapt, not code to copy."
         )
 
-    note_ids: list[int] = []
+    note_ids: list[int] = list(sync_note_ids)
     for item, marker, owner, foreign_lang in rendered:
         note_ids.append(int(item["id"]))
         lines.append(_prior_art_line(item, marker, owner, foreign_lang))
@@ -894,15 +947,24 @@ async def build_write_path_hint(
     # 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.
+    # helper in this exact file") the one arm nobody could measure. The sync
+    # class gets its own tag: its pull-through rate is the number that says
+    # whether edit-time record-sync actually happens (#2708's success measure).
     by_arm: dict[str, list[int]] = {}
+    if sync_note_ids:
+        by_arm["write_path_sync"] = list(sync_note_ids)
     for marker, item in menu:
-        arm = "write_path_place" if marker in ("here", "nearby") else "write_path_semantic"
+        arm = "write_path_place" if marker == "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}
+    return {
+        "context": "\n".join(lines),
+        "note_ids": note_ids,
+        "sync_note_ids": sync_note_ids,
+        "config": cfg,
+    }
 
 
 async def _topic_titles(topic_ids: set[int]) -> dict[int, str]:

tests/test_note_usage.py

@@ -156,16 +156,25 @@ async def test_no_ids_short_circuits_without_a_query():
 
 
 @pytest.mark.parametrize(
-    "marker, expected_source",
-    [("here", "write_path_place"), ("nearby", "write_path_place")],
+    "lookups, expected_source",
+    [
+        # A hit AT the exact file is the sync class (#2708); a hit from the
+        # directory query is the reuse-shaped place arm. Both are un-scored,
+        # and both must leave a usage trace under their own name.
+        ([(1, "hit"), (2, "empty")], "write_path_sync"),
+        ([(1, "empty"), (2, "hit")], "write_path_place"),
+    ],
+    ids=["at-the-file", "nearby"],
 )
-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."""
+async def test_unscored_location_arms_are_recorded(lookups, expected_source):
+    """The location arms carry no score, so they have no home in retrieval_logs
+    — they surfaced snippets while leaving no trace anywhere. That was the
+    blocker #2082 recorded against this task; this is the assertion that it's
+    closed, per class."""
     from scribe.services import plugin_context
 
     here = [{"id": 42, "title": "helper", "user_id": 1, "note_type": "snippet"}]
+    responses = [(here, 1) if kind == "hit" else ([], 0) for _n, kind in lookups]
     with (
         patch.object(
             plugin_context,
@@ -175,7 +184,7 @@ async def test_write_path_place_arm_is_recorded(marker, expected_source):
         patch.object(
             plugin_context.snippets_svc,
             "list_snippets",
-            AsyncMock(side_effect=[(here, 1), ([], 0)]),
+            AsyncMock(side_effect=responses),
         ),
         patch.object(
             plugin_context, "semantic_search_notes", AsyncMock(return_value=[])

tests/test_write_path_trigger.py

@@ -101,12 +101,14 @@ async def test_nothing_recorded_is_silent():
     assert out["context"] == "" and out["note_ids"] == []
 
 
-# --- arm 1: by place ---------------------------------------------------------
+# --- the sync class, and arm 1 by place --------------------------------------
 
 @pytest.mark.asyncio
-async def test_snippet_at_this_path_is_surfaced_without_a_score():
-    """A snippet recorded HERE is prior art by definition, not by resemblance —
-    it must not be subject to the similarity threshold."""
+async def test_snippet_at_this_path_becomes_the_sync_nudge():
+    """A snippet recorded AT the exact file is not a reuse suggestion — it IS
+    the record of the file being edited (#2708). It surfaces without a score,
+    framed as "updating the record is part of this edit", and is reported in
+    sync_note_ids so the hook feeds the sync dedup channel."""
     from scribe.services import plugin_context as pc
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
          patch.object(pc.snippets_svc, "list_snippets",
@@ -116,11 +118,59 @@ async def test_snippet_at_this_path_is_surfaced_without_a_score():
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
         out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     assert out["note_ids"] == [12]
-    assert '#12 [here] "debounce — rate-limit a callback"' in out["context"]
-    assert "get_snippet(id)" in out["context"]
+    assert out["sync_note_ids"] == [12]
+    assert '#12 [records this file] "debounce — rate-limit a callback"' in out["context"]
+    # The load-bearing sentence: the record's freshness belongs to this edit.
+    assert "update_snippet" in out["context"]
+    assert "verify_snippet" in out["context"]
     assert "src/x.py" in out["context"]
 
 
+@pytest.mark.asyncio
+async def test_reuse_dedup_never_suppresses_the_sync_nudge():
+    """THE bug #2708 names: a title surfaced as a reuse hint twenty turns ago
+    landed in exclude_ids — and then silenced "you are editing the recorded
+    file right now", the one claim that must fire at the moment of change. The
+    sync class dedups only against its own channel."""
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets",
+                      AsyncMock(return_value=([_snippet_item(12, "seen as reuse already")], 1))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE, exclude_ids=[12],
+        )
+    assert out["sync_note_ids"] == [12]
+    assert "[records this file]" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_sync_dedup_silences_a_sync_nudge_already_shown():
+    """Once the session has been told "snippet #12 records this file", every
+    further edit of the same file stays quiet about it — on every arm: the
+    directory query and the semantic arm would both re-surface it otherwise."""
+    from scribe.services import plugin_context as pc
+
+    async def _listing(uid, **kw):
+        # Recorded at the exact file, so it matches the file AND dir queries.
+        return ([_snippet_item(12, "already nudged")], 1)
+
+    search = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", _listing), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()):
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE, exclude_sync_ids=[12],
+        )
+    assert out["note_ids"] == [] and out["sync_note_ids"] == []
+    assert "#12" not in out["context"]
+    assert 12 in search.await_args.kwargs["exclude_ids"]
+
+
 @pytest.mark.asyncio
 async def test_directory_is_only_consulted_when_the_file_leaves_room():
     from scribe.services import plugin_context as pc
@@ -267,16 +317,24 @@ async def test_semantic_arm_only_asks_for_the_budget_the_place_arm_left():
 
 
 @pytest.mark.asyncio
-async def test_session_dedup_excludes_ids_from_both_arms():
+async def test_session_dedup_excludes_ids_from_the_reuse_arms():
+    """exclude_ids governs the REUSE classes — nearby and semantic. (The sync
+    class has its own channel; see the tests above.)"""
     from scribe.services import plugin_context as pc
+
+    async def _listing(uid, **kw):
+        # Recorded at a SIBLING file, so it reaches only the directory query.
+        if kw["path"] == "src":
+            return ([_snippet_item(12, "already shown")], 1)
+        return ([], 0)
+
     search = AsyncMock(return_value=[])
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
-         patch.object(pc.snippets_svc, "list_snippets",
-                      AsyncMock(return_value=([_snippet_item(12, "already shown")], 1))), \
+         patch.object(pc.snippets_svc, "list_snippets", _listing), \
          patch.object(pc, "semantic_search_notes", search), \
          patch.object(pc, "record_retrieval", MagicMock()):
         out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, exclude_ids=[12])
-    # The location hit was already surfaced this session → dropped, not repeated.
+    # The nearby hit was already surfaced this session → dropped, not repeated.
     assert out["note_ids"] == []
     assert 12 in search.await_args.kwargs["exclude_ids"]
 
@@ -315,6 +373,33 @@ async def test_telemetry_uses_its_own_source():
     assert rec.call_args.kwargs["project_id"] == 4
 
 
+@pytest.mark.asyncio
+async def test_sync_surfacing_is_measured_under_its_own_usage_source():
+    """The sync class's pull-through rate is #2708's scoreboard — whether
+    sessions actually update the record when told they're editing it. Folding
+    it into write_path_place would make that unmeasurable."""
+    from scribe.services import plugin_context as pc
+
+    async def _listing(uid, **kw):
+        if kw["path"] == "src/x.py":
+            return ([_snippet_item(12, "records me")], 1)
+        return ([_snippet_item(9, "sibling")], 1)
+
+    surfaced = MagicMock()
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", _listing), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "record_surfaced", surfaced), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
+    by_source = {
+        c.kwargs["source"]: c.kwargs["note_ids"] for c in surfaced.call_args_list
+    }
+    assert by_source["write_path_sync"] == [12]
+    assert by_source["write_path_place"] == [9]
+
+
 @pytest.mark.asyncio
 async def test_config_has_its_own_switch_and_threshold_but_shares_top_k():
     from scribe.services import plugin_context as pc
@@ -693,11 +778,11 @@ def test_route_reads_every_arg_the_hook_sends():
 
     from scribe.routes import plugin as routes
     src = inspect.getsource(routes.write_path_prior_art)
-    for arg in ("path", "code", "repo", "project_id", "exclude_ids"):
+    for arg in ("path", "code", "repo", "project_id", "exclude_ids", "exclude_sync_ids"):
         assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"
 
     hook = HOOK.read_text()
-    for arg in ("path=", "code=", "repo=", "exclude_ids="):
+    for arg in ("path=", "code=", "repo=", "exclude_ids=", "exclude_sync_ids="):
         assert arg in hook, f"hook never sends {arg}"
 
 
@@ -788,7 +873,21 @@ def test_plugin_version_bumped_with_the_hook():
     ship."""
     manifest = json.loads((PLUGIN / ".claude-plugin" / "plugin.json").read_text())
     version = tuple(int(p) for p in manifest["version"].split("."))
-    assert version >= (0, 1, 18)
+    assert version >= (0, 1, 31)
+
+
+def test_hook_keeps_sync_and_reuse_dedup_apart():
+    """#2708's dedup audit, pinned: the hook holds TWO per-session id files and
+    feeds each its own class — sync ids (snippets recording the edited file) to
+    the sync file, the rest to the reuse file. One shared file is exactly the
+    bug this replaced: a reuse hint early in the session silencing the record-
+    sync nudge when the recorded file is edited later."""
+    src = HOOK.read_text()
+    assert ".sync.ids" in src                       # its own state file
+    assert "exclude_sync_ids=" in src               # its own query channel
+    # The reuse file must NOT swallow sync ids — the write-back subtracts them.
+    assert "(.note_ids // []) - (.sync_note_ids // [])" in src
+    assert "(.sync_note_ids // [])[]?" in src
 
 
 def _hook_runtime_env():