FabledScribe/tests/test_rule_via_lesson.py

"""A rule reached through its lessons (milestone 440, #4633).

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 scribe.services import plugin_context as pc
from scribe.services import rule_usage
from scribe.services.retrieval_registry import POINTS

# 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_runs_only_when_the_arm_ran_and_never_leaks_its_key():
    step = AsyncMock(return_value=(["Standing rule … Reached through lesson #41"], [7]))
    with patch.object(pc, "_rules_via_lessons", step):
        idle = await pc._add_rules_via_lessons(
            1, {"context": "", "rule_ids": []}, project_id=2,
            exclude_rule_ids=[3], held_rule_ids=[], where="here",
        )
        ran = await pc._add_rules_via_lessons(
            1, {"context": "direct line", "rule_ids": [5], "_via_query": "q"},
            project_id=2, exclude_rule_ids=[3], held_rule_ids=[], where="here",
        )
    assert idle == {"context": "", "rule_ids": []}
    assert step.await_count == 1
    assert step.await_args.kwargs["skip"] == {3, 5}
    assert "_via_query" not in ran
    assert ran["rule_ids"] == [5, 7]
    assert ran["context"].startswith("direct line\n")


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