src/scribe/services/plugin_context.py

@@ -644,7 +644,7 @@ async def _reserve_slot_for_reuse(
     cfg: dict,
     *,
     project_id: int | None,
-    exclude_ids: set[int],
+    already: set[int],
 ) -> list:
     """Guarantee the reuse-shaped kinds one slot, if one clears threshold (#2246).
 
@@ -671,16 +671,23 @@ async def _reserve_slot_for_reuse(
     top_k = cfg["top_k"]
     _t0 = time.perf_counter()
     _rep: dict = {}
+    # THE LEDGER IS NOT AN EXCLUSION HERE EITHER (#4101) — but what is already
+    # in `kept` still is, and the two are different claims. A record sitting in
+    # this call's own menu must not be shown twice in it; a record shown in an
+    # EARLIER call is exactly what the slot should be allowed to spend itself
+    # on, because a snippet that was relevant then and is relevant now is the
+    # reuse case rather than a duplicate of it.
     reuse = await semantic_search_notes(
         user_id, query,
         limit=1,
         threshold=cfg["threshold"],
         project_id=project_id,
-        exclude_ids=exclude_ids | {int(n.id) for _s, n in kept},
+        exclude_ids={int(n.id) for _s, n in kept},
         note_type=_REUSE_KINDS,
         scope="browse",
         report=_rep,
     )
+    fresh_reuse = [(s, n) for s, n in reuse if int(n.id) not in already]
     # A real semantic query competing for a menu slot — logged like the scored
     # arm it displaces. Before this, the hit it PUSHED OUT was in
     # retrieval_logs and the query that pushed it out was not, so the slot
@@ -689,16 +696,17 @@ async def _reserve_slot_for_reuse(
     record_retrieval(
         user_id=user_id, source="reuse_slot", query=query,
         threshold=cfg["threshold"], limit=1, project_id=project_id,
-        is_task=None, results=reuse,
+        is_task=None, results=fresh_reuse,
         best_available=_rep.get("best_available_score"),
         best_available_id=_rep.get("best_available_id"),
         searched=bool(_rep.get("searched", True)),
+        suppressed=len(reuse) - len(fresh_reuse),
         duration_ms=(time.perf_counter() - _t0) * 1000.0,
     )
     # Verify the kind rather than trusting the query that asked for it, and
-    # dedup on top of exclude_ids. This slot exists FOR reuse kinds — a slot
-    # silently spent on something else is worse than no slot, because the line
-    # is indistinguishable from one that earned its place on score.
+    # dedup against this call's own menu. This slot exists FOR reuse kinds — a
+    # slot silently spent on something else is worse than no slot, because the
+    # line is indistinguishable from one that earned its place on score.
     kept_ids = {int(n.id) for _s, n in kept}
     fresh = [
         (s, n) for s, n in reuse
@@ -726,7 +734,8 @@ async def build_autoinject_hint(
     The four anti-bloat gates (see the module + milestone-93 design):
       1. high-confidence threshold (stricter than pull) — set per-user;
       2. margin gate — keep only hits within _AUTOINJECT_BAND of the top score;
-      3. session dedup — caller passes already-injected ids as `exclude_ids`;
+      3. session marking — caller passes already-injected ids as `exclude_ids`
+         and they are rendered again with `[seen]`, never withheld (#4101);
       4. title-first payload — id + kind + title + score only, never bodies.
     Disabled, blank-query, or nothing-clears-the-gates all return empty context,
     so most turns inject nothing.
@@ -741,6 +750,24 @@ async def build_autoinject_hint(
     if not cfg["enabled"] or not q:
         return empty
 
+    # THE LEDGER LEAVES THE SEARCH (#4101). `exclude_ids` used to go into
+    # `semantic_search_notes` itself, so a record this session had already been
+    # shown was removed from the candidate set — which is #3750's defect, on the
+    # arm that fires most. Three things followed from it, none of them intended:
+    #
+    #   - the second time a note was the best answer, the session got SILENCE,
+    #     indistinguishable from "nothing matched";
+    #   - a compaction made that permanent, because the ledger outlived the
+    #     context it described (fixed one layer down in this same step);
+    #   - and the score the search reported was measured against a candidate set
+    #     the caller had already edited, so `best_available` could name a bar
+    #     that turned nothing away (#3739).
+    #
+    # Now the ledger is a RENDERING fact, not a retrieval one: every repeat is
+    # still ranked, still shown, and carries a marker saying it was surfaced
+    # before. A note line is a title and a score — the repeat costs about as
+    # much as the comma in this sentence — so there is nothing here to save.
+    already = {int(i) for i in (exclude_ids or [])}
     t0 = time.perf_counter()
     _rep_ai: dict = {}
     hits = await semantic_search_notes(
@@ -748,7 +775,6 @@ async def build_autoinject_hint(
         limit=cfg["top_k"],
         threshold=cfg["threshold"],
         project_id=(project_id or None),
-        exclude_ids=set(exclude_ids or []),
         # Injection is the one retrieval nobody asked for, so it takes the BROWSE
         # scope: never a record shared one-to-one with the operator. What can
         # still appear is a collaborator's note inside a shared project — legible
@@ -756,24 +782,37 @@ async def build_autoinject_hint(
         scope="browse",
         report=_rep_ai,
     )
+    # `results=fresh` and `suppressed` together, on the rule arms' contract
+    # (#3752): a rendered repeat is not a new surfacing, so it stays out of the
+    # row's result set and is COUNTED instead. That keeps this source's surfaced
+    # set identical to its own log row (#3668) while making a zero-result call
+    # readable — `result_count == 0` with `suppressed_count > 0` is "everything
+    # that matched, this session has already seen", which is a different fact
+    # about the bar from "nothing cleared it" and used to be unreportable here.
+    fresh = [(s, n) for s, n in hits if int(n.id) not in already]
     record_retrieval(
         user_id=user_id, source="auto_inject", query=q,
         threshold=cfg["threshold"], limit=cfg["top_k"],
-        project_id=(project_id or None), is_task=None, results=hits,
+        project_id=(project_id or None), is_task=None, results=fresh,
         best_available=_rep_ai.get("best_available_score"),
         best_available_id=_rep_ai.get("best_available_id"),
         searched=bool(_rep_ai.get("searched", True)),
+        suppressed=len(hits) - len(fresh),
         duration_ms=(time.perf_counter() - t0) * 1000.0,
     )
     if not hits:
         return empty
 
-    # Margin gate: keep only hits close to the strongest one.
+    # Margin gate: keep only hits close to the strongest one. Computed over ALL
+    # hits, repeats included — the band measures distance from the top SCORE,
+    # and letting the ledger move that cutoff would make "you were shown this"
+    # change what counts as relevant, which is the axis independence the rule
+    # band keeps for the same reason.
     top_score = hits[0][0]
     kept = [(s, n) for s, n in hits if s >= top_score - _AUTOINJECT_BAND]
     kept = await _reserve_slot_for_reuse(
         user_id, q, kept, cfg, project_id=(project_id or None),
-        exclude_ids=set(exclude_ids or []),
+        already=already,
     )
 
     # A collaborator's note can reach this menu via a shared project, and the
@@ -786,10 +825,15 @@ async def build_autoinject_hint(
     # "records", not "notes" — the menu can hold snippets, processes and tasks
     # too, and the kind marker on each line is only legible if the header doesn't
     # already claim they're all one thing.
+    # "injected once per session" was true and is not any more (#4101): a repeat
+    # is shown again with a marker rather than withheld, so the header must stop
+    # promising the old contract. It now says what the marker means instead,
+    # once, rather than each repeated line having to explain itself.
     lines = [
         "> Possibly relevant from your Scribe records — open any in full with "
         "`get_note(id)`, or `get_snippet` / `get_process` for those kinds "
-        "(titles only; injected once per session):",
+        "(titles only; a line marked `seen` was surfaced earlier this session "
+        "and may no longer be in context):",
     ]
     # A superseded record is DEMOTED, not removed (#278) — so one can still reach
     # this menu, and when it does the reader has to be told. An agent handed
@@ -802,6 +846,13 @@ async def build_autoinject_hint(
         note_ids.append(int(note.id))
         title = (note.title or "(untitled)").replace("\n", " ").strip()
         line = f"> - #{note.id} [{_record_kind(note)}] \"{title}\" ({score:.2f})"
+        # ONE WORD, NOT A SENTENCE, and deliberately not the rule arms' phrasing.
+        # A rule line says "before deciding it does not apply", which is the
+        # voice of a record that BINDS; a note binds nothing, and borrowing that
+        # tone would tell the reader a dev-log has authority it does not have.
+        # The header carries the meaning, so the line carries only the flag.
+        if int(note.id) in already:
+            line += " [seen]"
         if int(note.id) in stale:
             line += " — SUPERSEDED, a later record covers this; check that first"
         if note.user_id != user_id:
@@ -813,7 +864,17 @@ async def build_autoinject_hint(
     # 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")
+    #
+    # FRESH ONLY, which is the same cut the log row above takes (#4101). A
+    # repeat is rendered but is not a new surfacing, and counting it again would
+    # make this table disagree with `retrieval_logs` about the same call —
+    # #3668's identity, which is the cheapest true statement available about
+    # this pair of tables and is not worth a marker's convenience.
+    record_surfaced(
+        user_id=user_id,
+        note_ids=[i for i in note_ids if i not in already],
+        source="auto_inject",
+    )
 
     return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
 
@@ -1468,12 +1529,16 @@ async def build_write_path_hint(
     #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).
+    The two classes track the session 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). The two
+    channels also now ACT differently, which is #4101: a reuse repeat is
+    rendered again with a `seen` marker, while the sync class still shows
+    once, because its claim is about an edit in progress rather than about a
+    record's continuing relevance and repeating it would be nagging.
 
-    Carries auto-inject's anti-bloat gates (margin, session dedup,
+    Carries auto-inject's anti-bloat gates (margin, session marking,
     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
@@ -1592,16 +1657,28 @@ async def build_write_path_hint(
         query = concept_query(query) or query
     if remaining > 0 and query:
         t0 = time.perf_counter()
-        # Pulled-and-seen ids stay in the query (as evidence) but never in
-        # the menu — the dedup contract holds, the resemblance still lands.
-        pulled_seen = seen & set(pulled)
+        # `seen` CARRIES TWO DIFFERENT CLAIMS, and only one of them is a reason
+        # to withhold (#4101). It was built as the ledger plus everything this
+        # call has already rendered, and the arm treated both the same way:
+        #
+        #   - already in THIS menu (the sync block, the place arm) — listing it
+        #     twice in one hint is noise with no reader it could help;
+        #   - on the LEDGER from an earlier call — a different claim entirely,
+        #     and the one #3750 says must be rendered rather than dropped.
+        #
+        # Splitting them is the whole change here. `in_menu` keeps its
+        # exclusion; the ledger becomes a marker on the line.
+        in_menu = seen - excluded
+        # Pulled-and-already-listed ids stay in the query (as evidence for
+        # `resembles`) but never in the menu, so the limit has to cover them.
+        pulled_in_menu = in_menu & set(pulled)
         _rep_wp: dict = {}
         hits = await semantic_search_notes(
             user_id, query,
-            limit=remaining + len(pulled_seen),
+            limit=remaining + len(pulled_in_menu),
             threshold=cfg["threshold"],
             project_id=scope_project,
-            exclude_ids=seen - pulled_seen,
+            exclude_ids=in_menu - set(pulled),
             # Snippets AND recorded experience (#2246). This arm was
             # snippets-only, which is auto-inject's mistake inverted: an issue
             # saying "we tried this and it deadlocked", or a dev-log recording
@@ -1625,41 +1702,50 @@ async def build_write_path_hint(
             int(note.id): float(score) for score, note in hits
             if int(note.id) in pulled
         }
-        shown = [(s, n) for s, n in hits if int(n.id) not in seen]
+        shown = [(s, n) for s, n in hits if int(n.id) not in in_menu]
         # WHAT THIS ARM WITHHELD AFTER THE SEARCH ANSWERED, and the reason
         # `best_available_score` cannot always be reported here (#3739 again,
         # from the side its fix did not reach).
         #
         # This arm is the one note arm that filters TWICE. `exclude_ids` takes
-        # `seen - pulled_seen` into the search, but the pulled-and-seen ids stay
-        # in the query deliberately — `resembles` above needs them — and are
-        # dropped in the line above instead. So the score the search reported is
-        # PRE that drop while the row's `result_count` is POST it, and a record
-        # the session had already been shown could be logged as something the
-        # BAR turned away. Live proof on the first read after #3739 shipped:
-        # write_path's near-miss max was 0.822 while the lowest score it ever
-        # RETURNED was 0.6857 — a "rejection" that beat every acceptance.
-        #
-        # The suppression column cannot rescue it the way it does for the rule
-        # arms: this arm's count would be PARTIAL, covering only the drops made
-        # here and not the ones `exclude_ids` made inside the search, and a
-        # partial number under a name that reads as complete is the substitution
-        # this whole milestone exists to stop.
+        # the already-listed ids into the search, but the PULLED ones among them
+        # stay in the query deliberately — `resembles` above needs them — and
+        # are dropped in the line above instead. So the score the search
+        # reported is PRE that drop while the row's `result_count` is POST it,
+        # and a record already listed in this same menu could be logged as
+        # something the BAR turned away. Live proof on the first read after
+        # #3739 shipped: write_path's near-miss max was 0.822 while the lowest
+        # score it ever RETURNED was 0.6857 — a "rejection" that beat every
+        # acceptance.
         #
         # So the honest answer is null — "not measured on this call" — whenever
         # this filter removed anything, because then the bar is not the only
         # thing that turned something away and the reported score may belong to
         # a record we withheld ourselves. Calls where nothing was dropped keep
         # reporting it, which is most of them.
+        #
+        # THE LEDGER IS NO LONGER PART OF THIS (#4101), and that is why the
+        # suppression column can now be filled in where it could not before.
+        # The old objection was that a count here would be PARTIAL — covering
+        # the drops made in this function but not the ones `exclude_ids` made
+        # inside the search — and a partial number under a name that reads as
+        # complete is the substitution this milestone exists to stop. That was
+        # right while the ledger was one of the things `exclude_ids` carried.
+        # It no longer is: every ledger repeat comes back from the search and is
+        # rendered, so `suppressed` counts all of them and none are hidden
+        # inside the query. What `exclude_ids` still removes is this call's own
+        # menu, which is not suppression at all — those records ARE being shown,
+        # one block further up.
         withheld_here = len(hits) - len(shown)
         hits = shown[:remaining]
+        fresh = [(s, n) for s, n in hits if int(n.id) not in excluded]
         record_retrieval(
             user_id=user_id, source="write_path", query=query,
             threshold=cfg["threshold"], limit=remaining,
             # is_task is None, not False: this arm now returns issues too, and
             # recording it as a notes-only retrieval would misdescribe the
             # candidate set the threshold is being tuned against.
-            project_id=scope_project, is_task=None, results=hits,
+            project_id=scope_project, is_task=None, results=fresh,
             best_available=(
                 None if withheld_here else _rep_wp.get("best_available_score")
             ),
@@ -1670,6 +1756,7 @@ async def build_write_path_hint(
                 None if withheld_here else _rep_wp.get("best_available_id")
             ),
             searched=bool(_rep_wp.get("searched", True)),
+            suppressed=len(hits) - len(fresh),
             duration_ms=(time.perf_counter() - t0) * 1000.0,
         )
         if hits:
@@ -1683,9 +1770,19 @@ async def build_write_path_hint(
                 # and an unlabelled line would be read as "here is code to
                 # reuse", which is the opposite of what it says.
                 kind = _record_kind(note)
-                scored.append((
+                marker = (
                     f"similar {score:.2f}" if kind == "snippet"
-                    else f"similar {score:.2f} · {kind}",
+                    else f"similar {score:.2f} · {kind}"
+                )
+                # The repeat marker rides the same dotted list as the kind, so a
+                # reference costs four characters and needs no second line
+                # (#4101). Same word as the auto-inject menu deliberately: a
+                # reader meeting `seen` on two different surfaces should not
+                # have to work out whether they mean the same thing.
+                if int(note.id) in excluded:
+                    marker += " · seen"
+                scored.append((
+                    marker,
                     {
                         "id": int(note.id), "title": note.title, "user_id": note.user_id,
                         # Carried so the line can disclose a cross-language hit
@@ -1811,7 +1908,8 @@ async def build_write_path_hint(
             "`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):"
+            "(titles only; a line marked `seen` was surfaced earlier this "
+            "session and may no longer be in context):"
         )
     # 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
@@ -1848,6 +1946,12 @@ async def build_write_path_hint(
     if sync_note_ids:
         by_arm["write_path_sync"] = list(sync_note_ids)
     for marker, item in menu:
+        # A rendered repeat is not a new surfacing (#4101) — same cut the log
+        # row takes, so this table and `retrieval_logs` keep agreeing about the
+        # same call (#3668). The place arm never reaches here with one: a nearby
+        # snippet on the ledger is skipped before it is ever placed.
+        if int(item["id"]) in excluded:
+            continue
         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():

tests/test_ledger_references_not_silence.py

@@ -0,0 +1,240 @@
+"""A repeat on a note arm is referenced, not withheld (#4101).
+
+WHY THIS EXISTS
+
+#3750 settled the argument for rules: a record the session was told about an
+hour ago is not a record in front of the reader now, so the second time it is
+the best answer the session gets the line again with a tail saying so — never
+silence, which is indistinguishable from "nothing matched".
+
+The note and snippet arms never got that fix. Worse, their ledger went into
+`semantic_search_notes` as `exclude_ids`, so the repeat left the candidate set
+entirely, which had three consequences:
+
+  - the session got silence the second time, on the arms that fire most;
+  - a compaction made it permanent, because the ledger outlived the context it
+    described (the same step fixes that one layer down);
+  - and `best_available` was measured against a candidate set the caller had
+    already edited, so the bar could be blamed for a record the caller withheld
+    (#3739, from the side its fix never reached).
+
+WHAT THIS PINS
+
+  1. The ledger does not reach the search. Asserted on the call's kwargs,
+     because this is the difference between a reference and silence and every
+     behavioural assertion below rests on it.
+  2. The repeat is rendered, and marked. The wording is NOT the rule arms' —
+     "before deciding it does not apply" is the voice of a record that binds,
+     and a dev-log borrowing it would claim authority it does not have — so
+     what is pinned is that the line appears and is distinguishable, not its
+     prose.
+  3. The telemetry splits the two (#3752 / #3668): the row's result set and the
+     surfaced table both take FRESH only, and the repeat is COUNTED in
+     `suppressed` instead. This is what makes a zero-result call readable —
+     `result_count == 0` with `suppressed_count > 0` says "everything that
+     matched, the session has already seen", which was unreportable here.
+  4. The deliberate exception: the write-path SYNC class still shows once. Its
+     claim is about an edit in progress, not about a record's continuing
+     relevance, and repeating it would be nagging rather than reminding.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from tests.helpers import fake_note
+
+REAL_CODE = '''def debounce(fn, wait=0.25):
+    """Rate-limit a callback so it fires once after the last call."""
+    timer = None
+
+    def wrapped(*a, **kw):
+        nonlocal timer
+        if timer:
+            timer.cancel()
+        timer = threading.Timer(wait, fn, a, kw)
+        timer.start()
+
+    return wrapped
+'''
+
+
+def _wp_cfg(**over):
+    base = {"enabled": True, "threshold": 0.68, "top_k": 3, "rule_threshold": 0.72}
+    base.update(over)
+    return base
+
+
+def _snippet_item(nid, title, user_id=1):
+    return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}
+
+
+# ── auto-inject ────────────────────────────────────────────────────────────
+
+async def _autoinject(hits, exclude_ids, *, rec=None, surf=None):
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=hits)
+    with patch.object(pc, "get_autoinject_config", AsyncMock(
+            return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", rec or MagicMock()), \
+         patch.object(pc, "record_surfaced", surf or MagicMock()):
+        out = await pc.build_autoinject_hint(
+            1, "postgres pool", project_id=2, exclude_ids=exclude_ids)
+    return out, search
+
+
+@pytest.mark.asyncio
+async def test_the_ledger_never_reaches_the_search():
+    """The load-bearing one. A ledger inside the query IS the silence."""
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
+    _out, search = await _autoinject(hits, [11])
+    # The menu arm runs first; the reuse slot's own call may legitimately
+    # exclude this call's menu, which is a different claim (see below).
+    menu_call = search.call_args_list[0]
+    assert not menu_call.kwargs.get("exclude_ids"), (
+        "the session ledger was passed into the search, so a repeat is "
+        "withheld rather than referenced"
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_repeat_is_shown_again_and_marked():
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
+            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
+    out, _ = await _autoinject(hits, [11])
+    assert "#11" in out["context"] and "#22" in out["context"]
+    # Distinguishable, without pinning the word's neighbours in the sentence.
+    line_11 = next(ln for ln in out["context"].splitlines() if "#11" in ln)
+    line_22 = next(ln for ln in out["context"].splitlines() if "#22" in ln)
+    assert "seen" in line_11 and "seen" not in line_22
+
+
+@pytest.mark.asyncio
+async def test_the_repeat_is_counted_not_reported_as_a_result():
+    rec = MagicMock()
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
+            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
+    await _autoinject(hits, [11], rec=rec)
+    row = next(c.kwargs for c in rec.call_args_list
+               if c.kwargs["source"] == "auto_inject")
+    assert [int(n.id) for _s, n in row["results"]] == [22]
+    assert row["suppressed"] == 1
+
+
+@pytest.mark.asyncio
+async def test_a_call_where_everything_was_already_seen_is_readable():
+    """The fact that could not be expressed before.
+
+    Previously this call logged zero results with no suppression count, so it
+    was indistinguishable from a bar nothing cleared — and the operator tuning
+    that bar would have been reading the wrong number.
+    """
+    rec = MagicMock()
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
+    out, _ = await _autoinject(hits, [11], rec=rec)
+    row = next(c.kwargs for c in rec.call_args_list
+               if c.kwargs["source"] == "auto_inject")
+    assert row["results"] == [] and row["suppressed"] == 1
+    # And the session still gets the line, which is the whole point.
+    assert "#11" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_repeat_is_not_recorded_as_a_fresh_surfacing():
+    """#3668's identity: this table and the log row describe the same call."""
+    surf = MagicMock()
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
+            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
+    await _autoinject(hits, [11], surf=surf)
+    rows = [c.kwargs for c in surf.call_args_list
+            if c.kwargs.get("source") == "auto_inject"]
+    assert rows and rows[0]["note_ids"] == [22]
+
+
+@pytest.mark.asyncio
+async def test_the_header_no_longer_promises_once_per_session():
+    """A contract stated in the prose is a contract, and this one changed.
+
+    Cheap to forget and invisible when wrong: the menu would carry a marker
+    the header had never explained, and a reader would have to guess.
+    """
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
+    out, _ = await _autoinject(hits, [])
+    header = out["context"].splitlines()[0]
+    assert "once per session" not in header
+    assert "seen" in header
+
+
+# ── the write path ─────────────────────────────────────────────────────────
+
+async def _write_path(hits, exclude_ids, *, here=(), rec=None, sync_exclude=()):
+    from scribe.services import plugin_context as pc
+    search = AsyncMock(return_value=hits)
+    listing = AsyncMock(side_effect=[(list(here), len(here)), ([], 0)])
+    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_wp_cfg())), \
+         patch.object(pc.snippets_svc, "list_snippets", listing), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", rec or MagicMock()), \
+         patch.object(pc, "record_surfaced", MagicMock()):
+        out = await pc.build_write_path_hint(
+            1, "src/x.py", code=REAL_CODE,
+            exclude_ids=list(exclude_ids),
+            exclude_sync_ids=list(sync_exclude),
+        )
+    return out, search
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_ledger_does_not_reach_its_search_either():
+    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1))]
+    _out, search = await _write_path(hits, [11])
+    assert 11 not in (search.call_args.kwargs.get("exclude_ids") or set())
+
+
+@pytest.mark.asyncio
+async def test_a_write_path_repeat_is_rendered_with_a_marker():
+    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1))]
+    out, _ = await _write_path(hits, [11])
+    assert "#11" in out["context"]
+    assert "seen" in next(ln for ln in out["context"].splitlines() if "#11" in ln)
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_counts_its_repeat():
+    rec = MagicMock()
+    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1)),
+            (0.80, fake_note(id=22, title="throttle helper", user_id=1))]
+    await _write_path(hits, [11], rec=rec)
+    row = next(c.kwargs for c in rec.call_args_list
+               if c.kwargs["source"] == "write_path")
+    assert [int(n.id) for _s, n in row["results"]] == [22]
+    assert row["suppressed"] == 1
+
+
+@pytest.mark.asyncio
+async def test_this_calls_own_menu_is_still_excluded_from_its_search():
+    """The claim that stayed an exclusion, and must not be lost with the other.
+
+    A snippet already listed by PLACE in this same hint has nothing to gain
+    from a second line in it. That is same-call duplication, not a repeat
+    across calls, and the two were the same variable until this step.
+    """
+    hits = []
+    _out, search = await _write_path(
+        hits, [], here=[_snippet_item(7, "records this file")])
+    assert 7 in (search.call_args.kwargs.get("exclude_ids") or set())
+
+
+@pytest.mark.asyncio
+async def test_the_sync_class_still_shows_only_once():
+    """The deliberate exception, pinned so it reads as a decision.
+
+    The sync nudge says "you are editing the file this record describes, so
+    updating it is part of the edit". Repeated every write to the same file it
+    is nagging, and unlike a reuse suggestion it is not a claim whose relevance
+    can return — it either got acted on or it did not.
+    """
+    out, _ = await _write_path(
+        [], [], here=[_snippet_item(7, "records this file")], sync_exclude=[7])
+    assert out["sync_note_ids"] == []
+    assert "#7" not in out["context"]