refactor(retrieval): the completion-report preferences run through the pipeline (milestone 456 step 3, #4905)

reply_preferences.completion_preferences was the last hand-written
rule search + record_retrieval + record_rule_surfaced triple outside the
pipeline. It is now REPORT_PREFERENCE, a RuleArm with kind="preference"
over COMPLETION_QUERY, read back as records rather than as lines.

- RuleArm gains `kind`. _ranked asks the ranker for that kind and drops
  anything else on the way out, before the row is logged.
- RuleResult gains `shown`, the (score, rule) pairs behind the lines, for
  callers that return records.
- RuleMoment.project_id may be None: logged as given, searched as
  `project_id or None`. report_preference rows keep their NULL project.
- Guards: reply_preferences now has 0 direct rule searches. The registry
  constant-source example moves from report_preference to preference_slot,
  the pipeline slot that records through a module constant.

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

src/scribe/services/reply_preferences.py

@@ -58,8 +58,8 @@ bar can be wrong forever without a single call looking unusual.
 from __future__ import annotations
 
 import logging
-import time
 
+from scribe.services import retrieval_pipeline as rp
 from scribe.services.embeddings import semantic_search_rules
 from scribe.services.retrieval_surfaces import SURFACES, budget_for, floor_for
 from scribe.services.retrieval_telemetry import record_retrieval
@@ -67,7 +67,7 @@ from scribe.services.rule_usage import record_rule_surfaced
 
 logger = logging.getLogger(__name__)
 
-SOURCE = "report_preference"
+SOURCE = rp.REPORT_PREFERENCE.source
 
 # Written in the vocabulary of the MOMENT, because that is what a trigger is
 # written in and what this query is scored against. Domain-neutral on purpose
@@ -114,34 +114,27 @@ async def completion_preferences(user_id: int, *, project_id: int | None = None)
     happened, and a lookup that errors must not turn it into a failure.
     """
     try:
-        threshold = await _threshold(user_id)
-        limit = await _limit(user_id)
-        report: dict = {}
-        t0 = time.perf_counter()
-        hits = await semantic_search_rules(
-            user_id, COMPLETION_QUERY, limit=limit, threshold=threshold,
-            kind="preference", report=report, project_id=project_id,
-        )
-        hits = [(score, rule) for score, rule in hits if rule.kind == "preference"]
-        record_retrieval(
-            user_id=user_id, source=SOURCE, query=COMPLETION_QUERY,
-            threshold=threshold, limit=limit, project_id=project_id,
-            is_task=None, results=hits,
-            best_available=report.get("best_available_score"),
-            best_available_id=report.get("best_available_id"),
-            searched=bool(report.get("searched", True)),
-            duration_ms=(time.perf_counter() - t0) * 1000.0,
-        )
-        if not hits:
-            return []
+        # The stages — the kind filter, the unconditional call row, the
+        # fresh-only surfacing rows — are the one pipeline's (milestone 456).
         # RANKED: this surface chose what it showed, so the name is in
         # rule_usage.RANKED_SOURCES and its hits count toward pull-through.
-        record_rule_surfaced(
-            user_id=user_id, rule_ids=[rule.id for _s, rule in hits], source=SOURCE,
+        # There is no session ledger here: a completion report is written
+        # once, so nothing it could repeat has been shown before.
+        result = await rp.run_rule_arm(
+            rp.REPORT_PREFERENCE,
+            rp.RuleMoment(
+                user_id=user_id, query=COMPLETION_QUERY, project_id=project_id,
+            ),
+            floor=await _threshold(user_id), budget=await _limit(user_id),
+            io=rp.RuleIO(
+                search=semantic_search_rules,
+                record_retrieval=record_retrieval,
+                record_rule_surfaced=record_rule_surfaced,
+            ),
         )
         return [
             {"id": rule.id, "title": rule.title, "statement": rule.statement, "kind": "preference"}
-            for _score, rule in hits
+            for _score, rule in result.shown
         ]
     except Exception:  # noqa: BLE001 - a decoration never breaks its payload
         logger.warning("completion preference lookup failed", exc_info=True)

src/scribe/services/retrieval_pipeline.py

@@ -351,6 +351,9 @@ class RuleArm:
     preference_slot: bool
     """Reserve one line for a preference that lost the ranking (#3894)."""
 
+    kind: str | None = None
+    """Ask the ranker for one record kind only ("preference"), or every kind."""
+
 
 # Today's differences, reproduced exactly (milestone 456 step 2). Whether the
 # prompt arm should band, and whether the act arms should reserve a
@@ -367,7 +370,16 @@ WRITE_PATH_RULE = RuleArm(
     "write_path_rule", band=True, compact_tail=True, checkpoint=True,
     preference_slot=False,
 )
-RULE_ARMS: tuple[RuleArm, ...] = (WRITE_PATH_RULE, PRE_TOOL_RULE, PROMPT_RULE)
+# The completion report's preferences (milestone 409 step 4): a FIXED query,
+# preferences only, read by update_task as records rather than as lines. Its
+# query is `reply_preferences.COMPLETION_QUERY`.
+REPORT_PREFERENCE = RuleArm(
+    "report_preference", band=False, compact_tail=False, checkpoint=False,
+    preference_slot=False, kind="preference",
+)
+RULE_ARMS: tuple[RuleArm, ...] = (
+    WRITE_PATH_RULE, PRE_TOOL_RULE, PROMPT_RULE, REPORT_PREFERENCE,
+)
 
 PREFERENCE_SLOT_SOURCE = "preference_slot"
 
@@ -396,8 +408,11 @@ class RuleMoment:
 
     user_id: int
     query: str
-    project_id: int
-    where: str
+    project_id: int | None
+    """The bound project, 0 or None when unbound. Logged as given, searched as
+    `project_id or None` — global rules plus that project's own (milestone 414)."""
+
+    where: str = ""
     """How a line names the moment: "here", "to this Bash call", …"""
 
     checkpoint_where: str = ""
@@ -419,6 +434,10 @@ class RuleResult:
     shown_rule_ids: list[int] = field(default_factory=list)
     """Every rule LINE, repeats and the reserved slot included."""
 
+    shown: list = field(default_factory=list)
+    """The (score, rule) pairs behind those lines, best first — for a caller
+    that hands back records rather than lines (the completion report)."""
+
     checkpoint: dict = field(default_factory=dict)
 
 
@@ -462,6 +481,13 @@ async def _ranked(
         # slot on it, indistinguishable from a line that earned its place.
         kwargs["kind"] = kind
     hits = await io.search(moment.user_id, moment.query, **kwargs)
+    if kind:
+        # And checked on the way out: a record of another kind that slipped
+        # through must not be logged, shown or counted under a source whose
+        # name claims the kind — a rule handed back as "how the operator likes
+        # this done" asserts a force the record does not have.
+        hits = [(score, rule) for score, rule in hits
+                if getattr(rule, "kind", None) == kind]
     kept = _rule_band(hits) if band else hits
     fresh = [(score, rule) for score, rule in kept if rule.id not in moment.exclude]
     try:
@@ -578,7 +604,7 @@ async def run_rule_arm(
     try:
         _hits, kept, fresh = await _ranked(
             io, moment, source=arm.source, floor=floor, limit=budget,
-            band=arm.band,
+            band=arm.band, kind=arm.kind,
         )
         shown = kept
         if arm.preference_slot:
@@ -631,6 +657,7 @@ async def run_rule_arm(
             lines=lines, rule_ids=rule_ids,
             shown_rule_ids=[rule.id for _score, rule in shown],
             checkpoint=checkpoint,
+            shown=list(shown),
         )
     except Exception:  # noqa: BLE001 - a recall aid never breaks its act
         logger.debug("%s arm failed", arm.source, exc_info=True)

src/scribe/services/retrieval_registry.py

@@ -34,8 +34,8 @@ every tunable surface also appears here, so the two cannot drift apart.
 ADDING AN ARM MEANS ADDING A ROW HERE. `tests/test_retrieval_registry.py`
 walks the call sites with the ast module and fails on a source it cannot find
 below — deliberately not a grep, because two of the sources in this file
-(`wide_net`, `report_preference`) reach their recorder as `source=SOURCE`
-through a module constant and a grep for `source="` misses both. That is the
+(`wide_net`, `preference_slot`) reach their recorder through a
+module constant and a grep for `source="` misses both. That is the
 narrowing #3191 warns about, caught here in the act.
 """
 from __future__ import annotations