FabledGrimoire/tests/test_retrieval_registry.py

"""Every retrieval point is declared, derived from the CALL SITES.

WHY AST AND NOT GREP, demonstrated rather than asserted. Two of the sources in
this system reach their recorder as `source=SOURCE` through a module-level
constant — `wide_net` and `preference_slot` — so a grep for `source="` finds
neither. A registry test built on that grep would pass while being blind to
two arms — a check that looks
thorough and quietly covers less than it claims.

So the extractor below parses each module, resolves module-level string
constants, and reports anything it still cannot settle rather than dropping
it. `test_the_extractor_resolves_the_constant_sources` pins the specific case,
so that if someone later "simplifies" this to a text scan the suite says which
capability was lost instead of merely going red.

WHAT THIS CANNOT DO, stated because a guard trusted past its reach is worse
than none. Three call sites pass `source` as a variable — a loop variable over
a dict of write-path arms, and a forwarded parameter in `rules_payload`. The
values are not statically knowable without real dataflow analysis, so those
sites are DECLARED in `FAN_OUT_SITES` and this test pins the sites, not the
values. A fourth arm added inside one of them passes here. What catches that
is the `unregistered_source` warning, which fires the first time the arm
actually records anything.
"""
from __future__ import annotations

import ast
from pathlib import Path

from grimoire.services.retrieval_registry import (
    ASKED, FAN_OUT_SITES, POINTS, UNBIDDEN,
)
from grimoire.services.retrieval_surfaces import SURFACES

SRC = Path(__file__).resolve().parents[1] / "src"

# The recorders, plus the one FORWARDER. `rules_payload` takes a `source` and
# passes it to `record_rule_surfaced` on its caller's behalf,
# so its callers are the real declaration site and a scan of recorders alone
# would miss every ambient rule surfacing.
RECORDERS = {
    "record_retrieval", "record_surfaced", "record_pulled",
    "record_rule_surfaced", "record_rule_pulled", "rules_payload",
    # The rulings arm: its two recorders, and `rulings_for_paths`,
    # which forwards its caller's `source` the way `rules_payload` does.
    "record_system_surfaced", "record_system_pulled", "rulings_for_paths",
}


def _module_constants(tree: ast.Module) -> dict[str, str]:
    """Module-level NAME = "literal", so `source=SOURCE` resolves."""
    out: dict[str, str] = {}
    for node in tree.body:
        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant) \
           and isinstance(node.value.value, str):
            for t in node.targets:
                if isinstance(t, ast.Name):
                    out[t.id] = node.value.value
    return out


def call_sites() -> tuple[dict[str, set[str]], list[str]]:
    """Every `source=` reaching a recorder: resolved, and what could not be."""
    found: dict[str, set[str]] = {}
    unresolved: list[str] = []
    for path in sorted(SRC.rglob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        consts = _module_constants(tree)
        rel = path.relative_to(SRC).as_posix()
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            fn = node.func
            name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", None)
            if name not in RECORDERS:
                continue
            kw = next((k for k in node.keywords if k.arg == "source"), None)
            if kw is None:
                continue  # a recorder called without one is a different bug
            v = kw.value
            if isinstance(v, ast.Constant) and isinstance(v.value, str):
                found.setdefault(v.value, set()).add(rel)
            elif isinstance(v, ast.Name) and v.id in consts:
                found.setdefault(consts[v.id], set()).add(rel)
            elif isinstance(v, ast.Name):
                unresolved.append(f"{rel}::{name}(source={v.id})")
            else:
                unresolved.append(f"{rel}::{name}(source=<expression>)")
    return found, unresolved


def test_the_extractor_finds_something() -> None:
    """The guard has to be able to fail.

    An extractor that silently matched nothing would make every assertion
    below vacuously true — a green suite proving the opposite of what it
    claims.
    """
    found, _ = call_sites()
    assert len(found) >= 20, f"suspiciously few call sites found: {sorted(found)}"


def test_the_extractor_resolves_the_constant_sources() -> None:
    """The specific capability a grep would lose. See the module docstring.

    The pipeline's reserved slot records through a module constant, so it
    carries the case.
    """
    found, _ = call_sites()
    for via_constant in ("wide_net", "preference_slot"):
        assert via_constant in found, (
            f"{via_constant} reaches its recorder through a module constant; "
            f"an extractor that cannot resolve one is blind to it"
        )


def test_every_call_site_source_is_registered() -> None:
    """The point of the file: adding an arm means declaring it."""
    found, _ = call_sites()
    missing = {s: sorted(found[s]) for s in found if s not in POINTS}
    assert not missing, (
        "these sources record telemetry but are not in "
        f"retrieval_registry.POINTS: {missing}"
    )


def test_every_unresolved_call_site_is_declared() -> None:
    """A site passing a variable must be named, not silently skipped."""
    _, unresolved = call_sites()
    # EQUALITY, not prefix. The first spelling of this compared against a
    # prefix that could never match — the paths begin `grimoire/` — so the check
    # passed by matching nothing, which is the failure mode a guard is most
    # likely to have and least likely to show.
    undeclared = sorted({u for u in unresolved if u not in FAN_OUT_SITES})
    assert not undeclared, (
        "these call sites pass `source` as a value this test cannot resolve, "
        "and are not declared in FAN_OUT_SITES: " + repr(undeclared)
    )


def test_the_declared_fan_out_values_are_registered() -> None:
    """The sites are unresolvable; the values they claim to pass are not."""
    for site, sources in FAN_OUT_SITES.items():
        for s in sources:
            assert s in POINTS, f"{site} claims to emit {s!r}, which is not registered"


def test_every_tunable_surface_is_also_a_registered_point() -> None:
    """The two registries answer different questions and must not drift.

    `SURFACES` is what can be TUNED, `POINTS` is what can be MEASURED. A
    surface with a floor dial and no entry here would be adjustable and
    unjudgeable at the same time.
    """
    missing = [s for s in SURFACES if s not in POINTS]
    assert not missing, f"tunable but unregistered: {missing}"


def test_a_quiet_point_says_why() -> None:
    """A justified silence must carry its justification.

    Without the reason the reader cannot tell a decision from an oversight,
    which is the whole difference this field exists to record.
    """
    silent_without_reason = [
        s for s, p in POINTS.items() if not p.expects_traffic and not p.quiet_because
    ]
    assert not silent_without_reason, silent_without_reason


def test_every_point_declares_a_known_kind() -> None:
    from grimoire.services.retrieval_registry import AMBIENT, PULL

    for s, p in POINTS.items():
        assert p.kind in {UNBIDDEN, ASKED, AMBIENT, PULL}, (s, p.kind)


def test_the_reserved_slots_are_measurable_even_though_they_are_not_tunable() -> None:
    """The case that motivated a second registry rather than reusing SURFACES.

    A budget of 1 is the reserved slots' feature, so they are deliberately
    absent from the tuning registry — but "never places a hit" is exactly the
    kind of thing this readout exists to notice.
    """
    for slot in ("preference_slot", "reuse_slot", "lesson_slot"):
        assert slot not in SURFACES, f"{slot} became tunable; re-read the decision"
        assert slot in POINTS
        assert POINTS[slot].kind == UNBIDDEN