"""Protocol tests; synthetic report fixtures never qualify model inference.""" import copy import json from pathlib import Path import pytest from hessian_recipes import gym_experiment_evaluation as experiment def write(path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False) + "\n") def rows(path, values): path.write_text("".join(json.dumps(value) + "\n" for value in values)) @pytest.fixture def prepared(tmp_path): baseline, tokenizer = tmp_path / "baseline", tmp_path / "tokenizer" baseline.mkdir() tokenizer.mkdir() (baseline / "config.json").write_text("{}") (baseline / "model.safetensors").write_bytes(b"not-a-real-model") (tokenizer / "chat_template.jinja").write_text("{{ messages }}") train, dev, mgsm = (tmp_path / name for name in ("train.jsonl", "dev.jsonl", "mgsm.jsonl")) rows(train, [{"id": "train0", "source_id": "gsm8k/train/00000"}]) rows( dev, [ { "id": f"dev{i}", "source_id": f"gsm8k/train/{i + 1000:05d}", "prompt": f"Aufgabe {i}", "solution": str(i), } for i in range(128) ], ) rows( mgsm, [ { "id": f"mgsm-de-test-{i:03d}", "source_id": f"mgsm/de/test/{i:03d}", "prompt": f"Prüfung {i}", "answer": str(i), "scorer": "final_numeric", "category": "German math", } for i in range(250) ], ) output = tmp_path / "evaluation-plan" plan = experiment.prepare(baseline, tokenizer, dev, mgsm, output) checkpoints = {} for step in experiment.CHECKPOINTS: checkpoint = tmp_path / f"model{step}" checkpoint.mkdir() (checkpoint / "config.json").write_text(str(step)) checkpoints[step] = checkpoint return output / "plan.json", plan, baseline, checkpoints def test_prepare_freezes_new_templates_splits_and_real_evaluator_specs(prepared): path, plan, baseline, _ = prepared dev, test = [ experiment._rows(plan["datasets"][f"agent_{split}"]["path"])[0] for split in ("dev", "test") ] assert len(dev) == 64 and len(test) == 256 assert len({row["task"]["template_id"] for row in dev}) == 4 assert len({row["task"]["template_id"] for row in test}) == 8 assert not ({row["id"] for row in dev} & {row["id"] for row in test}) assert all("20261020" in row["id"] and "Werkzeugprotokoll:" in row["prompt"] for row in test) assert set(experiment.DEV_TEMPLATES).isdisjoint(experiment.TEST_TEMPLATES) assert plan["language_audit"]["agent_ids"] == [row["id"] for row in test[:10]] entries = experiment.write_specs(path, baseline, "baseline") for entry in entries: if entry["kind"] == "agent": spec, samples, _ = experiment.agent._load(entry["spec_path"]) else: spec, samples, _ = experiment.numeric.load_spec(entry["spec_path"]) assert len(samples) in (64, 128) and spec["sampling"]["seed"] == 42 with pytest.raises(ValueError, match="selection first"): experiment.write_specs(path, baseline, "premature", "test") def test_source_overlap_and_changed_protocol_fail_before_evaluation(prepared): path, plan, baseline, _ = prepared dev_path = Path(plan["source_inputs"]["math_dev"]["path"]) data, _ = experiment._rows(dev_path) data[0]["source_id"] = "gsm8k/train/00000" rows(dev_path, data) with pytest.raises(ValueError, match="overlaps"): experiment.prepare( baseline, plan["tokenizer"]["path"], dev_path, plan["source_inputs"]["mgsm"]["path"], path.parent.parent / "bad", ) with pytest.raises(ValueError, match="input changed"): experiment.verify_plan(path) @pytest.fixture def evidence(prepared, monkeypatch): # Expensive native verifiers have independent replay tests. These test the # new selection/protocol layer with explicitly fabricated evidence only. monkeypatch.setattr(experiment.agent, "verify_report", lambda report: None) monkeypatch.setattr(experiment.numeric, "_verify_report", lambda report: None) path, plan, baseline, checkpoints = prepared def suite( checkpoint, label, *, split="dev", selection=None, agent_correct=10, math_correct=80, errors=5, ): entries = experiment.write_specs(path, checkpoint, label, split, selection_path=selection) for entry in entries: kind, seed = entry["kind"], entry["seed"] spec, _ = experiment._read(entry["spec_path"]) data, _ = experiment._rows(plan["datasets"][f"{kind}_{split}"]["path"]) correct = agent_correct if kind == "agent" else math_correct if isinstance(correct, dict): correct = correct[seed] provenance = {"spec": spec, "packages": {"unit-test": "fixture"}} if kind == "agent": provenance.update( effective_template_sha256="template", tool_schemas=["tool"], sources={"native": "fixed"}, runtime={"device": "cuda", "dtype": "bfloat16"}, tasks_snapshot=Path(plan["datasets"][f"agent_{split}"]["path"]).read_text(), ) report = { "source_kind": "captured", "provenance": provenance, "episodes": [ { "id": row["id"], "trial": 0, "seed": (seed + int(experiment._sha(row["id"].encode())[:8], 16)) % (2**32), "wall_seconds": 0.01, "outcome": { "success": i < correct, "workflow_valid": i < correct, "final_answer": str( row["task"]["private_facts"]["quantity"] * row["task"]["private_facts"]["unit_price"] ) if i < correct else None, }, } for i, row in enumerate(data) ], "metrics": { "episodes": len(data), "successful": correct, "tool_calls": 10, "tool_errors": errors, "malformed_final_outputs": len(data) - correct, "turn_cap_episodes": 1, "token_cap_episodes": 1, "assistant_tokens": 100, "prompt_tokens": 1000, "completion_stream_tokens": 130, "wall_seconds": 1.0, }, } else: provenance.update(execution_mode="transformers", scorer_sha256="scorer") report = { "provenance": provenance, "results": [ { "id": row["id"], "trial": 0, "outcome": "correct" if i < correct else "incorrect", "output": row["answer"] if i < correct else "malformed", "usage": { "prompt_tokens": 5, "completion_tokens": 768 if i == 0 else 10, }, "wall_seconds": 0.1, } for i, row in enumerate(data) ], } write(Path(entry["path"]), report) return Path(entries[0]["spec_path"]).parent / "manifest.json" return path, plan, baseline, checkpoints, suite def dev_manifests(evidence, *, all_fail=False): path, plan, baseline, checkpoints, suite = evidence base = suite(baseline, "baseline-dev") candidates = { 64: suite(checkpoints[64], "step64-dev", agent_correct=20, math_correct=79), 128: suite( checkpoints[128], "step128-dev", agent_correct=15 if not all_fail else 10, math_correct=80, ), 256: suite( checkpoints[256], "step256-dev", agent_correct=15 if not all_fail else 10, math_correct=80, ), } output = path.parent.parent / "selection" decision = experiment.select(path, base, candidates, output) return decision, output / "selection.json", base, candidates def test_dev_selection_rejects_math_drop_and_uses_earliest_tie_without_mgsm(evidence): path, plan, _, _, _ = evidence # An unrelated MGSM result can never enter this function's candidate inputs. write(path.parent.parent / "unused-MGSM-report.json", {"candidate_accuracy": 1.0}) decision, _, _, _ = dev_manifests(evidence) assert decision["selected_step"] == 128 and decision["dev_acceptance"] assert decision["candidates"][0]["admissible"] is False assert all("/dev/" in row["path"] for row in decision["baseline_dev"]["evidence"]) assert not decision["retain_baseline"] def test_failed_dev_screen_assesses256_but_recommends_baseline(evidence): _, plan, _, checkpoints, _ = evidence decision, _, _, _ = dev_manifests(evidence, all_fail=True) assert decision["selected_step"] == 256 and decision["selected"]["path"] == str( checkpoints[256] ) assert not decision["dev_acceptance"] and decision["retain_baseline"] assert decision["recommendation"] == plan["baseline"] def test_final_reports_gate_each_seed_even_when_pooled_math_improves(evidence): path, _, baseline, checkpoints, suite = evidence decision, selection, _, _ = dev_manifests(evidence) base = suite( baseline, "baseline-final", split="test", selection=selection, agent_correct=40, math_correct={42: 100, 20261020: 100}, ) candidate = suite( checkpoints[128], "candidate-final", split="test", selection=selection, agent_correct=60, math_correct={42: 99, 20261020: 104}, ) summary = experiment.summarize(path, selection, base, candidate, path.parent.parent / "summary") assert summary["math"]["candidate_correct"] == 203 > summary["math"]["baseline_correct"] assert not summary["observed_zero_drop_gate"]["passed"] assert summary["observed_zero_drop_gate"]["per_seed"] == {"42": False, "20261020": True} assert summary["agent"]["wins"] == 60 and summary["agent"]["losses"] == 0 assert summary["agent"]["items"] == 256 and summary["agent"]["paired_trials"] == 768 assert summary["counts"]["baseline"]["math"]["hit_token_cap_trials"] == 2 assert summary["per_seed_counts"]["baseline"]["math"]["42"]["hit_token_cap_trials"] == 1 assert not summary["overall_acceptance"] and not summary["release_eligible"] assert summary["retain_baseline"] and not summary["dev_retain_baseline"] def test_diagnostic_candidate_cannot_be_promoted_by_final_test(evidence): path, _, baseline, checkpoints, suite = evidence _, selection, _, _ = dev_manifests(evidence, all_fail=True) base = suite( baseline, "baseline-final", split="test", selection=selection, agent_correct=40, math_correct=100, ) candidate = suite( checkpoints[256], "diagnostic-final", split="test", selection=selection, agent_correct=60, math_correct=110, ) summary = experiment.summarize(path, selection, base, candidate, path.parent.parent / "summary") assert summary["observed_joint_goal_met"] and not summary["overall_acceptance"] assert summary["retain_baseline"] and not summary["dev_acceptance"] @pytest.mark.parametrize("fault", ["seed", "missing", "budget", "fixture", "coverage"]) def test_missing_wrong_seed_budget_fixture_or_coverage_fails_closed(evidence, fault): path, plan, baseline, _, suite = evidence manifest = suite(baseline, "baseline-dev") entries, _ = experiment._read(manifest) if fault == "seed": entries[0]["seed"] = 1729 elif fault == "missing": entries.pop() elif fault == "budget": spec, _ = experiment._read(entries[0]["spec_path"]) spec["max_completion_length"] = 1024 write(Path(entries[0]["spec_path"]), spec) else: report, _ = experiment._read(entries[0]["path"]) if fault == "fixture": report["source_kind"] = "fixture" else: report["episodes"].pop() write(Path(entries[0]["path"]), report) write(manifest, entries) with pytest.raises(ValueError): experiment._load_suite(plan, experiment._read(path)[1], manifest, "dev") def test_selection_tamper_or_unselected_model_prevents_final_specs(evidence): path, _, _, checkpoints, _ = evidence decision, selection, _, _ = dev_manifests(evidence) with pytest.raises(ValueError, match="neither baseline nor"): experiment.write_specs( path, checkpoints[64], "not-selected", "test", selection_path=selection ) modified = copy.deepcopy(decision) modified["selected"] = modified["candidates"][0]["model"] write(selection, modified) with pytest.raises(ValueError, match="selection changed"): experiment.write_specs(path, checkpoints[64], "tampered", "test", selection_path=selection) def test_item_bootstrap_keeps_repeated_generation_seeds_in_one_cluster(): a = {(item, seed): False for item in ("a", "b") for seed in (1, 2, 3)} b = {(item, seed): item == "a" for item, seed in a} paired = experiment._paired(a, b, {"seed": 7, "resamples": 1000, "method": "item"}) assert paired["items"] == 2 and paired["paired_trials"] == 6 assert paired["wins"] == 3 and paired["delta"] == 0.5 assert paired["bootstrap"]["ci_95"] == [0, 1]