Module 19. Evaluating Modern LLM Systems: From Benchmarks to Decisions

import hashlib
import json
import math
import re
import string
import unicodedata
from collections import defaultdict
from dataclasses import dataclass
from typing import Callable, Mapping, Sequence

REFERENCE_JSON = '{"golden":{"canary_guess_probability":2.938735877055719e-39,"canary_hex_characters":32,"cost":{"completion_tokens":150,"dense_flops_floor":6183100000000000.0,"effective_flops_per_second":495000000000000.0,"examples":700,"parameter_count":8030000000,"prompt_tokens":400,"seconds_floor":12.491111111111111,"total_tokens":385000},"expected_rate":0.8,"n_for_3pp":683,"n_for_5pp":246},"judge":{"averaged_a_rate":0.5,"calls_per_pair":6,"forward_a_wins":650,"judges":3,"orders":2,"pairs":1000,"position_bias":0.15,"swapped_a_wins":350},"leaderboard":{"aggregate_approximation":{"ci_high":0.03136546073938403,"ci_low":-0.021365460739384022,"delta":0.0050000000000000044,"half_width":0.026365460739384027,"se":0.013451765683359198,"z_score":0.3716984162298756},"first_rate":0.955,"paired_demo":{"both_correct":150,"both_wrong":20,"delta":0.05,"first_accuracy":0.85,"first_only":20,"mcnemar_p":0.09873714670538902,"second_accuracy":0.8,"second_only":10,"total":200},"required_tasks":{"0.005":14599,"0.01":3650,"0.02":913,"0.05":146},"sample_size":500,"second_rate":0.95},"lost_in_middle":{"dip":0.35,"edge":0.95,"mean":0.7749999999999999,"threshold":0.9,"working_share":0.2467517144288497,"working_tokens_of_1m":246752},"pass_at_k":{"correct_samples":4,"naive":{"1":0.19999999999999998,"10":0.8926258176,"5":0.67232},"total_samples":20,"unbiased":{"1":0.19999999999999998,"10":0.9566563467492261,"5":0.718266253869969}},"schema_version":1}'
REFERENCE_SHA256 = "d89b860e2938e3c662854b634cbf9bf5ded99aded5af80555a3a9f76d154f9ef"
FIXTURE_JSON = '{"examples":[{"expected":"paris","id":"geo_fr","prompt":"capital of France","slice":"facts"},{"expected":"4","id":"math_2","prompt":"2+2","slice":"math"},{"expected":"au","id":"chem_au","prompt":"chemical symbol of gold","slice":"facts"},{"expected":"1969","id":"hist_moon","prompt":"year of the Moon landing","slice":"facts"},{"expected":"42","id":"math_42","prompt":"what is 7*6","slice":"math"},{"expected":"tokyo","id":"geo_jp","prompt":"capital of Japan","slice":"facts"},{"expected":"9","id":"math_sqrt","prompt":"square root of 81","slice":"math"},{"expected":"cold","id":"lang_antonym","prompt":"antonym of hot","slice":"language"}],"model_a":["Paris.","4","AU","1969","42","Warsaw","9","cold"],"model_b":["paris","four","ag","1969","42","Tokyo","9","cold."]}'
FIXTURE_SHA256 = "1b6e970a2eb3a032424a6688ac9dabb5b99f7c865dab35de95814b741fdd0758"


def canonical_json(value) -> str:
    """Serialize JSON deterministically for embedded-contract verification."""

    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


REFERENCE = json.loads(REFERENCE_JSON)
FIXTURE = json.loads(FIXTURE_JSON)
assert hashlib.sha256(canonical_json(REFERENCE).encode("utf-8")).hexdigest() == REFERENCE_SHA256
assert hashlib.sha256(canonical_json(FIXTURE).encode("utf-8")).hexdigest() == FIXTURE_SHA256
assert REFERENCE["schema_version"] == 1
print('Embedded M19 contracts verified')

def _positive_int(name: str, value: int) -> int:
    """Validate and return a strictly positive, non-boolean integer."""
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ValueError(f"{name} must be a positive integer")
    return value

def _non_negative_int(name: str, value: int) -> int:
    """Validate and return a non-negative, non-boolean integer."""
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{name} must be a non-negative integer")
    return value

def _probability(name: str, value: float) -> float:
    """Validate and return a finite probability in the closed unit interval."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be a real number")
    number = float(value)
    if not math.isfinite(number) or not 0.0 <= number <= 1.0:
        raise ValueError(f"{name} must lie in [0, 1]")
    return number

def _positive_finite(name: str, value: float) -> float:
    """Validate and return a finite positive real number."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be a real number")
    number = float(value)
    if not math.isfinite(number) or number <= 0.0:
        raise ValueError(f"{name} must be finite and positive")
    return number

def wilson_interval(successes: int, total: int, z_value: float = 1.96) -> tuple[float, float]:
    """Return the Wilson score interval for a binomial proportion."""
    n = _positive_int("total", total)
    s = _non_negative_int("successes", successes)
    if s > n:
        raise ValueError("successes cannot exceed total")
    z_value = _positive_finite("z_value", z_value)
    p = s / n
    z2 = z_value**2
    denominator = 1.0 + z2 / n
    centre = (p + z2 / (2.0 * n)) / denominator
    radius = z_value * math.sqrt(p * (1.0 - p) / n + z2 / (4.0 * n**2)) / denominator
    return max(0.0, centre - radius), min(1.0, centre + radius)

def mcnemar_exact_pvalue(first_only: int, second_only: int) -> float:
    """Return the two-sided exact McNemar p-value for discordant paired outcomes."""
    first_only = _non_negative_int("first_only", first_only)
    second_only = _non_negative_int("second_only", second_only)
    discordant = first_only + second_only
    if discordant == 0:
        return 1.0
    tail = min(first_only, second_only)
    cdf = sum(math.comb(discordant, k) for k in range(tail + 1)) / (1 << discordant)
    return min(1.0, 2.0 * cdf)

L = REFERENCE["leaderboard"]
A = L["aggregate_approximation"]
assert abs(A["z_score"] - 0.3716984162298756) < 1e-12
assert A["ci_low"] < 0.0 < A["ci_high"]
assert L["required_tasks"]["0.005"] == 14599
P = L["paired_demo"]
assert abs(mcnemar_exact_pvalue(P["first_only"], P["second_only"]) - P["mcnemar_p"]) < 1e-15
lo, hi = wilson_interval(7, 8)
assert 0.0 <= lo < 7 / 8 < hi <= 1.0
print(f"planning probabilities: delta={A['delta']*100:.1f} pp, z={A['z_score']:.2f}, CI=[{A['ci_low']*100:.1f}, {A['ci_high']*100:.1f}] pp")
print(f"paired demo: {P['first_accuracy']:.0%} vs {P['second_accuracy']:.0%}, exact McNemar p={P['mcnemar_p']:.3f}")
print('Aggregate approximation, Wilson interval, and exact McNemar verified')

def pass_at_k_unbiased(total_samples: int, correct_samples: int, k: int) -> float:
    """Return the finite-sample unbiased pass@k estimator of Chen et al."""
    n = _positive_int("total_samples", total_samples)
    c = _non_negative_int("correct_samples", correct_samples)
    k = _positive_int("k", k)
    if c > n:
        raise ValueError("correct_samples cannot exceed total_samples")
    if k > n:
        raise ValueError("k cannot exceed total_samples")
    if n - c < k:
        return 1.0
    # Use the shorter symmetric product and avoid cancellation near zero.
    if c == 0:
        return 0.0
    count, removed = (k, c) if k <= c else (c, k)
    if removed / n == 1.0:  # the success probability rounds to one at float precision
        return 1.0
    return -math.expm1(sum(math.log1p(-removed / (n - i)) for i in range(count)))

def pass_at_k_naive(total_samples: int, correct_samples: int, k: int) -> float:
    """Return the plug-in estimate ``1 - (1 - c/n)**k`` for comparison."""
    n = _positive_int("total_samples", total_samples)
    c = _non_negative_int("correct_samples", correct_samples)
    k = _positive_int("k", k)
    if c > n:
        raise ValueError("correct_samples cannot exceed total_samples")
    if k > n:
        raise ValueError("k cannot exceed total_samples")
    return 1.0 if c / n == 1.0 else -math.expm1(k * math.log1p(-c / n))

R = REFERENCE["pass_at_k"]
n, c = R["total_samples"], R["correct_samples"]
for k in (1, 5, 10):
    unbiased = pass_at_k_unbiased(n, c, k)
    naive = pass_at_k_naive(n, c, k)
    assert abs(unbiased - R["unbiased"][str(k)]) < 1e-15
    assert abs(naive - R["naive"][str(k)]) < 1e-15
    print(f"pass@{k:>2}: unbiased={unbiased:.4f}, plug-in={naive:.4f}, delta={(naive-unbiased)*100:+.1f} pp")
assert pass_at_k_unbiased(20, 0, 10) == 0.0
assert pass_at_k_unbiased(20, 20, 1) == 1.0
assert all(pass_at_k_unbiased(n, c, k) <= pass_at_k_unbiased(n, c, k + 1) for k in range(1, n))
print('Unbiased pass@k and edge cases verified')

@dataclass(frozen=True)
class EvalExample:
    """One immutable golden-set example with a predeclared slice."""

    example_id: str
    prompt: str
    expected: str
    slice_name: str

    def __post_init__(self) -> None:
        """Validate that every identifying field is a non-empty string."""

        for name, value in (
            ("example_id", self.example_id),
            ("prompt", self.prompt),
            ("expected", self.expected),
            ("slice_name", self.slice_name),
        ):
            if not isinstance(value, str) or not value.strip():
                raise ValueError(f"{name} must be a non-empty string")


@dataclass(frozen=True)
class ModelResponse:
    """One response with optional measured accounting; unknown values remain None."""

    text: str
    latency_ms: float | None = None
    output_tokens: int | None = None

    def __post_init__(self) -> None:
        """Validate response text and optional accounting fields."""

        if not isinstance(self.text, str):
            raise TypeError("text must be a string")
        if self.latency_ms is not None and (isinstance(self.latency_ms, bool) or not isinstance(self.latency_ms, (int, float)) or not math.isfinite(self.latency_ms) or self.latency_ms < 0):
            raise ValueError("latency_ms must be finite and non-negative")
        if self.output_tokens is not None:
            _non_negative_int("output_tokens", self.output_tokens)


@dataclass(frozen=True)
class PredictionRecord:
    """Stored response and all metric values for one example."""

    example_id: str
    slice_name: str
    expected: str
    prediction: str
    scores: Mapping[str, float | None]
    latency_ms: float | None
    output_tokens: int | None
    error: str | None = None


@dataclass(frozen=True)
class EvaluationReport:
    """Deterministic evaluation report built from call-once prediction records."""

    records: tuple[PredictionRecord, ...]
    aggregate: Mapping[str, float | None]
    by_slice: Mapping[str, Mapping[str, float | None]]
    model_calls: int
    error_count: int


def normalize_answer(text: str) -> str:
    """Normalize case, whitespace, and terminal punctuation without deleting signs."""

    if not isinstance(text, str):
        raise TypeError("answer must be a string")
    text = unicodedata.normalize("NFKC", text).casefold()
    text = text.strip().rstrip(".!?。！？")  # task-specific terminal sentence punctuation only
    return " ".join(text.split())


def exact_match(prediction: str, expected: str) -> float:
    """Return strict case-insensitive exact match after outer whitespace removal."""

    if not isinstance(prediction, str) or not isinstance(expected, str):
        raise TypeError("prediction and expected must be strings")
    return float(prediction.strip().casefold() == expected.strip().casefold())


def normalized_exact_match(prediction: str, expected: str) -> float:
    """Return exact match after the explicit normalization above."""

    return float(normalize_answer(prediction) == normalize_answer(expected))


def _validate_dataset(dataset: Sequence[EvalExample]) -> tuple[EvalExample, ...]:
    """Freeze a non-empty dataset and reject duplicate example IDs."""

    items = tuple(dataset)
    if not items:
        raise ValueError("dataset must be non-empty")
    if not all(isinstance(item, EvalExample) for item in items):
        raise TypeError("every dataset item must be EvalExample")
    ids = [item.example_id for item in items]
    if len(ids) != len(set(ids)):
        raise ValueError("example_id values must be unique")
    return items


def evaluate(
    model_fn: Callable[[str], str | ModelResponse],
    dataset: Sequence[EvalExample],
    metrics: Mapping[str, Callable[[str, str], float]],
    *,
    max_response_chars: int = 10_000,
) -> EvaluationReport:
    """Call the model exactly once per example and score the stored response."""

    items = _validate_dataset(dataset)
    if not callable(model_fn):
        raise TypeError("model_fn must be callable")
    if not metrics or any(not isinstance(name, str) or not name for name in metrics):
        raise ValueError("metrics must be a non-empty mapping with non-empty names")
    if any(not callable(metric) for metric in metrics.values()):
        raise TypeError("every metric must be callable")
    max_response_chars = _positive_int("max_response_chars", max_response_chars)

    records: list[PredictionRecord] = []
    calls = 0
    for example in items:
        calls += 1
        try:
            raw = model_fn(example.prompt)  # the only model invocation for this example
            response = raw if isinstance(raw, ModelResponse) else ModelResponse(raw)
            if len(response.text) > max_response_chars:
                raise ValueError("model response exceeds max_response_chars")
        except Exception as exc:
            records.append(PredictionRecord(
                example_id=example.example_id,
                slice_name=example.slice_name,
                expected=example.expected,
                prediction="",
                scores={name: 0.0 for name in metrics},
                latency_ms=None,
                output_tokens=None,
                error=f"model: {type(exc).__name__}: {exc}",
            ))
            continue

        try:
            values: dict[str, float] = {}
            for name, metric in metrics.items():
                value = metric(response.text, example.expected)
                if isinstance(value, bool) or not isinstance(value, (int, float)):
                    raise TypeError(f"metric {name!r} must return a real number")
                value = float(value)
                if not math.isfinite(value) or not 0.0 <= value <= 1.0:
                    raise ValueError(f"metric {name!r} must return a finite value in [0, 1]")
                values[name] = value
            error = None
        except Exception as exc:
            # Keep the single model response visible even when a metric implementation fails.
            values = {name: None for name in metrics}
            error = f"metric: {type(exc).__name__}: {exc}"

        records.append(PredictionRecord(
            example_id=example.example_id,
            slice_name=example.slice_name,
            expected=example.expected,
            prediction=response.text,
            scores=values,
            latency_ms=response.latency_ms,
            output_tokens=response.output_tokens,
            error=error,
        ))

    aggregate = {
        name: (None if any(record.scores[name] is None for record in records) else
               sum(record.scores[name] for record in records) / len(records))
        for name in metrics
    }
    groups: dict[str, list[PredictionRecord]] = defaultdict(list)
    for record in records:
        groups[record.slice_name].append(record)
    by_slice = {
        slice_name: {
            name: (None if any(record.scores[name] is None for record in group) else
                   sum(record.scores[name] for record in group) / len(group))
            for name in metrics
        }
        for slice_name, group in sorted(groups.items())
    }
    return EvaluationReport(
        records=tuple(records),
        aggregate=aggregate,
        by_slice=by_slice,
        model_calls=calls,
        error_count=sum(record.error is not None for record in records),
    )


class CountingTableModel:
    """Deterministic mock model that exposes how many times it was called."""

    def __init__(self, prompts: Sequence[str], outputs: Sequence[str]) -> None:
        """Build a one-to-one prompt table and initialize the call counter."""

        if len(prompts) != len(outputs):
            raise ValueError("prompts and outputs must have equal length")
        self._table = dict(zip(prompts, outputs))
        if len(self._table) != len(prompts):
            raise ValueError("prompts must be unique")
        self.calls = 0

    def __call__(self, prompt: str) -> ModelResponse:
        """Return one deterministic response and increment the visible counter."""

        self.calls += 1
        if prompt not in self._table:
            raise KeyError(prompt)
        text = self._table[prompt]
        return ModelResponse(text=text)


DATASET = tuple(EvalExample(
    example_id=item["id"], prompt=item["prompt"], expected=item["expected"], slice_name=item["slice"]
) for item in FIXTURE["examples"])
METRICS = {"exact": exact_match, "normalized_exact": normalized_exact_match}
MODEL_A = CountingTableModel([e.prompt for e in DATASET], FIXTURE["model_a"])
REPORT_A = evaluate(MODEL_A, DATASET, METRICS)
assert MODEL_A.calls == len(DATASET) == REPORT_A.model_calls
assert REPORT_A.error_count == 0
assert REPORT_A.aggregate == {"exact": 0.75, "normalized_exact": 0.875}
assert [r.prediction for r in REPORT_A.records] == FIXTURE["model_a"]
print(REPORT_A.aggregate)
print(REPORT_A.by_slice)
print('One-call-per-example report built')

class AlternatingModel:
    """Return a correct answer on odd calls and a wrong answer on even calls."""

    def __init__(self, correct: str) -> None:
        """Store the correct answer and reset the stateful call counter."""

        self.correct = correct
        self.calls = 0

    def __call__(self, prompt: str) -> str:
        """Alternate between the correct and deliberately wrong response."""

        self.calls += 1
        return self.correct if self.calls % 2 else "__wrong__"


example = DATASET[0]
broken = AlternatingModel(example.expected)
shown_prediction = broken(example.prompt)
score_from_second_call = normalized_exact_match(broken(example.prompt), example.expected)
assert normalize_answer(shown_prediction) == normalize_answer(example.expected)
assert score_from_second_call == 0.0 and broken.calls == 2

fixed = AlternatingModel(example.expected)
fixed_report = evaluate(fixed, [example], {"normalized_exact": normalized_exact_match})
assert fixed.calls == 1
assert fixed_report.records[0].prediction == shown_prediction
assert fixed_report.aggregate["normalized_exact"] == 1.0
print("broken displayed:", shown_prediction, "scored:", score_from_second_call, "calls:", broken.calls)
print("fixed displayed:", fixed_report.records[0].prediction, "score:", fixed_report.aggregate["normalized_exact"], "calls:", fixed.calls)
print('The double-call defect was reproduced and removed')

def binary_scores(report: EvaluationReport, metric_name: str) -> dict[str, int]:
    """Return one 0/1 score per example ID for a binary metric."""

    if not report.records:
        raise ValueError("paired report must be non-empty")
    result: dict[str, int] = {}
    for record in report.records:
        if record.example_id in result:
            raise ValueError("duplicate report ID")
        value = record.scores[metric_name]
        if value not in {0.0, 1.0}:
            raise ValueError("paired binary comparison requires 0/1 metric values")
        result[record.example_id] = int(value)
    return result


def paired_contingency(first: EvaluationReport, second: EvaluationReport, metric_name: str) -> dict[str, int]:
    """Build the four paired correctness cells for two reports on the same IDs."""

    a = binary_scores(first, metric_name)
    b = binary_scores(second, metric_name)
    if set(a) != set(b):
        raise ValueError("reports must contain the same example IDs")
    metadata = lambda report: {r.example_id: (r.expected, r.slice_name) for r in report.records}
    if metadata(first) != metadata(second):
        raise ValueError("paired reports must use the same targets and slices")
    cells = {"both_correct": 0, "first_only": 0, "second_only": 0, "both_wrong": 0}
    for example_id in sorted(a):
        pair = a[example_id], b[example_id]
        key = {
            (1, 1): "both_correct",
            (1, 0): "first_only",
            (0, 1): "second_only",
            (0, 0): "both_wrong",
        }[pair]
        cells[key] += 1
    return cells


MODEL_B = CountingTableModel([e.prompt for e in DATASET], FIXTURE["model_b"])
REPORT_B = evaluate(MODEL_B, DATASET, METRICS)
assert MODEL_B.calls == len(DATASET)
assert REPORT_B.aggregate == {"exact": 0.625, "normalized_exact": 0.75}

cells = paired_contingency(REPORT_A, REPORT_B, "normalized_exact")
p_value = mcnemar_exact_pvalue(cells["first_only"], cells["second_only"])
assert cells == {"both_correct": 5, "first_only": 2, "second_only": 1, "both_wrong": 0}
assert p_value == 1.0
successes_a = sum(binary_scores(REPORT_A, "normalized_exact").values())
ci_a = wilson_interval(successes_a, len(DATASET))
print("A:", REPORT_A.aggregate, "Wilson normalized_exact:", tuple(round(v, 3) for v in ci_a))
print("B:", REPORT_B.aggregate)
print("paired cells:", cells, "exact McNemar p=", p_value)
print('Slice and paired reports built')

"""CPU calibration experiment over authored, saved response records (no LLM)."""
import json
import math
from pathlib import Path
from tempfile import TemporaryDirectory


def validate_records(records):
    if not records:
        raise ValueError("records must be nonempty")
    ids = set()
    for row in records:
        ident = row["id"]
        if not isinstance(ident, str) or not ident or ident in ids:
            raise ValueError("unique nonempty string ids required")
        ids.add(ident)
        if not all(isinstance(row[key], str) for key in ("prediction", "expected")):
            raise ValueError("responses and targets must be strings")
        q = row["confidence"]
        if isinstance(q, bool) or not isinstance(q, (float, int)) or not math.isfinite(q) or not 0 <= q <= 1:
            raise ValueError("confidence must be finite and in [0, 1]")


def correctness(row):
    # Same explicitly declared policy as the lecture's exact_match.
    return int(row["prediction"].strip().casefold() == row["expected"].strip().casefold())


def brier_score(records):
    validate_records(records)
    return sum((row["confidence"] - correctness(row)) ** 2 for row in records) / len(records)


def reliability_groups(records):
    """Group identical authored scores; general data require declared bin edges."""
    validate_records(records)
    result = []
    for q in sorted({row["confidence"] for row in records}):
        group = [row for row in records if row["confidence"] == q]
        result.append(dict(confidence=q, count=len(group),
                           accuracy=sum(map(correctness, group)) / len(group)))
    return result


def selective_report(records, threshold):
    validate_records(records)
    if isinstance(threshold, bool) or not isinstance(threshold, (int, float)) or not math.isfinite(threshold) or not 0 <= threshold <= 1:
        raise ValueError("threshold must be finite and in [0, 1]")
    accepted = [row for row in records if row["confidence"] >= threshold]
    errors = sum(1 - correctness(row) for row in accepted)
    return dict(threshold=threshold, accepted=len(accepted), errors=errors,
                coverage=len(accepted) / len(records),
                risk=errors / len(accepted) if accepted else None)


def authored_records():
    # These counts define a teaching fixture, not sampled model behavior.
    # Confidence is part of each authored response; it is not fitted here.
    records = []
    for q, count, successes in [(0.95, 4, 4), (0.8, 10, 6), (0.6, 6, 3)]:
        for j in range(count):
            i = len(records)
            records.append(dict(id=f"cal-{i:02}", prompt=f"{i} + 1 = ?",
                                prediction=str(i + 1 if j < successes else i + 2),
                                expected=str(i + 1), confidence=q))
    return records


def run_calibration_lab():
    with TemporaryDirectory(dir=Path.cwd()) as folder:
        path = Path(folder) / "calibration_responses.json"
        path.write_text(json.dumps(authored_records()), encoding="utf-8")
        saved = json.loads(path.read_text(encoding="utf-8"))
    # Every score below comes from the same saved records, without new calls.
    score = brier_score(saved)
    groups = reliability_groups(saved)
    curve = [selective_report(saved, t) for t in (0.0, 0.6, 0.8, 0.9, 1.0)]
    assert sum(map(correctness, saved)) == 13
    assert math.isclose(score, 0.2185, abs_tol=1e-14)
    assert groups[1] == dict(confidence=0.8, count=10, accuracy=0.6)
    assert curve[2]["accepted"] == 14 and curve[2]["errors"] == 4
    assert curve[3]["accepted"] == 4 and curve[3]["risk"] == 0
    assert curve[4]["coverage"] == 0 and curve[4]["risk"] is None
    print("Brier:", score, "accuracy:", 13 / 20)
    print("Confidence groups:", groups)
    print("Coverage and selective risk:", curve)
    return saved, groups, curve


if __name__ == "__main__":
    run_calibration_lab()

@dataclass(frozen=True)
class Candidate:
    """Candidate identity and response text for pairwise judging."""

    candidate_id: str
    text: str

    def __post_init__(self) -> None:
        """Reject candidates without a stable identity or response text."""

        if any(not isinstance(v, str) or not v.strip() for v in (self.candidate_id, self.text)):
            raise ValueError("candidate_id and text must be non-empty")


@dataclass(frozen=True)
class SwapDecision:
    """Both order-specific labels, mapped identities, and consensus."""

    first_order_label: str
    swapped_order_label: str
    first_order_winner: str | None
    swapped_order_winner: str | None
    consensus: str | None
    status: str


def map_positional_label(label: str, first: Candidate, second: Candidate) -> str | None:
    """Map first/second/tie back to a stable candidate identity."""

    if label == "first":
        return first.candidate_id
    if label == "second":
        return second.candidate_id
    if label == "tie":
        return None
    raise ValueError("judge label must be 'first', 'second', or 'tie'")


def run_swap_protocol(
    judge_fn: Callable[[str, Candidate, Candidate], str],
    prompt: str,
    candidate_a: Candidate,
    candidate_b: Candidate,
) -> SwapDecision:
    """Judge A/B and B/A exactly once each, then require identity consensus."""

    if candidate_a.candidate_id == candidate_b.candidate_id:
        raise ValueError("candidate IDs must differ")
    label_ab = judge_fn(prompt, candidate_a, candidate_b)
    label_ba = judge_fn(prompt, candidate_b, candidate_a)
    winner_ab = map_positional_label(label_ab, candidate_a, candidate_b)
    winner_ba = map_positional_label(label_ba, candidate_b, candidate_a)
    consensus = winner_ab if winner_ab is not None and winner_ab == winner_ba else None
    status = "winner" if consensus is not None else ("tie" if label_ab == label_ba == "tie" else "abstain")
    return SwapDecision(label_ab, label_ba, winner_ab, winner_ba, consensus, status)


class FirstPositionJudge:
    """A maximally position-biased judge used as a negative control."""

    def __init__(self) -> None:
        """Initialize the visible number of judge invocations."""

        self.calls = 0

    def __call__(self, prompt: str, first: Candidate, second: Candidate) -> str:
        """Always select the first position, ignoring candidate content."""

        self.calls += 1
        return "first"


class ContentJudge:
    """A deterministic judge that selects the candidate containing a marker."""

    def __init__(self, marker: str) -> None:
        """Store the content marker and initialize the call counter."""

        self.marker = marker
        self.calls = 0

    def __call__(self, prompt: str, first: Candidate, second: Candidate) -> str:
        """Select the unique candidate containing the marker, otherwise tie."""

        self.calls += 1
        in_first = self.marker in first.text
        in_second = self.marker in second.text
        if in_first == in_second:
            return "tie"
        return "first" if in_first else "second"


candidate_a = Candidate("A", "answer with VERIFIED fact")
candidate_b = Candidate("B", "unsupported answer")
position_judge = FirstPositionJudge()
position_decision = run_swap_protocol(position_judge, "prompt", candidate_a, candidate_b)
assert position_judge.calls == 2
assert position_decision.first_order_winner == "A"
assert position_decision.swapped_order_winner == "B"
assert position_decision.consensus is None and position_decision.status == "abstain"

content_judge = ContentJudge("VERIFIED")
content_decision = run_swap_protocol(content_judge, "prompt", candidate_a, candidate_b)
assert content_judge.calls == 2 and content_decision.consensus == "A"

J = REFERENCE["judge"]
assert J["forward_a_wins"] == 650 and J["swapped_a_wins"] == 350
assert J["averaged_a_rate"] == 0.5 and J["calls_per_pair"] == 6
print("position control:", position_decision)
print("content consensus:", content_decision)
print("analytic pure-position aggregate: 65% first order -> 50% after two-order averaging")
print('The swap protocol separates candidate identity from position')

def golden_sample_size(margin: float, expected_rate: float = 0.8, z_value: float = 1.96) -> int:
    """Return a normal-approximation planning size for a target half-width."""
    margin = _positive_finite("margin", margin)
    p = _probability("expected_rate", expected_rate)
    z_value = _positive_finite("z_value", z_value)
    if p in {0.0, 1.0}:
        raise ValueError("expected_rate must be strictly between 0 and 1")
    return math.ceil(p * (1.0 - p) * (z_value / margin) ** 2)

def deterministic_demo_canary(namespace: str, hex_characters: int = 32) -> str:
    """Create a reproducible demo canary; use ``secrets.token_hex`` for a real private set."""

    if not isinstance(namespace, str) or not namespace:
        raise ValueError("namespace must be non-empty")
    length = _positive_int("hex_characters", hex_characters)
    if length > 64:
        raise ValueError("this SHA-256 demo supports at most 64 hex characters")
    return hashlib.sha256(namespace.encode("utf-8")).hexdigest()[:length]


def contains_canary(canary: str, texts: Sequence[str]) -> bool:
    """Return whether an exact canary occurs in any supplied text."""

    if not canary:
        raise ValueError("canary must be non-empty")
    if any(not isinstance(text, str) for text in texts):
        raise TypeError("texts must contain strings")
    return any(canary in text for text in texts)


G = REFERENCE["golden"]
assert golden_sample_size(0.03) == G["n_for_3pp"] == 683
assert golden_sample_size(0.05) == G["n_for_5pp"] == 246
canary = deterministic_demo_canary("modern-llms-m19-demo", G["canary_hex_characters"])
assert len(canary) == 32 and all(ch in string.hexdigits.lower() for ch in canary)
assert not contains_canary(canary, [record.prediction for record in REPORT_A.records])
assert abs(16.0**-32 - G["canary_guess_probability"]) < 1e-50
cost = G["cost"]
assert cost["total_tokens"] == 385_000
assert abs(cost["seconds_floor"] - 12.491111111111111) < 1e-12
print(f"planning size: ±3 pp -> {G['n_for_3pp']}; ±5 pp -> {G['n_for_5pp']}")
assert contains_canary(canary, ["planted control " + canary])
print("PUBLIC demo canary:", canary, "random-secret reference probability:", f"{G['canary_guess_probability']:.2e}")
print(f"conditional arithmetic equivalent: {cost['total_tokens']:,} tokens, {cost['dense_flops_floor']:.3e} FLOP, {cost['seconds_floor']:.2f} s")
print('Golden-set sizing, canary, and compute estimate verified')

def working_context_share(edge: float, dip: float, threshold: float) -> float:
    """Return the fraction of positions above threshold for a synthetic U profile."""
    edge = _probability("edge", edge)
    threshold = _probability("threshold", threshold)
    dip = _probability("dip", dip)
    if edge - dip < 0.0:
        raise ValueError("edge - dip must remain a valid probability")
    if threshold > edge:
        return 0.0
    if threshold <= edge - dip:
        return 1.0
    return 2.0 * math.asin(math.sqrt((edge - threshold) / dip)) / math.pi

U = REFERENCE["lost_in_middle"]
share = working_context_share(U["edge"], U["dip"], U["threshold"])
assert abs(share - U["working_share"]) < 1e-15
assert U["working_tokens_of_1m"] == round(share * 1_000_000)
assert abs(U["mean"] - (U["edge"] - U["dip"] / 2)) < 1e-15
print(f"synthetic mean={U['mean']:.3f}; positions above {U['threshold']:.2f}: {share:.1%} ({U['working_tokens_of_1m']:,} of 1M)")
print('Synthetic working-context profile verified')

def expect_error(error_type: type[BaseException], fn: Callable[[], object]) -> str:
    """Run a negative test and return its expected error message."""

    try:
        fn()
    except error_type as exc:
        return str(exc)
    except Exception as exc:
        raise AssertionError(f"expected {error_type.__name__}, got {type(exc).__name__}: {exc}") from exc
    raise AssertionError(f"expected {error_type.__name__}")


errors = []
errors.append(expect_error(ValueError, lambda: evaluate(lambda _: "x", [], METRICS)))
duplicate = [DATASET[0], DATASET[0]]
errors.append(expect_error(ValueError, lambda: evaluate(lambda _: "x", duplicate, METRICS)))
errors.append(expect_error(ValueError, lambda: pass_at_k_unbiased(4, 5, 1)))
errors.append(expect_error(ValueError, lambda: pass_at_k_unbiased(4, 1, 5)))
errors.append(expect_error(ValueError, lambda: wilson_interval(9, 8)))
errors.append(expect_error(ValueError, lambda: map_positional_label("left", candidate_a, candidate_b)))

bad_type_report = evaluate(lambda _: 123, [DATASET[0]], METRICS)
assert bad_type_report.error_count == 1 and "TypeError" in bad_type_report.records[0].error
oversized_report = evaluate(lambda _: "x" * 20, [DATASET[0]], METRICS, max_response_chars=10)
assert oversized_report.error_count == 1 and "max_response_chars" in oversized_report.records[0].error
bad_metric_report = evaluate(lambda _: "ok", [DATASET[0]], {"bad": lambda p, e: math.nan})
assert bad_metric_report.error_count == 1 and "finite" in bad_metric_report.records[0].error
assert bad_metric_report.records[0].prediction == "ok"  # metric failure must not erase the model output
assert bad_metric_report.aggregate["bad"] is None
assert normalized_exact_match("-1", "1") == 0.0
assert normalized_exact_match("1.2", "12") == 0.0
assert len(errors) == 6
print("expected errors:", errors)
print('Negative contract tests passed')

from __future__ import annotations
import importlib.util, subprocess, sys
for package in ("numpy", "matplotlib"):
    if importlib.util.find_spec(package) is None:
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package])


import math
import random
import time
from collections import Counter, defaultdict

import matplotlib.pyplot as plt
import numpy as np

SEED = 20260801
random.seed(SEED)
np.random.seed(SEED)
print(f"reproducible smoke tier; seed={SEED}")
DEMO_LANGUAGE = 'en'

import math
from dataclasses import dataclass
from typing import Callable, Mapping, Sequence


@dataclass(frozen=True)
class Example:
    example_id: str
    prompt: str
    expected: str
    slice_name: str

    def __post_init__(self):
        if any(not isinstance(v, str) or not v.strip()
               for v in (self.example_id, self.prompt, self.expected, self.slice_name)):
            raise ValueError("example fields must be non-empty strings")


def exact_match(prediction: str, expected: str) -> float:
    return float(prediction.strip().casefold() == expected.strip().casefold())


def evaluate(
    model_fn: Callable[[str], str],
    dataset: Sequence[Example],
    metrics: Mapping[str, Callable[[str, str], float]],
) -> dict:
    dataset = tuple(dataset)
    if not callable(model_fn) or not metrics or any(not isinstance(k, str) or not k or not callable(v) for k, v in metrics.items()):
        raise ValueError("supply a callable model and named callable metrics")
    if not dataset:
        raise ValueError("dataset must be non-empty")
    if any(not isinstance(ex, Example) for ex in dataset):
        raise TypeError("expected Example records")
    if len({ex.example_id for ex in dataset}) != len(dataset):
        raise ValueError("example_id values must be unique")

    records = []
    for ex in dataset:
        prediction = model_fn(ex.prompt)  # the only model call for this item
        if not isinstance(prediction, str):
            raise TypeError("model response must be text")
        scores = {
            name: metric(prediction, ex.expected)
            for name, metric in metrics.items()
        }
        if any(isinstance(v, bool) or not isinstance(v, (float, int)) or not math.isfinite(v) or not 0 <= v <= 1 for v in scores.values()):
            raise ValueError("metric scores must be finite numbers in [0, 1]")
        records.append({
            "id": ex.example_id,
            "slice": ex.slice_name,
            "expected": ex.expected,
            "prediction": prediction,
            "scores": scores,
        })

    aggregate = {
        name: sum(row["scores"][name] for row in records) / len(records)
        for name in metrics
    }
    return {
        "records": records,
        "aggregate": aggregate,
        "model_calls": len(records),
    }

def _positive_int(name: str, value: int) -> int:
    """Validate and return a strictly positive, non-boolean integer."""
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ValueError(f"{name} must be a positive integer")
    return value

def _non_negative_int(name: str, value: int) -> int:
    """Validate and return a non-negative, non-boolean integer."""
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{name} must be a non-negative integer")
    return value

def _positive_finite(name: str, value: float) -> float:
    """Validate and return a finite positive real number."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be a real number")
    number = float(value)
    if not math.isfinite(number) or number <= 0.0:
        raise ValueError(f"{name} must be finite and positive")
    return number

def wilson_interval(successes: int, total: int, z_value: float = 1.96) -> tuple[float, float]:
    """Return the Wilson score interval for a binomial proportion."""
    n = _positive_int("total", total)
    s = _non_negative_int("successes", successes)
    if s > n:
        raise ValueError("successes cannot exceed total")
    z_value = _positive_finite("z_value", z_value)
    p = s / n
    z2 = z_value**2
    denominator = 1.0 + z2 / n
    centre = (p + z2 / (2.0 * n)) / denominator
    radius = z_value * math.sqrt(p * (1.0 - p) / n + z2 / (4.0 * n**2)) / denominator
    return max(0.0, centre - radius), min(1.0, centre + radius)

def mcnemar_exact_pvalue(first_only: int, second_only: int) -> float:
    """Return the two-sided exact McNemar p-value for discordant paired outcomes."""
    first_only = _non_negative_int("first_only", first_only)
    second_only = _non_negative_int("second_only", second_only)
    discordant = first_only + second_only
    if discordant == 0:
        return 1.0
    tail = min(first_only, second_only)
    cdf = sum(math.comb(discordant, k) for k in range(tail + 1)) / (1 << discordant)
    return min(1.0, 2.0 * cdf)
import json, tempfile
from pathlib import Path
# Authored fictional policy: sealed items may be returned through day 30 inclusive.
# Unique requests, but generated fixtures are NOT a sample of real customers.
dataset = tuple(Example(
    f"ordinary-{i:03d}", json.dumps({"days": i % 50, "sealed": True}),
    "refund" if i % 50 <= 30 else "deny", "ordinary"
) for i in range(200))
def policy_a(prompt):
    item = json.loads(prompt)
    return "refund" if item["days"] <= 30 else "deny"  # ignores sealed flag
def policy_b(prompt):
    item = json.loads(prompt)
    return "refund" if item["sealed"] and item["days"] < 30 else "deny"  # boundary bug

reports = {name: evaluate(fn, dataset, {"exact": exact_match})
           for name, fn in [("A", policy_a), ("B", policy_b)]}
passport = {"fixture": "fictional-refunds-v1", "systems": ["A-v1", "B-v1"],
            "trial": 0, "metric": "casefold-exact-v1", "seed": 1901}
payload = {"passport": passport, "reports": reports}
with tempfile.TemporaryDirectory(dir=Path.cwd()) as folder:
    file = Path(folder)/"saved_responses.json"
    file.write_text(json.dumps(payload), encoding="utf-8")
    saved = json.loads(file.read_text(encoding="utf-8"))
# All rescoring below uses saved responses, making no new policy calls.
for name in ("A", "B"):
    rows = saved["reports"][name]["records"]
    assert all(exact_match(r["prediction"], r["expected"]) == r["scores"]["exact"] for r in rows)
a = np.array([r["scores"]["exact"] for r in saved["reports"]["A"]["records"]])
b = np.array([r["scores"]["exact"] for r in saved["reports"]["B"]["records"]])
assert len(a) == 200 and a.sum() == 200 and b.sum() == 196
rng = np.random.default_rng(1901)
differences = a - b
bootstrap = differences[rng.integers(0, len(a), size=(5000, len(a)))].mean(axis=1)
interval = np.quantile(bootstrap, [.025, .975])
first_only, second_only = int(((a==1)&(b==0)).sum()), int(((a==0)&(b==1)).sum())
p_value = mcnemar_exact_pvalue(first_only, second_only)
assert (first_only, second_only) == (4, 0) and p_value == .125
print("A/B:", a.mean(), b.mean(), "paired delta:", differences.mean(),
      "bootstrap:", interval, "exact McNemar:", p_value)
print("Calls:", {n: r["model_calls"] for n, r in reports.items()},
      "latency/cost: unmeasured; deterministic rules, not LLMs")
assert np.isclose(interval[0], .005) and np.isclose(interval[1], .04)

# Predeclared independent gates: ordinary >= .99, critical = 1, trajectory = 1.
# They illustrate a decision rule; they are not a justified production risk policy.
critical = tuple(Example(
    f"critical-{i}", json.dumps({"days": i, "sealed": False}), "deny", "critical"
) for i in range(20))
quality = {name: {
    "ordinary": reports[name]["aggregate"]["exact"],
    "critical": evaluate(fn, critical, {"exact":exact_match})["aggregate"]["exact"]
} for name, fn in [("A",policy_a),("B",policy_b)]}

def run_trajectory(authorized, version):
    """Execute two delivery attempts; inspect effects rather than final words."""
    ledger, seen, events = [], set(), []
    for _ in range(2):
        operation_id = "refund-17"
        if version == "B":
            if not authorized:
                events.append("denied")
                continue
            if operation_id in seen:
                events.append("replay")
                continue
        ledger.append({"operation":operation_id, "amount":7})
        seen.add(operation_id)
        events.append("effect")
    expected = [{"operation":"refund-17", "amount":7}] if authorized else []
    return {"ok":ledger == expected, "ledger":ledger, "events":events}

traces = {name: [run_trajectory(allowed,name) for allowed in [True,False]]
          for name in ("A","B")}
for name in ("A","B"):
    quality[name]["trajectory"] = sum(t["ok"] for t in traces[name])/2
gates = {"ordinary":.99, "critical":1., "trajectory":1.}
decision = {name: all(scores[k] >= gates[k] for k in gates) for name,scores in quality.items()}
assert quality == {"A":{"ordinary":1., "critical":0., "trajectory":0.},
                   "B":{"ordinary":.98, "critical":1., "trajectory":1.}}
assert decision == {"A":False,"B":False}
print("suites:", quality, "release gates:", decision)
print("actual traces:", traces)
# There is no justified universal average over these deliberately different suites.
# Fix B's boundary and rerun on development data; keep a fresh final audit set.

# A public reproducible canary verifies the detector, not privacy or training data.
import hashlib
canary = hashlib.sha256(b"public-m19-positive-control").hexdigest()[:32]
assert canary not in "ordinary answer"
assert canary in ("planted positive control " + canary)
print("canary detector: negative and planted positive controls pass")

# Actual search, with a deliberately limited reader as the negative control.
# Length counts lines, NOT model tokens. No model is involved.
positions = np.linspace(0, 1, 9)
measured = {}
for length in (128, 512, 1024):
    full, edges = [], []
    for fraction in positions:
        pos = round(fraction * (length-1))
        lines = [f"irrelevant line {j}" for j in range(length)]
        needle = "TARGET: blue-orchid"
        lines[pos] = needle
        extract = lambda selected: next((s.split(": ",1)[1] for s in selected if s.startswith("TARGET: ")), "")
        full.append(extract(lines) == "blue-orchid")
        edges.append(extract(lines[:32]+lines[-32:]) == "blue-orchid")
    assert all(full)
    measured[length] = edges
    plt.plot(positions, edges, "o-", label=f"{length} "+("строк" if DEMO_LANGUAGE=="ru" else "lines"))
plt.plot(positions, full, "--", color="black", label="full scan")
plt.xlabel("позиция иголки" if DEMO_LANGUAGE=="ru" else "needle position")
plt.ylabel("успех извлечения (0/1)" if DEMO_LANGUAGE=="ru" else "retrieval success (0/1)")
plt.title("Проверка ограниченного чтения" if DEMO_LANGUAGE=="ru" else "Testing a limited reader")
plt.legend(); plt.tight_layout(); plt.show()
assert measured[128] == [True,True,False,False,False,False,False,True,True]
assert measured[512] == measured[1024] == [True,False,False,False,False,False,False,False,True]