praxis/tests/test_mastery_integration.py

"""SLICE-07 TASK-07-03 — mastery integration test (end-to-end scoring flow).

Simulates a session with turns → runs the mastery flow → verifies the scenario
score, IRT theta update, path progress advancement, and the mastery_gate_event
audit row. The LLM for evidence extraction is mocked. Verifies determinism
(same input → same scores) and the scoring_inconclusive short-circuit path.
"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock

import pytest

from db.store import PraxisStore, HARDCODED_LEARNER_ID
from server.mastery.irt import IRTEngine, DEFAULT_THETA, DEFAULT_SIGMA_SQ
from server.mastery.rubric_loader import clear_cache, load_rubric
from server.paths.engine import PathEngine
from server.scenarios.loader import load as load_scenario
from server.session_recorder import MasteryFlowDeps, SessionRecorder

_RUBRICS_DIR = Path(__file__).resolve().parent.parent / "rubrics"
_SCENARIOS_DIR = Path(__file__).resolve().parent.parent / "scenarios"
_PATHS_DIR = Path(__file__).resolve().parent.parent / "paths"


def _turns() -> list[dict]:
    return [
        {"role": "customer", "content": "My order arrived cracked and I'm furious."},
        {
            "role": "learner",
            "content": (
                "I'm really sorry the bowl arrived cracked — that's genuinely "
                "frustrating. I can refund the full amount to your original card "
                "within 3 business days, or send a replacement first class tomorrow. "
                "Which would you prefer?"
            ),
        },
        {"role": "customer", "content": "Just refund it."},
        {
            "role": "learner",
            "content": (
                "Of course — I've issued a full refund of $42.99 to your Visa ending "
                "4421. You'll see it in 2-3 business days. Is there anything else I "
                "can help with today?"
            ),
        },
    ]


def _canned_good() -> str:
    t1 = _turns()[1]["content"]
    t2 = _turns()[3]["content"]
    return json.dumps(
        [
            {"criterion_id": "empathy", "quote": t1, "signals": ["named_emotion_in_own_words", "acknowledged_specific"]},
            {"criterion_id": "resolution", "quote": t1, "signals": ["concrete_method", "concrete_amount_or_channel", "concrete_next_step"]},
            {"criterion_id": "de_escalation", "quote": t1, "signals": ["explicit_acknowledge_reframe_offer"]},
            {"criterion_id": "professionalism", "quote": t2, "signals": ["plain_language", "in_role_throughout", "no_prohibited_advice"]},
        ]
    )


def _canned_bad() -> str:
    return json.dumps(
        [
            {"criterion_id": "empathy", "quote": "I apologize for the inconvenience, dear customer.", "signals": ["named_emotion_in_own_words"]},
            {"criterion_id": "resolution", "quote": "I will issue a refund shortly.", "signals": ["concrete_method"]},
        ]
    )


def _make_llm(raws: list[str]) -> AsyncMock:
    llm = AsyncMock()
    llm.chat_full = AsyncMock(side_effect=[(r, {"model": "test"}) for r in raws])
    return llm


def _deps(llm: AsyncMock, scenario_id: str = "cs_refund_ca_v01") -> MasteryFlowDeps:
    clear_cache()
    return MasteryFlowDeps(
        llm=llm,
        irt=IRTEngine(),
        path_engine=PathEngine(paths_dir=_PATHS_DIR),
        load_rubric=lambda: load_rubric("customer_service", rubrics_dir=_RUBRICS_DIR),
        load_scenario=lambda: load_scenario(scenario_id, scenarios_dir=_SCENARIOS_DIR),
        load_path=lambda: PathEngine(paths_dir=_PATHS_DIR).load_path("customer_service"),
    )


@pytest.fixture
def tmp_db(tmp_path: Path) -> Path:
    return tmp_path / "test_mastery_int.db"


def _run(coro):
    return asyncio.run(coro)


@pytest.mark.asyncio
async def test_mastery_flow_end_to_end_scored(tmp_db: Path):
    store = PraxisStore(tmp_db)
    await store.init()
    llm = _make_llm([_canned_good()])
    deps = _deps(llm)

    rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
    await rec.start()
    rec.set_mastery_turns(_turns())
    rec.set_branch_path(["accept_resolution"])
    await rec.end(outcome="success", debrief_text="nicely done")

    result = await rec.run_mastery_flow(deps)

    assert result["status"] == "scored"
    assert result["scenario_id"] == "cs_refund_ca_v01"
    assert result["passed"] is True
    assert result["weighted_mean"] >= 3.0

    # Theta moved up after a passing scenario against difficulty 1.
    assert result["theta"] > DEFAULT_THETA
    assert result["observations"] == 1
    assert result["gate_open"] is False  # only 1 distinct passed
    assert result["week"] == 1
    assert result["new_week"] == 1

    # Persistence: ability + progress rows.
    ability = await store.get_ability(HARDCODED_LEARNER_ID, "customer_service")
    assert ability is not None
    assert ability["theta"] == pytest.approx(result["theta"])
    assert ability["observations"] == 1

    progress = await store.get_progress(HARDCODED_LEARNER_ID, "customer_service")
    assert progress is not None
    assert progress["current_week"] == 1
    assert json.loads(progress["scenarios_passed_json"]) == ["cs_refund_ca_v01"]

    # Audit log: exactly one gate event recorded, with the rubric scores.
    events = await store.list_gate_events(HARDCODED_LEARNER_ID, "customer_service")
    assert len(events) == 1
    ev = events[0]
    assert ev["week"] == 1
    assert ev["gate_open"] == 0
    assert json.loads(ev["scenarios_passed_json"]) == ["cs_refund_ca_v01"]
    rubric_scores = json.loads(ev["rubric_scores_json"])
    assert len(rubric_scores) == 4
    assert {r["criterion_id"] for r in rubric_scores} == {
        "empathy",
        "resolution",
        "de_escalation",
        "professionalism",
    }


@pytest.mark.asyncio
async def test_mastery_flow_is_deterministic(tmp_path: Path):
    """Same input + same starting state → same scores + same theta delta."""
    import shutil

    async def _one(db_path: Path) -> dict[str, Any]:
        store = PraxisStore(db_path)
        await store.init()
        rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
        await rec.start()
        rec.set_mastery_turns(_turns())
        await rec.end(outcome="success")
        return await rec.run_mastery_flow(_deps(_make_llm([_canned_good()])))

    db1 = tmp_path / "det1.db"
    db2 = tmp_path / "det2.db"
    r1 = await _one(db1)
    r2 = await _one(db2)
    assert r1["weighted_mean"] == r2["weighted_mean"]
    assert r1["passed"] == r2["passed"]
    assert r1["theta"] == pytest.approx(r2["theta"])
    assert r1["sigma_sq"] == pytest.approx(r2["sigma_sq"])
    assert r1["gate_open"] == r2["gate_open"]


@pytest.mark.asyncio
async def test_mastery_flow_scoring_inconclusive_no_score_no_gate_event(tmp_db: Path):
    store = PraxisStore(tmp_db)
    await store.init()
    # Three bad-quote responses → 1 initial + 2 re-extractions = 3 attempts → inconclusive.
    llm = _make_llm([_canned_bad(), _canned_bad(), _canned_bad()])
    deps = _deps(llm)

    rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
    await rec.start()
    rec.set_mastery_turns(_turns())
    await rec.end(outcome="success")

    result = await rec.run_mastery_flow(deps)

    assert result["status"] == "scoring_inconclusive"
    assert result["retry_advised"] is True
    assert result["attempts"] == 3

    # No ability row written (theta unchanged / absent).
    ability = await store.get_ability(HARDCODED_LEARNER_ID, "customer_service")
    assert ability is None

    # No progress row written.
    progress = await store.get_progress(HARDCODED_LEARNER_ID, "customer_service")
    assert progress is None

    # No gate event recorded.
    events = await store.list_gate_events(HARDCODED_LEARNER_ID, "customer_service")
    assert events == []


@pytest.mark.asyncio
async def test_mastery_flow_failure_does_not_add_to_passed(tmp_db: Path):
    store = PraxisStore(tmp_db)
    await store.init()
    # Empathy at level 1 (scripted line only) + others weak → conjunctive floor
    # or mean failure. Use signals that map to low levels.
    weak = json.dumps(
        [
            {"criterion_id": "empathy", "quote": _turns()[1]["content"], "signals": ["scripted_empathy_line"]},
            {"criterion_id": "resolution", "quote": _turns()[1]["content"], "signals": ["resolution_missing_specifics"]},
            {"criterion_id": "de_escalation", "quote": _turns()[1]["content"], "signals": ["avoidance_or_deflection"]},
            {"criterion_id": "professionalism", "quote": _turns()[3]["content"], "signals": ["uses_jargon", "breaks_tone_once"]},
        ]
    )
    llm = _make_llm([weak])
    deps = _deps(llm)

    rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
    await rec.start()
    rec.set_mastery_turns(_turns())
    await rec.end(outcome="failure")

    result = await rec.run_mastery_flow(deps)

    assert result["status"] == "scored"
    assert result["passed"] is False

    progress = await store.get_progress(HARDCODED_LEARNER_ID, "customer_service")
    assert progress is not None
    assert json.loads(progress["scenarios_passed_json"]) == []
    assert progress["gate_open"] == 0

    # Theta moves down after a failed scenario.
    assert result["theta"] < DEFAULT_THETA

    events = await store.list_gate_events(HARDCODED_LEARNER_ID, "customer_service")
    assert len(events) == 1
    assert events[0]["gate_open"] == 0