feat(rules): rules before tools — a PreToolUse arm keyed on the action (#3476)

The only just-in-time rule surface was registered on `Write|Edit` and queried
with `code or path`, so a rule could be retrieved at the moment of a code
write and nowhere else. Every rule about which tool to reach for — don't curl
the forge, don't stand up a stack, don't run the suite locally, don't branch —
was unreachable exactly when it mattered, and residency in the always-on
preload was the only surface it had. That is the pressure that grew the
resident set to 31 against #3089's ceiling of ~23; it was never a judgment
anybody made.

A reflex generates no query, so an instruction to check the rules cannot catch
one. A mechanical trigger can: the tool call IS the query, and a reflex has to
become a tool call before it can do anything.

`build_tool_rule_hint` is deliberately tool-agnostic — a name and a string —
so widening the matcher later is a hooks.json edit with no server change. The
hook starts on Bash, which is where the action reflexes live.

The two pre-tool arms share ONE session ledger of already-named rules
(`<state>/<sid>.rules.ids`). Two ledgers would mean a rule named by one arm
gets re-offered by the other, and the hint that fires most often is exactly
the one that must not repeat itself. A test asserts both scripts build the
same path, and another checks the shell hook and the Python route agree on
every query-arg name (rule 33) — a rename there fails silently, looking like
a surface that never finds anything rather than a broken one.

Deliberately silent on outage, unlike the prior-art hook: a write is
occasional, a Bash call is not, and an outage line before every command is
what gets a channel muted.

`tier="conditional"` matches the write arm and is the transition point — an
always-on rule is already resident, so re-tier one and it starts arriving here
instead of in every session's preamble. `pre_tool_rule` joins RANKED_SOURCES:
this arm chose what it showed, so a pull can settle whether the choice landed.

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

tests/test_rule_usage_wiring.py

@@ -401,3 +401,179 @@ def test_the_marker_paths_stay_silent():
     assert "record_rule_surfaced" not in staleness, (
         "the staleness arm emits a surfacing — it compares a marker, it shows nothing"
     )
+
+
+# ── The PRE-TOOL arm: rules keyed on the action (#3476) ────────────────
+#
+# The write-path arm can only be reached by a code write, so every rule about
+# which tool to reach for was unretrievable at the moment it mattered — which
+# is why they all had to be resident. These cover the surface that changes it.
+
+
+def _tool_patches(pc, hits, recorder, cfg=None):
+    return (
+        patch.object(pc, "get_writepath_config",
+                     AsyncMock(return_value=cfg or {
+                         "enabled": True, "threshold": 0.6,
+                         "top_k": 3, "rule_threshold": 0.6,
+                     })),
+        patch.object(pc, "semantic_search_rules", AsyncMock(return_value=hits)),
+        patch.object(pc, "record_retrieval", MagicMock()),
+        patch.object(pc, "record_rule_surfaced", recorder),
+    )
+
+
+async def _run_tool_arm(hits, recorder, command="curl -s https://git.example/api/v1/runs",
+                        tool="Bash", **kwargs):
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, hits, recorder):
+            stack.enter_context(ctx)
+        return await pc.build_tool_rule_hint(1, tool, command, **kwargs)
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_names_a_rule_for_the_command_about_to_run():
+    """The 2026-09-03 incident in one test: reaching for curl against the forge
+    API is a Bash call, and nothing watched Bash."""
+    rec = MagicMock()
+    hits = [(0.71, fake_rule(id=161,
+                             title="Reach the forge through its MCP tools, never curl",
+                             when_to_apply="whenever you need CI status"))]
+    out = await _run_tool_arm(hits, rec)
+
+    assert out["rule_ids"] == [161]
+    assert "Reach the forge through its MCP tools" in out["context"]
+    assert "get_rule(161)" in out["context"], "the hint must hand over the way to read it"
+    assert "Bash" in out["context"], "the hint names the tool it is about"
+    assert rec.call_args.kwargs["source"] == "pre_tool_rule"
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_is_a_ranked_source():
+    """It CHOSE what it showed, so a pull can settle whether the choice was any
+    good — unlike a preload, which chose nothing. If this drifts into the
+    ambient class the arm becomes unjudgeable, which is the state #3311
+    described and M333 existed to end."""
+    from scribe.services.rule_usage import is_ambient
+
+    assert not is_ambient("pre_tool_rule")
+
+
+@pytest.mark.asyncio
+async def test_a_rule_the_session_already_holds_is_not_re_offered():
+    rec = MagicMock()
+    hits = [(0.71, fake_rule(id=161, title="Reach the forge through its MCP tools")),
+            (0.70, fake_rule(id=12, title="Don't run a local stack unless asked"))]
+    out = await _run_tool_arm(hits, rec, exclude_rule_ids=[161])
+
+    assert out["rule_ids"] == [12]
+    assert "161" not in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_an_empty_command_asks_the_ranker_nothing():
+    """Every Bash call reaches this. A blank payload must cost no embedding
+    query at all, not merely return nothing after paying for one."""
+    from scribe.services import plugin_context as pc
+
+    search = AsyncMock(return_value=[])
+    rec = MagicMock()
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        out = await pc.build_tool_rule_hint(1, "Bash", "   ")
+
+    assert out == {"context": "", "rule_ids": []}
+    search.assert_not_called()
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_fails_open():
+    """A recall aid may never break the operator's action. A ranker that raises
+    must cost the hint, not the command."""
+    from scribe.services import plugin_context as pc
+
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_writepath_config",
+                                         AsyncMock(side_effect=RuntimeError("boom"))))
+        out = await pc.build_tool_rule_hint(1, "Bash", "docker compose up -d")
+
+    assert out == {"context": "", "rule_ids": []}
+
+
+@pytest.mark.asyncio
+async def test_a_long_command_is_bounded_before_it_reaches_the_ranker():
+    """A heredoc or a pasted script would push the verb and its target — the
+    part a rule is about — out of the embedding window."""
+    from scribe.services import plugin_context as pc
+
+    search = AsyncMock(return_value=[])
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, [], MagicMock()):
+            stack.enter_context(ctx)
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        await pc.build_tool_rule_hint(1, "Bash", "git tag v1 && " + "x" * 5000)
+
+    sent = search.call_args.args[1]
+    assert len(sent) <= pc._TOOL_QUERY_CHARS
+    assert sent.startswith("git tag v1"), "the head of the command is the signal"
+
+
+def test_the_two_pre_tool_arms_share_one_session_rule_ledger():
+    """The integration point most worth guarding.
+
+    Two ledgers would mean a rule named by the write arm gets re-offered by the
+    tool arm — and the hint that fires most often is exactly the one that must
+    not repeat itself. Asserted on the FILENAME both scripts build, because
+    that is the shared thing; a copy of the path in each is how they drift.
+    """
+    prior = Path("plugin/hooks/scribe_prior_art.sh").read_text()
+    tool = Path("plugin/hooks/scribe_tool_rules.sh").read_text()
+
+    for src, name in ((prior, "scribe_prior_art.sh"), (tool, "scribe_tool_rules.sh")):
+        assert '"${TMPDIR:-/tmp}/scribe-priorart"' in src, f"{name}: state dir moved"
+        assert '.rules.ids' in src, f"{name}: rules ledger filename moved"
+        assert "exclude_rule_ids" in src, f"{name}: does not send the exclusion"
+
+
+def test_the_tool_arm_is_registered_on_bash():
+    """A hook that exists and is not registered runs never — and reads exactly
+    like a surface nobody needed."""
+    import json
+
+    manifest = json.loads(Path("plugin/hooks/hooks.json").read_text())
+    pre = manifest["hooks"]["PreToolUse"]
+    entries = {
+        m.get("matcher"): [h["command"] for h in m["hooks"]] for m in pre
+    }
+    assert "Bash" in entries, "nothing watches Bash — the reflex surface is unguarded"
+    assert any("scribe_tool_rules.sh" in c for c in entries["Bash"])
+    # The write arm keeps its own matcher; this is an addition, not a move.
+    assert any("scribe_prior_art.sh" in c for c in entries["Write|Edit"])
+
+
+def test_the_hook_and_the_route_agree_on_every_parameter_name():
+    """Rule 33, on a brand-new integration between layers.
+
+    The hook is shell and the route is Python; nothing but this test connects
+    them. A renamed query arg fails SILENTLY — the route reads an absent value,
+    the arm quietly searches nothing, and the surface looks like one that never
+    finds anything rather than one that is broken.
+    """
+    import re
+
+    hook = Path("plugin/hooks/scribe_tool_rules.sh").read_text()
+    route = Path("src/scribe/routes/plugin.py").read_text()
+    handler = route.split("async def pre_tool_rules")[1].split("\n@plugin_bp")[0]
+
+    sent = set(re.findall(r"[?&]([a-z_]+)=", hook))
+    assert sent == {"tool", "command", "repo", "exclude_rule_ids"}, sent
+
+    # `repo` is read by the shared _project_scope() helper, not inline.
+    assert "_project_scope()" in handler
+    for arg in ("tool", "command", "exclude_rule_ids"):
+        assert f'request.args.get("{arg}")' in handler, (
+            f"the hook sends {arg!r} and the route never reads it"
+        )