"""Verify and summarize a completed bounded agent-RL pilot without copying raw outputs.""" from __future__ import annotations import argparse from collections import defaultdict import hashlib import json import math from pathlib import Path import re import statistics import subprocess from . import agent_rl_evaluation as agent_eval from .agent_rl_pilot import require_learning_update, verify_inputs from .agent_rl_tasks import GermanAgentEnv, score_episode from .evaluation import _verify_report, checkpoint_sha256, compare as compare_math, sha256_file from .german_report import _counts as math_counts, _paired as math_paired def _check(condition, message): if not condition: raise ValueError(message) def _read(path): raw = Path(path).read_bytes() return json.loads(raw), hashlib.sha256(raw).hexdigest() def _jsonl(path): return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()] def _audit_training(manifest, root): """Re-execute trusted tools, recompute rewards, and prove sampled IDs/context/masks.""" records = _jsonl(root / "training/rollout-evidence.jsonl") tasks = {row["id"]: row["task"] for row in _jsonl(root / "train.jsonl")} groups, totals = ( defaultdict(list), { "rollouts": 0, "sampled_spans": 0, "assistant_tokens": 0, "external_tokens": 0, "tool_calls": 0, }, ) for record in records: task = tasks[record["task_id"]] _check( record["split"] == "train" and task["episode_seed"] == record["episode_seed"], "training task/seed identity mismatch", ) env = GermanAgentEnv() env.reset(task=task) messages = record["completion_messages"] for index, message in enumerate(messages): if message["role"] != "assistant" or not message.get("tool_calls"): continue observations = [] for following in messages[index + 1 :]: if following["role"] != "tool": break observations.append(following) if not observations: # Native tool-iteration limit may leave a final unexecuted call. continue _check( len(observations) == len(message["tool_calls"]), "native tool observation coverage mismatch", ) for call, observation in zip(message["tool_calls"], observations, strict=True): name = call.get("function", {}).get("name", "unknown") try: _check(call["type"] == "function", "unsupported tool type") if name not in ("lookup", "calculator"): raise ValueError(f"Tool {name} not found.") result = {"lookup": env.lookup, "calculator": env.calculator}[name]( **call["function"]["arguments"] ) except Exception as error: result = {"error": str(error)} _check( observation["name"] == name and observation["content"] == str(result), "native tool observation replay mismatch", ) totals["tool_calls"] += 1 score = score_episode(env, messages) _check(score == record["score"], "training strict outcome/trace mismatch") reward = max( 0.0, min( 1.0, 0.15 * score["lookup_success"] + 0.25 * score["calculation_success"] + 0.60 * score["success"] - 0.05 * sum(not row["ok"] for row in score["trace"]), ), ) _check( math.isclose(reward, record["training_reward"], abs_tol=1e-12), "training milestone reward mismatch", ) _check( record["reward_components"]["training_reward"] == record["training_reward"], "stored reward components mismatch", ) audit = record["policy_audit"] prompt, completion, mask = audit["prompt_ids"], audit["completion_ids"], audit["tool_mask"] _check( len(mask) == len(completion) <= manifest["training"]["max_completion_length"], "retained training stream budget mismatch", ) full, proven = prompt + completion, [False] * len(completion) for span in audit["sampled_spans"]: offset = len(span["prompt_ids"]) - len(prompt) retained = min(len(span["sampled_completion_ids"]), len(completion) - offset) _check( 0 <= offset < len(completion) and retained == span["retained_tokens"] and offset == span["offset"], "sampled span boundary mismatch", ) _check( full[: len(span["prompt_ids"])] == span["prompt_ids"] and completion[offset : offset + retained] == span["sampled_completion_ids"][:retained], "sampled token IDs or actual conditioning context mismatch", ) proven[offset : offset + retained] = [True] * retained totals["sampled_spans"] += 1 _check( all(type(bit) is int and bit in (0, 1) for bit in mask) and [bool(bit) for bit in mask] == proven, "tool/assistant training loss mask mismatch", ) _check( audit["assistant_tokens"] == sum(mask) and audit["external_tokens"] == len(mask) - sum(mask), "token audit aggregate mismatch", ) totals["rollouts"] += 1 totals["assistant_tokens"] += sum(mask) totals["external_tokens"] += len(mask) - sum(mask) groups[record["optimizer_step"]].append(record) _check( len(groups) == manifest["training"]["max_steps"] and sorted(groups) == list(range(len(groups))), "optimizer group coverage mismatch", ) _check(len(manifest["reward_groups"]) == len(groups), "reward-group evidence missing") for group in manifest["reward_groups"]: rows = sorted(groups[group["optimizer_step"]], key=lambda row: row["rollout_index"]) rewards = [row["training_reward"] for row in rows] _check( len(rows) == manifest["training"]["num_generations"] and len({row["task_id"] for row in rows}) == 1, "GRPO same-task group identity mismatch", ) _check( [row["rollout_index"] for row in rows] == list(range(len(rows))), "rollout indices missing or repeated", ) _check( group["task_ids"] == [row["task_id"] for row in rows] and group["rewards"] == rewards and math.isclose(group["reward_std"], statistics.pstdev(rewards)), "reward group aggregates mismatch", ) _check( group["strict_rewards"] == [row["score"]["reward"] for row in rows], "strict reward group mismatch", ) _check( all( row["policy_version"]["global_step"] == group["optimizer_step"] and row["policy_version"]["base_model_sha256"] == manifest["model"]["sha256"] for row in rows ), "sampled policy-version identity mismatch", ) unique = sorted({row["task_id"] for row in records}) coverage = manifest["task_coverage"] _check( coverage["consumed_unique_task_ids"] == unique and coverage["total_rollouts"] == len(records) and coverage["training_pool_rows"] == len(tasks), "training task coverage aggregate mismatch", ) mixed = sum( statistics.pstdev([row["training_reward"] for row in group]) > 0 for group in groups.values() ) gradients = sum(row.get("grad_norm", 0) > 0 for row in manifest["train_log_history"]) _check( mixed == manifest["update_evidence"]["mixed_reward_groups"] and gradients == manifest["update_evidence"]["nonzero_gradient_steps"], "learning-update evidence aggregates mismatch", ) return { "configured_pool": len(tasks), "consumed_unique_tasks": len(unique), "total_rollouts": len(records), "optimizer_groups": len(groups), "generations_per_group": manifest["training"]["num_generations"], "strict_train_successes": sum(row["score"]["success"] for row in records), "shaping_mean": statistics.mean(row["training_reward"] for row in records), "token_audit": totals, "update_evidence": manifest["update_evidence"], } def _agent_counts(report): tasks = { row["id"]: row["task"] for row in ( json.loads(line) for line in report["provenance"]["tasks_snapshot"].splitlines() if line.strip() ) } correct = sum( episode["outcome"]["final_answer"] == str( tasks[episode["id"]]["private_facts"]["quantity"] * tasks[episode["id"]]["private_facts"]["unit_price"] ) for episode in report["episodes"] ) return { **report["metrics"], "final_answer_correct": correct, "final_answer_correct_rate": correct / len(report["episodes"]), "final_answer_note": "Accepted terminal numeric answer correctness independently of required tool workflow; this never replaces strict workflow success.", } def _verify_launch(launch, plan): _check( launch["source_sha256"] == plan["source_sha256"], "launch and consumed protocol source hashes differ", ) commit = launch["source_commit"] _check( isinstance(commit, str) and re.fullmatch(r"[0-9a-f]{40}", commit), "launch requires an immutable source commit", ) repo = Path(__file__).resolve().parents[2] for name, expected in plan["source_sha256"].items(): try: blob = subprocess.check_output( ["git", "-C", str(repo), "show", f"{commit}:src/hessian_recipes/{name}.py"], stderr=subprocess.PIPE, ) except subprocess.CalledProcessError as error: raise ValueError("launch source commit is unavailable in this checkout") from error _check( hashlib.sha256(blob).hexdigest() == expected, "launch commit does not contain consumed source", ) def summarize(run_dir: str | Path, output_dir: str | Path) -> dict: root, output = Path(run_dir).resolve(), Path(output_dir).resolve() _check(not output.exists(), "choose a fresh summary output directory") names = { "plan": "plan.json", "run": "run-manifest.json", "launch": "launch-provenance.json", "training": "training/training-manifest.json", "merge": "merged-checkpoint.merge-manifest.json", "baseline": "baseline/report.json", "candidate": "candidate/report.json", "retention": "retention/report.json", } loaded = {key: _read(root / path) for key, path in names.items()} plan, status, launch, training, merge, baseline, candidate, retention = ( loaded[key][0] for key in names ) _check( status["status"] == "completed" and set(status["completed_stages"]) >= {"baseline", "grpo", "merge", "candidate", "retention"}, "agent RL run is not complete", ) _check(status["plan_sha256"] == loaded["plan"][1], "completed run plan hash mismatch") verify_inputs(root, plan) _verify_launch(launch, plan) require_learning_update(training) _check(merge["status"] == "completed", "merge is not complete") _check( training["config_sha256"] == plan["training_config_sha256"] and training["dataset"]["sha256"] == plan["split_sha256"]["train"], "training input ancestry mismatch", ) _check( training["model"]["sha256"] == merge["base_model"]["sha256"] == plan["model_sha256"], "SFT parent ancestry mismatch", ) _check( training["helper_sha256"] == plan["source_sha256"]["agent_rl_training"] and training["environment_sha256"] == plan["source_sha256"]["agent_rl_tasks"], "training source identity mismatch", ) _check( training["checkpoint_sha256"] == merge["adapter"]["sha256"] == checkpoint_sha256(training["checkpoint"]), "RL adapter ancestry/hash mismatch", ) _check( Path(merge["adapter"]["path"]).resolve() == Path(training["checkpoint"]).resolve(), "merged adapter path mismatch", ) merged_sha = checkpoint_sha256(root / "merged-checkpoint") _check(merged_sha == merge["output"]["sha256"], "merged RL checkpoint hash mismatch") for report, identity in ((baseline, plan["model_sha256"]), (candidate, merged_sha)): agent_eval.verify_report(report) spec = report["provenance"]["spec"] _check( spec["trials"] == 1 and report["metrics"]["episodes"] == plan["split_sizes"]["test"], "heldout trial count differs from bounded plan", ) _check( spec["model"]["sha256"] == identity and spec["tasks"]["sha256"] == plan["split_sha256"]["test"] and spec["tokenizer"]["sha256"] == plan["tokenizer_sha256"], "heldout/model/tokenizer identity mismatch", ) _check( all( spec[key] == plan[key] for key in ( "sampling", "chat_template_kwargs", "max_completion_length", "max_tool_calling_iterations", ) ), "heldout protocol differs from frozen plan", ) _check( training["policy_format"]["helper_sha256"] == plan["source_sha256"]["agent_rl_training"], "training tokenizer helper identity mismatch", ) _check( all( report["provenance"]["effective_template_sha256"] == training["policy_format"]["chat_template_sha256"] for report in (baseline, candidate) ), "training/evaluation effective template identity mismatch", ) agent_gate = agent_eval.compare(root / names["baseline"], root / names["candidate"]) _check( agent_gate.get("paired") is not None and status["comparison"] == agent_gate, "agent comparison evidence mismatch", ) prior_root = Path(plan["retention_run"]) prior, prior_sha = _read(prior_root / "candidate/report.json") _verify_report(prior) _verify_report(retention) _check( prior_sha == plan["retention_baseline_report_sha256"] and prior["provenance"]["spec"]["model"]["revision"] == plan["model_sha256"], "MGSM baseline ancestry mismatch", ) _check( retention["provenance"]["spec"]["model"]["revision"] == merged_sha and retention["provenance"]["spec"]["benchmark"]["sha256"] == plan["retention_tasks_sha256"], "MGSM candidate/model/data identity mismatch", ) math_gate = compare_math(prior_root / "candidate/report.json", root / names["retention"]) _check( all("regression exceeds allowed drop" in reason for reason in math_gate["reasons"]) and status["retention_comparison"] == math_gate, "MGSM matched comparison evidence mismatch", ) _check( training["evidence_sha256"] == sha256_file(root / "training/rollout-evidence.jsonl"), "training rollout evidence hash mismatch", ) learning = _audit_training(training, root) for key in ( "max_completion_length", "max_tool_calling_iterations", "temperature", "top_p", "top_k", "min_p", ): expected = plan[key] if key in plan else plan["sampling"][key] _check( training["training"][key] == expected, "training/evaluation retained budget or sampling mismatch", ) _check( training["training"]["max_steps"] == plan["steps"] and training["chat_template_kwargs"] == plan["chat_template_kwargs"], "training step/thinking plan mismatch", ) summary = { "schema_version": 1, "status": "ready", "release_eligible": False, "quality_gates": { "agent": "passed" if agent_gate["passed"] else "regression", "MGSM_retention": "passed" if math_gate["passed"] else "regression", }, "source_commit": launch["source_commit"], "reporter_sha256": sha256_file(__file__), "identity": { "source_hashes": plan["source_sha256"], "effective_template_sha256": training["policy_format"]["chat_template_sha256"], "SFT_parent_sha256": plan["model_sha256"], "RL_adapter_sha256": training["checkpoint_sha256"], "merged_RL_sha256": merged_sha, "train_task_sha256": plan["split_sha256"]["train"], "heldout_task_sha256": plan["split_sha256"]["test"], "retention_task_sha256": plan["retention_tasks_sha256"], }, "training": { **learning, "optimizer_steps": plan["steps"], "parameters": training["training"], "lora": training["lora"], "shaping_protocol": training["reward_protocol"], "dependencies": training["dependencies"], }, "agent_eval": { "source_kind": baseline["source_kind"], "baseline": _agent_counts(baseline), "candidate": _agent_counts(candidate), "paired": agent_gate["paired"], "gate": agent_gate, "sampling": plan["sampling"], "max_completion_length": plan["max_completion_length"], "max_tool_calling_iterations": plan["max_tool_calling_iterations"], "language_audit": "unassessed; tool JSON and numeric-only final answers are language-neutral", }, "MGSM_retention": { "baseline": math_counts(prior), "candidate": math_counts(retention), "execution_mode": prior["provenance"]["execution_mode"], "paired": math_paired(prior, retention), "gate": math_gate, }, "worker": launch.get("gpu_preflight", {}), "evidence_sha256": { **{key: value[1] for key, value in loaded.items()}, "rollouts": training["evidence_sha256"], "retention_baseline": prior_sha, }, "limitations": "Training milestone reward is not strict task success; partial workflows can receive credit. This is a small synthetic lookup/calculator family, with one training seed and one stochastic heldout trial per item. Paired item bootstrap is descriptive and excludes training/generation-seed variance. No general reasoning, German fluency, or release readiness claim. Native training may generate and discard tails beyond the retained stream cap; matched agent evaluations generate only the remaining retained budget. MGSM scores also reflect final-answer format and token-cap effects.", } output.mkdir(parents=True) (output / "summary.json").write_text( json.dumps(summary, indent=2, ensure_ascii=False, allow_nan=False) + "\n" ) lines = [ "# German agent RL pilot", "", "Verified completed evidence; release eligible: **false**.", "", "| Evaluation | Baseline | RL candidate |", "| --- | ---: | ---: |", f"| Agent strict workflow success | {baseline['metrics']['successful']}/{baseline['metrics']['episodes']} | {candidate['metrics']['successful']}/{candidate['metrics']['episodes']} |", f"| Agent accepted final answer correct (independent of workflow) | {summary['agent_eval']['baseline']['final_answer_correct']}/{baseline['metrics']['episodes']} | {summary['agent_eval']['candidate']['final_answer_correct']}/{candidate['metrics']['episodes']} |", f"| MGSM strict final-answer protocol accuracy | {prior['metrics']['correct']}/{prior['metrics']['expected']} | {retention['metrics']['correct']}/{retention['metrics']['expected']} |", "", f"Actual training coverage: {learning['consumed_unique_tasks']} unique tasks from a pool of {learning['configured_pool']}; {learning['total_rollouts']} rollouts. Strict training successes: {learning['strict_train_successes']}. Mean shaped training reward: {learning['shaping_mean']:.4f}.", "", f"Agent gate: {summary['quality_gates']['agent']}; MGSM retention gate: {summary['quality_gates']['MGSM_retention']}. Agent paired wins/losses: {agent_gate['paired']['wins']}/{agent_gate['paired']['losses']}; delta {agent_gate['paired']['accuracy_delta']:+.1%}, descriptive 95% item-bootstrap interval {agent_gate['paired']['descriptive_bootstrap_95']}.", "", f"MGSM paired delta {summary['MGSM_retention']['paired']['accuracy_delta']:+.1%}, descriptive 95% item-bootstrap interval {summary['MGSM_retention']['paired']['bootstrap']['ci_95']}.", "", summary["limitations"], "", ] (output / "summary.md").write_text("\n".join(lines)) return summary def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() result = summarize(args.run, args.output) print( json.dumps( {"status": result["status"], "release_eligible": False, "summary": str(args.output)}, indent=2, ) ) if __name__ == "__main__": main()