Capstone: Fine-Tuned Classifier

label_version = "escalation-policy-v1"
fixtures = [
    {
        "id": "access_override_exception",
        "text": "The production API key is disabled, but the incident freeze still blocks restore.",
        "gold_route": "human_review_now",
        "reason": "access_override_exception",
    },
    {
        "id": "incident_freeze_exception",
        "text": "The production API key was disabled during an incident freeze. Please allow an exception.",
        "gold_route": "human_review_now",
        "reason": "incident_freeze_exception",
    },
    {
        "id": "account_takeover",
        "text": "Someone changed my recovery email and I can't sign in.",
        "gold_route": "human_review_now",
        "reason": "account_takeover",
    },
    {
        "id": "routine_key_rotation",
        "text": "Where is the key rotation guide?",
        "gold_route": "guarded_agent",
        "reason": "routine_key_rotation",
    },
    {
        "id": "access_policy_question",
        "text": "What is the access policy for disabled keys?",
        "gold_route": "guarded_agent",
        "reason": "access_policy_question",
    },
]

allowed_routes = {"human_review_now", "guarded_agent"}
assert all(row["gold_route"] in allowed_routes for row in fixtures)
assert {row["reason"] for row in fixtures if row["gold_route"] == "human_review_now"} == {
    "access_override_exception",
    "incident_freeze_exception",
    "account_takeover",
}

print("label guide:", label_version)
for row in fixtures:
    print(f"{row['id']}: {row['gold_route']} ({row['reason']})")

records = [
    {"ticket": "t01", "account": "acme", "month": 1, "split": "train"},
    {"ticket": "t02", "account": "acme", "month": 2, "split": "train"},
    {"ticket": "t03", "account": "north", "month": 2, "split": "train"},
    {"ticket": "t04", "account": "north", "month": 2, "split": "train"},
    {"ticket": "v01", "account": "summit", "month": 3, "split": "validation"},
    {"ticket": "v02", "account": "summit", "month": 3, "split": "validation"},
    {"ticket": "x01", "account": "harbor", "month": 4, "split": "test"},
    {"ticket": "x02", "account": "harbor", "month": 4, "split": "test"},
]

accounts = {
    split: {row["account"] for row in records if row["split"] == split}
    for split in ("train", "validation", "test")
}
assert accounts["train"].isdisjoint(accounts["validation"])
assert accounts["train"].isdisjoint(accounts["test"])
assert accounts["validation"].isdisjoint(accounts["test"])
assert len({row["ticket"] for row in records}) == len(records)
months = {
    split: [row["month"] for row in records if row["split"] == split]
    for split in ("train", "validation", "test")
}
assert max(months["train"]) < min(months["validation"])
assert max(months["validation"]) < min(months["test"])

for split in ("train", "validation", "test"):
    count = sum(row["split"] == split for row in records)
    names = ",".join(sorted(accounts[split]))
    print(f"{split}: {count} tickets, accounts={names}")

validation = [
    ("explicit_access_override_exception", "Production API key disabled and the incident freeze blocks restore.", 1),
    ("indirect_incident_freeze_exception", "The production API key was disabled during the incident freeze.", 1),
    ("routine_key_rotation", "Where is the key rotation guide?", 0),
    ("routine_policy", "Where can I read the access policy?", 0),
]
terms = ("key disabled", "can't sign in", "admin override")

predictions = []
for fixture_id, text, label in validation:
    predicted = int(any(term in text.lower() for term in terms))
    predictions.append((fixture_id, label, predicted))

positives = sum(label for _, label, _ in predictions)
true_positives = sum(label == 1 and predicted == 1 for _, label, predicted in predictions)
missed = [fixture_id for fixture_id, label, predicted in predictions if label == 1 and predicted == 0]

print(f"positive recall: {true_positives}/{positives}")
print("missed positive:", missed[0])
print("next experiment: fit a text-feature baseline")

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline

baseline_train = [
    ("prod key disabled incident freeze", 1),
    ("prod key disabled during incident freeze", 1),
    ("recovery email changed cannot sign in", 1),
    ("where is key rotation guide", 0),
    ("show key rotation guide", 0),
    ("show access policy", 0),
]
baseline_validation = [
    ("Production API key disabled and the incident freeze blocks restore.", 1),
    ("The production API key was disabled during the incident freeze.", 1),
    ("Where is the key rotation guide?", 0),
    ("Where can I read the access policy?", 0),
]
baseline = make_pipeline(
    TfidfVectorizer(ngram_range=(1, 2)),
    LogisticRegression(C=1.0, max_iter=500, random_state=7),
)
baseline.fit([text for text, _ in baseline_train], [gold for _, gold in baseline_train])
predicted = baseline.predict([text for text, _ in baseline_validation])
tp = sum(gold == 1 and pred == 1 for (_, gold), pred in zip(baseline_validation, predicted))
fn = sum(gold == 1 and pred == 0 for (_, gold), pred in zip(baseline_validation, predicted))
fp = sum(gold == 0 and pred == 1 for (_, gold), pred in zip(baseline_validation, predicted))
print(f"TF-IDF baseline: tp={tp} fn={fn} fp={fp}")
print("validation predictions:", predicted.tolist())

from math import exp, isfinite

examples = [
    ("incident_freeze", [-1.0, 2.0]),
    ("key_rotation", [1.0, -1.0]),
]

def softmax(logits: list[float]) -> list[float]:
    def finite_logit(value: object) -> bool:
        try:
            return type(value) in (int, float) and isfinite(value)
        except OverflowError:
            return False

    if type(logits) is not list or not logits or not all(finite_logit(value) for value in logits):
        raise ValueError("Expected nonempty finite logits")
    offset = max(logits)
    values = [exp(value - offset) for value in logits]
    total = sum(values)
    return [value / total for value in values]

for fixture_id, logits in examples:
    routine, human = softmax(logits)
    print(f"{fixture_id}: human_review_score={human:.2f}, routine_score={routine:.2f}")

import torch
from torch import nn
from io import BytesIO

torch.manual_seed(7)
torch.set_num_threads(1)

training_rows = [
    ("prod key disabled incident freeze", 1),
    ("prod key disabled during incident freeze", 1),
    ("recovery email changed cannot sign in", 1),
    ("where is key rotation guide", 0),
    ("show key rotation guide", 0),
    ("show access policy", 0),
]
vocabulary = {"<pad>": 0, "<unk>": 1}
for text, _ in training_rows:
    for token in text.split():
        vocabulary.setdefault(token, len(vocabulary))

def encode(texts: list[str]) -> torch.Tensor:
    if type(texts) is not list or not texts or any(type(text) is not str or not text.strip() for text in texts):
        raise ValueError("Expected a nonempty list of nonblank ticket strings")
    rows = [[vocabulary.get(token, 1) for token in text.lower().split()] for text in texts]
    width = max(1, max(len(row) for row in rows))
    return torch.tensor([row + [0] * (width - len(row)) for row in rows])

inputs = encode([text for text, _ in training_rows])
labels = torch.tensor([label for _, label in training_rows])

class TinyEncoderClassifier(nn.Module):
    def __init__(self, vocab_size: int) -> None:
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, 8, padding_idx=0)
        self.head = nn.Linear(8, 2)

    def forward(self, token_ids: torch.Tensor) -> torch.Tensor:
        mask = (token_ids != 0).unsqueeze(-1)
        embedded = self.embedding(token_ids) * mask
        pooled = embedded.sum(dim=1) / mask.sum(dim=1).clamp_min(1)
        return self.head(pooled)

model = TinyEncoderClassifier(len(vocabulary))
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.05)
initial_loss = criterion(model(inputs), labels).item()
original_head = model.head.weight.detach().clone()

for _ in range(80):
    loss = criterion(model(inputs), labels)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

model.eval()
with torch.inference_mode():
    final_logits = model(inputs)
    final_loss = criterion(final_logits, labels).item()
print("logit shape:", tuple(final_logits.shape))
print("loss decreased:", final_loss < initial_loss)
print("classification head updated:", not torch.equal(original_head, model.head.weight))

# Evaluate after fitting. This set was not used for gradient updates.
held_out = [
    ("prod key was disabled during incident freeze", 1),
    ("recovery email changed cannot sign in today", 1),
    ("where is the key rotation guide", 0),
    ("show the access policy", 0),
]
with torch.inference_mode():
    predictions = model(encode([text for text, _ in held_out])).argmax(dim=1).tolist()
tp = sum(gold == 1 and pred == 1 for (_, gold), pred in zip(held_out, predictions))
fn = sum(gold == 1 and pred == 0 for (_, gold), pred in zip(held_out, predictions))
fp = sum(gold == 0 and pred == 1 for (_, gold), pred in zip(held_out, predictions))
print(f"tiny held-out check: tp={tp} fn={fn} fp={fp}")

# Verify a state-dict round trip without writing a deployment artifact.
buffer = BytesIO()
torch.save(model.state_dict(), buffer)
buffer.seek(0)
restored = TinyEncoderClassifier(len(vocabulary))
restored.load_state_dict(torch.load(buffer, weights_only=True))
restored.eval()
with torch.inference_mode():
    print("restored logits match:", torch.equal(final_logits, restored(inputs)))
    original = model(encode(["prod key disabled incident freeze"]))
    reordered = model(encode(["freeze incident disabled key prod"]))
    print("reordered tokens give the same scores:", torch.allclose(original, reordered))

uv init --bare --python 3.12 ticket-classifier
cd ticket-classifier
uv add torch 'transformers==5.17.0' datasets accelerate numpy

import hashlib
import json
from importlib.metadata import version
from pathlib import Path
import numpy as np
from datasets import load_dataset
from transformers import (
    AutoModelForSequenceClassification,
    AutoTokenizer,
    DataCollatorWithPadding,
    Trainer,
    TrainingArguments,
    set_seed,
)

checkpoint = "distilbert/distilbert-base-uncased"
revision = "12040accade4e8a0f71eabdb258fecc2e7e948be"
run_root = Path("artifacts/encoder_v1")
if run_root.exists():
    raise FileExistsError("Use a new versioned directory for each training run")
set_seed(7)  # before initializing the new classification head
tokenizer = AutoTokenizer.from_pretrained(checkpoint, revision=revision)
development_files = {name: f"data/{name}.jsonl" for name in ("train", "validation")}
# Check original JSON values before a dataset library can coerce column types.
development_rows = {
    name: [json.loads(line) for line in Path(filename).read_text().splitlines() if line.strip()]
    for name, filename in development_files.items()
}
seen_ids = set()
seen_text = set()
for partition in development_rows.values():
    for row in partition:
        if type(row) is not dict or not {"fixture_id", "text", "label"}.issubset(row):
            raise ValueError("Missing development row fields")
        if type(row["label"]) is not int or row["label"] not in (0, 1) or type(row["text"]) is not str or not row["text"].strip():
            raise ValueError("Invalid label or blank text")
        fixture_id = row["fixture_id"]
        if type(fixture_id) is not str or not fixture_id.strip() or fixture_id != fixture_id.strip():
            raise ValueError("Invalid fixture ID")
        if row["fixture_id"] in seen_ids:
            raise ValueError("Repeated fixture ID across development data")
        seen_ids.add(row["fixture_id"])
        normalized = " ".join(row["text"].lower().split())
        if normalized in seen_text:
            raise ValueError("Repeated normalized text in development data")
        seen_text.add(normalized)
    if {row["label"] for row in partition} != {0, 1}:
        raise ValueError("Each development partition must contain both classes")
dataset = load_dataset("json", data_files=development_files)

def tokenize(batch):
    encoded = tokenizer([text.strip() for text in batch["text"]], truncation=False)
    if any(len(ids) > 256 for ids in encoded["input_ids"]):
        raise ValueError("Oversize ticket requires the manual-input policy")
    return encoded

encoded = dataset.map(tokenize, batched=True)
model = AutoModelForSequenceClassification.from_pretrained(
    checkpoint,
    revision=revision,
    num_labels=2,
    id2label={0: "guarded_agent", 1: "human_review_now"},
    label2id={"guarded_agent": 0, "human_review_now": 1},
)
def compute_metrics(eval_pred):
    logits, labels = eval_pred
    predicted = np.argmax(logits, axis=-1)
    tp = np.sum((predicted == 1) & (labels == 1))
    fn = np.sum((predicted == 0) & (labels == 1))
    fp = np.sum((predicted == 1) & (labels == 0))
    return {"recall_at_argmax": float(tp / (tp + fn)), "false_positives_at_argmax": int(fp)}

arguments = TrainingArguments(
    output_dir="artifacts/encoder_v1",
    learning_rate=2e-5,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=32,
    num_train_epochs=3,
    eval_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    greater_is_better=False,
    seed=7,
    data_seed=7,
    report_to=[],
    push_to_hub=False,
)
trainer = Trainer(
    model=model,
    args=arguments,
    train_dataset=encoded["train"],
    eval_dataset=encoded["validation"],
    processing_class=tokenizer,
    data_collator=DataCollatorWithPadding(tokenizer=tokenizer),
    compute_metrics=compute_metrics,
)
trainer.train()
print(trainer.evaluate(encoded["validation"]))
export = Path("artifacts/encoder_v1/model")
trainer.save_model(str(export))
tokenizer.save_pretrained(str(export))
files = {
    path.name: hashlib.sha256(path.read_bytes()).hexdigest()
    for path in sorted(export.iterdir()) if path.is_file()
}
manifest = {
    "base_checkpoint": checkpoint, "base_revision": revision,
    "selected_checkpoint": trainer.state.best_model_checkpoint,
    "label2id": model.config.label2id, "max_length": 256,
    "seed": 7, "sha256": files,
    "input_policy_version": "access-intake-input-v1",
    "packages": {name: version(name) for name in ("torch", "transformers", "datasets", "accelerate", "numpy")},
    "development_files": {
        name: hashlib.sha256(Path(f"data/{name}.jsonl").read_bytes()).hexdigest()
        for name in ("train", "validation")
    },
}
Path("artifacts/encoder_v1/export-manifest.json").write_text(json.dumps(manifest, indent=2))

import hashlib
import json
from importlib.metadata import version
from pathlib import Path
import re
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

export = Path("artifacts/encoder_v1/model")
manifest_path = Path("artifacts/encoder_v1/export-manifest.json")
manifest = json.loads(manifest_path.read_text())
if (
    type(manifest) is not dict
    or manifest.get("label2id") != {"guarded_agent": 0, "human_review_now": 1}
    or type(manifest.get("max_length")) is not int
    or manifest["max_length"] != 256
    or manifest.get("input_policy_version") != "access-intake-input-v1"
):
    raise ValueError("Unsupported scoring policy")
digests = manifest.get("sha256")
if type(digests) is not dict or not {"config.json", "model.safetensors", "tokenizer.json"}.issubset(digests):
    raise ValueError("Incomplete export digest inventory")
for name, digest in digests.items():
    if (
        type(name) is not str or not name or name in (".", "..")
        or "/" in name or "\\" in name
        or type(digest) is not str or re.fullmatch(r"[0-9a-f]{64}", digest) is None
    ):
        raise ValueError("Invalid export digest entry")
if any(not path.is_file() or path.is_symlink() for path in export.iterdir()):
    raise ValueError("Expected a flat export without symbolic links")
if {path.name for path in export.iterdir()} != set(digests):
    raise ValueError("Export file inventory changed")
for name, expected in manifest["sha256"].items():
    if hashlib.sha256((export / name).read_bytes()).hexdigest() != expected:
        raise ValueError(f"Export mismatch: {name}")
tokenizer = AutoTokenizer.from_pretrained(export, local_files_only=True)
classifier = AutoModelForSequenceClassification.from_pretrained(export, local_files_only=True)
if (
    classifier.config.label2id != {"guarded_agent": 0, "human_review_now": 1}
    or classifier.config.id2label != {0: "guarded_agent", 1: "human_review_now"}
    or classifier.config.num_labels != 2
):
    raise ValueError("Positive-class mapping changed")
classifier.eval()
fixture_path = Path("gate-fixtures.json")
fixtures = json.loads(fixture_path.read_text())
if type(fixtures) is not list or not fixtures:
    raise ValueError("Expected a nonempty gate fixture list")
seen_ids = set()
for fixture in fixtures:
    if type(fixture) is not dict or set(fixture) != {"fixture_id", "text"}:
        raise ValueError("Expected only fixture ID and ticket text")
    fixture_id = fixture["fixture_id"]
    if type(fixture_id) is not str or not fixture_id.strip() or fixture_id != fixture_id.strip():
        raise ValueError("Invalid gate fixture ID")
    if fixture_id in seen_ids:
        raise ValueError("Repeated gate fixture ID")
    seen_ids.add(fixture_id)
    if type(fixture["text"]) is not str or not fixture["text"].strip():
        raise ValueError("Invalid gate input text")
scores = []
for fixture in fixtures:
    text = fixture["text"].strip()
    if not text:
        raise ValueError("Blank gate input")
    encoded = tokenizer(text, return_tensors="pt", truncation=False)
    if encoded["input_ids"].shape[1] > manifest["max_length"]:
        raise ValueError("Gate input exceeds serving policy; route manually")
    with torch.inference_mode():
        score = torch.softmax(classifier(**encoded).logits, dim=-1)[0, 1].item()
    scores.append({"fixture_id": fixture["fixture_id"], "score": score})
Path("gate-scores.json").write_text(json.dumps(scores, indent=2, allow_nan=False))
Path("gate-scoring-metadata.json").write_text(json.dumps({
    "fixtures_sha256": hashlib.sha256(fixture_path.read_bytes()).hexdigest(),
    "manifest_sha256": hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
    "input_policy_version": manifest["input_policy_version"],
    "max_length": manifest["max_length"],
    "packages": {name: version(name) for name in ("torch", "transformers")},
}, indent=2))

# Selection set: used only to propose a threshold. Not the release gate.
selection_rows = [
    ("access_override_exception", 1, 0.89),
    ("incident_freeze_exception", 1, 0.42),
    ("account_takeover", 1, 0.64),
    ("angry_key_rotation", 0, 0.71),
    ("routine_key_rotation", 0, 0.23),
    ("access_policy_question", 0, 0.18),
]

def summarize(threshold: float) -> tuple[int, int, int, float]:
    if type(threshold) not in (int, float) or not 0 <= threshold <= 1:
        raise ValueError("Expected a numeric threshold in [0, 1]")
    tp = fp = fn = 0
    for _, gold, score in selection_rows:
        predicted = int(score >= threshold)
        tp += predicted == 1 and gold == 1
        fp += predicted == 1 and gold == 0
        fn += predicted == 0 and gold == 1
    recall = tp / (tp + fn)
    return tp, fp, fn, recall

for threshold in (0.35, 0.50, 0.75):
    tp, fp, fn, recall = summarize(threshold)
    print(f"threshold={threshold:.2f} tp={tp} fp={fp} fn={fn} recall={recall:.2f}")

from collections import Counter
from dataclasses import dataclass
from types import MappingProxyType

EXPECTED_IDENTITY = {
    "dataset_version": "access-intake-gate-v1",
    "label_version": "escalation-policy-v1",
    "split_manifest_version": "access-intake-split-2026-05",
    "model_version": "encoder_v1",
    "threshold_selection_set": "access-intake-selection-v1",
    "gate_policy_version": "access-intake-gate-policy-v2",
}
# The frozen gate receipt uses fixture IDs that never appeared in the threshold
# selection set above. It is scored only after the threshold is locked.
SELECTION_FIXTURES = {
    "access_override_exception",
    "incident_freeze_exception",
    "account_takeover",
    "angry_key_rotation",
    "routine_key_rotation",
    "access_policy_question",
}
TRUSTED_BUNDLES = MappingProxyType({
    "intake_bundle_v1": MappingProxyType({"threshold": 0.50}),
    "intake_bundle_v2": MappingProxyType({"threshold": 0.35}),
})

@dataclass(frozen=True)
class FixtureSpec:
    text: str
    slice: str
    gold: int

# This in-process fixture mapping is read-only. That is not authentication:
# the production evaluator and scoring worker need separate trust boundaries.
AUTHORITATIVE_FIXTURES = MappingProxyType({
    "after_hours_access_override": FixtureSpec(
        "The production API key is disabled outside business hours. Please make an exception.",
        "access_override_exception",
        1,
    ),
    "freeze_window_ticket": FixtureSpec(
        "The production API key was disabled during an incident freeze. Please restore it.",
        "incident_freeze_exception",
        1,
    ),
    "takeover_recovery_request": FixtureSpec(
        "Someone changed the recovery email and I can't sign in.",
        "account_takeover",
        1,
    ),
    "routine_rotation_followup": FixtureSpec(
        "The key rotation guide failed. Can I see it again?",
        "routine_key_rotation",
        0,
    ),
    "routine_access_question": FixtureSpec(
        "Which document explains who may rotate a service key?",
        "access_policy_question",
        0,
    ),
    "routine_status_request": FixtureSpec(
        "Can you tell me whether the key rotation completed?",
        "routine_key_rotation",
        0,
    ),
})
EXPECTED_FIXTURES = frozenset(AUTHORITATIVE_FIXTURES)
REQUIRED_SLICES = frozenset(
    spec.slice for spec in AUTHORITATIVE_FIXTURES.values() if spec.gold == 1
)
MAX_FALSE_POSITIVES = 1  # queue capacity for this teaching receipt
# Production scores must come from the pinned model runner. The values below
# are synthetic fixtures for the evaluator, not output from a trained model.
SUBMITTED_SCORE_FIELDS = frozenset({"fixture_id", "score"})
RECEIPT_FIELDS = frozenset(EXPECTED_IDENTITY) | {"bundle_version", "threshold", "rows"}
gate_scores = [
    {"fixture_id": "after_hours_access_override", "score": 0.87},
    {"fixture_id": "freeze_window_ticket", "score": 0.41},
    {"fixture_id": "takeover_recovery_request", "score": 0.62},
    {"fixture_id": "routine_rotation_followup", "score": 0.69},
    {"fixture_id": "routine_access_question", "score": 0.21},
    {"fixture_id": "routine_status_request", "score": 0.14},
]
assert REQUIRED_SLICES == {
    "access_override_exception",
    "incident_freeze_exception",
    "account_takeover",
}

def receipt(bundle_version: str, rows: list[dict[str, object]], **overrides: object) -> dict[str, object]:
    return {
        **EXPECTED_IDENTITY,
        "bundle_version": bundle_version,
        "threshold": TRUSTED_BUNDLES[bundle_version]["threshold"],
        "rows": rows,
        **overrides,
    }

def release_decision(run: dict[str, object]) -> tuple[str, str]:
    if type(run) is not dict:
        return "hold", "invalid:receipt"
    if set(run) != RECEIPT_FIELDS:
        return "hold", "invalid:receipt_fields"
    for field, expected in EXPECTED_IDENTITY.items():
        actual = run.get(field)
        if actual != expected:
            return "hold", f"{field}:{actual}"
    bundle_version = run.get("bundle_version")
    if type(bundle_version) is not str:
        return "hold", f"bundle_version:{bundle_version}"
    bundle = TRUSTED_BUNDLES.get(bundle_version)
    if bundle is None:
        return "hold", f"bundle_version:{bundle_version}"
    # A submitted threshold is metadata, not policy. Use the evaluator's
    # pinned bundle manifest for every comparison and hold a report that disagrees.
    if run.get("threshold") != bundle["threshold"]:
        return "hold", f"threshold:{run.get('threshold')}"
    rows = run.get("rows")
    if type(rows) is not list or any(type(row) is not dict for row in rows):
        return "hold", "invalid:rows"

    scores_by_fixture = {}
    fixture_ids = []
    for row in rows:
        if set(row) != SUBMITTED_SCORE_FIELDS:
            fields = sorted(str(field) for field in set(row) ^ SUBMITTED_SCORE_FIELDS)
            return "hold", f"invalid:submitted_fields:{','.join(fields)}"
        fixture_id = row["fixture_id"]
        if type(fixture_id) is not str or not fixture_id.strip():
            return "hold", "invalid:fixture_id"
        if fixture_id in SELECTION_FIXTURES:
            return "hold", f"selection_reuse:{fixture_id}"
        if fixture_id not in AUTHORITATIVE_FIXTURES:
            return "hold", f"unexpected:{fixture_id}"
        raw_score = row["score"]
        # JSON numbers only: booleans and numeric-looking strings are not scores.
        if type(raw_score) not in (int, float) or not 0.0 <= raw_score <= 1.0:
            return "hold", f"invalid:score:{fixture_id}"
        score = float(raw_score)
        fixture_ids.append(fixture_id)
        scores_by_fixture[fixture_id] = score

    counts = Counter(fixture_ids)
    missing = sorted(EXPECTED_FIXTURES - set(fixture_ids))
    unexpected = sorted(set(fixture_ids) - EXPECTED_FIXTURES)
    duplicated = sorted(fixture_id for fixture_id, count in counts.items() if count != 1)
    if missing:
        return "hold", f"missing:{','.join(missing)}"
    if unexpected:
        return "hold", f"unexpected:{','.join(unexpected)}"
    if duplicated:
        return "hold", f"duplicate:{','.join(duplicated)}"
    threshold = bundle["threshold"]
    misses = [
        fixture_id
        for fixture_id in fixture_ids
        if AUTHORITATIVE_FIXTURES[fixture_id].slice in REQUIRED_SLICES
        and AUTHORITATIVE_FIXTURES[fixture_id].gold == 1
        and scores_by_fixture[fixture_id] < threshold
    ]
    if misses:
        return "hold", f"missed:{','.join(misses)}"
    false_positives = sum(
        1
        for fixture_id, spec in AUTHORITATIVE_FIXTURES.items()
        if spec.gold == 0 and scores_by_fixture[fixture_id] >= threshold
    )
    if false_positives > MAX_FALSE_POSITIVES:
        return "hold", f"false_positives:{false_positives}>max:{MAX_FALSE_POSITIVES}"
    return "eligible_for_shadow", "exact_receipt_pass"

runs = {
    "intake_bundle_v1": receipt("intake_bundle_v1", gate_scores),
    "intake_bundle_v2": receipt("intake_bundle_v2", gate_scores),
    "intake_bundle_incomplete": receipt(
        "intake_bundle_v2",
        [row for row in gate_scores if row["fixture_id"] != "takeover_recovery_request"],
    ),
    "intake_bundle_padded": receipt(
        "intake_bundle_v2",
        [*gate_scores, {"fixture_id": "easy_status_extra", "score": 0.02}],
    ),
    "intake_bundle_duplicated": receipt("intake_bundle_v2", [*gate_scores, gate_scores[-1]]),
    "intake_bundle_drifted": receipt("intake_bundle_v2", gate_scores, dataset_version="access-intake-gate-v2"),
    "intake_bundle_wrong_threshold": receipt("intake_bundle_v2", gate_scores, threshold=0.20),
    "intake_bundle_tampered_labels": receipt(
        "intake_bundle_v2",
        [
            {**row, "slice": "routine_key_rotation", "gold": 0}
            if row["fixture_id"] == "freeze_window_ticket"
            else row
            for row in gate_scores
        ],
    ),
    "intake_bundle_invalid_score": receipt(
        "intake_bundle_v2",
        [*gate_scores[:-1], {"fixture_id": "routine_status_request", "score": float("nan")}],
    ),
    "intake_bundle_string_score": receipt(
        "intake_bundle_v2",
        [*gate_scores[:-1], {"fixture_id": "routine_status_request", "score": "0.14"}],
    ),
    "intake_bundle_fp_flood": receipt(
        "intake_bundle_v2",
        [
            {
                **row,
                "score": (
                    0.90
                    if AUTHORITATIVE_FIXTURES[row["fixture_id"]].gold == 0
                    else row["score"]
                ),
            }
            for row in gate_scores
        ],
    ),
}
for bundle_version, run in runs.items():
    decision, reason = release_decision(run)
    print(f"{bundle_version}: decision={decision} reason={reason}")

invented_scores = [
    {"fixture_id": fixture_id, "score": float(spec.gold)}
    for fixture_id, spec in AUTHORITATIVE_FIXTURES.items()
]
invented_decision, _ = release_decision(receipt("intake_bundle_v2", invented_scores))
print("invented scores pass arithmetic:", invented_decision == "eligible_for_shadow")

from math import exp, log

scored_rows = [
    (0.89, 1),
    (0.64, 1),
    (0.42, 1),
    (0.71, 0),
    (0.23, 0),
    (0.18, 0),
]
buckets = {
    "low [0.0,0.4)": [(score, gold) for score, gold in scored_rows if score < 0.4],
    "middle [0.4,0.7)": [(score, gold) for score, gold in scored_rows if 0.4 <= score < 0.7],
    "high [0.7,1.0]": [(score, gold) for score, gold in scored_rows if score >= 0.7],
}

for name, bucket in buckets.items():
    if not bucket:
        print(f"{name}: no observations")
        continue
    average_score = sum(score for score, _ in bucket) / len(bucket)
    observed_rate = sum(gold for _, gold in bucket) / len(bucket)
    print(f"{name}: n={len(bucket)}, score={average_score:.2f}, observed={observed_rate:.2f}")
print("decision: diagnostic only; collect more held-out labels")

# Illustrate a supplied positive temperature, not a fitted calibrator.
before = 0.30
after = 1 / (1 + exp(-log(before / (1 - before)) / 2))
print("argmax stays routine:", before < 0.5 and after < 0.5)
print("route changes at 0.35:", before < 0.35 <= after)

from types import MappingProxyType

SERVING_POLICY = MappingProxyType({
    "bundle_version": "intake_bundle_v2",
    "label_version": "escalation-policy-v1",
    "gate_receipt_version": "access-intake-gate-v1",
    "split_manifest_version": "access-intake-split-2026-05",
    "tokenizer_version": "encoder_v1",
    "model_version": "encoder_v1",
    "input_policy_version": "access-intake-input-v1",
    "max_length": 256,
    "threshold": 0.35,
    "fallback_route": "human_review_now",
})
bundle = dict(SERVING_POLICY)

def route_ticket(text: object, score: object, token_count: object) -> tuple[str, str]:
    if (
        type(bundle) is not dict
        or set(bundle) != set(SERVING_POLICY)
        or any(type(bundle[key]) is not type(expected) or bundle[key] != expected
               for key, expected in SERVING_POLICY.items())
    ):
        return "human_review_now", "invalid_bundle"
    if not isinstance(text, str):
        return bundle["fallback_route"], "invalid_text"
    if not text.strip():
        return bundle["fallback_route"], "empty_text"
    if type(token_count) is not int or token_count <= 0:
        return bundle["fallback_route"], "invalid_token_count"
    if token_count > bundle["max_length"]:
        return bundle["fallback_route"], "input_too_long"
    if score is None:
        return bundle["fallback_route"], "model_unavailable"
    if type(score) not in (int, float) or not 0.0 <= score <= 1.0:
        return bundle["fallback_route"], "invalid_score"
    if score >= bundle["threshold"]:
        return "human_review_now", "threshold"
    return "guarded_agent", "below_threshold"

checks = [
    ("Production API key disabled during the incident freeze", 0.72, 12, "human_review_now", "threshold"),
    ("Where is the access policy?", 0.18, 8, "guarded_agent", "below_threshold"),
    ("", 0.04, 2, "human_review_now", "empty_text"),
    ("Where is the key rotation guide?", None, 9, "human_review_now", "model_unavailable"),
    ("Malformed model score", float("nan"), 5, "human_review_now", "invalid_score"),
    ("Boolean is not a score", False, 8, "human_review_now", "invalid_score"),
    ("Oversize ticket fixture", 0.01, 257, "human_review_now", "input_too_long"),
]
print("bundle:", bundle["bundle_version"], bundle["gate_receipt_version"])
for text, score, token_count, expected_route, expected_reason in checks:
    route, reason = route_ticket(text, score, token_count)
    assert (route, reason) == (expected_route, expected_reason)
    print(f"{route}: {reason}")

handoff = {
    "producer": "support_ticket_escalation_classifier",
    "bundle_version": "intake_bundle_v2",
    "model_version": "encoder_v1",
    "threshold": 0.35,
    "allowed_agent_route": "guarded_agent",
    "blocked_or_manual_route": "human_review_now",
    "policy_evidence_service": "document_qa_v2",
    "evaluation_report": "classifier_dashboard_run_intake_bundle_v2",
}

agent_inputs = [
    {"ticket_id": "r-104", "bundle_version": "intake_bundle_v2", "route": "guarded_agent"},
    {"ticket_id": "r-105", "bundle_version": "intake_bundle_v2", "route": "human_review_now"},
    {"ticket_id": "r-106", "bundle_version": "intake_bundle_v1", "route": "guarded_agent"},
]

def admitted_by_pinned_bundle(row: dict[str, str]) -> bool:
    return (
        isinstance(row, dict)
        and isinstance(row.get("ticket_id"), str)
        and bool(row["ticket_id"].strip())
        and row.get("bundle_version") == handoff["bundle_version"]
        and row.get("route") == handoff["allowed_agent_route"]
    )

accepted = [
    row["ticket_id"]
    for row in agent_inputs
    if admitted_by_pinned_bundle(row)
]
blocked = [
    row["ticket_id"]
    for row in agent_inputs
    if not admitted_by_pinned_bundle(row)
]

print("agent accepts:", ",".join(accepted))
print("agent blocked:", ",".join(blocked))
print("agent authority: draft_with_evidence_and_request_approval")
forged = {"ticket_id": "invented-ticket", "bundle_version": "intake_bundle_v2", "route": "guarded_agent"}
print("matching fabricated artifact passes checks:", admitted_by_pinned_bundle(forged))