FabledGrimoire/tests/test_retrieval_surfaces.py

"""The surface registry, and the identity the whole tuning story rests on.

WHY THIS EXISTS

The retrieval floors MOVE: the model using an arm chooses and adjusts them, and
a person can too. A tuning surface cannot be model-consistent across arms that
each own a settings key, a default and a limit spelled their own way, so the
arms read `{floor, budget}` from one registry.

WHAT THIS PINS

  1. **A surface's name IS its telemetry source.** This is the load-bearing
     one. The tool that moves a floor and the table that reports what the floor
     did have to be naming the same arm, and nothing in the type system says so
     — `record_retrieval(source="pre_tool_rule")` is a string literal in a
     different file. Renaming one without the other produces a surface that can
     be tuned and cannot be measured, or measured and not tuned, and both fail
     silently.
  2. **Keys are unique.** Two surfaces sharing a settings key move together
     whenever either is tuned — one dial wearing two labels.
  3. **An unknown surface is refused.** Settings keys are free-form strings in
     a generic table, so a typo'd name would write a key nothing reads: a
     change that appears to succeed, reports a new value, and alters nothing.
  4. **The budget clamps at 1, never 0.** A zero-budget arm searches, logs a
     retrieval and renders nothing — indistinguishable in the telemetry from a
     bar nothing cleared, which is the exact confusion this milestone exists to
     remove. "Off" is what the `enabled` switch is for.
  5. **`write_path` inherits auto-inject's budget when it has none.** It used
     to share that key outright; giving it its own without a fallback would
     silently reset the budget on every install that had tuned the shared one.
"""
from unittest.mock import AsyncMock, patch

import pytest

from grimoire.services import retrieval_surfaces as rs

SERVICES = ("plugin_context",)


def _service_source(name: str) -> str:
    from pathlib import Path
    root = Path(__file__).resolve().parents[1] / "src" / "grimoire" / "services"
    return (root / f"{name}.py").read_text()


def test_every_surface_name_is_a_real_telemetry_source():
    """The join key, checked against the arms that emit it.

    Asserted on the source text rather than by calling the arms, because what
    can rot here is the literal: a surface renamed in the registry and not in
    the `record_retrieval(source=…)` call still runs, still logs, and still
    tunes — just against two different names, so the readout an operator uses
    to justify a change describes a different arm from the one the change hits.
    """
    blob = "\n".join(_service_source(n) for n in SERVICES)
    # The rule arms record through the one pipeline, where
    # `source` is the spec's own field — so for them the spec IS the string
    # `record_retrieval` receives, and the join key is checked against it.
    from grimoire.services.retrieval_pipeline import NOTE_ARMS, RULE_ARMS

    via_pipeline = {arm.source for arm in (*RULE_ARMS, *NOTE_ARMS)}
    missing = [
        s.name for s in rs.SURFACES.values()
        if f'source="{s.name}"' not in blob
        and s.name not in via_pipeline
    ]
    assert not missing, (
        f"these surfaces can be tuned but never measured: {missing}. The "
        "registry name must match the string the arm passes to record_retrieval."
    )


def test_a_shared_settings_key_is_one_bar_with_one_default():
    """One dial, one label. Surfaces may share a floor key, as the rule arms
    do, but then they must ship the same default: otherwise the one number
    means something different on each surface until somebody sets it."""
    by_key: dict[str, set[float]] = {}
    for s in rs.SURFACES.values():
        by_key.setdefault(s.floor_key, set()).add(s.floor_default)
    split = {k: v for k, v in by_key.items() if len(v) > 1}
    assert not split, f"a shared floor key ships two defaults: {split}"

    budgets = [s.budget_key for s in rs.SURFACES.values()]
    assert len(set(budgets)) == len(budgets), f"duplicate budget key in {budgets}"
    assert not (set(by_key) & set(budgets)), "a floor key doubles as a budget key"


def test_the_rule_arms_that_annotate_share_one_bar_and_the_reply_stop_does_not():
    from grimoire.services.retrieval_pipeline import RULE_FLOOR_KEY

    assert sorted(rs.sharing_floor("prompt_rule")) == [
        "pre_tool_rule", "prompt_rule", "write_path_rule",
    ]
    assert rs.SURFACES["prompt_rule"].floor_key == RULE_FLOOR_KEY
    assert rs.sharing_floor("reply_rule") == ["reply_rule"]


def test_every_surface_says_what_it_asks_over_what_and_how_often():
    """The prose is rendered, not decorative.

    A floor cannot be moved responsibly by anyone — model or human — who does
    not know the query shape, the corpus, or how often the arm costs something.
    An empty string here reaches the tuning tool and the Settings UI as a blank.
    """
    for s in rs.SURFACES.values():
        for field in ("asks", "over", "fires"):
            assert getattr(s, field).strip(), f"{s.name}.{field} is empty"


def test_an_unknown_surface_is_refused_rather_than_written():
    with pytest.raises(ValueError) as e:
        rs.get_surface("pretool_rule")          # a real typo for pre_tool_rule
    # The message has to name the alternatives, or the caller's next move is a
    # second guess.
    assert "pre_tool_rule" in str(e.value)


@pytest.mark.asyncio
@pytest.mark.parametrize("stored, expected", [
    ("0.8", 0.8),
    ("5", 1.0),            # clamped, not believed
    ("-3", 0.0),
    ("banana", 0.55),      # unparseable falls back to the surface's default
    ("", 0.55),
])
async def test_a_floor_is_clamped_to_the_unit_interval(stored, expected):
    with patch.object(rs, "get_setting", AsyncMock(return_value=stored)):
        assert await rs.floor_for(1, "auto_inject") == expected


@pytest.mark.asyncio
@pytest.mark.parametrize("stored, expected", [
    ("4", 4),
    ("999", rs.MAX_BUDGET),
    ("0", 1),
    ("-2", 1),
    ("banana", 3),
])
async def test_a_budget_is_clamped_with_a_floor_of_one(stored, expected):
    """Zero is the value that must not get through.

    An arm with a budget of 0 runs its search, writes a `retrieval_logs` row
    with `result_count == 0`, and renders nothing — which reads in the telemetry
    exactly like a bar nothing cleared. Turning a surface off is the `enabled`
    switch's job, and that one says so.
    """
    with patch.object(rs, "get_setting", AsyncMock(return_value=stored)):
        assert await rs.budget_for(1, "auto_inject") == expected


@pytest.mark.asyncio
async def test_the_write_path_budget_falls_back_to_auto_injects():
    """The migration this fallback exists for.

    `write_path` had no budget key of its own — it read auto-inject's. An
    install that had tuned that shared knob to 6 must not silently drop to the
    new key's default the day this ships; nothing would look broken and the
    operator would never know to look.
    """
    async def _setting(_uid, key, default=""):
        return {"kb_writepath_top_k": "", "kb_autoinject_top_k": "6"}.get(key, default)

    with patch.object(rs, "get_setting", AsyncMock(side_effect=_setting)):
        assert await rs.budget_for(1, "write_path") == 6


@pytest.mark.asyncio
async def test_its_own_budget_wins_once_it_is_set():
    """And the fallback must not become a permanent override."""
    async def _setting(_uid, key, default=""):
        return {"kb_writepath_top_k": "2", "kb_autoinject_top_k": "6"}.get(key, default)

    with patch.object(rs, "get_setting", AsyncMock(side_effect=_setting)):
        assert await rs.budget_for(1, "write_path") == 2


@pytest.mark.asyncio
async def test_the_write_path_config_carries_every_arm_it_drives():
    """One hook request drives three arms, and each reads its pair out of this.

    A missing key does not raise where a reader would see it — the rule arms
    fail open, so a KeyError becomes an empty hint, and the arm reads as "fired
    and found nothing". That is the failure this asserts against, and it is the
    reason tests build the dict from the registry rather than by hand.
    """
    from grimoire.services import plugin_context as pc

    # Both modules read settings: `pc` for the enabled switch, `rs` for the
    # pairs. Patching one and not the other reaches a real database.
    with patch.object(pc, "get_setting", AsyncMock(return_value="")), \
         patch.object(rs, "get_setting", AsyncMock(return_value="")):
        cfg = await pc.get_writepath_config(1)

    for key in ("threshold", "top_k", "rule_threshold", "rule_top_k",
                "tool_rule_threshold", "tool_rule_top_k",
                # Not a surface, and that is why it is spelled out here: it
                # has no floor/budget pair to derive from, so nothing else in
                # this file would notice it going missing.
                "checkpoint_threshold"):
        assert key in cfg, f"{key} missing — its arm will silently no-op"