FabledScribe/tests/test_rule_usage_wiring.py

"""Both ends of the rule-usage loop are actually wired (milestone 333 step 2).

Step 1 built the table and the service. A counter nobody calls reads zero and
looks exactly like a surface nobody uses — which is #2663's shape and the whole
reason this milestone exists. So this file is about the CALL SITES, not the
storage.

Cross-cutting on purpose: the surfaced end lives in `plugin_context`, the pull
end in two different doors, and the property under test is that they meet. Split
across three module-shaped files, "both ends are wired" is a thing no single
test asserts.
"""
from contextlib import ExitStack
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from tests.helpers import fake_note, fake_rule

# The MCP tool layer reads its caller from a ContextVar the HTTP transport sets
# per request; a unit test has no request, so it binds the caller itself. The
# arm tests do not need it — build_write_path_hint takes user_id directly — but
# the module-level mark is how every tool-layer test file in this repo opts in.
pytestmark = pytest.mark.usefixtures("_bind_user")


# ── The surfaced end ───────────────────────────────────────────────────
#
# conftest's autouse `_no_rule_arm` stubs `semantic_search_rules` so unrelated
# plugin-context tests don't pull a real embedding model through this arm. Its
# docstring says a test that wants the arm live can re-patch it — that is what
# each of these does.


# The write-path hint returns early when a write matched nothing at all — no
# staleness, no synced record, no prior-art menu, no shape signal. The rule arm
# sits deliberately on the FAR side of that guard, because it runs a semantic
# search and moving it above would mean an embedding query on every write in
# the session (#3311's closing note, and the reason its gating is a separate
# question from precision).
#
# So a fixture that stubs every other arm to empty never reaches the rule arm
# at all — which is what the first run of this file did. The note hit below is
# not decoration: it is the condition the arm requires in order to fire.
_PRIOR_ART = [(0.72, fake_note(id=9, title="debounce helper", user_id=1,
                               note_type="snippet"))]


def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None):
    """The minimum stubbing that lets the rule arm run and nothing else.

    `cfg` and `rule_search` are overridable so a caller can inspect what the
    arm ASKED for rather than only what it did with the answer — patching them
    a second time on top would work, but reads as an accident.
    """
    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.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))),
        patch.object(pc, "semantic_search_notes",
                     AsyncMock(return_value=_PRIOR_ART if prior_art is None
                               else prior_art)),
        patch.object(pc, "semantic_search_rules",
                     rule_search or AsyncMock(return_value=hits)),
        patch.object(pc, "record_retrieval", MagicMock()),
        patch.object(pc, "record_surfaced", MagicMock()),
        patch.object(pc, "record_rule_surfaced", recorder),
        patch.object(pc, "owner_names_for", AsyncMock(return_value={})),
        patch.object(pc, "concept_query", MagicMock(return_value="a deadline on a fetch")),
    )


async def _run_arm(hits, recorder, prior_art=None, **kwargs):
    from scribe.services import plugin_context as pc
    with ExitStack() as stack:
        for ctx in _arm_patches(pc, hits, recorder, prior_art):
            stack.enter_context(ctx)
        return await pc.build_write_path_hint(
            1, "frontend/src/api/client.ts", code="x" * 400, **kwargs
        )


@pytest.mark.asyncio
async def test_the_arm_records_what_it_showed():
    """The claim being measured. Without this call the arm keeps producing
    scores in retrieval_logs and no evidence that any hint was ever read."""
    rec = MagicMock()
    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
    await _run_arm(hits, rec)

    assert rec.call_count == 1
    kw = rec.call_args.kwargs
    assert kw["rule_ids"] == [156]
    assert kw["source"] == "write_path_rule"


@pytest.mark.asyncio
async def test_a_rule_the_session_already_holds_is_not_counted_as_surfaced():
    """`exclude_rule_ids` drops what the session already has, and the recorded
    set must be what was SHOWN, not what was considered.

    Counting the excluded ones would inflate the denominator with claims the
    agent never saw — the ratio would fall for a reason that has nothing to do
    with whether the hints landed, which is precisely the misreading this
    milestone exists to prevent.
    """
    rec = MagicMock()
    hits = [
        (0.71, fake_rule(id=156, title="A wait with no deadline is a bug")),
        (0.70, fake_rule(id=157, title="A loop re-arms in a finally")),
    ]
    await _run_arm(hits, rec, exclude_rule_ids=[157])

    assert rec.call_args.kwargs["rule_ids"] == [156]


@pytest.mark.asyncio
async def test_nothing_is_recorded_when_every_hit_was_already_held():
    """No surfacing happened, so no surfacing is recorded. A zero-row batch
    would still be a call, and a call that says "we showed nothing" pollutes
    the count of times the arm spoke."""
    rec = MagicMock()
    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
    await _run_arm(hits, rec, exclude_rule_ids=[156])

    assert rec.call_count == 0


@pytest.mark.asyncio
async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
    """The consuming half of step 4. `get_writepath_config` assembling a
    separate `rule_threshold` means nothing if the arm still passes
    `cfg["threshold"]` to its search — the split would exist in the config and
    not in the behaviour, and #3311 would be exactly where it was.

    The two values are deliberately different here so the assertion can tell
    them apart.
    """
    from scribe.services import plugin_context as pc

    search = AsyncMock(return_value=[])
    with ExitStack() as stack:
        for ctx in _arm_patches(
            pc, [], MagicMock(), rule_search=search,
            cfg={"enabled": True, "threshold": 0.60,
                 "top_k": 3, "rule_threshold": 0.81},
        ):
            stack.enter_context(ctx)
        await pc.build_write_path_hint(
            1, "frontend/src/api/client.ts", code="x" * 400,
        )

    kw = search.await_args.kwargs
    assert kw["threshold"] == 0.81, "the arm is still using the code threshold"
    assert kw["limit"] == pc.RULEHINT_LIMIT
    assert kw["tier"] == "conditional"


@pytest.mark.asyncio
async def test_the_arm_does_not_fire_on_a_write_that_matched_nothing():
    """The gate, pinned — because the fixture above now depends on it and a
    silent change would make every other test here pass vacuously.

    A write matching no prior art returns before the rule arm runs. That is
    deliberate: the arm is a semantic search, and ungating it means an
    embedding query on every write in the session. #3311 is explicit that the
    gate stays until the arm's precision is fixed, so this failing is a signal
    to go read that issue rather than to update the assertion.
    """
    rec = MagicMock()
    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
    await _run_arm(hits, rec, prior_art=[])

    assert rec.call_count == 0


@pytest.mark.asyncio
async def test_a_failing_recorder_does_not_break_the_write():
    """Telemetry must never take down the surface it observes. The arm is
    already wrapped in a fail-open try/except; this pins that the new call is
    INSIDE it rather than after."""
    rec = MagicMock(side_effect=RuntimeError("telemetry is down"))
    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
    out = await _run_arm(hits, rec)

    assert "context" in out


# ── The pull end ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_mcp_get_rule_records_an_agent_pull():
    """THE pull that matters: the arm's own message ends "Read it with
    get_rule(N)", so this is the exact action a landed hint produces."""
    rec = MagicMock()
    rule = fake_rule(id=156, title="A wait with no deadline is a bug")
    with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule",
               AsyncMock(return_value=rule)), \
         patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail",
               AsyncMock(return_value={"id": 156})), \
         patch("scribe.mcp.tools.rulebooks.record_rule_pulled", rec):
        from scribe.mcp.tools.rulebooks import get_rule
        await get_rule(rule_id=156)

    assert rec.call_args.kwargs["rule_id"] == 156
    assert rec.call_args.kwargs["source"] == "mcp_get_rule"


@pytest.mark.asyncio
async def test_a_rule_that_cannot_be_read_is_not_a_pull():
    """Recorded after the access check. A refused read is not a pull, and
    counting it would credit the arm for a hint nobody could open."""
    rec = MagicMock()
    with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule",
               AsyncMock(return_value=None)), \
         patch("scribe.mcp.tools.rulebooks.record_rule_pulled", rec):
        from scribe.mcp.tools.rulebooks import get_rule
        with pytest.raises(ValueError):
            await get_rule(rule_id=156)

    assert rec.call_count == 0


# ── Completeness: every door, and only the doors ───────────────────────


def _source_of(module_path: str) -> str:
    return (Path(__file__).resolve().parents[1] / module_path).read_text()


def test_every_rule_detail_door_records_a_pull():
    """The task's own warning, made mechanical: miss a door and the ratio
    reads low for a reason that is not about the rules.

    Source inspection rather than behaviour, because the REST door has no
    live-HTTP harness in the unit lane (see test_routes_rulebooks.py's own
    note). What it can still prove is that the handler names the recorder —
    which is the thing that gets forgotten when a door is added.
    """
    rest = _source_of("src/scribe/routes/rulebooks.py")
    mcp = _source_of("src/scribe/mcp/tools/rulebooks.py")
    assert 'source="rest_rule"' in rest, (
        "the REST rule-detail route does not record a pull"
    )
    assert 'source="mcp_get_rule"' in mcp, (
        "the MCP get_rule tool does not record a pull"
    )


def test_the_bulk_loaders_are_not_counted_as_pulls():
    """`list_always_on_rules` and `enter_project` hand over every applicable
    rule at once. That is delivery, not somebody choosing to open one record,
    and counting it would swamp the signal with exactly the ambient surfacing
    the ratio exists to distinguish from.

    Stated as a test because it is the tempting addition: both put rules in
    front of an agent, so "surely those are pulls too" is the reading someone
    arrives at without the argument.
    """
    for path in ("src/scribe/mcp/tools/rulebooks.py",
                 "src/scribe/mcp/tools/projects.py"):
        src = _source_of(path)
        for door in ("list_always_on_rules", "enter_project"):
            if f"async def {door}" not in src:
                continue
            body = src.split(f"async def {door}", 1)[1].split("\nasync def ", 1)[0]
            assert "record_rule_pulled" not in body, (
                f"{door} records a pull. It is a bulk resident load — every "
                "applicable rule at once — so counting it would drown the "
                "surfaced:pulled ratio in ambient delivery."
            )


# ── The AMBIENT end: bulk deliveries (#3473) ───────────────────────────
#
# The preload was the largest rule surface in the product and emitted nothing,
# so its cost was certain and its usefulness unfalsifiable. These assert the
# three delivery shapes now emit — and, just as importantly, that the two
# lookalike call sites which show nobody anything do NOT.


@pytest.mark.asyncio
async def test_the_session_start_preload_records_what_it_delivered():
    """The block every session opens with. Chosen by nobody, paid for every
    turn — and until it emitted, invisible to the scoreboard that judges every
    other surface."""
    from scribe.services import plugin_context as pc

    rec = MagicMock()
    rules = [fake_rule(id=1, title="`dev` is home"),
             fake_rule(id=2, title="`main` — never without explicit request")]
    with ExitStack() as stack:
        stack.enter_context(
            patch.object(pc.rulebooks_svc, "list_always_on_rules",
                         AsyncMock(return_value=rules))
        )
        stack.enter_context(
            patch.object(pc.rulebooks_svc, "excluded_always_on_rulebooks",
                         AsyncMock(return_value=[]))
        )
        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
        stack.enter_context(
            patch.object(pc, "_topic_titles", AsyncMock(return_value={}))
        )
        await pc.build_session_context(1, project_id=0)

    assert rec.call_count == 1, "the preload recorded nothing"
    kw = rec.call_args.kwargs
    assert kw["rule_ids"] == [1, 2]
    assert kw["source"] == "session_start"


@pytest.mark.asyncio
async def test_the_always_on_tool_records_what_it_handed_over():
    from scribe.mcp.tools import rulebooks as tools

    rec = MagicMock()
    rules = [fake_rule(id=3, title="No GitHub — Fabled-Git only")]
    with ExitStack() as stack:
        stack.enter_context(
            patch.object(tools.rulebooks_svc, "list_always_on_rules",
                         AsyncMock(return_value=rules))
        )
        stack.enter_context(
            patch.object(tools.rulebooks_svc, "rules_etag",
                         MagicMock(return_value="etag"))
        )
        stack.enter_context(patch.object(tools, "record_rule_surfaced", rec))
        await tools.list_always_on_rules()

    assert rec.call_args.kwargs["rule_ids"] == [3]
    assert rec.call_args.kwargs["source"] == "list_always_on_rules"


def test_rules_payload_records_both_the_family_and_project_halves():
    """One emit site for all five `rules_payload` surfaces.

    Per-caller emission would be five sites to remember, and #3430 gap 2 is
    what that costs: the process→skill sync went un-emitted through an entire
    dedicated telemetry survey because nothing forced its surface to be
    accounted for.
    """
    from scribe.services import rulebooks as svc

    rec = MagicMock()
    with patch.object(svc, "record_rule_surfaced", rec):
        svc.rules_payload(
            {
                "rules": [{"id": 10}, {"id": 11}],
                "project_rules": [{"id": 12}],
                "truncated": False,
                "subscribed_rulebooks": [],
            },
            user_id=1,
            source="enter_project",
        )

    kw = rec.call_args.kwargs
    assert kw["rule_ids"] == [10, 11, 12], "project-scoped rules were delivered too"
    assert kw["source"] == "enter_project"


def test_every_rules_payload_caller_names_itself():
    """`source` is the CALLER's name, so the readout can still separate the
    session handshake from a mid-session milestone read. A shared constant here
    would collapse five distinguishable surfaces into one."""
    import re

    seen = set()
    for path in Path("src/scribe").rglob("*.py"):
        for m in re.finditer(r"rules_payload\([^)]*source=\"([a-z_]+)\"", path.read_text()):
            seen.add(m.group(1))
    assert seen == {
        "enter_project", "get_project", "get_milestone",
        "start_planning", "get_task",
    }, f"a rules_payload caller is missing or misnamed: {sorted(seen)}"


def test_the_marker_paths_stay_silent():
    """The two call sites that read the rules and show NOBODY anything.

    `rules_etag_for` and the write-path staleness arm both call
    `list_always_on_rules` to build or compare a marker. Emitting there would
    put rules in the denominator that no agent ever saw — the exact inflation
    `record_rule_surfaced`'s docstring forbids, arriving from the one direction
    nothing else guards.
    """
    svc_src = Path("src/scribe/services/rulebooks.py").read_text()
    etag_fn = svc_src.split("async def rules_etag_for")[1].split("\ndef ")[0]
    assert "record_rule_surfaced" not in etag_fn, (
        "rules_etag_for emits a surfacing — it builds a marker, it shows nothing"
    )

    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
    staleness = pc_src.split("if rules_etag:")[1].split("# The guard sits BELOW")[0]
    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"
        )