Модуль 19. Оценка современных LLM: от бенчмарка к инженерному решению

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":"париж","id":"geo_fr","prompt":"столица Франции","slice":"facts"},{"expected":"4","id":"math_2","prompt":"2+2","slice":"math"},{"expected":"au","id":"chem_au","prompt":"химический символ золота","slice":"facts"},{"expected":"1969","id":"hist_moon","prompt":"год высадки на Луну","slice":"facts"},{"expected":"42","id":"math_42","prompt":"сколько будет 7*6","slice":"math"},{"expected":"токио","id":"geo_jp","prompt":"столица Японии","slice":"facts"},{"expected":"9","id":"math_sqrt","prompt":"квадратный корень из 81","slice":"math"},{"expected":"холодный","id":"lang_antonym","prompt":"антоним слова горячий","slice":"language"}],"model_a":["Париж.","4","AU","1969","42","Варшава","9","холодный"],"model_b":["париж","четыре","ag","1969","42","Токио","9","холодный."]}'
FIXTURE_SHA256 = "c7528a138a167486717962112054664f8f548356a10b8c583633f49d41209535"


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('Встроенные контракты М19 проверены')

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('Агрегированная аппроксимация, интервал Уилсона и точный McNemar проверены')

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('Несмещённый pass@k и крайние случаи проверены')

@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('Отчёт с одним вызовом на пример построен')

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('Дефект двойного вызова воспроизведён и устранён')

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('Срезы и парное сравнение построены')

"""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('Перестановочный протокол отделяет идентичность от позиции')

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('Размер золотого набора, canary и вычислительная оценка проверены')

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('Синтетический профиль рабочего контекста проверен')

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('Негативные тесты контрактов пройдены')

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"воспроизводимый тестовый запуск; seed={SEED}")
DEMO_LANGUAGE = 'ru'

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]