refactor(retrieval): the three rule arms run through one pipeline (milestone 456 step 2, #4904)

prompt_rule, pre_tool_rule and the rule half of the write path each wrote
the same steps out by hand: search, band, split fresh from repeats, log the
call before any early return, reserve a slot, render, record surfacings.
The copies drifted, and #3497, #3750 and #3752 were fixed one copy at a time.

- New services/retrieval_pipeline.py. run_rule_arm runs the stages in one
  order. RuleArm is the spec (source, band, compact_tail, checkpoint,
  preference_slot). RuleMoment is the query and the session ledger.
- The I/O (ranker and recorders) is passed in as RuleIO. plugin_context
  resolves it from its own names at call time, so existing patch points
  still apply.
- _rule_band, _rule_hint_line, checkpoint_for, checkpoint_reason, the band
  constant and the preference slot moved into the pipeline unchanged.
  plugin_context re-exports them.
- A recorder that raises now costs only its row, never a rendered line.
- The flags reproduce today exactly. Whether prompt_rule should band, and
  whether the act arms should reserve a preference, are step 7.
- Registry: the pipeline call sites are declared in FAN_OUT_SITES, with
  values read from the specs.
- The #3497 structural guard now checks the one implementation, and that
  plugin_context writes no rule-source row of its own.

plugin_context.py: 3,558 -> 2,980 lines.

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

tests/test_rule_usage_wiring.py

@@ -887,80 +887,76 @@ async def test_a_command_the_arm_never_searched_writes_no_row_at_all():
     log.assert_not_called()
 
 
-def test_neither_rule_arm_logs_its_call_behind_a_results_guard():
-    """Structural, on top of the behavioural pair above, because the defect was
-    one level of indentation and it appeared INDEPENDENTLY in two places — the
-    pre-tool arm inherited it by being modelled on its sibling. The third arm
-    modelled on either of them is the one this catches.
+def test_no_rule_arm_logs_its_call_behind_a_results_guard():
+    """Structural, on top of the behavioural tests above, because the defect
+    was one level of indentation and it appeared INDEPENDENTLY in two arms —
+    the pre-tool arm inherited it by being modelled on its sibling (#3497).
+
+    Since milestone 456 there is one implementation to guard rather than a
+    copy per arm, so the property is checked where it is written: the
+    pipeline's ranked-search stage writes the call row unconditionally, the
+    arm runner calls that stage before it can return, and the surfacing row
+    alone stays behind `if rule_ids:`. And no arm writes a rule-source row of
+    its own any more — a fourth copy would be the defect's way back in.
     """
-    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
+    from scribe.services import retrieval_pipeline as rp
 
-    # Write-path arm: what remains inside `if fresh:` is the SURFACING log only.
-    guarded = pc_src.split('source="write_path_rule", query=code or path')[1]
-    guarded = guarded.split("if fresh:")[1].split("except Exception:")[0]
-    assert "record_rule_surfaced" in guarded, "the surfacing log must stay guarded"
-    assert "record_retrieval" not in guarded, (
-        "the call log is back inside the results guard — a call that found "
-        "nothing is the only evidence a threshold is set too high"
+    src = Path("src/scribe/services/retrieval_pipeline.py").read_text()
+    tree = ast.parse(src)
+    fns = {n.name: n for n in ast.walk(tree) if isinstance(n, ast.AsyncFunctionDef)}
+
+    def _calls(fn, name):
+        return [
+            n.lineno for n in ast.walk(fn)
+            if isinstance(n, ast.Call)
+            and (getattr(n.func, "attr", None) == name or getattr(n.func, "id", None) == name)
+        ]
+
+    # 1. The stage logs, and nothing in it can return before it does.
+    ranked = fns["_ranked"]
+    logged_at = _calls(ranked, "record_retrieval")
+    assert logged_at, "the ranked-search stage no longer writes the call row"
+    early = [n.lineno for n in ast.walk(ranked)
+             if isinstance(n, ast.Return) and n.lineno < min(logged_at)]
+    assert not early, (
+        f"the ranked-search stage can return at line {early[0]} before logging "
+        f"its call — a call that found nothing is the only evidence a bar is "
+        f"too high (#3497)"
     )
 
-    # Pre-tool arm: the call log comes BEFORE the early return.
-    #
-    # WALKED, NOT SUBSTRING-MATCHED (rule 167). This assertion used to read
-    # `body.index("if not fresh:")`, which pinned the name of a local variable
-    # rather than the property. #3750 changed that guard to `if not hits:` —
-    # the arm still logs before returning, so the property held perfectly, and
-    # a name-matching assertion would have raised ValueError and reported the
-    # #3497 defect as back. A guard that cries regression when the thing it
-    # protects is intact is the failure mode rule 167 names.
-    #
-    # The property is positional: between the search and the first guard that
-    # can return early, the call row has already been written.
-    # The arm's BODY, which since #4633 lives in `_tool_rule_hint`; the public
-    # `build_tool_rule_hint` is a wrapper that adds the via-lesson step and
-    # searches nothing itself. Located by what it does — the function that
-    # calls the rule search under the pre_tool_rule source — so the next
-    # rename moves the guard with it instead of emptying it.
-    fn = next(
-        n for n in ast.walk(ast.parse(pc_src))
-        if isinstance(n, ast.AsyncFunctionDef)
-        and any(
-            isinstance(c, ast.Call) and getattr(c.func, "id", None) == "semantic_search_rules"
-            for c in ast.walk(n)
-        )
-        and any(
-            isinstance(k, ast.keyword) and k.arg == "source"
-            and isinstance(k.value, ast.Constant) and k.value.value == "pre_tool_rule"
-            for k in ast.walk(n)
-        )
-    )
-    search_at = min(
-        n.lineno for n in ast.walk(fn)
-        if isinstance(n, ast.Call)
-        and getattr(n.func, "id", None) == "semantic_search_rules"
-    )
-    logged_at = min(
-        n.lineno for n in ast.walk(fn)
-        if isinstance(n, ast.Call)
-        and getattr(n.func, "id", None) == "record_retrieval"
-    )
-    # Every `if <cond>: return ...` after the search — whatever it tests.
+    # 2. The runner reaches the stage before its first early return.
+    run = fns["run_rule_arm"]
+    staged_at = min(_calls(run, "_ranked"))
     bailouts = [
-        n.lineno for n in ast.walk(fn)
-        if isinstance(n, ast.If) and n.lineno > search_at
-        and any(isinstance(b, ast.Return) for b in n.body)
+        n.lineno for n in ast.walk(run)
+        if isinstance(n, ast.If) and any(isinstance(b, ast.Return) for b in n.body)
     ]
     assert bailouts, (
-        "no early return found after the search in the pre-tool arm — the "
-        "guard has nothing left to protect, which means this test is now "
-        "passing vacuously rather than the arm being correct"
+        "no early return found in run_rule_arm — this check has nothing left "
+        "to protect and would pass vacuously"
     )
-    assert logged_at < min(bailouts), (
-        f"the pre-tool arm returns at line {min(bailouts)} before logging its "
-        f"call at line {logged_at} — a surface with no rows at all cannot be "
-        f"told apart from a hook that never fired (#3497)"
+    assert staged_at < min(bailouts), (
+        f"run_rule_arm returns at line {min(bailouts)} before the call row is "
+        f"written at line {staged_at}"
     )
 
+    # 3. The surfacing row is the guarded one.
+    guards = [n for n in ast.walk(run) if isinstance(n, ast.If)
+              and isinstance(n.test, ast.Name) and n.test.id == "rule_ids"]
+    assert guards and any(_calls(g, "record_rule_surfaced") for g in guards), (
+        "the surfacing row must stay behind `if rule_ids:` — nothing was shown, "
+        "so no surfacing occurred"
+    )
+
+    # 4. No arm keeps a private copy: outside the pipeline, nothing writes a
+    #    call row under a rule arm's source.
+    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
+    for arm in rp.RULE_ARMS:
+        assert f'source="{arm.source}"' not in pc_src, (
+            f"plugin_context writes a {arm.source} row of its own again — the "
+            f"arm is meant to be a spec over the one pipeline"
+        )
+
 
 # ── Suppression: which zeros were the ranker, which were repeats (#3497) ──
 #