"""The review pass: judging whether an injected line RELATED (#4772).

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.
"""
import ast
import inspect
import textwrap
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest
import pytest_asyncio

from scribe.models import async_session
from scribe.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 ───────────────────────────────────────────


def _search_kwargs(fn) -> dict:
    """The constant keywords `fn` passes to semantic_search_notes."""
    tree = ast.parse(textwrap.dedent(inspect.getsource(fn)))
    for node in ast.walk(tree):
        if (isinstance(node, ast.Call) and getattr(node.func, "id", "")
                == "semantic_search_notes"):
            return {
                kw.arg: kw.value.value for kw in node.keywords
                if isinstance(kw.value, ast.Constant)
            }
    return {}


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. Read from both call sites, so a change to the
    arm's search fails here until the review follows it."""
    from scribe.services.plugin_context import build_autoinject_hint

    arm = _search_kwargs(build_autoinject_hint)
    rerun = _search_kwargs(review._rerun)
    assert arm, "found no semantic_search_notes call in the arm"
    for key in ("include_global_kinds", "scope"):
        assert key in arm, f"the arm no longer passes {key} as a constant"
        assert rerun.get(key) == arm[key], key


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

    async def search(user_id, query, **kw):
        assert kw["limit"] == 3 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("scribe.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


# ── 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 scribe.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 scribe.models.note_usage import NoteUsageEvent
    from scribe.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, "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 scribe.models.note_usage import NoteUsageEvent
    from scribe.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 scribe.services.retrieval_telemetry import retrieval_summary

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