Merge pull request 'The prompt-boundary rule arm, and a slot a preference cannot lose (#3852, #3894)' (#151) from dev into main

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": "2026.09.11.0319",
+  "version": "2026.09.11.1154",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/hooks/scribe_autoinject.sh

@@ -7,6 +7,18 @@
 # only — never bodies; the agent calls get_note(id) to pull anything it judges
 # relevant. Most turns inject nothing.
 #
+# TWO ARMS SINCE #3852, on one request. Rules and preferences are retrieved
+# against the same prompt and returned in the same payload, ahead of the notes
+# menu. That arm exists because the two act arms are keyed on a file write or
+# a command, so a rule governing what to SAY — extract intent from loose
+# phrasing, raise a conflict before acting, end a finding with an offer — had
+# no moment to fire at. The operator's message is the only query that exists
+# before a response is composed.
+#
+# The two arms are gated separately server-side: turning the notes menu off
+# leaves rules arriving, because they are different claims with different
+# costs of being missed.
+#
 # Best-effort enrichment ONLY: unlike the SessionStart channel there is no
 # static floor here. If the instance is unconfigured/unreachable, or anything
 # fails, the hook stays SILENT and exits 0 — it must never block a prompt.
@@ -65,16 +77,32 @@ fi
 # Per-session dedup: ids already injected this session are skipped.
 state_dir="${TMPDIR:-/tmp}/scribe-autoinject"
 mkdir -p "$state_dir" 2>/dev/null || true
+# RULES DEDUP IN A DIFFERENT DIRECTORY, and it has to be this one. The rule
+# ledger is SHARED by every arm that can name a rule — the two PreToolUse
+# hooks already keep it under scribe-priorart — so that one session keeps ONE
+# list and a rule named here is not re-announced before the next Bash call.
+# A private copy here would make each arm's "already seen" mean something
+# different, which is the state #3749/#3750 exist to keep coherent. The
+# directory name is the prior-art hook's history, not a scope claim.
+rule_state_dir="${TMPDIR:-/tmp}/scribe-priorart"
+mkdir -p "$rule_state_dir" 2>/dev/null || true
 idfile=""
+rulefile=""
 exclude_q=""
 if [ -n "$session_id" ]; then
   # session_id is an opaque token from Claude Code; keep only filename-safe chars.
   safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
   idfile="$state_dir/${safe_sid}.ids"
+  rulefile="$rule_state_dir/${safe_sid}.rules.ids"
   if [ -f "$idfile" ]; then
     seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
     [ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
   fi
+  # AGED, not read flat: an exclusion that never expires means a rule surfaced
+  # once in a long session is silenced for the rest of it, even as the session
+  # stops holding what it was told. scribe_rules_live carries the reasoning.
+  rule_seen=$(scribe_rules_live "$rulefile")
+  [ -n "$rule_seen" ] && exclude_q="${exclude_q}&exclude_rule_ids=${rule_seen}"
 fi
 
 body=$(curl -fsS --max-time 5 \
@@ -89,6 +117,14 @@ context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || exit 0
 if [ -n "$idfile" ]; then
   printf '%s' "$body" | jq -r '.note_ids[]? // empty' 2>/dev/null >> "$idfile" || true
 fi
+# Rules onto the SHARED ledger, stamped so they can age out. Only FRESH ids
+# come back in rule_ids (#3752) — a rule rendered as a repeat is already on
+# the ledger, and re-appending it would keep pushing its stamp forward so it
+# never aged at all.
+if [ -n "$rulefile" ]; then
+  printf '%s' "$body" | jq -r '.rule_ids[]? // empty' 2>/dev/null \
+    | scribe_rules_append "$rulefile"
+fi
 
 jq -n --arg c "$context" \
   '{hookSpecificOutput: {hookEventName: "UserPromptSubmit", additionalContext: $c}}'

src/scribe/routes/plugin.py

@@ -91,13 +91,37 @@ async def autoinject_retrieve():
         project_id (opt)   — explicit project scope override (ad-hoc/testing).
         exclude_ids (opt)  — comma-separated note ids already injected this
                              session; skipped so each note injects at most once.
+        exclude_rule_ids   — comma-separated rule ids already surfaced this
+            (opt)            session. SHARED with /prior-art and /tool-rules on
+                             purpose: one session keeps ONE rule ledger, so a
+                             rule named by any arm is not re-announced by
+                             another. Ages out (#3751), so salience decays.
+
+    TWO ARMS, TWO SETS OF GATES. Rules ride the same hook and the same query
+    but nothing else: the notes menu can be disabled, thresholded and top-k'd
+    by the operator without touching whether a rule reaches them. Composed
+    here rather than inside either builder so neither one's early return can
+    silently suppress the other.
+
+    Rules come FIRST in the payload. A rule or preference governing the answer
+    is more consequential than a menu of things that might be worth reading,
+    and a reader who stops after the first block should have stopped after
+    the right one.
     """
     q = (request.args.get("q") or "").strip()
     project_id, _repo, _unbound = await _project_scope()
     exclude_ids = _int_list(request.args.get("exclude_ids"))
+    exclude_rule_ids = _int_list(request.args.get("exclude_rule_ids"))
+
+    rules = await plugin_ctx_svc.build_prompt_rule_hint(
+        g.user.id, q, project_id=project_id, exclude_rule_ids=exclude_rule_ids
+    )
     result = await plugin_ctx_svc.build_autoinject_hint(
         g.user.id, q, project_id=project_id, exclude_ids=exclude_ids
     )
+    blocks = [b for b in (rules["context"], result["context"]) if b]
+    result["context"] = "\n\n".join(blocks)
+    result["rule_ids"] = rules["rule_ids"]
     return jsonify(result)
 
 

src/scribe/services/embeddings.py

@@ -818,6 +818,7 @@ async def semantic_search_rules(
     limit: int = 5,
     threshold: float = _SIMILARITY_THRESHOLD,
     tier: str | None = None,
+    kind: str | None = None,
     report: dict | None = None,
 ) -> list[tuple[float, "Rule"]]:
     """Return up to *limit* (score, rule) pairs most relevant to *query*.
@@ -855,6 +856,16 @@ async def semantic_search_rules(
     Pass a tier when a caller genuinely wants one class — a listing, an audit,
     a UI that renders the tiers apart. Not to approximate relevance.
 
+    `kind` narrows to `rule` or `preference`, and NONE is likewise the ordinary
+    case: a caller asking "what governs this" wants both, because the reader
+    needs to know what binds AND how the operator wants it done. The one place
+    it is passed is a RESERVED SLOT — a query that may only return a
+    preference, so the slot cannot be spent on something else. That is the
+    same reason `note_type` exists on the sibling search, and the same failure
+    it prevents: a slot silently filled by the wrong kind is worse than no
+    slot, because the line is indistinguishable from one that earned its place
+    on score.
+
     Collapses to best-chunk-per-rule like the note search, so a long rule split
     across chunks competes once rather than crowding the results with itself.
 
@@ -895,6 +906,7 @@ async def semantic_search_rules(
                         Project.user_id == user_id,
                     ),
                     *( [Rule.tier == tier] if tier else [] ),
+                    *( [Rule.kind == kind] if kind else [] ),
                 )
                 # Overfetch so collapsing chunks to their best row still fills
                 # the page — the same reason the note search overfetches.

src/scribe/services/plugin_context.py

@@ -314,6 +314,42 @@ _AUTOINJECT_BAND = 0.10
 # awareness menu (titles only), never a content dump.
 _AUTOINJECT_MAX_TOP_K = 10
 
+# --- the prompt-boundary rule arm (#3852) ------------------------------------
+#
+# Both existing rule arms are keyed on something the session is about to DO —
+# a file write, a command. A rule that governs what to SAY has no such moment.
+# Extract intent from loose phrasing, raise a conflict before acting, hand off
+# an action with its reason, end a finding with an offer: every one binds on a
+# RESPONSE, and no tool call precedes a response.
+#
+# The operator's message is the only query that exists before one is composed,
+# and this arm is what runs against it. Until now that hook searched notes
+# alone, so no rule had ever been retrieved against a thing the operator said.
+PROMPTRULE_THRESHOLD_KEY = "kb_promptrule_threshold"
+# INHERITED FROM THE ACT ARMS, AND NOT YET EARNED HERE. 0.72 was tuned against
+# code and shell commands. An operator's prose is a different query shape
+# against the same documents, and nothing yet says the two distributions line
+# up — triggers are written in the vocabulary of the MOMENT, which for most
+# rules is act vocabulary, so prose may well score lower across the board.
+#
+# Starting at the act arms' number anyway is deliberate: it is the only value
+# with evidence behind it, and guessing lower would put an unmeasured bar in
+# front of a corpus that binds. Every call is logged under `prompt_rule` from
+# the first deploy, so a few days of real traffic settles it — read
+# `near_miss_samples` (#3807) before moving this, not the percentile alone.
+PROMPTRULE_DEFAULT_THRESHOLD = 0.72
+
+# MORE THAN THE ACT ARMS' SINGLE SLOT, anchored on this hook's budget rather
+# than theirs. RULEHINT_LIMIT is 1 because that arm fires before EVERY Bash
+# call, where a second line is a second interruption per command. This arm
+# fires once per TURN, on the same hook whose notes menu already spends
+# AUTOINJECT_DEFAULT_TOP_K slots — so that is the comparable budget.
+#
+# And a prompt genuinely contains more than one act. "Merge to main and then
+# start on X" is two, governed by different rules; k=1 cannot serve that case
+# at all, where the act arms never face it because a command is one thing.
+PROMPTRULE_LIMIT = 3
+
 
 def _slugify(text: str) -> str:
     """kebab-case slug for a skill directory name (a-z0-9 + single hyphens)."""
@@ -642,6 +678,235 @@ async def build_autoinject_hint(
     return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
 
 
+async def _reserve_slot_for_preference(
+    user_id: int,
+    query: str,
+    hits: list,
+    *,
+    threshold: float,
+    project_id: int,
+    already: set[int],
+) -> tuple[list, int | None]:
+    """Guarantee a preference one slot, if one clears the bar (#3894).
+
+    THE ASYMMETRY THIS EXISTS FOR. A rule and a preference are not equally
+    served by a shared score contest, because their losses are not equal:
+
+      - a RULE crowded out here can still fire at the act arm. A `git push`
+        reaches `pre_tool_rule`, a file write reaches `write_path_rule`. The
+        prompt hit is a preview of a second chance.
+      - a PREFERENCE about how to answer has no second chance. There is no
+        later act — the response IS the act — so crowded out here it is never
+        delivered at all.
+
+    A straight ranking therefore favours the record whose loss is recoverable
+    over the one whose loss is total, and it does so INVISIBLY: the rule that
+    won is a legitimate hit, the telemetry looks healthy, and the only symptom
+    is a preference that quietly never arrives. `reuse_slot` exists for the
+    same shape one corpus over (#2463), where snippets kept losing to project
+    records that merely resembled the query.
+
+    THE SLOT BUYS POSITION, NOT A LOWER BAR — same as `reuse_slot`, which also
+    reserves at `cfg["threshold"]`. A weak preference cannot buy the slot, so
+    silence stays the default and the reserved line is never worse than the
+    ones it sits beside. If `preference_slot` later shows a stream of
+    near-misses, `best_available_id` (#3807) names which preference was
+    refused and a separate bar becomes an argument with evidence behind it
+    rather than a knob added on a guess.
+
+    LEDGER REPEATS STILL COUNT AS REPRESENTED. A preference already on the
+    session's ledger occupies the slot rather than being skipped for a fresh
+    one: it is still rendered (#3750), just with the tail that says so, and a
+    preference is the kind of record where being reminded is the point.
+
+    Returns the possibly-extended hit list, and the id the slot spent — the
+    caller needs that to keep each source's surfaced set matching its own log
+    row (#3668), since the slot logs under its own name.
+    """
+    if any(rule.kind == "preference" for _s, rule in hits):
+        return hits, None
+
+    _t0 = time.perf_counter()
+    _rep: dict = {}
+    # KIND-FILTERED, so the query can only answer with what the slot is for.
+    # Verifying the kind afterwards would be weaker: an unfiltered search that
+    # happened to return a rule would spend the slot on it, and the line would
+    # be indistinguishable from one that earned its place.
+    found = await semantic_search_rules(
+        user_id, query, limit=1, threshold=threshold,
+        kind="preference", report=_rep,
+    )
+    fresh = [(s, r) for s, r in found if r.id not in already]
+    # ITS OWN SOURCE, and both sides of the trade logged. #2463's own finding
+    # is the warning rather than the precedent here: the hit that slot pushed
+    # OUT was in retrieval_logs while the query that pushed it out was not, so
+    # the slot could never be judged against what it displaced. `results` is
+    # fresh-only, matching what gets recorded as surfaced below (#3752/#3668).
+    record_retrieval(
+        user_id=user_id, source="preference_slot", query=query,
+        threshold=threshold, limit=1, project_id=project_id,
+        is_task=None, results=fresh,
+        best_available=_rep.get("best_available_score"),
+        best_available_id=_rep.get("best_available_id"),
+        searched=bool(_rep.get("searched", True)),
+        suppressed=len(found) - len(fresh),
+        duration_ms=(time.perf_counter() - _t0) * 1000.0,
+    )
+    seen = {rule.id for _s, rule in hits}
+    slot = [(s, r) for s, r in found
+            if r.kind == "preference" and r.id not in seen][:1]
+    if not slot:
+        return hits, None
+
+    slot_id = int(slot[0][1].id)
+    if slot_id not in already:
+        record_rule_surfaced(
+            user_id=user_id, rule_ids=[slot_id], source="preference_slot",
+        )
+    # IT EXTENDS, IT NEVER DISPLACES — and here it parts company with
+    # `reuse_slot`, which evicts its menu's weakest hit. The reason is the
+    # ledger rather than taste. A displaced hit was RETURNED by the general
+    # search and is sitting in that call's `retrieval_logs` row, but would not
+    # have been shown — so `prompt_rule`'s surfaced set would stop matching
+    # its own log row, and #3668's identity would break for a reason nothing
+    # in the data explains. That identity is the cheapest true statement
+    # available about this pair of tables, and milestone #379 is what it costs
+    # to lose it: five steps planned against a gap that was two counters
+    # disagreeing, not a write path dropping rows.
+    #
+    # The price is one extra line, only when the general search already filled
+    # the limit AND a preference cleared the bar without placing. Cheap, and
+    # it buys a surface whose two tables can always be checked against each
+    # other.
+    return hits + slot, slot_id
+
+
+async def build_prompt_rule_hint(
+    user_id: int,
+    query: str,
+    *,
+    project_id: int = 0,
+    exclude_rule_ids: list[int] | None = None,
+) -> dict:
+    """Rules and preferences that may apply to what the operator just asked.
+
+    The third rule arm, and the one that closes a gap the other two cannot
+    reach. `write_path_rule` is keyed on code, `pre_tool_rule` on a command —
+    both are things the session is about to DO. A rule that governs what to
+    SAY has no such trigger, and residency was the only surface it ever had.
+    Removing residency (milestone 394) without this would drop that half of
+    the corpus on the floor.
+
+    A SEPARATE FUNCTION, not a branch inside build_autoinject_hint, and the
+    reason is its early returns. That arm bails when auto-inject is disabled,
+    when the query is blank, when nothing clears the note bar — and every one
+    of those is a statement about NOTES. Folded in, a user who turned the
+    notes menu off would silently lose their rules too, which is the kind of
+    coupling nothing downstream could see. Two functions, two sets of gates,
+    composed by the caller.
+
+    THE OUTPUT IS DELIBERATELY NOT QUOTED, where the notes menu is. The task
+    asked whether the two share a header; the answer is that neither needs
+    one. A note line is a bare title and needs the menu's header to say what
+    it is doing there, while a rule line names itself in its opening words
+    ("Standing rule that may apply…" / "Preference that may apply…"). Leaving
+    rules unquoted separates the two claims visually with no extra prose, and
+    matches how a rule line already renders on both act arms.
+
+    Fails open and returns empty context on any error, like its siblings: a
+    recall aid may never break the operator's prompt.
+    """
+    out: dict = {"context": "", "rule_ids": []}
+    q = (query or "").strip()
+    if not q:
+        return out
+
+    try:
+        try:
+            threshold = float(await get_setting(
+                user_id, PROMPTRULE_THRESHOLD_KEY,
+                str(PROMPTRULE_DEFAULT_THRESHOLD)))
+        except (TypeError, ValueError):
+            threshold = PROMPTRULE_DEFAULT_THRESHOLD
+        threshold = min(1.0, max(0.0, threshold))
+
+        t0 = time.perf_counter()
+        _rep: dict = {}
+        # NOT scoped to the project, and that is the corpus's own decision
+        # rather than an omission here — semantic_search_rules is scoped by
+        # OWNERSHIP on purpose, because "is there a rule about this" is asked
+        # across a whole rulebook. `project_id` below reaches the log row and
+        # nothing else.
+        hits = await semantic_search_rules(
+            user_id, q, limit=PROMPTRULE_LIMIT, threshold=threshold,
+            report=_rep,
+        )
+        duration_ms = (time.perf_counter() - t0) * 1000.0
+
+        already = set(exclude_rule_ids or [])
+        fresh = [(score, rule) for score, rule in hits if rule.id not in already]
+
+        # BEFORE the early return, for the reason both sibling arms spell out
+        # at length: a call that found nothing is the only evidence a bar is
+        # too high, and an arm that logs only the calls it liked reports a
+        # flawless clear-rate however badly it is tuned. This bar is inherited
+        # and unverified for this corpus, so the zero rows are the point.
+        record_retrieval(
+            user_id=user_id, source="prompt_rule", query=q,
+            threshold=threshold, limit=PROMPTRULE_LIMIT,
+            project_id=project_id,
+            is_task=None, results=fresh, duration_ms=duration_ms,
+            best_available=_rep.get("best_available_score"),
+            best_available_id=_rep.get("best_available_id"),
+            searched=bool(_rep.get("searched", True)),
+            suppressed=len(hits) - len(fresh),
+        )
+        # THE RESERVED SLOT RUNS BEFORE THE BAIL-OUT, and that ordering is
+        # load-bearing rather than tidy. An empty general result is not proof
+        # that no preference qualifies: the general search overfetches by
+        # distance and then collapses, so a preference ranked below that
+        # window is invisible to it while a kind-filtered query finds it at
+        # once. Bailing first would make the slot dead in exactly the corpus
+        # it exists for — one where rules outnumber preferences.
+        hits, slot_id = await _reserve_slot_for_preference(
+            user_id, q, hits, threshold=threshold,
+            project_id=project_id, already=already,
+        )
+
+        # `hits`, not `fresh` (#3750): a call whose only hit is a repeat still
+        # has something to say, it just says it differently.
+        if not hits:
+            return out
+
+        lines = [
+            _rule_hint_line(rule, where="to this request", seen=rule.id in already)
+            for _score, rule in hits
+        ]
+        # FRESH-ONLY (#3752). A reference is a rendering decision, not a
+        # retrieval outcome, and counting one here would inflate the
+        # denominator pull_through is read from.
+        #
+        # `fresh` is the PRE-SLOT list on purpose: it is exactly what this
+        # call's own `retrieval_logs` row recorded, so the two stay equal
+        # (#3668). The slot's hit is surfaced under `preference_slot` by the
+        # helper, against that source's own row.
+        rule_ids = [rule.id for _score, rule in fresh]
+
+        # RANKED, not ambient: this arm chose what it showed. The name is also
+        # in `rule_usage.RANKED_SOURCES`, and it has to be — a ranked source
+        # missing from that tuple is counted as a bulk delivery nobody decided
+        # on, which silently moves it out of the pull-through denominator.
+        if rule_ids:
+            record_rule_surfaced(
+                user_id=user_id, rule_ids=rule_ids, source="prompt_rule",
+            )
+        out["context"] = "\n".join(lines)
+        out["rule_ids"] = rule_ids
+    except Exception:
+        logger.debug("prompt rule arm failed", exc_info=True)
+    return out
+
+
 # --- Write-path trigger (#2082): prior art at the moment code is written ------
 # Auto-inject above fires on the operator's prompt. The moment reuse is actually
 # lost is later — when the AGENT decides mid-task to write a helper — and nothing

src/scribe/services/rule_usage.py

@@ -97,7 +97,15 @@ logger = logging.getLogger(__name__)
 # surfacing is a claim ("this rule may apply to what you are doing") that a pull
 # can confirm or refute, while an ambient one is a delivery nobody decided on.
 # Add a source here only when a ranker picked it.
-RANKED_SOURCES = ("write_path_rule", "pre_tool_rule")
+RANKED_SOURCES = (
+    "write_path_rule", "pre_tool_rule", "prompt_rule",
+    # A reserved slot is a ranker's choice twice over — it ran a query AND
+    # decided a kind was worth guaranteeing a place. Left out, its line would
+    # be counted as bulk delivery and drop out of the denominator, so the one
+    # surface built because a record class kept losing would be the one whose
+    # hits nobody could confirm.
+    "preference_slot",
+)
 
 
 def is_ambient(source: str) -> bool:

tests/test_rule_usage_wiring.py

@@ -444,6 +444,165 @@ async def _run_tool_arm(hits, recorder, command="curl -s https://git.example/api
         return await pc.build_tool_rule_hint(1, tool, command, **kwargs)
 
 
+# ── the prompt-boundary arm (#3852) ─────────────────────────────────────
+#
+# The third arm, and it joins _ARMS rather than getting a test file of its
+# own. That is the point of the shared parametrisation: #3497's history is
+# that the pre-tool arm inherited a defect from its sibling by being MODELLED
+# on it instead of sharing with it, and a third arm modelled on two is two
+# chances to repeat that. Everything in the family — repeat rendering,
+# fresh-only counting, the log-before-bailout order, the kind register, the
+# two recorders reading one list — is a property of every arm or of none.
+#
+# Fewer patches than its siblings because it does less: no prior-art menu, no
+# config object, no concept query. Just a bar, a search, and two recorders.
+
+
+def _prompt_patches(pc, hits, recorder, retrieval_log=None):
+    return (
+        # The arm reads its own threshold key rather than a shared config
+        # object — a third corpus with a bar nothing has yet tuned for it.
+        patch.object(pc, "get_setting", AsyncMock(return_value="0.6")),
+        patch.object(pc, "semantic_search_rules", AsyncMock(return_value=hits)),
+        patch.object(pc, "record_retrieval", retrieval_log or MagicMock()),
+        patch.object(pc, "record_rule_surfaced", recorder),
+    )
+
+
+async def _run_prompt_arm(hits, recorder, prompt="please merge to main",
+                          retrieval_log=None, **kwargs):
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        for ctx in _prompt_patches(pc, hits, recorder, retrieval_log=retrieval_log):
+            stack.enter_context(ctx)
+        return await pc.build_prompt_rule_hint(1, prompt, **kwargs)
+
+
+@pytest.mark.asyncio
+async def test_the_prompt_arm_retrieves_against_what_the_operator_SAID():
+    """The gap this arm closes, stated as the thing that had no trigger.
+
+    Both other arms are keyed on an act — a file write, a command. A rule that
+    governs what to SAY has no act in front of it: extract intent from loose
+    phrasing, raise a conflict before acting, end a finding with an offer all
+    bind on a response. Before this, the operator's message reached only
+    `semantic_search_notes`, so no rule had ever been retrieved against a
+    thing the operator actually said.
+    """
+    rec = MagicMock()
+    search = AsyncMock(return_value=[(0.79, fake_rule(
+        id=2, title="`main` — never without explicit request",
+        when_to_apply="opening or merging a dev→main pull request",
+    ))])
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        stack.enter_context(patch.object(pc, "record_retrieval", MagicMock()))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        out = await pc.build_prompt_rule_hint(1, "please merge to main")
+
+    # The PROMPT is the query — not a path, not a command.
+    assert search.call_args.args[1] == "please merge to main"
+    assert "get_rule(2)" in out["context"]
+    assert rec.call_args.kwargs["source"] == "prompt_rule"
+
+
+@pytest.mark.asyncio
+async def test_the_prompt_arm_addresses_the_request_not_a_tool_call():
+    """`where` has to name the moment, and this arm's moment is the asking.
+
+    "may apply to this Bash call" would be a lie here — there is no Bash call,
+    which is the entire reason the arm exists.
+    """
+    out = await _run_prompt_arm(
+        [(0.79, fake_rule(id=77, title="Extract intent from loose phrasing"))],
+        MagicMock(),
+    )
+    assert "may apply to this request" in out["context"], out["context"]
+
+
+@pytest.mark.asyncio
+async def test_the_prompt_arm_says_nothing_when_asked_nothing():
+    """A blank prompt is not a query, and searching on one would put a row in
+    retrieval_logs that no operator action produced."""
+    search = AsyncMock(return_value=[])
+    log = MagicMock()
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        stack.enter_context(patch.object(pc, "record_retrieval", log))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", MagicMock()))
+        out = await pc.build_prompt_rule_hint(1, "   ")
+
+    assert out == {"context": "", "rule_ids": []}
+    search.assert_not_called()
+    log.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_prompt_arm_logs_the_calls_that_found_nothing():
+    """#3497's defect, pinned on the arm that most needs it.
+
+    This bar is INHERITED from the act arms and unverified against prose. The
+    zero rows are therefore the whole evidence base for whether 0.72 belongs
+    here at all — an arm that logged only the calls it liked would report a
+    flawless clear-rate however wrong the number is.
+    """
+    log = MagicMock()
+    out = await _run_prompt_arm([], MagicMock(), retrieval_log=log)
+
+    assert out["context"] == ""
+    # BY SOURCE, not by count. The reserved slot (#3894) logs its own query on
+    # the same call, so a bare call_count would pin the number of arms rather
+    # than the property — and would go red the next time one is added, which
+    # is rule 167's false alarm about the very thing being protected.
+    general = [c for c in log.call_args_list
+               if c.kwargs.get("source") == "prompt_rule"]
+    assert len(general) == 1, (
+        "the prompt arm returned early without logging a call that found "
+        "nothing — the only evidence its inherited threshold is too high"
+    )
+    assert general[0].kwargs["results"] == []
+
+
+@pytest.mark.asyncio
+async def test_turning_off_the_notes_menu_does_not_turn_off_rules():
+    """Why this is a separate function and not a branch in the notes arm.
+
+    `build_autoinject_hint` returns early when auto-inject is disabled, when
+    the query is blank, and when nothing clears the note bar. Every one of
+    those is a statement about NOTES. Folded together, an operator who turned
+    the awareness menu off would silently stop receiving RULES — a coupling
+    with no symptom, since both failure modes look like a quiet hook.
+
+    Pinned as "the rule arm never asks the notes arm's config", which is the
+    structural fact rather than a simulation of the setting. A future refactor
+    that reaches for that config here fails, whatever it then does with it.
+    """
+    from scribe.services import plugin_context as pc
+
+    notes_cfg = AsyncMock(return_value={
+        "enabled": False, "threshold": 0.55, "top_k": 3,
+    })
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_autoinject_config", notes_cfg))
+        for ctx in _prompt_patches(
+            pc, [(0.79, fake_rule(id=2, title="`main` — never without explicit request"))],
+            MagicMock(),
+        ):
+            stack.enter_context(ctx)
+        out = await pc.build_prompt_rule_hint(1, "please merge to main")
+
+    assert "get_rule(2)" in out["context"], (
+        "the rule arm went quiet while the notes menu was disabled — the two "
+        "are different claims with different costs of being missed, and one "
+        "operator setting should not silence both"
+    )
+    notes_cfg.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_the_tool_arm_names_a_rule_for_the_command_about_to_run():
     """The 2026-09-03 incident in one test: reaching for curl against the forge
@@ -871,7 +1030,14 @@ _THREE_HITS = [
     (0.74, fake_rule(id=161, title="Reach the forge through its MCP tools")),
 ]
 
-_ARMS = [("write_path_rule", _run_arm), ("pre_tool_rule", _run_tool_arm)]
+_ARMS = [
+    ("write_path_rule", _run_arm),
+    ("pre_tool_rule", _run_tool_arm),
+    # The prompt arm joins the family rather than being modelled on it — see
+    # the block above _prompt_patches for why that distinction is the whole
+    # lesson of #3497.
+    ("prompt_rule", _run_prompt_arm),
+]
 
 
 def _both_ends(log, rec, source):
@@ -896,7 +1062,7 @@ def _both_ends(log, rec, source):
     return logged, surfaced
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.parametrize(
     ("excluded", "expected"),
     [([], 3), ([157], 2), ([156, 157, 161], 0)],
@@ -954,7 +1120,7 @@ _FRESH_TAIL = "not in this session's loaded set"
 _SEEN_TAIL = "You saw it earlier this session"
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_a_rule_already_on_the_ledger_still_produces_a_line(source, run):
     """THE REGRESSION, stated as the thing that used to be absent.
@@ -983,7 +1149,7 @@ async def test_a_rule_already_on_the_ledger_still_produces_a_line(source, run):
     )
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_the_two_tails_are_distinguishable_and_say_the_true_one(source, run):
     """One clause differs, and it is the clause that would otherwise be false.
@@ -1029,7 +1195,7 @@ _RULE_FORCE = "before deciding it does not apply"
 _PREF_FORCE = "for how this has been done before"
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_a_preference_does_not_speak_in_the_rules_voice(source, run):
     """The register, pinned on the two places force is actually asserted.
@@ -1053,7 +1219,7 @@ async def test_a_preference_does_not_speak_in_the_rules_voice(source, run):
     )
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_kind_and_seen_do_not_read_each_other(source, run):
     """The structural claim the design rests on: two INDEPENDENT axes.
@@ -1093,7 +1259,7 @@ async def test_kind_and_seen_do_not_read_each_other(source, run):
     )
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_neither_tail_injects_the_rule_statement(source, run):
     """The budget, pinned on both branches.
@@ -1138,7 +1304,7 @@ async def test_neither_tail_injects_the_rule_statement(source, run):
 # checkable, so it is asserted rather than described.
 
 
-@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool", "prompt"])
 @pytest.mark.asyncio
 async def test_a_reference_is_rendered_but_not_counted(source, run):
     """The counters must read exactly as they did before #3750."""
@@ -1174,3 +1340,217 @@ async def test_a_reference_is_rendered_but_not_counted(source, run):
         "whole context will not do: the write-path arm fills it from four "
         "other sources."
     )
+
+
+# ── the reserved preference slot (#3894) ────────────────────────────────
+#
+# A rule and a preference are not equally served by one ranking, because their
+# losses are not equal. A rule crowded out at the prompt boundary still fires
+# at an act arm — a push reaches pre_tool_rule, a write reaches
+# write_path_rule. A preference about how to ANSWER has no later act: the
+# response is the act, so crowded out here it is never delivered at all.
+#
+# The failure is invisible without this slot. The rule that won is a
+# legitimate hit, the telemetry reads healthy, and the only symptom is a
+# preference that quietly never arrives — which is `reuse_slot`'s shape one
+# corpus over (#2463), where snippets kept losing to project records that
+# merely resembled the query.
+
+
+def _search_by_kind(general, preference):
+    """Stand in for the two calls the arm makes against one corpus.
+
+    The arm searches twice: once across every kind, once filtered to
+    preferences for the slot. A single return value cannot tell those apart,
+    and a test that could not tell them apart would pass against an arm that
+    never filtered at all — which is the one thing making the slot a slot.
+    """
+    async def _search(*_args, **kwargs):
+        return preference if kwargs.get("kind") == "preference" else general
+    return AsyncMock(side_effect=_search)
+
+
+async def _run_slot(general, preference, recorder=None, retrieval_log=None, **kwargs):
+    from scribe.services import plugin_context as pc
+    rec = recorder or MagicMock()
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
+        stack.enter_context(patch.object(
+            pc, "semantic_search_rules", _search_by_kind(general, preference)))
+        stack.enter_context(patch.object(
+            pc, "record_retrieval", retrieval_log or MagicMock()))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        out = await pc.build_prompt_rule_hint(1, "please merge to main", **kwargs)
+    return out, rec
+
+
+_PREF_HIT = [(0.74, fake_rule(
+    id=140, kind="preference", title="Let each action land before the next",
+    when_to_apply="before starting an action while a previous one is settling",
+))]
+_RULES_FILLING_THE_LIMIT = [
+    (0.81, fake_rule(id=2, title="`main` — never without explicit request")),
+    (0.79, fake_rule(id=1, title="`dev` is home")),
+    (0.77, fake_rule(id=153, title="Merge dev→main with a plain merge commit")),
+]
+
+
+@pytest.mark.asyncio
+async def test_a_preference_that_lost_the_ranking_still_gets_a_line():
+    """The whole point. Three rules fill the limit; the preference places
+    fourth on score and would never be seen without the slot."""
+    out, _rec = await _run_slot(_RULES_FILLING_THE_LIMIT, _PREF_HIT)
+
+    assert "get_rule(140)" in out["context"], (
+        "a preference cleared the bar, placed behind the rules, and was "
+        "dropped — which is the outcome with no symptom: the rules that won "
+        "are legitimate hits and nothing in the telemetry looks wrong"
+    )
+    assert "Preference that may apply" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_the_slot_is_not_spent_when_a_preference_already_placed():
+    """A guaranteed slot is a floor, not a quota. A preference that earned its
+    place on score does not entitle the corpus to a second one."""
+    general = [(0.81, fake_rule(id=140, kind="preference", title="Let each action land"))]
+    search_log = MagicMock()
+    out, _rec = await _run_slot(general, _PREF_HIT, retrieval_log=search_log)
+
+    sources = [c.kwargs.get("source") for c in search_log.call_args_list]
+    assert "preference_slot" not in sources, (
+        "the slot ran while a preference had already placed — a second "
+        "reserved line for a kind already represented is noise the general "
+        "ranking had already decided against"
+    )
+    assert out["context"].count("Preference that may apply") == 1
+
+
+@pytest.mark.asyncio
+async def test_the_slot_query_can_only_answer_with_a_preference():
+    """Filtered at the QUERY, not verified afterwards.
+
+    An unfiltered search that happened to return a rule would spend the slot
+    on it, and that line would be indistinguishable from one that earned its
+    place on score — a slot silently spent on the wrong kind is worse than no
+    slot at all.
+    """
+    from scribe.services import plugin_context as pc
+    search = _search_by_kind(_RULES_FILLING_THE_LIMIT, _PREF_HIT)
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_setting", AsyncMock(return_value="0.6")))
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        stack.enter_context(patch.object(pc, "record_retrieval", MagicMock()))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", MagicMock()))
+        await pc.build_prompt_rule_hint(1, "please merge to main")
+
+    kinds = [c.kwargs.get("kind") for c in search.await_args_list]
+    assert "preference" in kinds, "the slot searched without filtering by kind"
+    assert kinds.count("preference") == 1, "the slot searched more than once"
+
+
+@pytest.mark.asyncio
+async def test_the_slot_runs_even_when_the_general_search_found_nothing():
+    """The ordering the arm gets wrong by default.
+
+    An empty general result is not proof no preference qualifies: that search
+    overfetches by distance and then collapses, so a preference ranked below
+    the window is invisible to it while a kind-filtered query finds it at
+    once. Bailing out first would make the slot dead in exactly the corpus it
+    exists for — one where rules outnumber preferences.
+    """
+    out, _rec = await _run_slot([], _PREF_HIT)
+    assert "get_rule(140)" in out["context"], (
+        "the arm returned early on an empty general result and never asked "
+        "for a preference"
+    )
+
+
+@pytest.mark.asyncio
+async def test_the_slot_extends_rather_than_displacing():
+    """Nothing the general search returned goes un-shown.
+
+    `reuse_slot` evicts its menu's weakest hit; this one does not, and the
+    reason is the ledger. A displaced hit sits in `prompt_rule`'s
+    retrieval_logs row while never being surfaced, so that source's two
+    tables stop agreeing — and #3668's identity is the cheapest true
+    statement available about this pair. Milestone #379 is what losing it
+    costs: five steps planned against two counters disagreeing.
+    """
+    log, rec = MagicMock(), MagicMock()
+    out, _ = await _run_slot(
+        _RULES_FILLING_THE_LIMIT, _PREF_HIT, recorder=rec, retrieval_log=log,
+    )
+
+    for rule_id in (2, 1, 153):
+        assert f"get_rule({rule_id})" in out["context"], (
+            f"rule {rule_id} was returned and logged, then pushed out by the "
+            f"slot — surfaced and logged now disagree for prompt_rule"
+        )
+
+    logged = [
+        r.id for c in log.call_args_list if c.kwargs.get("source") == "prompt_rule"
+        for _s, r in c.kwargs["results"]
+    ]
+    surfaced = [
+        rid for c in rec.call_args_list if c.kwargs.get("source") == "prompt_rule"
+        for rid in c.kwargs["rule_ids"]
+    ]
+    assert logged == surfaced, (
+        f"prompt_rule logged {logged} and surfaced {surfaced} — the identity "
+        f"#3668 pins, broken by the slot rather than by a write path"
+    )
+
+
+@pytest.mark.asyncio
+async def test_the_slot_accounts_for_itself_under_its_own_source():
+    """Both sides of the trade, logged.
+
+    #2463's own finding is the warning rather than the precedent: the hit
+    that slot pushed OUT was in retrieval_logs while the query that pushed it
+    out was not, so the slot could never be judged against what it displaced.
+    This one logs its query AND records its surfacing, under a source of its
+    own, so it can be evaluated separately from the ranking it bypassed.
+    """
+    log, rec = MagicMock(), MagicMock()
+    await _run_slot(_RULES_FILLING_THE_LIMIT, _PREF_HIT,
+                    recorder=rec, retrieval_log=log)
+
+    slot_logs = [c for c in log.call_args_list
+                 if c.kwargs.get("source") == "preference_slot"]
+    assert len(slot_logs) == 1, "the slot ran without logging its own query"
+    assert [r.id for _s, r in slot_logs[0].kwargs["results"]] == [140]
+
+    slot_surfacings = [c for c in rec.call_args_list
+                       if c.kwargs.get("source") == "preference_slot"]
+    assert len(slot_surfacings) == 1
+    assert slot_surfacings[0].kwargs["rule_ids"] == [140]
+
+
+@pytest.mark.asyncio
+async def test_a_preference_on_the_ledger_keeps_the_slot_and_is_not_recounted():
+    """Repeats hold the slot; they just do not count twice.
+
+    A preference is the kind of record where being reminded is the point, so
+    one already on the ledger occupies the slot rather than being skipped for
+    a fresh one — rendered with the repeat tail (#3750). What it must not do
+    is register a second surfacing, which would count one delivery twice in
+    the denominator pull-through is read from (#3752).
+    """
+    log, rec = MagicMock(), MagicMock()
+    out, _ = await _run_slot(
+        _RULES_FILLING_THE_LIMIT, _PREF_HIT,
+        recorder=rec, retrieval_log=log, exclude_rule_ids=[140],
+    )
+
+    assert "get_rule(140)" in out["context"]
+    assert _SEEN_TAIL in out["context"], "the repeat was rendered as a first surfacing"
+    assert not [c for c in rec.call_args_list
+                if c.kwargs.get("source") == "preference_slot"], (
+        "a preference the session had already been shown was counted as a "
+        "fresh surfacing"
+    )
+    assert 140 not in out["rule_ids"], (
+        "a repeat was written back to the hook's ledger, which would keep "
+        "pushing its stamp forward so it never aged out (#3751)"
+    )