FabledGrimoire/tests/test_rule_via_lesson.py

"""A rule reached through its lessons.

The step runs after each of the three rule arms. These pin, with the database
and the embedder stubbed: that nothing is searched when no lesson carries a
confirmed link; that a matching linked lesson brings its rule in rule voice,
naming the lesson; that suppression is by RULE; that the slot holds one line;
and that the call is logged under its own source. That a SUGGESTED link never
expands — the soft link proving itself — is pinned against Postgres in
tests/test_integration_lesson_rule_links.py, where the state filter lives.
"""
from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services import plugin_context as pc
from grimoire.services import rule_usage
from grimoire.services.retrieval_registry import POINTS
from grimoire.services.retrieval_pipeline import RuleMoment, RuleResult

# Bound before conftest's autouse stub replaces the module attribute.
_REAL = pc._rules_via_lessons


def _lesson(lid=41, title="An overrun run usually failed early"):
    return SimpleNamespace(
        id=lid, title=title, note_type="lesson", body="",
        data={"what": title, "when_to_apply": "a CI run overran"},
    )


def _rule(rid, title="Read the job log first", kind="rule"):
    return SimpleNamespace(id=rid, title=title, kind=kind, when_to_apply="a run overran")


def _stubs(*, confirmed, found, by_lesson):
    log, surfaced = MagicMock(), MagicMock()
    stack = [
        patch.object(pc.lesson_rules_svc, "confirmed_lessons", AsyncMock(return_value=confirmed)),
        patch.object(pc.lesson_rules_svc, "confirmed_rules_in_scope", AsyncMock(return_value=by_lesson)),
        patch.object(pc, "get_autoinject_config", AsyncMock(return_value={"threshold": 0.5})),
        patch.object(pc, "semantic_search_notes", AsyncMock(return_value=found)),
        patch.object(pc, "record_retrieval", log),
        patch.object(pc, "record_rule_surfaced", surfaced),
    ]
    return stack, log, surfaced


async def _run(stack, **kw):
    for p in stack:
        p.start()
    try:
        return await _REAL(1, "the CI run is still going", project_id=2,
                           skip=kw.get("skip", set()), held=set(), where="to this request")
    finally:
        for p in stack:
            p.stop()


@pytest.mark.asyncio
async def test_no_confirmed_link_means_no_search_and_no_row():
    search = AsyncMock()
    stack, log, _ = _stubs(confirmed=set(), found=[], by_lesson={})
    stack[3] = patch.object(pc, "semantic_search_notes", search)
    assert await _run(stack) == ([], [])
    search.assert_not_awaited()
    log.assert_not_called()


@pytest.mark.asyncio
async def test_a_matching_linked_lesson_brings_its_rule_in_rule_voice():
    stack, log, surfaced = _stubs(
        confirmed={41}, found=[(0.71, _lesson())], by_lesson={41: [_rule(7)]},
    )
    lines, ids = await _run(stack)
    assert ids == [7]
    assert lines[0].startswith("Standing rule")
    assert "Reached through lesson #41" in lines[0]
    assert log.call_args.kwargs["source"] == "rule_via_lesson"
    assert surfaced.call_args.kwargs == {"user_id": 1, "rule_ids": [7], "source": "rule_via_lesson"}


@pytest.mark.asyncio
async def test_a_lesson_without_a_confirmed_link_carries_nothing():
    """The search found a lesson, but it is not in the confirmed set — a
    suggested or unlinked lesson brings no rule."""
    stack, log, surfaced = _stubs(
        confirmed={99}, found=[(0.9, _lesson(41))], by_lesson={},
    )
    assert await _run(stack) == ([], [])
    surfaced.assert_not_called()
    assert log.call_args.kwargs["results"] == []


@pytest.mark.asyncio
async def test_suppression_is_by_rule_whichever_lesson_reached_it():
    stack, log, surfaced = _stubs(
        confirmed={41}, found=[(0.71, _lesson())], by_lesson={41: [_rule(7)]},
    )
    assert await _run(stack, skip={7}) == ([], [])
    assert log.call_args.kwargs["suppressed"] == 1
    surfaced.assert_not_called()


@pytest.mark.asyncio
async def test_the_slot_holds_one_line():
    stack, _log, _ = _stubs(
        confirmed={41, 42},
        found=[(0.8, _lesson(41)), (0.7, _lesson(42, "Another"))],
        by_lesson={41: [_rule(7), _rule(8)], 42: [_rule(9)]},
    )
    lines, ids = await _run(stack)
    assert ids == [7] and len(lines) == pc.VIA_LESSON_LIMIT == 1


@pytest.mark.asyncio
async def test_a_failure_brings_no_rule_and_raises_nothing():
    stack, _log, _ = _stubs(confirmed={41}, found=[], by_lesson={})
    stack[0] = patch.object(pc.lesson_rules_svc, "confirmed_lessons",
                            AsyncMock(side_effect=RuntimeError("db down")))
    assert await _run(stack) == ([], [])


@pytest.mark.asyncio
async def test_the_step_follows_the_arm_on_its_query_and_skips_what_it_named():
    """One composer for all three rule builders: the
    via-lesson step searches with the query the arm searched with, skips the
    session's ledger and every rule the arm's band named, and its lines and
    fresh ids join the arm's — while `shown_rule_ids` stays the direct band."""
    step = AsyncMock(return_value=(["Standing rule … Reached through lesson #41"], [7]))
    direct = RuleResult(lines=["direct line"], rule_ids=[5], shown_rule_ids=[5, 6])
    moment = RuleMoment(user_id=1, query="q", project_id=2, where="here",
                        exclude=frozenset({3, 6}))
    with patch.object(pc, "_rules_via_lessons", step), \
         patch.object(pc.rp, "run_rule_arm", AsyncMock(return_value=direct)):
        out = await pc._rule_moment(pc.rp.PROMPT_RULE, moment, floor=0.5, budget=3)
    assert step.await_args.args[1] == "q"
    assert step.await_args.kwargs["skip"] == {3, 5, 6}
    assert out.rule_ids == [5, 7]
    assert out.lines == ["direct line", "Standing rule … Reached through lesson #41"]
    assert out.shown_rule_ids == [5, 6]


@pytest.mark.asyncio
async def test_a_blank_prompt_runs_neither_the_arm_nor_the_step():
    step = AsyncMock(return_value=([], []))
    with patch.object(pc, "_rules_via_lessons", step):
        out = await pc.build_prompt_rule_hint(1, "   ")
    assert out == {"context": "", "rule_ids": []}
    step.assert_not_awaited()


def test_only_the_composer_runs_a_rule_arm_or_the_via_lesson_step():
    """Structural: every rule builder goes through `_rule_moment`, so a
    builder that runs the arm or the via-lesson step by hand is a second copy
    of the composer — and a count is what lets it fail."""
    import ast
    from pathlib import Path

    tree = ast.parse(Path("src/grimoire/services/plugin_context.py").read_text())
    callers: dict[str, set[str]] = {}
    for fn in ast.walk(tree):
        if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        for call in ast.walk(fn):
            if isinstance(call, ast.Call):
                name = getattr(call.func, "attr", None) or getattr(call.func, "id", None)
                if name in ("run_rule_arm", "_rules_via_lessons", "_rule_moment"):
                    callers.setdefault(name, set()).add(fn.name)
    assert callers["run_rule_arm"] == {"_rule_moment"}
    assert callers["_rules_via_lessons"] == {"_rule_moment"}
    assert callers["_rule_moment"] == {
        "build_prompt_rule_hint", "build_tool_rule_hint", "build_write_path_hint",
    }


def test_the_source_is_ranked_and_registered():
    assert "rule_via_lesson" in rule_usage.RANKED_SOURCES
    assert "rule_via_lesson" in POINTS