feat(prior-art): edit-time record-sync nudge — the sync class (#2708)

A snippet recorded AT the exact file being edited is not a reuse
suggestion — it IS the record of the file being changed. The write-path
hint now renders those as their own SYNC class: 'snippet #N records this
file — updating the record is part of the edit (update_snippet /
verify_snippet)'. Nearby and semantic hits stay the reuse menu.

The two classes dedup on separate per-session channels (exclude_ids vs
exclude_sync_ids, .ids vs .sync.ids in the hook), so a reuse hint shown
early in a session can no longer silence the record-sync nudge when the
recorded file itself is edited later. Sync surfacing is measured under
its own note_usage source (write_path_sync) — its pull-through rate is
the scoreboard for whether edit-time sync actually happens, per decision
#2707 (no forge connection; records stay current in the session that has
the context). Plugin 0.1.31.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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]: