refactor(retrieval): the rule builders compose one moment - _rule_moment runs the arm and its via-lesson step for all three (milestone 456 step 6, #4908)

build_prompt_rule_hint, build_tool_rule_hint and build_write_path_hint
each ran the rule arm and then the via-lesson step by hand, the first two
passing the query between them through a _via_query key on the payload.
They now call one composer, _rule_moment, and the split helpers
(_prompt_rule_hint, _tool_rule_hint, _add_rules_via_lessons) and the side
channel are deleted. Output shapes are unchanged: the prompt builder
returns no checkpoint key, the tool builder always does, and
shown_rule_ids stays the direct band.

A structural test pins that only _rule_moment runs a rule arm or the
via-lesson step, and that the three builders are its only callers.

plugin_context.py: 2,418 -> 2,361 lines this step; 3,558 when the
milestone began.

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

tests/test_rule_via_lesson.py

@@ -18,6 +18,7 @@ import pytest
 from scribe.services import plugin_context as pc
 from scribe.services import rule_usage
 from scribe.services.retrieval_registry import POINTS
+from scribe.services.retrieval_pipeline import RuleMoment, RuleResult
 
 # Bound before conftest's autouse stub replaces the module attribute.
 _REAL = pc._rules_via_lessons
@@ -123,23 +124,56 @@ async def test_a_failure_brings_no_rule_and_raises_nothing():
 
 
 @pytest.mark.asyncio
-async def test_the_step_runs_only_when_the_arm_ran_and_never_leaks_its_key():
+async def test_the_step_follows_the_arm_on_its_query_and_skips_what_it_named():
+    """One composer for all three rule builders (milestone 456 step 6): the
+    via-lesson step searches with the query the arm searched with, skips the
+    session's ledger and every rule the arm's band named, and its lines and
+    fresh ids join the arm's — while `shown_rule_ids` stays the direct band."""
     step = AsyncMock(return_value=(["Standing rule … Reached through lesson #41"], [7]))
+    direct = RuleResult(lines=["direct line"], rule_ids=[5], shown_rule_ids=[5, 6])
+    moment = RuleMoment(user_id=1, query="q", project_id=2, where="here",
+                        exclude=frozenset({3, 6}))
+    with patch.object(pc, "_rules_via_lessons", step), \
+         patch.object(pc.rp, "run_rule_arm", AsyncMock(return_value=direct)):
+        out = await pc._rule_moment(pc.rp.PROMPT_RULE, moment, floor=0.5, budget=3)
+    assert step.await_args.args[1] == "q"
+    assert step.await_args.kwargs["skip"] == {3, 5, 6}
+    assert out.rule_ids == [5, 7]
+    assert out.lines == ["direct line", "Standing rule … Reached through lesson #41"]
+    assert out.shown_rule_ids == [5, 6]
+
+
+@pytest.mark.asyncio
+async def test_a_blank_prompt_runs_neither_the_arm_nor_the_step():
+    step = AsyncMock(return_value=([], []))
     with patch.object(pc, "_rules_via_lessons", step):
-        idle = await pc._add_rules_via_lessons(
-            1, {"context": "", "rule_ids": []}, project_id=2,
-            exclude_rule_ids=[3], held_rule_ids=[], where="here",
-        )
-        ran = await pc._add_rules_via_lessons(
-            1, {"context": "direct line", "rule_ids": [5], "_via_query": "q"},
-            project_id=2, exclude_rule_ids=[3], held_rule_ids=[], where="here",
-        )
-    assert idle == {"context": "", "rule_ids": []}
-    assert step.await_count == 1
-    assert step.await_args.kwargs["skip"] == {3, 5}
-    assert "_via_query" not in ran
-    assert ran["rule_ids"] == [5, 7]
-    assert ran["context"].startswith("direct line\n")
+        out = await pc.build_prompt_rule_hint(1, "   ")
+    assert out == {"context": "", "rule_ids": []}
+    step.assert_not_awaited()
+
+
+def test_only_the_composer_runs_a_rule_arm_or_the_via_lesson_step():
+    """Structural (rule 167): every rule builder goes through `_rule_moment`,
+    so a builder that runs the arm or the via-lesson step by hand again is the
+    copy this step removed coming back — and a count is what lets it fail."""
+    import ast
+    from pathlib import Path
+
+    tree = ast.parse(Path("src/scribe/services/plugin_context.py").read_text())
+    callers: dict[str, set[str]] = {}
+    for fn in ast.walk(tree):
+        if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        for call in ast.walk(fn):
+            if isinstance(call, ast.Call):
+                name = getattr(call.func, "attr", None) or getattr(call.func, "id", None)
+                if name in ("run_rule_arm", "_rules_via_lessons", "_rule_moment"):
+                    callers.setdefault(name, set()).add(fn.name)
+    assert callers["run_rule_arm"] == {"_rule_moment"}
+    assert callers["_rules_via_lessons"] == {"_rule_moment"}
+    assert callers["_rule_moment"] == {
+        "build_prompt_rule_hint", "build_tool_rule_hint", "build_write_path_hint",
+    }
 
 
 def test_the_source_is_ranked_and_registered():

tests/test_system_rulings.py

@@ -22,6 +22,7 @@ import pytest
 
 from scribe.services import plugin_context as pc
 from scribe.services import system_rulings as sr
+from scribe.services.retrieval_pipeline import RuleResult
 # Bound at import, before conftest's autouse stub replaces the module attribute.
 from scribe.services.system_rulings import rulings_for_paths as real_rulings_for_paths
 from tests.helpers import writepath_cfg
@@ -198,8 +199,8 @@ RULING_LINE = "> Rulings for `src/a.py` — the operator's decisions about A (Sy
 @pytest.mark.asyncio
 async def test_the_tool_arm_leads_with_rulings_and_hands_back_what_it_showed():
     rulings = AsyncMock(return_value={"lines": [RULING_LINE], "system_ids": [3]})
-    with patch.object(pc, "_tool_rule_hint", AsyncMock(return_value={
-            "context": "> a rule line", "rule_ids": [8], "checkpoint": {}})), \
+    with patch.object(pc, "_rule_moment", AsyncMock(return_value=RuleResult(
+            lines=["> a rule line"], rule_ids=[8]))), \
          patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())), \
          patch.object(pc.system_rulings_svc, "rulings_for_paths", rulings):
         out = await pc.build_tool_rule_hint(
@@ -215,8 +216,7 @@ async def test_the_tool_arm_leads_with_rulings_and_hands_back_what_it_showed():
 
 @pytest.mark.asyncio
 async def test_the_tool_arm_adds_no_key_when_no_ruling_was_shown():
-    with patch.object(pc, "_tool_rule_hint", AsyncMock(return_value={
-            "context": "", "rule_ids": [], "checkpoint": {}})), \
+    with patch.object(pc, "_rule_moment", AsyncMock(return_value=RuleResult())), \
          patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())):
         out = await pc.build_tool_rule_hint(1, "Bash", "ls", project_id=2)
     assert "ruling_system_ids" not in out