"""The review pass: judging whether an injected line RELATED.

Unit tests pin the refusals and the re-run's fidelity to the arm it replays.
Integration tests pin the parts a mock would make true by construction: which
logged calls are offered, that a verdict lands and replaces, that "opened
after" reads the usage table, and that the readout counts by rank.
"""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio

from grimoire.models import async_session
from grimoire.services import retrieval_review as review
from tests.helpers import ensure_user, fake_note


# ── refusals ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_only_an_arm_that_can_be_re_run_exactly_is_offered():
    with pytest.raises(ValueError, match="source must be one of"):
        await review.menus_to_review(1, source="write_path")


@pytest.mark.asyncio
async def test_a_verdict_outside_the_vocabulary_is_refused():
    with pytest.raises(ValueError, match="verdict must be one of"):
        await review.judge_menu(1, log_id=1, verdicts=[
            {"record_id": 5, "verdict": "relevant", "reason": "x"},
        ])


@pytest.mark.asyncio
async def test_a_verdict_without_its_reason_is_refused():
    with pytest.raises(ValueError, match="needs its reason"):
        await review.judge_menu(1, log_id=1, verdicts=[
            {"record_id": 5, "verdict": "unrelated", "reason": "  "},
        ])


@pytest.mark.asyncio
async def test_an_empty_verdict_list_is_refused():
    with pytest.raises(ValueError, match="at least one"):
        await review.judge_menu(1, log_id=1, verdicts=[])


# ── the re-run replays the arm ───────────────────────────────────────────


_FILTERS = ("note_type", "task_kind", "include_global_kinds", "scope")


@pytest.mark.asyncio
async def test_the_re_run_searches_the_way_the_arm_does():
    """A re-run with different visibility or kinds from the arm's would judge
    a menu nobody was shown. Both searches are RUN and their keywords
    compared, so a change to the arm's spec fails here until the review
    follows it — and the arm has to have filters for the check to mean
    anything."""
    from grimoire.services import retrieval_pipeline as rp

    arm_kw: dict = {}

    async def arm_search(_uid, _q, **kw):
        arm_kw.update(kw)
        return []

    await rp.run_note_arm(
        rp.AUTO_INJECT, rp.NoteMoment(user_id=1, query="q", project_id=None),
        floor=0.5, budget=3,
        io=rp.NoteIO(search=arm_search, record_retrieval=MagicMock(),
                     record_surfaced=MagicMock()),
    )
    rerun_kw: dict = {}

    async def rerun_search(_uid, _q, **kw):
        rerun_kw.update(kw)
        return []

    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.5, limit_n=3, project_id=None,
                          created_at=CALL)
    with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=rerun_search):
        await review._rerun(1, row, 3)

    arm = {k: arm_kw[k] for k in _FILTERS if k in arm_kw}
    assert "scope" in arm and "include_global_kinds" in arm, (
        f"the arm no longer passes its visibility filters: {arm_kw}"
    )
    assert {k: rerun_kw[k] for k in _FILTERS if k in rerun_kw} == arm


CALL = datetime(2026, 9, 20, 12, 0, tzinfo=timezone.utc)
BEFORE = CALL - timedelta(days=1)


@pytest.mark.asyncio
async def test_the_re_run_ranks_from_one_and_marks_the_budget_cut():
    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=2, project_id=None,
                          created_at=CALL)
    hits = [
        (0.81, fake_note(id=11, title="first", created_at=BEFORE, updated_at=BEFORE)),
        (0.74, fake_note(id=12, title="second", created_at=BEFORE, updated_at=BEFORE)),
        (0.66, fake_note(id=13, title="third", created_at=BEFORE, updated_at=BEFORE)),
    ]

    async def search(user_id, query, **kw):
        assert kw["limit"] == 3 + review.POSTDATED_SLACK and kw["threshold"] == 0.55
        assert kw["project_id"] is None
        kw["report"]["best_chunk"] = {12: {"text": "second\nthe passage that matched"}}
        return hits

    with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=search):
        lines = await review._rerun(1, row, 3)
    assert [(ln["rank"], ln["record_id"], ln["within_budget"]) for ln in lines] == [
        (1, 11, True), (2, 12, True), (3, 13, False),
    ]
    assert lines[1]["passage"] == "the passage that matched"
    assert lines[0]["passage"] is None


@pytest.mark.asyncio
async def test_a_record_written_after_the_call_is_dropped_before_ranking():
    """The session that made a call goes on to write about it, and that record
    outranks everything in a re-run. The call could never have been offered
    it, so it must not take a rank — or a verdict — from what was."""
    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=1, project_id=None,
                          created_at=CALL)
    after = CALL + timedelta(hours=1)
    hits = [
        (0.90, fake_note(id=21, title="the decision", created_at=after, updated_at=after)),
        (0.80, fake_note(id=22, title="held then", created_at=BEFORE, updated_at=BEFORE)),
        (0.70, fake_note(id=23, title="edited since", created_at=BEFORE, updated_at=after)),
    ]

    async def search(user_id, query, **kw):
        return hits

    rep: dict = {}
    with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=search):
        lines = await review._rerun(1, row, 2, report=rep)
    assert rep["postdated"] == [21]
    assert [(ln["rank"], ln["record_id"], ln["within_budget"], ln["changed_since_call"])
            for ln in lines] == [(1, 22, True, False), (2, 23, False, True)]


# ── the rule arms ────────────────────────────────────────────────────────


def _rule(rid, title, created=BEFORE, updated=BEFORE):
    return SimpleNamespace(
        id=rid, title=title, kind="rule", when_to_apply=f"when {title}",
        statement="s", created_at=created, updated_at=updated,
    )


def test_the_reviewable_rule_arms_are_the_ones_that_annotate():
    """Every rule arm that shows lines can be reviewed; the reply backstop
    shows none, so there is no menu to judge."""
    from grimoire.services import retrieval_pipeline as rp

    assert set(review.RULE_REVIEWABLE) == {
        a.source for a in rp.RULE_ARMS if not a.stop_only
    }


@pytest.mark.asyncio
async def test_the_rule_re_run_searches_the_way_the_arm_does():
    """The floor and the project the arm searched with, run and compared."""
    from grimoire.services import retrieval_pipeline as rp

    arm_kw: dict = {}

    async def arm_search(_uid, _q, **kw):
        if "kind" not in kw:
            arm_kw.update(kw)
        return []

    await rp.run_rule_arm(
        rp.PROMPT_RULE, rp.RuleMoment(user_id=1, query="q", project_id=7),
        floor=0.7, budget=3,
        io=rp.RuleIO(search=arm_search, record_retrieval=MagicMock(),
                     record_rule_surfaced=MagicMock()),
    )
    rerun_kw: dict = {}

    async def rerun_search(_uid, _q, **kw):
        rerun_kw.update(kw)
        return []

    row = SimpleNamespace(source="prompt_rule", query="q", threshold=0.7, limit_n=3,
                          project_id=7, created_at=CALL)
    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=rerun_search):
        await review._rerun(1, row, 3)

    for key in ("threshold", "project_id"):
        assert rerun_kw[key] == arm_kw[key], key
    assert "kind" not in rerun_kw


@pytest.mark.asyncio
@pytest.mark.parametrize(("source", "within"), [
    # No band: the budget alone is the cut.
    ("prompt_rule", [True, True, True, False]),
    # Banded: the third sits outside the band of the top hit, so it was
    # not shown though the budget had room for it.
    ("pre_tool_rule", [True, True, False, False]),
])
async def test_a_rule_line_is_inside_the_cut_only_where_the_arm_showed_it(source, within):
    row = SimpleNamespace(source=source, query="git push", threshold=0.7, limit_n=3,
                          project_id=None, created_at=CALL)
    hits = [(0.86, _rule(1, "one")), (0.84, _rule(2, "two")),
            (0.76, _rule(3, "three")), (0.74, _rule(4, "four"))]

    async def search(_uid, _q, **kw):
        assert kw["threshold"] == 0.7 and kw["project_id"] is None
        return hits

    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
        lines = await review._rerun(1, row, 4)
    assert [ln["within_budget"] for ln in lines] == within
    assert lines[0]["passage"] == "when one"
    assert lines[0]["kind"] == "rule"


@pytest.mark.asyncio
async def test_a_rule_written_after_the_call_is_dropped_before_ranking():
    row = SimpleNamespace(source="write_path_rule", query="q", threshold=0.7, limit_n=2,
                          project_id=None, created_at=CALL)
    after = CALL + timedelta(hours=1)
    hits = [(0.9, _rule(31, "written after", created=after, updated=after)),
            (0.8, _rule(32, "held then"))]

    async def search(_uid, _q, **kw):
        return hits

    rep: dict = {}
    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
        lines = await review._rerun(1, row, 2, report=rep)
    assert rep["postdated"] == [31]
    assert [(ln["rank"], ln["record_id"]) for ln in lines] == [(1, 32)]


# ── against Postgres ─────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def logged_calls():
    """A reviewer's logged calls: two reviewable, one that offered nothing,
    one of another arm — and another user's reviewable call."""
    from grimoire.models.retrieval_log import RetrievalLog

    async with async_session() as s:
        me = await ensure_user(s, "review_owner")
        other = await ensure_user(s, "review_other")
        now = datetime.now(timezone.utc)

        def log(uid, source="auto_inject", count=2, query="how is the cache invalidated"):
            return RetrievalLog(
                user_id=uid, source=source, query=query, threshold=0.55,
                limit_n=3, result_count=count, created_at=now - timedelta(hours=2),
                result_ids=[{"id": 501, "score": 0.8, "rank": 0}],
            )

        rows = {
            "a": log(me.id), "b": log(me.id),
            "empty": log(me.id, count=0),
            "other_arm": log(me.id, source="write_path"),
            "theirs": log(other.id),
        }
        s.add_all(rows.values())
        await s.commit()
        ids = {k: int(v.id) for k, v in rows.items()}
        ids |= {"me": me.id, "other": other.id}
    yield ids

    from sqlalchemy import delete
    from grimoire.models.note_usage import NoteUsageEvent
    from grimoire.models.retrieval_judgment import RetrievalJudgment
    async with async_session() as s:
        logs = [ids[k] for k in ("a", "b", "empty", "other_arm", "theirs")]
        await s.execute(delete(RetrievalJudgment).where(
            RetrievalJudgment.retrieval_log_id.in_(logs)))
        await s.execute(delete(RetrievalLog).where(RetrievalLog.id.in_(logs)))
        await s.execute(delete(NoteUsageEvent).where(
            NoteUsageEvent.user_id == ids["me"], NoteUsageEvent.note_id.in_([501, 502])))
        await s.commit()


def _lines(*specs):
    """A stand-in re-run: (record_id, rank, within_budget)."""
    return AsyncMock(return_value=[
        {"record_id": rid, "rank": rank, "score": 0.9 - rank / 10,
         "within_budget": within, "changed_since_call": False,
         "kind": "note", "name": f"n{rid}", "passage": "p"}
        for rid, rank, within in specs
    ])


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_the_sample_offers_only_the_reviewers_unjudged_reviewable_calls(logged_calls):
    ids = logged_calls
    with patch.object(review, "_rerun", _lines((501, 1, True))):
        out = await review.menus_to_review(ids["me"], n=20, days=1)
        assert {m["log_id"] for m in out["menus"]} == {ids["a"], ids["b"]}
        assert out["remaining"] == 2

        await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
            {"record_id": 501, "verdict": "on_point", "reason": "names the cache"},
        ])
        out = await review.menus_to_review(ids["me"], n=20, days=1)
    assert {m["log_id"] for m in out["menus"]} == {ids["b"]}


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_another_users_call_cannot_be_judged(logged_calls):
    ids = logged_calls
    with patch.object(review, "_rerun", _lines((501, 1, True))), \
            pytest.raises(ValueError, match="not found"):
        await review.judge_menu(ids["me"], log_id=ids["theirs"], verdicts=[
            {"record_id": 501, "verdict": "unrelated", "reason": "r"},
        ])


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_a_record_the_re_run_does_not_hold_is_refused(logged_calls):
    ids = logged_calls
    with patch.object(review, "_rerun", _lines((501, 1, True))), \
            pytest.raises(ValueError, match=r"\[999\]"):
        await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
            {"record_id": 999, "verdict": "unrelated", "reason": "r"},
        ])


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_verdicts_land_replace_and_read_back_by_rank(logged_calls):
    from grimoire.models.note_usage import NoteUsageEvent
    from grimoire.services.retrieval_telemetry import retrieval_summary

    ids = logged_calls
    # The reviewer's agent opened 501 an hour inside the call; nothing opened 502.
    async with async_session() as s:
        s.add(NoteUsageEvent(
            user_id=ids["me"], note_id=501, event="pulled", source="mcp_get_note",
            created_at=datetime.now(timezone.utc) - timedelta(hours=1, minutes=30),
        ))
        await s.commit()

    rerun = _lines((501, 1, True), (502, 4, False))
    with patch.object(review, "_rerun", rerun):
        await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
            {"record_id": 501, "verdict": "adjacent", "reason": "first look"},
            {"record_id": 502, "verdict": "on_point", "reason": "the exact fix"},
        ])
        again = await review.judge_menu(ids["me"], log_id=ids["a"], verdicts=[
            {"record_id": 501, "verdict": "on_point", "reason": "on reflection"},
        ])
    assert again == {"log_id": ids["a"], "recorded": 1, "replaced": 1}

    block = (await retrieval_summary(ids["me"], days=1))["judged"]
    assert "judged_failed" not in block, "the readout did not execute"
    ai = block["auto_inject"]
    assert ai["judged_calls"] == 1 and ai["judged_lines"] == 2
    assert ai["within_budget"] == {
        "on_point": 1, "adjacent": 0, "unrelated": 0, "opened": 1}
    assert ai["beyond_budget"] == {
        "on_point": 1, "adjacent": 0, "unrelated": 0, "opened": 0}
    assert [r["rank"] for r in ai["by_rank"]] == [1, 4]
    # Related and left closed: the number an open rate could never show.
    assert ai["on_point_unopened"] == 1


@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.usefixtures("_dispose_engine")
async def test_a_fresh_install_reads_an_empty_judged_block_not_a_failure():
    from grimoire.services.retrieval_telemetry import retrieval_summary

    assert (await retrieval_summary(990041, days=30))["judged"] == {}