feat(rules): a slot a preference cannot lose (#3894)

Milestone 399 step 4. 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. The prompt
    hit is a preview of a second chance.
  - a PREFERENCE about how to answer has no second chance. The response IS
    the act, so crowded out there 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 does it INVISIBLY: the rule that won is
a legitimate hit, the telemetry reads healthy, and the only symptom is a
preference that quietly never arrives. reuse_slot exists for the same shape
one corpus over (#2463).

`semantic_search_rules` gains a `kind` filter, so the slot's query can only
answer with what the slot is for. Verifying afterwards would be weaker — 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.

THE SLOT BUYS POSITION, NOT A LOWER BAR, matching reuse_slot. A weak
preference cannot buy it, so silence stays the default. The task asked for a
separate threshold; I did not add one, and the reason is that the worry
behind it — reading a miss rate as a fact about preferences — is answered by
`preference_slot` being its own logged source, where best_available_id names
which preference was refused. A knob added on a guess is a way to
misconfigure the surface; a bar moved on evidence is an argument. The
evidence arrives on its own now.

IT EXTENDS, IT NEVER DISPLACES — and here it parts from reuse_slot, which
evicts its menu's weakest hit. 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 breaks for a reason nothing in the data
explains. Milestone #379 is what losing that identity costs: five steps
planned against two counters disagreeing, not a write path dropping rows. One
extra line in a rare case is the cheaper price.

It also runs BEFORE the bail-out. An empty general result is not proof no
preference qualifies: that search overfetches by distance then collapses, so
a preference ranked below the 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 existing assertion repinned from a bare call_count to a per-source
filter: the slot logs its own query on the same call, and a count would pin
the number of arms rather than the property — going red the next time one is
added, which is rule 167's false alarm about the thing it protects.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ

tests/test_rule_usage_wiring.py

@@ -554,12 +554,17 @@ async def test_the_prompt_arm_logs_the_calls_that_found_nothing():
     out = await _run_prompt_arm([], MagicMock(), retrieval_log=log)
 
     assert out["context"] == ""
-    assert log.call_count == 1, (
+    # 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 log.call_args.kwargs["source"] == "prompt_rule"
-    assert log.call_args.kwargs["results"] == []
+    assert general[0].kwargs["results"] == []
 
 
 @pytest.mark.asyncio
@@ -1335,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)"
+    )