Part F: Evaluation Track

# examples/cap_eval_starter.py
"""Evaluation track starter: assertion tier, judge tier with calibration, noise floor, CI gate.

System under test: an extractive drafter (retrieve the top article with BM25,
copy the two sentences that share the most words with the ticket). It is not
an LLM, so every number below is a real measurement of a real system; point
the same harness at your LLM drafter to evaluate it.
"""
from __future__ import annotations

import json
import random
import re
import statistics
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Protocol

import torch
from pydantic import ValidationError

from supportdesk.data import CATEGORIES, DATA_DIR, Ticket, get_article, load_articles, load_tickets
from supportdesk.kb_search import KBSearch, tokenize
from supportdesk.llm import ChatResult, chat
from supportdesk.schemas import DraftReply
from supportdesk.stand_in import ScriptedLLM
from supportdesk.tinylm import load

sys.path.insert(0, str(Path(__file__).resolve().parent))
from cap_adapt_starter import label_scores  # noqa: E402
from cap_baseline import wilson  # noqa: E402

KB_IDS = {a.id for a in load_articles()}
KB_TEXT = " ".join(a.body for a in load_articles())
RUBRIC = ("good = the reply answers the customer's actual question correctly from the help center, "
          "in the customer's language, and escalates what the help center cannot answer; bad = anything else.")


# System under test ---------------------------------------------------------------
GREETING = "Hi, thanks for contacting Brightlane."


def extractive_draft(ticket: Ticket, kb: KBSearch, query_mode: str = "full", n_sentences: int = 2,
                     greeting: str = GREETING) -> DraftReply:
    query = ticket.text if query_mode == "full" else ticket.subject
    hits = kb.search(query, k=1)
    if not hits:
        return DraftReply(reply="Thanks for writing in. An agent will follow up.", cited_articles=[], confidence="low")
    article = get_article(hits[0].article_id)
    words = set(tokenize(ticket.text))
    sentences = re.split(r"(?<=[.!?])\s+", article.body)
    best = sorted(range(len(sentences)), key=lambda i: (-len(words & set(tokenize(sentences[i]))), i))[:n_sentences]
    body = " ".join(sentences[i] for i in sorted(best))
    return DraftReply(reply=f"{greeting} {body}", cited_articles=[article.id],
                      confidence="medium" if hits[0].score > 5 else "low")


# Tier 1: deterministic assertions -----------------------------------------------------
FORBIDDEN = re.compile(r"\b(guarantee|we have refunded|unlocked your account|legal advice)\b", re.IGNORECASE)


def assertions(ticket: Ticket, draft: DraftReply) -> dict[str, bool]:
    reply = draft.reply
    amounts = re.findall(r"\d[\d,]*(?:\.\d+)? USD", reply)
    try:
        DraftReply.model_validate(draft.model_dump())
        schema_ok = True
    except ValidationError:
        schema_ok = False
    gold = ticket.gold["kb_article"]
    return {
        "schema_valid": schema_ok,
        "citations_exist": all(c in KB_IDS for c in draft.cited_articles),
        "no_forbidden_promises": not FORBIDDEN.search(reply),
        "prices_match_kb": all(a in KB_TEXT for a in amounts),
        "length_ok": len(reply) <= 600,
        "language_matches": ticket.language == "en",  # this drafter always writes English
        "cites_gold_or_escalates": (gold in draft.cited_articles) if gold else draft.confidence == "low",
    }


HARD = {"schema_valid", "citations_exist", "no_forbidden_promises", "prices_match_kb"}  # must never fail


# Tier 2: judge interface and calibration ------------------------------------------------
@dataclass
class Verdict:
    label: str  # "good" or "bad"
    reason: str


class Judge(Protocol):
    def __call__(self, ticket: Ticket, draft: DraftReply) -> Verdict: ...


class LLMJudge:
    """Pointwise rubric judge through any chat-shaped callable (llm.chat, ScriptedLLM, a Meter)."""

    def __init__(self, llm: Callable[..., ChatResult]) -> None:
        self.llm = llm

    def __call__(self, ticket: Ticket, draft: DraftReply) -> Verdict:
        messages = [
            {"role": "system", "content": f"You grade support replies. Rubric: {RUBRIC} "
                                          'Answer only JSON: {"label": "good" or "bad", "reason": "..."}'},
            {"role": "user", "content": f"<ticket>\n{ticket.text}\n</ticket>\n<reply>\n{draft.reply}\n</reply>"},
        ]
        raw = self.llm(messages, temperature=0.0, response_format={"type": "json_object"}).text
        try:
            data = json.loads(raw)
            label = data["label"] if data.get("label") in ("good", "bad") else "bad"
            return Verdict(label, str(data.get("reason", "")))
        except (json.JSONDecodeError, TypeError, KeyError):
            return Verdict("bad", f"unparseable judge output: {raw[:60]}")


class OverlapJudge:
    """A cheap heuristic judge: good if English and the reply shares 3+ content words with the ticket."""

    def __call__(self, ticket: Ticket, draft: DraftReply) -> Verdict:
        shared = set(tokenize(ticket.text)) & set(tokenize(draft.reply)) - {"brightlane", "thanks", "hi"}
        ok = ticket.language == "en" and len(shared) >= 3
        return Verdict("good" if ok else "bad", f"{len(shared)} shared words")


MIN_KAPPA, MAX_FALSE_GOOD = 0.6, 1  # trust a judge only above this agreement and below this leniency


def calibrate(judge: Judge, tickets: list[Ticket], drafts: list[DraftReply], human: list[str]) -> dict:
    """Compare a judge with human labels on the same items and decide whether to trust it."""
    labels = [judge(t, d).label for t, d in zip(tickets, drafts)]
    false_good = sum(j == "good" and h == "bad" for j, h in zip(labels, human))
    false_bad = sum(j == "bad" and h == "good" for j, h in zip(labels, human))
    kappa = cohen_kappa(human, labels)
    return {"agree": sum(j == h for j, h in zip(labels, human)), "n": len(human), "kappa": round(kappa, 3),
            "false_good": false_good, "false_bad": false_bad, "says_good": labels.count("good"),
            "trusted": kappa >= MIN_KAPPA and false_good <= MAX_FALSE_GOOD}


def cohen_kappa(a: list[str], b: list[str]) -> float:
    """Agreement beyond chance: (observed - expected) / (1 - expected)."""
    n = len(a)
    labels = sorted(set(a) | set(b))
    observed = sum(x == y for x, y in zip(a, b)) / n
    expected = sum((a.count(lab) / n) * (b.count(lab) / n) for lab in labels)
    return 1.0 if expected == 1 else (observed - expected) / (1 - expected)


# Noise floor ---------------------------------------------------------------------------
def label_distributions(tickets: list[Ticket]) -> list[torch.Tensor]:
    """Prompted TinyLM's label scores for each ticket, turned into probabilities with a softmax.

    Sampling from these imitates running the classifier with temperature 1: the
    same system, the same tickets, a different answer each run.
    """
    model, tok = load()
    return [torch.softmax(torch.tensor([label_scores(model, tok, t.text)[c] for c in CATEGORIES]), dim=0)
            for t in tickets]


def noise_floor(tickets: list[Ticket], runs: int = 10) -> list[int]:
    """Accuracy of the SAME system over repeated runs with sampling at temperature 1."""
    dists = label_distributions(tickets)
    scores = []
    for seed in range(runs):
        g = torch.Generator().manual_seed(seed)
        picks = [CATEGORIES[int(torch.multinomial(d, 1, generator=g))] for d in dists]
        scores.append(sum(p == t.gold["category"] for p, t in zip(picks, tickets)))
    return scores


def bootstrap_ci(passes: list[bool], reps: int = 2000, seed: int = 0) -> tuple[float, float]:
    rng = random.Random(seed)
    n = len(passes)
    means = sorted(sum(rng.choice(passes) for _ in range(n)) / n for _ in range(reps))
    return round(means[int(0.025 * reps)], 3), round(means[int(0.975 * reps)], 3)


# CI gate -------------------------------------------------------------------------------
def gate(current: dict[str, list[bool]], reference: dict[str, list[bool]], max_drop: int) -> list[str]:
    """Fail on any HARD assertion failure, or when a soft check loses more than `max_drop` items."""
    problems = []
    for name, results in current.items():
        if name in HARD and not all(results):
            problems.append(f"{name}: {results.count(False)} hard failures")
        lost = sum(r and not c for r, c in zip(reference[name], results))
        if name not in HARD and lost > max_drop:
            problems.append(f"{name}: {lost} items went pass -> fail (allowed {max_drop})")
    return problems


def run_assertions(tickets, kb, **variant) -> dict[str, list[bool]]:
    table: dict[str, list[bool]] = {}
    for t in tickets:
        for name, ok in assertions(t, extractive_draft(t, kb, **variant)).items():
            table.setdefault(name, []).append(ok)
    return table


def main() -> None:
    torch.set_num_threads(1)
    test, kb = load_tickets("test"), KBSearch()
    n = len(test)
    print(f"== Tier 1: assertions on {n} test tickets (extractive drafter v1)")
    v1 = run_assertions(test, kb)
    for name, results in v1.items():
        k = sum(results)
        print(f"   {name:24} {k:2}/{n} {'HARD' if name in HARD else 'soft'}  95% CI {wilson(k, n)}")

    print(f"\n== Tier 2: judges vs human labels (n={n}, one human rater); trust needs kappa >= {MIN_KAPPA} "
          f"and at most {MAX_FALSE_GOOD} false 'good'")
    human = json.loads((DATA_DIR / "cap_human_labels.json").read_text())["labels"]
    gold = [human[t.id] for t in test]
    drafts = [extractive_draft(t, kb) for t in test]
    lenient = ScriptedLLM(responder=lambda m, k: '{"label": "good", "reason": "Polite and relevant."}')
    judges: dict[str, Judge] = {"LLMJudge(ScriptedLLM always-good)": LLMJudge(lenient), "OverlapJudge": OverlapJudge()}
    if "--real" in sys.argv:
        judges["LLMJudge(llm.chat)"] = LLMJudge(chat)
    for name, judge in judges.items():
        c = calibrate(judge, test, drafts, gold)
        print(f"   {name:34} agree {c['agree']}/{c['n']}  kappa {c['kappa']:+.3f}  false good {c['false_good']}  "
              f"false bad {c['false_bad']}  says good {c['says_good']} (humans {gold.count('good')})  "
              f"{'TRUSTED' if c['trusted'] else 'not trusted'}")

    print("\n== Noise floor: prompted TinyLM category, sampled at temperature 1, 10 runs")
    scores = noise_floor(test)
    print(f"   correct per run: {scores}  mean {statistics.mean(scores):.1f}  sd {statistics.stdev(scores):.2f}  "
          f"range {min(scores)} to {max(scores)}")
    passes = v1["cites_gold_or_escalates"]
    print(f"   dataset noise: cites_gold_or_escalates {sum(passes)}/{n}, bootstrap 95% CI {bootstrap_ci(passes)}")

    print("\n== CI gate: three candidate changes, each compared with v1 (soft checks may lose at most 1 item)")
    candidates = {
        "v2a retrieval searches the subject only": {"query_mode": "subject"},
        "v2b reply keeps 1 sentence instead of 2": {"n_sentences": 1},
        "v2c friendlier greeting from marketing": {"greeting": "Hi! We guarantee a fast fix, thanks for writing."},
    }
    print(f"   v1 vs v1: {'PASS' if not gate(v1, v1, max_drop=1) else 'FAIL'}")
    for label, variant in candidates.items():
        problems = gate(run_assertions(test, kb, **variant), v1, max_drop=1)
        print(f"   {label}: {'PASS' if not problems else 'FAIL'} {problems}")
    judged = [OverlapJudge()(t, extractive_draft(t, kb, n_sentences=1)).label for t in test]
    print(f"   v2b seen by OverlapJudge: good {judged.count('good')}/{n} "
          f"(v1: {[OverlapJudge()(t, d).label for t, d in zip(test, drafts)].count('good')}/{n}); "
          f"assertions cannot see a shorter, less complete answer")


if __name__ == "__main__":
    main()
Code explained

In simple words: a quality lab with three instruments: a checklist that never gets tired, a taste tester who must first prove they agree with the head chef, and a scale that tells you how much the readings wobble on their own.

What happens:

# examples/cap_lab.py
"""Capstone Module Lab: the four shared requirements in one run.

  1. baseline       keyword triage + BM25 retrieval on the test split, Wilson intervals
  2. error analysis all real failures + a SYNTHETIC set, tagged, clustered, report written
  3. measurement    Meter around ScriptedLLM (priced; plumbing only) and TinyLM (real local latency)
  4. write-up       what did not work, pre-filled with the measured rule-fix experiments
Writes runs/error_report.md, runs/failures.jsonl, runs/capstone_evidence.md.
Replace each stand-in with your own system; the tooling stays the same.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import torch

from supportdesk.data import load_tickets
from supportdesk.stand_in import ScriptedLLM

sys.path.insert(0, str(Path(__file__).resolve().parent))
from cap_baseline import keyword_triage, run_baseline, score  # noqa: E402
from cap_error_analysis import analyze, report  # noqa: E402
from cap_measure import Meter, TinyChat, check_budgets  # noqa: E402
from cap_rule_fixes import FIXES, compare  # noqa: E402
from cap_synthetic_tickets import build  # noqa: E402

RUNS = Path("runs")
BUDGET_P95_MS, BUDGET_COST = 250, 0.0005


def main() -> None:
    torch.set_num_threads(1)
    RUNS.mkdir(exist_ok=True)
    lines = ["# Capstone evidence (generated by examples/cap_lab.py)", ""]

    # 1. Baseline to beat
    real_records = run_baseline(load_tickets())
    lines += ["## 1. Baseline to beat (test split, n=24)", "", "| Task | Score | 95% CI |", "|---|---|---|"]
    for task in ("category", "priority", "retrieval_top1"):
        s = score(real_records, task, "test")
        lines.append(f"| {task} | {s['k']}/{s['n']} = {s['acc']:.3f} | {s['ci95']} |")
        print(f"[1] baseline {task:15} {s['k']}/{s['n']} {s['ci95']}")

    # 2. Error analysis
    overrides_path = RUNS / "tag_overrides.json"
    overrides = json.loads(overrides_path.read_text()) if overrides_path.exists() else {}
    result = analyze(real_records + run_baseline(build()), overrides)
    (RUNS / "error_report.md").write_text(report(result) + "\n", encoding="utf-8")
    with (RUNS / "failures.jsonl").open("w", encoding="utf-8") as f:
        for row in result["failures"]:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    real = [f for f in result["failures"] if f["features"]["source"] == "real"]
    real_tags = {}
    for f in real:
        for tag in f["tags"]:
            real_tags[tag] = real_tags.get(tag, 0) + 1
    top = sorted(real_tags.items(), key=lambda kv: -kv[1])[:5]
    print(f"[2] failures {len(result['failures'])} ({len(real)} real, {len(result['failures']) - len(real)} SYNTHETIC); "
          f"top real causes {top}")
    lines += ["", "## 2. Error analysis", "",
              f"{len(real)} real failures (plus {len(result['failures']) - len(real)} on SYNTHETIC tickets). "
              "Top real causes (full report: runs/error_report.md):", ""]
    lines += [f"- {tag}: {n}" for tag, n in top]

    # 3. Cost and latency, measured
    test = load_tickets("test")
    responder = lambda m, k: json.dumps(dict(zip(("category", "priority"), keyword_triage(m[-1]["content"]))))  # noqa: E731
    scripted = Meter(ScriptedLLM(responder=responder), price_as="openai/gpt-oss-120b")
    tiny = TinyChat(max_new_tokens=24)
    tiny([{"role": "user", "content": "warm-up"}])
    local = Meter(tiny)
    for t in test:
        scripted([{"role": "user", "content": t.text}], task_id=t.id)
        local([{"role": "user", "content": t.body}], task_id=t.id)
    s1, s2 = scripted.summary(), local.summary()
    broken = check_budgets(s2, BUDGET_P95_MS, None)
    print(f"[3] ScriptedLLM priced as gpt-oss-120b: {s1['cost_per_task_mean']:.8f} USD/task, "
          f"{s1['tokens_per_task_mean']} tokens/task | TinyLM p50 {s2['call_ms_p50']:.0f} ms, "
          f"p95 {s2['call_ms_p95']:.0f} ms, budget broken: {broken or 'no'}")
    lines += ["", "## 3. Cost and latency (measured)", "",
              "| System | Calls | p50 ms | p95 ms | Tokens/task | Cost/task USD |", "|---|---|---|---|---|---|",
              f"| ScriptedLLM (plumbing only, priced as gpt-oss-120b) | {s1['calls']} | {s1['call_ms_p50']} | "
              f"{s1['call_ms_p95']} | {s1['tokens_per_task_mean']} | {s1['cost_per_task_mean']:.8f} |",
              f"| TinyLM local, 24 new tokens | {s2['calls']} | {s2['call_ms_p50']} | {s2['call_ms_p95']} | "
              f"{s2['tokens_per_task_mean']} | local compute, not priced |",
              "", f"Budgets: task p95 <= {BUDGET_P95_MS} ms, cost <= {BUDGET_COST} USD per task. "
              f"TinyLM broken budgets: {broken or 'none'}."]

    # 4. What did not work
    lines += ["", "## 4. What did not work", ""]
    for name in FIXES:
        r = compare(name)
        d, t = r["dev"], r["test"]
        verdict = "within noise, not adopted as a win" if max(d["p"], t["p"]) > 0.05 else "significant"
        lines.append(f"- Tried: {name}. Measured: dev {d['before']} -> {d['after']}/{d['n']} "
                     f"(fixed {len(d['fixed'])}, broke {len(d['broke'])}); test {t['before']} -> {t['after']}/{t['n']} "
                     f"(fixed {len(t['fixed'])}, broke {len(t['broke'])}). Verdict: {verdict}. "
                     "Why it failed: WRITE THIS FROM THE FLIPPED TICKETS.")
        print(f"[4] {name}: dev {d['before']}->{d['after']}, test {t['before']}->{t['after']}, {verdict}")
    (RUNS / "capstone_evidence.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    print("wrote runs/capstone_evidence.md and runs/error_report.md")


if __name__ == "__main__":
    main()

# Capstone evidence (generated by examples/cap_lab.py)

## 1. Baseline to beat (test split, n=24)

| Task | Score | 95% CI |
|---|---|---|
| category | 16/24 = 0.667 | (0.467, 0.82) |
| priority | 10/24 = 0.417 | (0.245, 0.612) |
| retrieval_top1 | 17/20 = 0.850 | (0.64, 0.948) |

## 2. Error analysis

69 real failures (plus 170 on SYNTHETIC tickets). Top real causes (full report: runs/error_report.md):

- question_rule: 20
- no_signal: 15
- non_english: 14
- vocab_gap: 9
- urgency_unseen: 8

## 3. Cost and latency (measured)

| System | Calls | p50 ms | p95 ms | Tokens/task | Cost/task USD |
|---|---|---|---|---|---|
| ScriptedLLM (plumbing only, priced as gpt-oss-120b) | 24 | 0.13 | 0.18 | 42.4 | 0.00001208 |
| TinyLM local, 24 new tokens | 24 | 22.83 | 25.26 | 57.0 | local compute, not priced |

Budgets: task p95 <= 250 ms, cost <= 0.0005 USD per task. TinyLM broken budgets: none.

## 4. What did not work

- Tried: A word boundaries (category). Measured: dev 31 -> 31/48 (fixed 1, broke 1); test 16 -> 16/24 (fixed 1, broke 1). Verdict: within noise, not adopted as a win. Why it failed: WRITE THIS FROM THE FLIPPED TICKETS.
- Tried: B no question rule (priority). Measured: dev 28 -> 28/48 (fixed 12, broke 12); test 10 -> 11/24 (fixed 6, broke 5). Verdict: within noise, not adopted as a win. Why it failed: WRITE THIS FROM THE FLIPPED TICKETS.

PYTHONPATH=. python -m pytest -q tests/test_cap_tooling.py