import json from pathlib import Path import pytest from hessian_recipes.evaluation import checkpoint_sha256, run, sha256_file from hessian_recipes.german_report import _paired, summarize def write(path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value)) def pilot(tmp_path, *, candidate_outputs=None, candidate_budget=8): root = tmp_path / "pilot" root.mkdir() data = root / "evaluation.jsonl" data.write_text( "".join( json.dumps( { "id": str(i), "prompt": "PRIVATE QUESTION " + str(i), "category": "math", "scorer": "final_numeric", "answer": str(i), } ) + "\n" for i in range(1, 5) ) ) (root / "train.jsonl").write_text("PRIVATE TRAINING CONTENT\n") write(root / "sft.json", {"test": "recipe"}) base = root / "base-model" base.mkdir() adapter = root / "training/checkpoint" adapter.mkdir(parents=True) (adapter / "adapter_model.safetensors").write_bytes(b"synthetic test adapter") tokenizer = "c" * 64 revisions = {} for label, outputs, budget in ( ("baseline", ["1", "wrong", "3", "wrong"], 8), ("candidate", candidate_outputs or ["1", "2", "wrong", "wrong"], candidate_budget), ): prediction = root / (label + "-predictions.jsonl") prediction.write_text( "".join( json.dumps( { "id": str(i), "trial": 0, "output": value, "usage": {"prompt_tokens": 3, "completion_tokens": 8 if i == 2 else 1}, } ) + "\n" for i, value in enumerate(outputs, 1) ) ) revisions[label] = sha256_file(prediction) spec = { "schema_version": 1, "benchmark": { "id": "synthetic-german-report-test", "version": "1", "status": "development", "count": 4, "path": "evaluation.jsonl", "sha256": sha256_file(data), }, "model": { "id": label, "revision": revisions[label], "tokenizer_id": "test-tokenizer", "tokenizer_revision": tokenizer, "chat_template": "{{ prompt }}", "chat_template_kwargs": {"enable_thinking": True}, }, "sampling": {"temperature": 0.6, "seed": 42, "max_new_tokens": budget}, "trials": 1, "adapter": {"type": "fixture", "path": prediction.name, "sha256": revisions[label]}, "scorer_version": "1", "harness_version": "1", } write(root / (label + "-spec.json"), spec) run(root / (label + "-spec.json"), root / label) plan = { "model": str(base), "model_sha256": revisions["baseline"], "train_sha256": sha256_file(root / "train.jsonl"), "recipe_sha256": sha256_file(root / "sft.json"), "evaluation_sha256": sha256_file(data), "tokenizer_sha256": tokenizer, "chat_template_kwargs": {"enable_thinking": True}, } write(root / "plan.json", plan) write( root / "run-manifest.json", { "status": "completed", "completed_stages": ["baseline", "sft", "merge", "candidate"], "plan_sha256": sha256_file(root / "plan.json"), }, ) write( root / "training/recipe-manifest.json", { "status": "completed", "method": "sft", "model": {"name": str(base)}, "checkpoint": str(adapter), "dataset_sha256": plan["train_sha256"], "config_sha256": plan["recipe_sha256"], "chat_template_kwargs": plan["chat_template_kwargs"], "training": {"max_steps": 1}, "rows": 2, }, ) write( root / "merged-checkpoint.merge-manifest.json", { "status": "completed", "base_model": {"sha256": revisions["baseline"]}, "output": {"sha256": revisions["candidate"]}, "adapter": {"path": str(adapter), "sha256": checkpoint_sha256(adapter)}, }, ) return root def test_completed_summary_counts_usage_clipping_pairing_and_no_private_text(tmp_path): root = pilot(tmp_path) summary = summarize(root, tmp_path / "summary") assert summary["status"] == "ready" assert summary["quality_gate"] == "passed" assert summary["baseline"]["correct"] == summary["candidate"]["correct"] == 2 assert summary["baseline"]["malformed_outputs"] == 2 assert summary["baseline"]["hit_token_cap_trials"] == 1 assert summary["baseline"]["token_cap_unknown_trials"] == 0 assert summary["baseline"]["usage"]["completion_tokens"] == 11 assert summary["paired"]["candidate_wins"] == summary["paired"]["candidate_losses"] == 1 assert summary["paired"]["accuracy_delta"] == 0 assert ( summary["paired"]["bootstrap"]["ci_95"][0] < 0 < summary["paired"]["bootstrap"]["ci_95"][1] ) assert summary["release_eligible"] is False assert "PRIVATE" not in (tmp_path / "summary/summary.json").read_text() assert "PRIVATE" not in (tmp_path / "summary/summary.md").read_text() assert all(not Path(value["path"]).is_absolute() for value in summary["evidence"].values()) assert summarize(root, tmp_path / "second")["paired"] == summary["paired"] with pytest.raises(ValueError, match="fresh"): summarize(root, tmp_path / "summary") def test_complete_regression_is_reported_without_discarding_run(tmp_path): root = pilot(tmp_path, candidate_outputs=["wrong"] * 4) summary = summarize(root, tmp_path / "summary") assert summary["status"] == "ready" assert summary["quality_gate"] == "regression" assert summary["paired"]["candidate_losses"] == 2 assert summary["paired"]["accuracy_delta"] == -0.5 def test_mismatched_generation_budget_rejected(tmp_path): root = pilot(tmp_path, candidate_budget=9) with pytest.raises(ValueError, match="incomparable"): summarize(root, tmp_path / "summary") assert not (tmp_path / "summary").exists() def test_modified_training_identity_rejected(tmp_path): root = pilot(tmp_path) path = root / "training/recipe-manifest.json" training = json.loads(path.read_text()) training["dataset_sha256"] = "f" * 64 write(path, training) with pytest.raises(ValueError, match="training data identity"): summarize(root, tmp_path / "summary") def test_paired_bootstrap_matches_id_not_output_order_and_groups_trials(): baseline = { "results": [{"id": "a", "trial": trial, "outcome": "incorrect"} for trial in range(2)] + [{"id": "b", "trial": trial, "outcome": "correct"} for trial in range(2)] } candidate = { "results": [{"id": "b", "trial": trial, "outcome": "incorrect"} for trial in range(2)] + [{"id": "a", "trial": trial, "outcome": "correct"} for trial in range(2)] } paired = _paired(baseline, candidate) assert paired["items"] == 2 and paired["trials"] == 4 assert paired["candidate_wins"] == paired["candidate_losses"] == 2 assert paired["bootstrap"]["ci_95"] == [-1, 1] candidate["results"].pop() with pytest.raises(ValueError, match="identities"): _paired(baseline, candidate) def test_tampered_output_rejected_by_existing_evidence_verifier(tmp_path): root = pilot(tmp_path) path = root / "candidate/report.json" report = json.loads(path.read_text()) report["results"][0]["output"] = "999" write(path, report) with pytest.raises(ValueError, match="re-scored"): summarize(root, tmp_path / "summary") def test_numeric_wrong_answers_distinguished_from_malformed_outputs(tmp_path): root = pilot(tmp_path, candidate_outputs=["1", "2", "77", "wrong"]) summary = summarize(root, tmp_path / "summary") assert summary["candidate"]["correct"] == 2 assert summary["candidate"]["malformed_outputs"] == 1 assert summary["candidate"]["wrong_accepted_answers"] == 1