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_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.


def _arm_patches(pc, hits, recorder):
    """The minimum stubbing that lets the rule arm run and nothing else."""
    return (
        patch.object(pc, "get_writepath_config",
                     AsyncMock(return_value={"enabled": True, "threshold": 0.6,
                                             "top_k": 3})),
        patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))),
        patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])),
        patch.object(pc, "semantic_search_rules", 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, **kwargs):
    from scribe.services import plugin_context as pc
    with ExitStack() as stack:
        for ctx in _arm_patches(pc, hits, recorder):
            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_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."
            )