align(4909c): the command and write rule arms reserve a line for a preference

PRE_TOOL_RULE and WRITE_PATH_RULE now take preference_slot=True, like
PROMPT_RULE. When no preference placed on score, the arm asks once more for
the best preference over the same bar, and shows it as one extra line.

Operator ruling (2026-10-05): yes. A preference about how to act ("let each
action land before the next") is ranked against every rule whose trigger
names the same command; the act is its last chance to shape anything.

Cost: one kind-filtered search per call that placed no preference. The
query embedding is memoised per request on both hook routes, so no extra
model call. Expect a new stream of preference_slot rows from the two act
arms; preference_slot's call count will rise roughly by pre_tool_rule's and
write_path_rule's (~23k calls over 14 days before this change), and its
zero-result share will rise with it, since most commands have no matching
preference.

Tests: the tool-arm call-row tests count rows under pre_tool_rule only (the
slot logs under its own name), and the write-arm bar test reads the arm's
own search, which is the first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/grimoire/services/retrieval_pipeline.py

@@ -474,9 +474,10 @@ class RuleArm:
 RULE_FLOOR_KEY = "kb_rule_threshold"
 RULE_FLOOR_DEFAULT = 0.70
 
-# The arms differ on purpose where these flags differ. Whether the prompt arm
-# should band, and whether the act arms should reserve a preference, are open
-# questions to settle on telemetry, not by symmetry.
+# The arms differ on purpose where these flags differ: the prompt arm takes no
+# band, so its budget alone decides how many hits it shows, and the act arms
+# band and may hold the act. Every arm that annotates reserves a line for a
+# preference.
 PROMPT_RULE = RuleArm(
     "prompt_rule", band=False, compact_tail=False, checkpoint=False,
     preference_slot=True,
@@ -494,7 +495,7 @@ PROMPT_RULE = RuleArm(
 )
 PRE_TOOL_RULE = RuleArm(
     "pre_tool_rule", band=True, compact_tail=True, checkpoint=True,
-    preference_slot=False,
+    preference_slot=True,
     tuning=Surface(
         name="pre_tool_rule",
         floor_key=RULE_FLOOR_KEY,
@@ -509,7 +510,7 @@ PRE_TOOL_RULE = RuleArm(
 )
 WRITE_PATH_RULE = RuleArm(
     "write_path_rule", band=True, compact_tail=True, checkpoint=True,
-    preference_slot=False,
+    preference_slot=True,
     tuning=Surface(
         name="write_path_rule",
         floor_key=RULE_FLOOR_KEY,
@@ -703,6 +704,13 @@ async def _reserve_slot_for_preference(
     same shape one corpus over, where snippets lose to project records that
     merely resemble the query.
 
+    The act arms reserve the slot for the same reason one step later. A
+    preference about how to act ("let each action land before the next") is
+    ranked against every rule whose trigger names the same command, and the
+    act is its last chance: once the command has run there is nothing left
+    for it to shape. The slot's query reuses the arm's query, so it costs one
+    kind-filtered search and no extra embedding.
+
     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

tests/test_rule_usage_wiring.py

@@ -161,7 +161,8 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
             1, "frontend/src/api/client.ts", code="x" * 400,
         )
 
-    kw = search.await_args.kwargs
+    # The first search is the arm's own; the preference slot asks after it.
+    kw = search.await_args_list[0].kwargs
     assert kw["threshold"] == 0.81, "the arm is still using the code threshold"
     assert kw["limit"] == pc.RULEHINT_LIMIT
     # NO CATEGORY FILTER, on the axis that exists: the act arms narrow by
@@ -834,10 +835,11 @@ async def test_the_tool_arm_logs_the_call_that_found_nothing():
     out = await _run_tool_arm([], rec, retrieval_log=log)
 
     assert out == {"context": "", "rule_ids": [], "checkpoint": {}}
-    assert log.call_count == 1
-    assert log.call_args.kwargs["source"] == "pre_tool_rule"
-    assert log.call_args.kwargs["results"] == []
-    assert log.call_args.kwargs["query"] == "curl -s https://git.example/api/v1/runs", (
+    # The arm's own row; the preference slot logs beside it under its own name.
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
+    assert len(rows) == 1
+    assert rows[0].kwargs["results"] == []
+    assert rows[0].kwargs["query"] == "curl -s https://git.example/api/v1/runs", (
         "the query is the point of the row: it is what a threshold is tuned against"
     )
     rec.assert_not_called()
@@ -850,8 +852,9 @@ async def test_the_tool_arm_logs_a_call_whose_only_hit_was_already_shown():
     out = await _run_tool_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[161])
 
     assert out["rule_ids"] == []
-    assert log.call_count == 1
-    assert log.call_args.kwargs["results"] == []
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
+    assert len(rows) == 1
+    assert rows[0].kwargs["results"] == []
     rec.assert_not_called()