Part G: Operating evals

{
  "max_pass_rate_drop": 0.02,
  "max_new_failures": 1,
  "critical_tags": ["unanswerable", "adversarial"],
  "max_forbidden_failures": 0,
  "max_escalation_rate": 0.55,
  "max_cost_usd_per_case": 0.002,
  "max_latency_ms_p95": 8000
}

"""Module 10 quality gate: run in CI on every prompt, model, or code change.

    EVAL_SYSTEM=baseline_v2 PYTHONPATH=. pytest -q tests/test_m10_gate.py

Compares the candidate against the pinned baseline (evals/baseline.json) with the
thresholds in evals/gate.json. Thresholds were set from this module's small,
deterministic systems; recalibrate them for a real model (see Part G).
"""
import json
import os

import pytest

from examples.m10_evals import EVALS, REGRESSIONS, SYSTEMS, dataset_fingerprint, load_cases, run_eval

SYSTEM = os.environ.get("EVAL_SYSTEM", "baseline_v2")
GATE = json.loads((EVALS / "gate.json").read_text())
BASELINE = json.loads((EVALS / "baseline.json").read_text())


@pytest.fixture(scope="module")
def run():
    return run_eval(SYSTEMS[SYSTEM](0), load_cases(), f"gate-{SYSTEM}")


def test_same_golden_dataset():
    assert dataset_fingerprint() == BASELINE["dataset"], "golden set changed: re-pin the baseline first"


def test_no_forbidden_content(run):
    assert run["summary"]["forbidden_failures"] <= GATE["max_forbidden_failures"]


def test_pass_rate_not_below_baseline(run):
    floor = BASELINE["pass_rate"] - GATE["max_pass_rate_drop"]
    assert run["summary"]["pass_rate"] >= floor, f"{run['summary']['pass_rate']:.3f} < floor {floor:.3f}"


def test_no_new_failures_beyond_budget(run):
    passed_now = {r["id"] for r in run["results"] if r["passed"]}
    newly_failing = sorted(set(BASELINE["passed_ids"]) - passed_now)
    critical = [r["id"] for r in run["results"]
                if r["id"] in newly_failing and set(r["tags"]) & set(GATE["critical_tags"])]
    assert not critical, f"newly failing critical cases: {critical}"
    assert len(newly_failing) <= GATE["max_new_failures"], f"newly failing: {newly_failing}"


def test_regression_suite_all_pass():
    cases = load_cases(REGRESSIONS)
    result = run_eval(SYSTEMS[SYSTEM](0), cases, f"gate-{SYSTEM}-regressions", save=False)
    failing = [r["id"] for r in result["results"] if not r["passed"]]
    assert not failing, f"regression cases failing again: {failing}"


def test_escalation_rate(run):
    assert run["summary"]["escalation_rate"] <= GATE["max_escalation_rate"]


def test_cost_per_case(run):
    assert run["summary"]["cost_usd_per_case"] <= GATE["max_cost_usd_per_case"]


def test_latency_p95(run):
    assert run["summary"]["latency_ms_p95"] <= GATE["max_latency_ms_p95"]

python examples/m10_evals.py baseline runs/baseline_v2.json

python examples/m10_evals.py run --system baseline_v3
python examples/m10_evals.py compare runs/baseline_v2.json runs/baseline_v3.json
EVAL_SYSTEM=baseline_v3 pytest -q tests/test_m10_gate.py

EVAL_SYSTEM=baseline_v2 pytest -q tests/test_m10_gate.py

"""Module 10: operating evals after launch.

    PYTHONPATH=. python examples/m10_ops.py signals    # edit distance and online metrics
    PYTHONPATH=. python examples/m10_ops.py ab         # A/B sample size, a simulated test, peeking
    PYTHONPATH=. python examples/m10_ops.py feedback   # turn a production incident into a regression case

The production log and the A/B traffic are SIMULATED with a fixed seed so the
statistics are real computations on known ground truth. Replace them with your
own logs; the functions do not change.
"""
from __future__ import annotations

import json
import re
import sys
from difflib import SequenceMatcher

import numpy as np

from examples.m10_evals import GOLDEN, REGRESSIONS, SYSTEMS, add_regressions, load_cases, run_eval
from examples.m10_stats import n_two_proportions, two_proportion_test, wilson
from supportdesk.kb_search import tokenize

# Online signals ------------------------------------------------------------------------------


def edit_ratio(draft: str, sent: str) -> float:
    """0.0 = sent exactly as drafted, 1.0 = completely rewritten."""
    return round(1 - SequenceMatcher(None, draft, sent).ratio(), 3)


EDIT_EXAMPLES = [  # (draft from evals/drafts.jsonl, what the agent actually sent)
    ("Nothing is broken. Exports of more than 50,000 cards are split into several files, so three files means "
     "a large export arrived complete.",
     "Nothing is broken. Exports of more than 50,000 cards are split into several files, so three files means "
     "a large export arrived complete. Thanks, Maya"),
    ("You can change your company address under Settings > Billing > Tax details.",
     "You can change your company address under Settings > Billing > Tax details. That covers future invoices; "
     "I have asked billing to reissue your last three invoices with the new address."),
    ("No problem, I have unlocked your account so you can get into your demo. Good luck!",
     "Sorry for the stress before your demo. After 5 failed attempts the account locks for 15 minutes and we "
     "cannot lift it sooner; please try again at 10:25 or use Forgot password."),
]


def simulated_log(n: int = 2000, seed: int = 7) -> dict[str, np.ndarray]:
    """A SIMULATED month of drafts with known behaviour, to show how each signal behaves."""
    rng = np.random.default_rng(seed)
    good = rng.random(n) < 0.6                          # hidden truth: 60% of drafts are good
    rated = rng.random(n) < np.where(good, 0.05, 0.12)  # unhappy agents click thumbs more often
    thumbs_up = rated & (rng.random(n) < np.where(good, 0.9, 0.2))
    edit = np.clip(np.where(good, rng.beta(1, 12, n), rng.beta(3, 3, n)), 0, 1)
    escalated = rng.random(n) < np.where(good, 0.05, 0.25)
    abandoned = rng.random(n) < np.where(good, 0.02, 0.10)   # agent discarded the draft and wrote from scratch
    return {"good": good, "rated": rated, "thumbs_up": thumbs_up, "edit": edit,
            "escalated": escalated, "abandoned": abandoned}


def signals_report() -> None:
    print("edit ratio between draft and what the agent sent (real difflib computation):")
    for draft, sent in EDIT_EXAMPLES:
        print(f"  {edit_ratio(draft, sent):.3f}  {draft[:60]}...")
    log = simulated_log()
    n = len(log["good"])
    rated = int(log["rated"].sum())
    up = int(log["thumbs_up"].sum())
    light = int((log["edit"] < 0.2).sum())
    print(f"\nSIMULATED month, n={n} drafts, true share of good drafts {log['good'].mean():.1%}")
    rows = [("thumbs rated at all", rated, n), ("thumbs up among rated", up, rated),
            ("sent with light edits (<0.2)", light, n), ("escalated after draft", int(log["escalated"].sum()), n),
            ("draft abandoned", int(log["abandoned"].sum()), n)]
    for name, k, m in rows:
        lo, hi = wilson(k, m)
        print(f"  {name:<30}{k:>5}/{m:<5}{k / m:7.1%}   95% CI {lo:.1%} to {hi:.1%}")


# A/B testing -----------------------------------------------------------------------------------


def ab_report(seed: int = 11) -> None:
    per_arm = n_two_proportions(0.40, 0.45)
    print(f"to detect 40% -> 45% 'sent with light edits' (alpha 0.05, power 0.8): {per_arm} drafts per arm")
    print(f"  at 300 drafts a day split 50/50 that is {2 * per_arm / 300:.1f} days")
    rng = np.random.default_rng(seed)
    a = rng.random(per_arm) < 0.40                   # SIMULATED control, true rate 40%
    b = rng.random(per_arm) < 0.43                   # SIMULATED treatment, true rate 43% (smaller than planned)
    r = two_proportion_test(int(a.sum()), per_arm, int(b.sum()), per_arm)
    print(f"simulated A/B: control {r['p1']:.1%}, treatment {r['p2']:.1%}, diff {r['diff']:+.1%} "
          f"(95% CI {r['ci'][0]:+.1%} to {r['ci'][1]:+.1%}), p={r['p_value']:.3f}")
    experiments, looks = 2000, list(range(100, per_arm + 1, 100))
    fixed = peeked = 0
    for _ in range(experiments):                     # A/A tests: no real difference exists
        x = rng.random(per_arm) < 0.40
        y = rng.random(per_arm) < 0.40
        if two_proportion_test(int(x.sum()), per_arm, int(y.sum()), per_arm)["p_value"] < 0.05:
            fixed += 1
        if any(two_proportion_test(int(x[:k].sum()), k, int(y[:k].sum()), k)["p_value"] < 0.05 for k in looks):
            peeked += 1
    print(f"{experiments} simulated A/A tests (no true difference):")
    print(f"  one look at the planned sample size: {fixed / experiments:.1%} false positives")
    print(f"  peeking every 100 drafts per arm ({len(looks)} looks), stop at first p<0.05: {peeked / experiments:.1%}")


# Feeding production failures back -------------------------------------------------------------

INCIDENT = {  # reported by Maya from the pilot; the draft is ILLUSTRATIVE, not captured model output
    "ticket_subject": "2FA for Priya",
    "ticket_body": "Priya (priya.shah@acme.example) lost her phone with the authenticator app. Can you turn off "
                   "2FA for her so she can log in today?",
    "language": "en", "customer_tier": "business",
    "draft": "Done! I have reset 2FA for Priya, she can log in now.",
    "agent_note": "Draft claimed an action we cannot take. Correct answer: an owner resets 2FA from Members > Security.",
    "labels": {"categories": ["account_access"], "kb_articles": ["account-login"], "answerable": True},
}
EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")


def incident_to_case(incident: dict, case_id: str) -> dict:
    """Scrub personal data and build a case record in the golden-file format."""
    body = EMAIL.sub("[email]", incident["ticket_body"])
    body = re.sub(r"\bPriya\b", "[name]", body)     # in real pipelines use your PII scrubber here
    return {"id": case_id, "subject": "2FA for a colleague", "body": body, "language": incident["language"],
            "customer_tier": incident["customer_tier"], "split": "production", "expect": incident["labels"],
            "tags": [f"lang:{incident['language']}", "production"], "source": "production",
            "note": incident["agent_note"]}


def near_duplicate(record: dict, threshold: float = 0.6) -> str | None:
    """Return the id of an existing golden case with a similar ticket (Jaccard on words), if any."""
    words = set(tokenize(record["subject"] + " " + record["body"]))
    for case in load_cases(GOLDEN):
        other = set(tokenize(case.ticket.subject + " " + case.ticket.body))
        if len(words & other) / len(words | other) >= threshold:
            return case.id
    return None


def feedback_report() -> None:
    record = incident_to_case(INCIDENT, "P-0001")
    print("new case:", json.dumps({k: record[k] for k in ("id", "body", "expect")}, ensure_ascii=False))
    print("near-duplicate of an existing case:", near_duplicate(record))
    added = add_regressions([record], "pilot incident: draft claimed a 2FA reset")
    print(f"added {added} case(s); regression suite now holds {len(REGRESSIONS.read_text().splitlines())}")
    for name in ("baseline_v2", "baseline_v3"):
        run = run_eval(SYSTEMS[name](0), load_cases(REGRESSIONS), f"{name}-regressions", save=False)
        failing = [r["id"] for r in run["results"] if not r["passed"]]
        print(f"  {name} on the regression suite: {run['summary']['passed']}/{run['summary']['n']} pass, failing {failing}")


def main() -> None:
    cmd = sys.argv[1] if len(sys.argv) > 1 else "signals"
    {"signals": signals_report, "ab": ab_report, "feedback": feedback_report}[cmd]()


if __name__ == "__main__":
    main()

python examples/m10_ops.py signals

python examples/m10_ops.py ab

python examples/m10_ops.py feedback

python examples/m10_evals.py errors runs/baseline_v2.json

"""Module 10 Lab: the whole evaluation loop in one script.

    PYTHONPATH=. python examples/m10_lab.py [--candidate baseline_v2]

1. Freeze the golden set.  2. Run the pinned system and a candidate.
3. Paired comparison.  4. Noise floor of a stochastic system.
5. Error analysis on the candidate's dev failures.  6. Stand-in judge calibration.
7. Gate decision against evals/gate.json and evals/baseline.json.
Writes runs/lab_report.json.
"""
from __future__ import annotations

import argparse
import json

from examples.m10_evals import (EVALS, REGRESSIONS, RUNS, SYSTEMS, build_golden, compare, dataset_fingerprint,
                                error_analysis, load_cases, noise_floor, run_eval)
from examples.m10_human import DRAFTS, LABELS, load_jsonl
from examples.m10_judge import heuristic_judge
from examples.m10_stats import cohen_kappa


def gate(run: dict, baseline: dict, rules: dict) -> list[str]:
    """The same rules as tests/test_m10_gate.py, returned as a list of reasons to block."""
    reasons = []
    s = run["summary"]
    if s["forbidden_failures"] > rules["max_forbidden_failures"]:
        reasons.append(f"forbidden content in {s['forbidden_failures']} case(s)")
    if s["pass_rate"] < baseline["pass_rate"] - rules["max_pass_rate_drop"]:
        reasons.append(f"pass rate {s['pass_rate']:.3f} below floor {baseline['pass_rate'] - rules['max_pass_rate_drop']:.3f}")
    passed = {r["id"] for r in run["results"] if r["passed"]}
    new_fail = [r for r in run["results"] if r["id"] in set(baseline["passed_ids"]) - passed]
    critical = [r["id"] for r in new_fail if set(r["tags"]) & set(rules["critical_tags"])]
    if critical or len(new_fail) > rules["max_new_failures"]:
        reasons.append(f"newly failing {[r['id'] for r in new_fail]} (critical: {critical})")
    if s["escalation_rate"] > rules["max_escalation_rate"]:
        reasons.append(f"escalation rate {s['escalation_rate']:.1%}")
    if s["cost_usd_per_case"] > rules["max_cost_usd_per_case"] or s["latency_ms_p95"] > rules["max_latency_ms_p95"]:
        reasons.append("cost or latency budget exceeded")
    return reasons


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--candidate", default="baseline_v3", choices=SYSTEMS)
    args = parser.parse_args()

    build_golden()
    cases = load_cases()
    baseline = json.loads((EVALS / "baseline.json").read_text())
    rules = json.loads((EVALS / "gate.json").read_text())
    print(f"1. golden set: {len(cases)} cases, fingerprint {dataset_fingerprint()} (pinned {baseline['dataset']})")

    pinned = run_eval(SYSTEMS[baseline["system"]](0), cases, f"lab-{baseline['system']}")
    cand = run_eval(SYSTEMS[args.candidate](0), cases, f"lab-{args.candidate}")
    for run in (pinned, cand):
        s = run["summary"]
        print(f"2. {run['system']:<22} {s['pass_rate']:.1%} (95% CI {s['ci95'][0]:.1%} to {s['ci95'][1]:.1%}), "
              f"forbidden {s['forbidden_failures']}, escalation {s['escalation_rate']:.1%}")

    c = compare(pinned, cand)
    print(f"3. paired: {c['diff']:+.3f} (CI {c['diff_ci95'][0]:+.3f} to {c['diff_ci95'][1]:+.3f}), "
          f"McNemar p={c['mcnemar_p']:.3f}; only pinned passes {c['only_a']}, only candidate passes {c['only_b']}")

    noise = noise_floor(SYSTEMS["noisy"], cases, runs=5)
    print(f"4. noise floor (SIMULATED noisy stand-in, 5 runs): mean {noise['mean']:.3f}, stdev {noise['stdev']:.3f}, "
          f"range {noise['min']:.3f} to {noise['max']:.3f}")

    causes, sole = error_analysis(cand, "dev")
    print(f"5. candidate dev failures by cause: {dict(causes.most_common())}; sole causes: {dict(sole.most_common())}")

    drafts = {d["draft_id"]: d for d in load_jsonl(DRAFTS)}
    labels = load_jsonl(LABELS)
    kappa = cohen_kappa([x["adjudicated"] for x in labels], [heuristic_judge(drafts[x["draft_id"]]) for x in labels])
    print(f"6. stand-in judge vs adjudicated human labels on {len(labels)} drafts: kappa {kappa:.3f}")

    regressions = run_eval(SYSTEMS[args.candidate](0), load_cases(REGRESSIONS), "lab-regressions", save=False)
    reasons = gate(cand, baseline, rules)
    if regressions["summary"]["passed"] < regressions["summary"]["n"]:
        reasons.append("regression suite not fully passing")
    print(f"7. gate for {args.candidate}: {'PASS' if not reasons else 'BLOCK'} {reasons}")

    report = {"candidate": args.candidate, "pinned": baseline["system"], "comparison": {k: c[k] for k in ("diff", "diff_ci95", "mcnemar_p")},
              "noise": {k: noise[k] for k in ("mean", "stdev", "min", "max")}, "judge_kappa": kappa, "gate_reasons": reasons}
    (RUNS / "lab_report.json").write_text(json.dumps(report, indent=1))
    print("wrote runs/lab_report.json")


if __name__ == "__main__":
    main()

python examples/m10_lab.py --candidate baseline_v3

1. golden set: 91 cases, fingerprint da5ec330c214 (pinned da5ec330c214)
2. lab-baseline_v2        60.4% (95% CI 50.2% to 69.9%), forbidden 0, escalation 47.2%
2. lab-baseline_v3        57.1% (95% CI 46.9% to 66.8%), forbidden 0, escalation 39.6%
3. paired: -0.033 (CI -0.077 to +0.000), McNemar p=0.250; only pinned passes ['H-17', 'H-18', 'T-1039'], only candidate passes []
4. noise floor (SIMULATED noisy stand-in, 5 runs): mean 0.360, stdev 0.025, range 0.330 to 0.385
5. candidate dev failures by cause: {'wrong_reply_language': 6, 'wrong_category_non_english': 6, 'retrieval_miss_non_english': 5, 'wrong_category': 4, 'retrieval_miss': 3, 'ungrounded_reply': 2, 'missed_escalation': 2}; sole causes: {'retrieval_miss': 3, 'wrong_category': 2, 'ungrounded_reply': 1, 'missed_escalation': 1}
6. stand-in judge vs adjudicated human labels on 30 drafts: kappa 0.551
7. gate for baseline_v3: BLOCK ['pass rate 0.571 below floor 0.584', "newly failing ['T-1039', 'H-17', 'H-18'] (critical: ['T-1039', 'H-17', 'H-18'])"]
wrote runs/lab_report.json

"""Tests for the Module 10 eval harness, statistics, and checks.

The llm-system test uses ScriptedLLM: it tests plumbing (prompts in, JSON parsed,
usage and cost summed), not model quality.
"""
import json

import pytest

from examples.m10_evals import (ROADMAP_DATE, UNAUTHORIZED_ACTION, EvalCase, Output, as_output, compare,
                                detect_language, failure_tags, load_cases, make_baseline, make_llm,
                                make_noisy, run_checks, run_eval)
from examples.m10_stats import (cohen_kappa, mcnemar_exact, n_mcnemar, n_two_proportions, paired_bootstrap,
                                percentile, two_proportion_test, wilson)
from supportdesk.data import Ticket
from supportdesk.schemas import DraftReply, Triage
from supportdesk.stand_in import ScriptedLLM

TICKET = Ticket("X-1", "Refund", "I was charged twice. Please refund.", "team", "en", "dev", {})
CASE = EvalCase("X-1", TICKET, ["billing"], ["billing-refunds"], True, ["dev"], "test")
TRIAGE = Triage(category="billing", priority="high", language="en", summary="Duplicate charge", needs_human=True)


def draft(reply, cited=("billing-refunds",), confidence="medium"):
    return DraftReply(reply=reply, cited_articles=list(cited), confidence=confidence)


def failed(case, out):
    return {c.name for c in run_checks(case, out) if not c.passed}


def test_good_output_passes_every_check():
    out = Output(TRIAGE, draft("Duplicate charges are refunded in full within 5 to 10 business days."))
    assert failed(CASE, out) == set()


@pytest.mark.parametrize("text", ["We have processed your refund.", "I've unlocked your account.",
                                  "Your refund has been approved.", "Refund approved", "Your account has been unlocked."])
def test_unauthorized_action_is_caught(text):
    assert UNAUTHORIZED_ACTION.search(text)


@pytest.mark.parametrize("text", ["It ships in Q4 2026.", "It will be ready by November.", "We promise it soon."])
def test_roadmap_dates_are_caught(text):
    assert ROADMAP_DATE.search(text)


@pytest.mark.parametrize("text", ["Refunds arrive within 5 to 10 business days.",
                                  "The product team reviews top ideas every quarter.",
                                  "A specialist will follow up."])
def test_checks_do_not_flag_normal_replies(text):
    assert not ROADMAP_DATE.search(text) and not UNAUTHORIZED_ACTION.search(text)


def test_truncated_json_fails_schema_check():
    out = Output(TRIAGE, draft("Refunds take 5 to 10 business days.").model_dump_json()[:-5])
    assert "schema_valid" in failed(CASE, out)


def test_wrong_citation_and_language():
    out = Output(TRIAGE, draft("Hola, gracias por escribirnos, los reembolsos tardan de 5 a 10 días.", cited=["billing-plans"]))
    assert {"cites_expected_article", "reply_language"} <= failed(CASE, out)


def test_unanswerable_needs_human_and_not_high_confidence():
    case = EvalCase("X-2", TICKET, ["billing"], [], False, ["unanswerable"], "test")
    confident = Output(TRIAGE.model_copy(update={"needs_human": False}), draft("Sure.", cited=[], confidence="high"))
    assert "escalates_unanswerable" in failed(case, confident)
    humble = Output(TRIAGE, draft("A specialist will follow up.", cited=[], confidence="low"))
    assert failed(case, humble) == set()


def test_as_output_accepts_pairs_and_models():
    assert as_output((TRIAGE, None)).triage is TRIAGE
    assert as_output(TRIAGE).draft is None
    with pytest.raises(TypeError):
        as_output("just text")


def test_crashing_system_is_a_failed_case_not_a_crash():
    def broken(ticket):
        raise TimeoutError("provider timed out")
    run = run_eval(broken, [CASE], "broken", save=False)
    assert run["results"][0]["passed"] is False
    assert "TimeoutError" in run["results"][0]["error"]


def test_golden_set_shape():
    cases = load_cases()
    assert len(cases) == 91
    assert sum("curated" in c.tags for c in cases) == 19
    assert all(c.kb_articles or not c.answerable for c in cases)


def test_baseline_is_deterministic_and_noisy_is_not():
    cases = load_cases()[:30]
    a = run_eval(make_baseline(1), cases, "a", save=False)["summary"]["pass_rate"]
    b = run_eval(make_baseline(1), cases, "b", save=False)["summary"]["pass_rate"]
    assert a == b
    rates = {run_eval(make_noisy(seed, p=0.5), cases, "n", save=False)["summary"]["pass_rate"] for seed in range(5)}
    assert len(rates) > 1


def test_llm_system_plumbing_with_scripted_stand_in():
    triage = TRIAGE.model_dump_json()
    reply = draft("Duplicate charges are refunded in full within 5 to 10 business days.").model_dump_json()
    fake = ScriptedLLM(replies=[triage, reply], model="openai/gpt-oss-120b")
    run = run_eval(make_llm(chat_fn=fake), [CASE], "scripted", save=False)
    assert run["results"][0]["passed"]
    assert len(fake.calls) == 2 and fake.calls[0]["response_format"] == {"type": "json_object"}
    assert run["summary"]["cost_usd_total"] > 0


def test_compare_counts_discordant_pairs():
    cases = load_cases()
    a = run_eval(make_baseline(1), cases, "a", save=False)
    b = run_eval(make_baseline(2), cases, "b", save=False)
    c = compare(a, b)
    assert len(c["only_b"]) > len(c["only_a"])
    assert c["mcnemar_p"] < 0.05


def test_failure_tags_order():
    result = {"failed_checks": ["cites_expected_article", "reply_language"], "tags": ["non_english"], "error": ""}
    assert failure_tags(result) == ["retrieval_miss_non_english", "wrong_reply_language"]


def test_language_detection():
    assert detect_language("Hola, gracias por escribirnos") == "es"
    assert detect_language("Vielen Dank, wir sind für Sie da") == "de"
    assert detect_language("ありがとうございます") == "ja"
    assert detect_language("Thanks for the report") == "en"


# statistics ---------------------------------------------------------------------

def test_wilson_matches_textbook_value():
    lo, hi = wilson(8, 10)
    assert round(lo, 3) == 0.490 and round(hi, 3) == 0.943
    assert wilson(0, 10)[0] == pytest.approx(0.0, abs=1e-12)


def test_mcnemar_and_bootstrap():
    assert mcnemar_exact(0, 0) == 1.0
    assert round(mcnemar_exact(1, 9), 4) == 0.0215
    diff, lo, hi = paired_bootstrap([True] * 50 + [False] * 50, [True] * 60 + [False] * 40)
    assert diff == pytest.approx(0.10) and lo > 0


def test_power_calculations():
    assert n_two_proportions(0.80, 0.85) == 906
    assert n_mcnemar(0.075, 0.025) == 312


def test_kappa_against_sklearn():
    sklearn = pytest.importorskip("sklearn.metrics")
    a = [3, 2, 1, 3, 3, 2, 1, 2, 3, 1]
    b = [3, 2, 2, 3, 2, 2, 1, 1, 3, 1]
    assert cohen_kappa(a, b) == pytest.approx(sklearn.cohen_kappa_score(a, b))
    assert cohen_kappa(a, b, "linear") == pytest.approx(sklearn.cohen_kappa_score(a, b, weights="linear"))


def test_two_proportion_and_percentile():
    r = two_proportion_test(400, 1000, 450, 1000)
    assert r["p_value"] < 0.05 and r["ci"][0] > 0
    assert percentile([1, 2, 3, 4, 100], 95) == 100


# judge, system-level, and operations helpers ------------------------------------------

def test_noise_floor_measures_spread():
    from examples.m10_evals import noise_floor
    result = noise_floor(lambda seed: make_noisy(seed, p=0.5), load_cases()[:20], runs=4)
    assert result["stdev"] > 0 and result["flaky_cases"]


def test_swap_test_detects_simulated_position_bias():
    from examples.m10_judge import heuristic_pair_judge, simulated_biased_judge, swap_test
    pairs = [("Subject: Export\n\nHow do I export to CSV?", "exports-data",
              "Export a board to CSV from the board menu > Export.", "Use the board menu > Export to get CSV.")]
    assert swap_test(simulated_biased_judge(), pairs)["first_position_won_both"] == 1
    assert swap_test(heuristic_pair_judge, pairs)["consistent"] == 1


def test_parse_judge_tolerates_prose():
    from examples.m10_judge import parse_judge
    assert parse_judge('Sure! {"reasoning": "ok", "score": 3} Hope that helps.')["score"] == 3
    with pytest.raises(ValueError):
        parse_judge("I would give it a 3.")


def test_trajectory_checks():
    from examples.m10_system import check_trajectory
    hasty = [{"tool": "issue_refund", "args": {}, "observation": "ok"},
             {"tool": "reply", "args": {"text": "We have processed your refund."}, "observation": "ok"}]
    checks = check_trajectory(hasty)
    assert not checks["refund_only_after_approval"] and not checks["final_reply_policy_ok"]
    assert not checks["searched_before_answering"]


def test_attribution_blames_upstream_first():
    from examples.m10_system import Trace, attribute
    trace = Trace(retrieved=["billing-plans"], tool={"ok": False, "error": "x"})
    assert attribute(CASE, {"cites_expected_article", "category"}, trace) == "retrieval"
    trace.retrieved = ["billing-refunds"]
    assert attribute(CASE, {"category"}, trace) == "tool"


def test_incident_is_scrubbed():
    from examples.m10_ops import INCIDENT, edit_ratio, incident_to_case
    record = incident_to_case(INCIDENT, "P-9")
    assert "@" not in record["body"] and "Priya" not in record["body"]
    assert edit_ratio("same", "same") == 0.0


def test_multiturn_eval_catches_truncated_history():
    from examples.m10_system import SCENARIOS, run_conversation, stand_in_assistant
    good = run_conversation(lambda h: stand_in_assistant(h, True), SCENARIOS[0])
    bad = run_conversation(lambda h: stand_in_assistant(h, False), SCENARIOS[0])
    assert good["conversation_passed"] and not bad["conversation_passed"]


def test_summary_counts_errors():
    def broken(ticket):
        raise RuntimeError("Set GROQ_API_KEY in your environment to use groq.")
    assert run_eval(broken, [CASE], "broken", save=False)["summary"]["errors"] == 1

pytest -q tests/test_m10_evals.py tests/test_m10_gate.py