FabledGrimoire/tests/test_wide_net.py

"""The wide net: many candidates, no bar, and a cost that stays bounded.

WHY THIS EXISTS

"Do not exclude potentially important data" belongs on a pull, not the push:
fifty candidates pushed would be a wall nobody reads, and pulled they arrive
only when a session asks for them.

Three things have to hold, and each has a way of quietly failing:

  1. **No bar.** The moment this tool serves is the one where the caller does
     not trust a threshold to decide for them. A default floor creeping in
     here would turn the wide net into the narrow one and nothing would look
     wrong — the results would simply be fewer.
  2. **A bounded cost.** Fifty candidates do not cost nothing: a line with
     its trigger rendered is ~143 tokens, so fifty is ~7,000. The graduated
     shape is what keeps this affordable, and a change that renders every trigger in full
     would pass every other test here.
  3. **A pull, logged as one.** The push arms' near-miss distributions are the
     evidence their floors are tuned from. A pull mixed into them moves those
     numbers.
"""
from unittest.mock import AsyncMock, patch

import pytest

from grimoire.mcp.tools import wide_net
from tests.helpers import FakeMCP, fake_rule, tool_doc


def _hits(n=3, trigger="Running git push", **over):
    """n (score, rule) pairs, descending, the shape the service returns."""
    return [
        (0.9 - i * 0.01,
         fake_rule(id=i + 1, title=f"rule {i + 1}", when_to_apply=trigger, **over))
        for i in range(n)
    ]


def _patched(hits, report=None):
    """Patch the search and the telemetry sink; hand back the search mock."""
    async def _search(*a, **kw):
        if report is not None and "report" in kw and kw["report"] is not None:
            kw["report"].update(report)
        return hits
    return patch.object(wide_net, "semantic_search_rules", AsyncMock(side_effect=_search))


@pytest.mark.asyncio
async def test_no_bar_reaches_the_search():
    """THE POINT OF THE TOOL. A floor here would narrow the net silently."""
    with _patched(_hits()) as search, \
         patch.object(wide_net, "record_retrieval"), \
         patch.object(wide_net, "current_user_id", lambda: 1):
        await wide_net.what_might_apply("push to dev")

    assert search.await_args.kwargs["threshold"] == 0.0
    assert wide_net.THRESHOLD == 0.0


@pytest.mark.asyncio
@pytest.mark.parametrize("asked, expected", [(999, 50), (0, 1), (-5, 1), (25, 25)])
async def test_the_limit_is_clamped_to_the_cap(asked, expected):
    with _patched(_hits()) as search, \
         patch.object(wide_net, "record_retrieval"), \
         patch.object(wide_net, "current_user_id", lambda: 1):
        await wide_net.what_might_apply("q", limit=asked)

    assert search.await_args.kwargs["limit"] == expected


@pytest.mark.asyncio
async def test_the_head_carries_its_trigger_whole_and_the_tail_is_cut():
    """The graduated shape, which is what makes fifty affordable.

    A regression that rendered every trigger in full would satisfy every other
    assertion in this file, so the cut is pinned on both sides: the head is
    NOT marked truncated and the tail IS.
    """
    long_trigger = " ".join(["running a git command before pushing anything"] * 12)
    with _patched(_hits(6, trigger=long_trigger)), \
         patch.object(wide_net, "record_retrieval"), \
         patch.object(wide_net, "current_user_id", lambda: 1):
        out = await wide_net.what_might_apply("q", detail=2)

    head, tail = out["candidates"][:2], out["candidates"][2:]
    assert all(c["truncated"] is False for c in head)
    assert all(c["when_to_apply"] == " ".join(long_trigger.split()) for c in head)
    assert all(c["truncated"] is True for c in tail)
    assert all(len(c["when_to_apply"]) <= wide_net._TEASER_CHARS + 1 for c in tail)
    assert out["detailed"] == 2


@pytest.mark.asyncio
async def test_a_short_trigger_is_never_marked_truncated():
    """`truncated` is a claim about this row, not about its rank. A tail row
    whose trigger already fits must not claim a cut that did not happen."""
    with _patched(_hits(4, trigger="Running git push")), \
         patch.object(wide_net, "record_retrieval"), \
         patch.object(wide_net, "current_user_id", lambda: 1):
        out = await wide_net.what_might_apply("q", detail=1)

    assert all(c["truncated"] is False for c in out["candidates"])


def test_the_cut_breaks_on_a_word_and_says_it_was_cut():
    """A raw slice ends mid-word and reads as the whole thing."""
    text, cut = wide_net._teaser("alpha beta gamma delta " * 40)
    assert cut is True
    assert text.endswith("…")
    # Broke on a word, so no partial token sits before the marker.
    assert not text.removesuffix("…").rstrip().endswith(("alph", "bet", "gam"))


def test_one_unbroken_word_still_yields_text_rather_than_a_bare_marker():
    """`textwrap.shorten` returns just "…" here, which would render a teaser
    carrying no information at all."""
    text, cut = wide_net._teaser("x" * 500)
    assert cut is True and text != "…"
    assert len(text) == wide_net._TEASER_CHARS


@pytest.mark.asyncio
async def test_every_row_says_its_force_and_its_scope():
    """`kind` is never inferred — a rule must be followed, a preference guides
    — and `scope` says how far a reader should generalise from it."""
    hits = [
        (0.8, fake_rule(id=1, kind="rule", project_id=None)),
        (0.7, fake_rule(id=2, kind="preference", project_id=44)),
    ]
    with _patched(hits), patch.object(wide_net, "record_retrieval"), \
         patch.object(wide_net, "current_user_id", lambda: 1):
        out = await wide_net.what_might_apply("q")

    assert [c["kind"] for c in out["candidates"]] == ["rule", "preference"]
    assert [c["scope"] for c in out["candidates"]] == ["global", "project"]


# ── telemetry: a pull, and never mistaken for a push ────────────────────────

@pytest.mark.asyncio
async def test_the_call_is_logged_under_its_own_pull_source():
    with _patched(_hits()), \
         patch.object(wide_net, "record_retrieval") as rec, \
         patch.object(wide_net, "current_user_id", lambda: 1):
        await wide_net.what_might_apply("push to dev")

    kw = rec.call_args.kwargs
    assert kw["source"] == wide_net.SOURCE == "wide_net"
    assert kw["threshold"] == 0.0


def test_the_wide_net_is_not_one_of_the_tunable_push_surfaces():
    """THE GUARD that keeps the push arms' tuning evidence clean.

    The registry holds the PUSH arms — the ones with a floor and a budget the
    model tunes. This source must not appear there: a pull folded into those
    rows would move the near-miss distributions that step is arguing from, and
    would offer a floor to tune on a tool whose whole point is not having one.
    """
    from grimoire.services.retrieval_surfaces import SURFACES

    assert wide_net.SOURCE not in SURFACES


def test_the_wide_net_is_not_ambient_either():
    """Ambient means a delivery nobody chose. This one is chosen by definition
    — somebody called the tool — so counting it as ambient would make a
    deliberate ask read as a bulk hand-over."""
    from grimoire.services.note_usage import AMBIENT_SOURCES

    assert wide_net.SOURCE not in AMBIENT_SOURCES


@pytest.mark.asyncio
async def test_a_search_that_never_ran_is_not_reported_as_a_decline():
    """An empty query, a dead embedder and a failed query all return
    nothing, and none of them is a ranker declining."""
    with _patched([], report={"searched": False, "best_available_score": None}), \
         patch.object(wide_net, "record_retrieval") as rec, \
         patch.object(wide_net, "current_user_id", lambda: 1):
        out = await wide_net.what_might_apply("")

    assert rec.call_args.kwargs["searched"] is False
    assert out["searched"] is False


@pytest.mark.asyncio
async def test_what_the_bar_turned_away_is_carried_through():
    """There is no bar here, but `best_available` still answers "did the corpus
    have anything at all" for a call that came back empty."""
    with _patched([], report={"searched": True, "best_available_score": 0.31,
                              "best_available_id": 7}), \
         patch.object(wide_net, "record_retrieval") as rec, \
         patch.object(wide_net, "current_user_id", lambda: 1):
        await wide_net.what_might_apply("q")

    assert rec.call_args.kwargs["best_available"] == 0.31
    assert rec.call_args.kwargs["best_available_id"] == 7


# ── the contract a session actually reads ───────────────────────────────────

def test_the_docstring_says_when_to_reach_for_it():
    """The step's done-when, and the load-bearing half of this tool.

    A wide net nobody knows to call is worth nothing, so the docstring has to
    name the MOMENT, not just the parameters — including the one that prompted
    it, where a session hands work back rather than finishing it.
    """
    doc = tool_doc("grimoire.mcp.tools.wide_net", "what_might_apply").lower()

    assert "consequential" in doc
    assert "handing work back" in doc
    # Says the tail is expected to be noise — otherwise the first caller reads
    # a low-scoring list as the tool being broken.
    assert "noise" in doc
    # And names its own limit: a pull only helps if somebody asks.
    assert "only helps if you ask" in doc


def test_it_is_registered_and_readable_with_a_read_key():
    from grimoire.mcp.server import _READ_ONLY_TOOLS

    mcp = FakeMCP()
    wide_net.register(mcp)
    assert mcp.names == ["what_might_apply"]
    assert "what_might_apply" in _READ_ONLY_TOOLS


def test_the_instruction_surfaces_point_at_it():
    """The instruction surfaces ARE the specification for product
    behaviour, so a tool the reflex never learns about is not shipped."""
    import pathlib

    from grimoire.mcp import server

    assert "what_might_apply" in server._INSTRUCTIONS

    skill = (pathlib.Path(__file__).resolve().parents[1]
             / "plugin" / "skills" / "using-grimoire" / "SKILL.md").read_text()
    assert "what_might_apply" in skill