"""Matched, local multi-turn evaluation for the bounded synthetic German agent pilot. Tools are deterministic in-memory methods. Captured policy tokens are never reconstructed from displayed text. Evidence is private and is not release certification. """ from __future__ import annotations import argparse import copy import hashlib import inspect import importlib.metadata import json import math from pathlib import Path import platform import random from statistics import mean import time from .evaluation import checkpoint_sha256, sha256_file from .trajectories import validate_trajectory, write_trajectories def _digest(value) -> str: return hashlib.sha256( json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode() ).hexdigest() def source_hashes() -> dict: root = Path(__file__).parent return { name: sha256_file(root / name) for name in ( "agent_rl_evaluation.py", "agent_rl_tasks.py", "agent_rl_training.py", "trajectories.py", ) } def _keys(value: dict, allowed: set, label: str) -> None: if not isinstance(value, dict) or set(value) - allowed: raise ValueError(f"unsupported {label} fields") def _load(spec_path: str | Path) -> tuple[dict, list[dict], dict]: path = Path(spec_path).resolve() spec_bytes = path.read_bytes() spec = json.loads(spec_bytes) _keys( spec, { "schema_version", "model", "tokenizer", "tasks", "sampling", "max_completion_length", "max_tool_calling_iterations", "chat_template_kwargs", "trials", "device", "dtype", }, "spec", ) if spec.get("schema_version") != 1: raise ValueError("schema_version must be 1") for name in ("max_completion_length", "max_tool_calling_iterations", "trials"): if type(spec.get(name)) is not int or spec[name] < 1: raise ValueError(f"{name} must be a positive integer") # The authoritative environment itself has four assistant turns. if spec["max_tool_calling_iterations"] > 3: raise ValueError("at most three tool-calling iterations are supported") sampling = spec["sampling"] _keys(sampling, {"temperature", "top_p", "top_k", "min_p", "seed"}, "sampling") for key, low, high in (("temperature", 0, float("inf")), ("top_p", 0, 1), ("min_p", 0, 1)): value = sampling.get(key) if ( type(value) not in (int, float) or not math.isfinite(value) or not low <= value <= high or (key == "top_p" and value == 0) ): raise ValueError(f"invalid sampling.{key}") for key in ("top_k", "seed"): if type(sampling.get(key)) is not int or sampling[key] < 0: raise ValueError(f"sampling.{key} must be a nonnegative integer") if spec.get("device", "cpu") not in ("cpu", "cuda") or spec.get("dtype", "float32") not in ( "float32", "bfloat16", "float16", ): raise ValueError("unsupported device/dtype") kwargs = spec.get("chat_template_kwargs", {}) _keys(kwargs, {"enable_thinking"}, "chat_template_kwargs") if "enable_thinking" in kwargs and type(kwargs["enable_thinking"]) is not bool: raise ValueError("enable_thinking must be boolean") resolved = {} for key in ("model", "tokenizer", "tasks"): _keys(spec[key], {"path", "sha256"}, key) resolved[key] = (path.parent / spec[key]["path"]).resolve() raw = resolved["tasks"].read_bytes() if hashlib.sha256(raw).hexdigest() != spec["tasks"]["sha256"]: raise ValueError("task SHA256 mismatch") rows = [json.loads(line) for line in raw.decode().splitlines() if line.strip()] if not rows or len({row["id"] for row in rows}) != len(rows): raise ValueError("task identities must be unique and nonempty") from .agent_rl_tasks import GermanAgentEnv for row in rows: if ( not isinstance(row.get("prompt"), str) or not row["prompt"].strip() or row["id"] != row["task"]["task_id"] ): raise ValueError("invalid task prompt/identity") GermanAgentEnv().reset(**row) return ( spec, rows, { "spec_snapshot": spec_bytes.decode(), "tasks_snapshot": raw.decode(), "spec_sha256": hashlib.sha256(spec_bytes).hexdigest(), "resolved": resolved, }, ) class LocalPolicy: """Lazy HF policy using the same native TRL parser and template as training.""" def __init__(self, spec: dict, paths: dict): import torch from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig, set_seed from transformers.utils import get_json_schema from trl.chat_template_utils import parse_response from .agent_rl_training import configure_policy_tokenizer from .agent_rl_tasks import GermanAgentEnv from .training import require_versions self.packages = require_versions(("torch", "transformers", "trl")) self.packages["jmespath"] = importlib.metadata.version("jmespath") if self.packages["jmespath"] != "1.0.1": raise ValueError("the native Qwen parser requires pinned jmespath==1.0.1") self.device = spec.get("device", "cpu") dtype = spec.get("dtype", "float32") if self.device == "cuda" and ( not torch.cuda.is_available() or (dtype == "bfloat16" and not torch.cuda.is_bf16_supported()) ): raise ValueError("requested CUDA/BF16 device is unavailable") if (paths["model"] / "adapter_config.json").exists(): raise ValueError("evaluate a full or merged local checkpoint, not a standalone adapter") self.tokenizer = AutoTokenizer.from_pretrained( str(paths["tokenizer"]), local_files_only=True, trust_remote_code=False ) self.template_metadata = configure_policy_tokenizer(self.tokenizer) self.tools = [ member for name, member in inspect.getmembers(GermanAgentEnv(), inspect.ismethod) if name != "reset" and not name.startswith("_") ] if {method.__name__ for method in self.tools} != {"lookup", "calculator"}: raise ValueError("unexpected native environment methods") self.tool_schemas = [get_json_schema(method) for method in self.tools] self.model = AutoModelForCausalLM.from_pretrained( str(paths["model"]), local_files_only=True, trust_remote_code=False, dtype=getattr(torch, dtype), ) self.model.to(self.device).eval() self.spec, self.torch, self.GenerationConfig, self.set_seed = ( spec, torch, GenerationConfig, set_seed, ) self.parse_response = parse_response self.runtime = { "device": self.device, "dtype": dtype, "cuda": torch.version.cuda, "device_name": torch.cuda.get_device_name() if self.device == "cuda" else "CPU", } def encode(self, messages: list[dict]) -> list[int]: return self.tokenizer.apply_chat_template( messages, tools=self.tools, add_generation_prompt=True, tokenize=True, return_dict=False, **self.spec.get("chat_template_kwargs", {}), ) def __call__(self, messages: list[dict], *, seed: int, max_new_tokens: int) -> dict: ids = self.encode(messages) inputs = self.torch.tensor([ids], dtype=self.torch.long, device=self.device) sampling = self.spec["sampling"] self.set_seed(seed) config = self.GenerationConfig( max_new_tokens=max_new_tokens, do_sample=sampling["temperature"] > 0, temperature=sampling["temperature"] if sampling["temperature"] else 1.0, top_p=sampling["top_p"], top_k=sampling["top_k"], min_p=sampling["min_p"] if sampling["temperature"] else None, num_beams=1, repetition_penalty=1.0, eos_token_id=self.tokenizer.eos_token_id, pad_token_id=self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id, ) with self.torch.inference_mode(): result = self.model.generate( input_ids=inputs, attention_mask=self.torch.ones_like(inputs), generation_config=config, ) generated = result[0, len(ids) :].detach().cpu().tolist() assistant = self.parse_response(self.tokenizer, generated) assistant["role"] = "assistant" return { "assistant": assistant, "prompt_token_ids": ids, "response_token_ids": generated, "finish_reason": "stop" if generated and generated[-1] == self.tokenizer.eos_token_id else "length", } def _execute_tools(env, assistant: dict) -> list[dict]: from .agent_rl_tasks import step_action observations = [] for call in assistant.get("tool_calls") or []: function = call.get("function", {}) if call.get("type") == "function" else {} arguments = function.get("arguments", {}) if not isinstance(arguments, dict): arguments = {} name = function.get("name", "unknown") # Arguments cannot override the selected tool name. action = {**arguments, "tool": name} if "tool" in arguments: action["__invalid_arguments__"] = True result = step_action(env, action) observations.append({"role": "tool", "name": name, "content": result["observation"]}) return observations def _episode(row: dict, trial: int, spec: dict, policy) -> dict: from .agent_rl_tasks import GermanAgentEnv, score_episode env = GermanAgentEnv() env.reset(**row) messages = [{"role": "user", "content": row["prompt"]}] initial = policy.encode(messages) calls, completions, reason, stream_tokens = [], [], "terminal", 0 base_seed = ( spec["sampling"]["seed"] + int(hashlib.sha256(row["id"].encode()).hexdigest()[:8], 16) + trial ) % (2**32) started = time.perf_counter() for turn in range(spec["max_tool_calling_iterations"] + 1): prompt = policy.encode(messages) if prompt[: len(initial)] != initial: raise ValueError("effective tool template is not prefix preserving") remaining = spec["max_completion_length"] - (len(prompt) - len(initial)) if remaining <= 0: reason = "token_cap" stream_tokens = spec["max_completion_length"] break generated = policy( copy.deepcopy(messages), seed=(base_seed + turn) % (2**32), max_new_tokens=remaining ) response = generated["response_token_ids"] if ( generated["prompt_token_ids"] != prompt or not response or any(type(token) is not int or token < 0 for token in response) or len(response) > remaining ): raise ValueError("policy capture token/prompt budget contract violated") assistant = generated["assistant"] if not isinstance(assistant, dict) or assistant.get("role") != "assistant": raise ValueError("policy must return a parsed native assistant message") stream_tokens = len(prompt) - len(initial) + len(response) call = { "prompt_token_ids": prompt, "response_token_ids": response, "response_loss_mask": [1] * len(response), "assistant": copy.deepcopy(assistant), "finish_reason": generated["finish_reason"], "context": copy.deepcopy(messages), "observations": [], } calls.append(call) messages.append(assistant) completions.append(assistant) if not assistant.get("tool_calls"): reason = "token_cap" if generated["finish_reason"] == "length" else "terminal" break if turn == spec["max_tool_calling_iterations"]: reason = "turn_cap" break observations = _execute_tools(env, assistant) call["observations"] = observations messages.extend(observations) completions.extend(observations) call["post_tool_prompt_token_ids"] = policy.encode(messages) result = score_episode(env, completions) return { "id": row["id"], "trial": trial, "seed": base_seed, "status": "complete", "calls": calls, "termination": reason, "completion_stream_tokens": stream_tokens, "outcome": result, "wall_seconds": time.perf_counter() - started, } def _metrics(episodes: list[dict]) -> dict: complete = [episode for episode in episodes if episode["status"] == "complete"] outcomes = [episode["outcome"] for episode in complete] all_complete = len(complete) == len(episodes) return { "episodes": len(episodes), "scored": len(complete), "infrastructure_errors": len(episodes) - len(complete), "successful": sum(value["success"] for value in outcomes), "success_rate": mean(value["success"] for value in outcomes) if all_complete else None, "workflow_rate": mean(value["workflow_valid"] for value in outcomes) if all_complete else None, "mean_reward": mean(value["reward"] for value in outcomes) if all_complete else None, "tool_calls": sum( len(call["observations"]) for episode in complete for call in episode["calls"] ), "tool_errors": sum( isinstance(json.loads(observation["content"]), dict) and "error" in json.loads(observation["content"]) for episode in complete for call in episode["calls"] for observation in call["observations"] ), "malformed_final_outputs": sum(value["final_answer"] is None for value in outcomes), "turn_cap_episodes": sum(episode["termination"] == "turn_cap" for episode in complete), "token_cap_episodes": sum(episode["termination"] == "token_cap" for episode in complete), "assistant_tokens": sum( len(call["response_token_ids"]) for episode in complete for call in episode["calls"] ), "prompt_tokens": sum( len(call["prompt_token_ids"]) for episode in complete for call in episode["calls"] ), "completion_stream_tokens": sum( episode["completion_stream_tokens"] for episode in complete ), "wall_seconds": sum(episode.get("wall_seconds", 0) for episode in episodes), } def run(spec_path: str | Path, output_dir: str | Path, *, generator=None) -> dict: output = Path(output_dir) if output.exists(): raise ValueError("choose a fresh evaluation output directory") spec, rows, snapshot = _load(spec_path) paths = snapshot.pop("resolved") for key in ("model", "tokenizer"): if checkpoint_sha256(paths[key]) != spec[key]["sha256"]: raise ValueError(f"{key} directory SHA256 mismatch") sources = source_hashes() fixture = generator is not None policy = generator if fixture else LocalPolicy(spec, paths) for key in ("model", "tokenizer"): if checkpoint_sha256(paths[key]) != spec[key]["sha256"]: raise ValueError(f"{key} changed during policy loading") episodes = [] for row in rows: for trial in range(spec["trials"]): try: episode = _episode(row, trial, spec, policy) episode["capture_sha256"] = _digest(episode["calls"]) except Exception as error: episode = { "id": row["id"], "trial": trial, "status": "infrastructure_error", "error": {"type": type(error).__name__, "message": str(error)}, } episodes.append(episode) if sources != source_hashes(): raise ValueError("evaluation/harness/template source changed during run") report = { "schema_version": 1, "status": "complete" if all(ep["status"] == "complete" for ep in episodes) else "failed", "release_eligible": False, "source_kind": "fixture" if fixture else "captured", "provenance": { **snapshot, "spec": spec, "sources": sources, "python": platform.python_version(), "effective_template_sha256": getattr(policy, "template_metadata", {}).get( "chat_template_sha256" ), "tool_schemas": getattr(policy, "tool_schemas", []), "runtime": getattr(policy, "runtime", {"mode": "fixture"}), "packages": getattr(policy, "packages", {}), }, "metrics": _metrics(episodes), "episodes": episodes, "language_audit": { "status": "unassessed", "fixed_episode_ids": [row["id"] for row in rows[:8]], "note": "Machine tool JSON and numeric-only final answers are language-neutral. Qualitatively audit any natural-language reasoning separately; no fluency score is inferred.", }, "limitations": "Synthetic hidden-fact lookup/calculator workflow only. No general reasoning or German fluency claim. Completion-stream cap includes tool observations and template framing; assistant tokens count only sampled policy tokens. Independent evaluation samples only the remaining retained stream budget, while native training may generate then discard excess tokens. Private reports contain task facts and transcripts.", } output.mkdir(parents=True) (output / "report.json").write_text( json.dumps(report, indent=2, ensure_ascii=False, allow_nan=False) + "\n" ) _export_trajectories(report, output / "trajectories.jsonl") (output / "report.md").write_text( f"# Synthetic German agent workflow evaluation\n\nStatus: {report['status']}. Release eligible: false.\n\n```json\n{json.dumps(report['metrics'], indent=2)}\n```\n\n{report['limitations']}\n" ) return report def _export_trajectories(report: dict, path: Path) -> None: records = [] spec = report["provenance"]["spec"] revision = report["provenance"]["sources"]["agent_rl_tasks.py"] for episode in report["episodes"]: if episode["status"] != "complete" or not episode["calls"]: continue calls = [ { "call_id": str(index), "prompt_token_ids": call["prompt_token_ids"], "response_token_ids": call["response_token_ids"], "response_loss_mask": call["response_loss_mask"], "response_role": "assistant", "finish_reason": call["finish_reason"], "context": [ { "role": msg["role"], "content": msg.get("raw_response") or msg.get("content") or json.dumps(msg.get("tool_calls", [])), } for msg in call["context"] ], } for index, call in enumerate(episode["calls"]) ] record = { "schema_version": "hessian.trajectory.v1", "episode_id": f"{episode['id']}:{episode['trial']}", "group_id": episode["id"], "task_id": episode["id"], "seed": episode["seed"], "source_kind": report["source_kind"], "policy": { "model_id": spec["model"]["path"], "revision": spec["model"]["sha256"], "tokenizer_id": spec["tokenizer"]["path"], "tokenizer_revision": spec["tokenizer"]["sha256"], "chat_template_sha256": report["provenance"].get("effective_template_sha256") or "0" * 64, }, "harness": {"name": "GermanAgentEnv", "version": "1", "revision": revision}, "reward": { "value": episode["outcome"]["reward"], "scorer": "workflow-and-final", "revision": revision, }, "calls": calls, } records.append(validate_trajectory(record)) if records: write_trajectories(path, records) def verify_report(report: dict) -> None: from .agent_rl_tasks import GermanAgentEnv, score_episode provenance, spec = report["provenance"], report["provenance"]["spec"] if ( report.get("schema_version") != 1 or report.get("status") != "complete" or provenance["sources"] != source_hashes() ): raise ValueError("incomplete report or incompatible evaluator/harness/template source") if ( hashlib.sha256(provenance["spec_snapshot"].encode()).hexdigest() != provenance["spec_sha256"] or json.loads(provenance["spec_snapshot"]) != spec ): raise ValueError("spec snapshot mismatch") if hashlib.sha256(provenance["tasks_snapshot"].encode()).hexdigest() != spec["tasks"]["sha256"]: raise ValueError("task snapshot mismatch") rows = [json.loads(line) for line in provenance["tasks_snapshot"].splitlines() if line.strip()] mapping = {row["id"]: row for row in rows} expected = {(row["id"], trial) for row in rows for trial in range(spec["trials"])} episodes = report["episodes"] if len(episodes) != len(expected) or {(ep["id"], ep["trial"]) for ep in episodes} != expected: raise ValueError("episode coverage mismatch") for episode in episodes: if ( episode["status"] != "complete" or not episode["calls"] or _digest(episode["calls"]) != episode["capture_sha256"] ): raise ValueError("incomplete or modified episode capture") env, completions = GermanAgentEnv(), [] env.reset(**mapping[episode["id"]]) context = [{"role": "user", "content": mapping[episode["id"]]["prompt"]}] initial = episode["calls"][0]["prompt_token_ids"] for index, call in enumerate(episode["calls"]): if call["prompt_token_ids"][: len(initial)] != initial or call[ "response_loss_mask" ] != [1] * len(call["response_token_ids"]): raise ValueError("captured prompt prefix or assistant mask mismatch") stream = len(call["prompt_token_ids"]) - len(initial) + len(call["response_token_ids"]) if ( stream > spec["max_completion_length"] or index > spec["max_tool_calling_iterations"] ): raise ValueError("captured token/turn budget exceeded") if call["context"] != context: raise ValueError( "policy context contains unobserved facts or differs from tool replay" ) assistant = call["assistant"] context.append(assistant) completions.append(assistant) observations = ( _execute_tools(env, assistant) if assistant.get("tool_calls") and index < spec["max_tool_calling_iterations"] else [] ) if observations != call["observations"]: raise ValueError("tool observation disagrees with deterministic replay") completions.extend(observations) context.extend(observations) last = episode["calls"][-1] token_limit = ( last["finish_reason"] == "length" or len(last.get("post_tool_prompt_token_ids", [])) - len(initial) >= spec["max_completion_length"] ) turn_limit = ( bool(last["assistant"].get("tool_calls")) and len(episode["calls"]) == spec["max_tool_calling_iterations"] + 1 ) expected_termination = ( "turn_cap" if turn_limit else "token_cap" if token_limit else "terminal" ) expected_stream = ( spec["max_completion_length"] if token_limit and last.get("post_tool_prompt_token_ids") else stream ) if ( episode["termination"] != expected_termination or episode["completion_stream_tokens"] != expected_stream ): raise ValueError("episode termination or completion stream accounting mismatch") if score_episode(env, completions) != episode["outcome"]: raise ValueError("episode outcome disagrees with re-scored workflow") if _metrics(episodes) != report["metrics"]: raise ValueError("aggregate metrics disagree with episodes") def compare( baseline_report_path: str | Path, candidate_report_path: str | Path, max_drop: float = 0.0 ) -> dict: if type(max_drop) not in (int, float) or not math.isfinite(max_drop) or not 0 <= max_drop <= 1: raise ValueError("max_drop must be a finite fraction in [0,1]") reports = [ json.loads(Path(path).read_text()) for path in (baseline_report_path, candidate_report_path) ] reasons = [] for label, report in zip(("baseline", "candidate"), reports): try: verify_report(report) except (ValueError, KeyError, TypeError) as error: reasons.append(f"{label}: {error}") if reasons: return {"passed": False, "release_eligible": False, "reasons": reasons, "paired": None} a, b = (report["provenance"]["spec"] for report in reports) for key in ( "sampling", "max_completion_length", "max_tool_calling_iterations", "chat_template_kwargs", "trials", "device", "dtype", ): if a.get(key) != b.get(key): reasons.append(f"incompatible {key}") for key in ("tasks", "tokenizer"): if a[key]["sha256"] != b[key]["sha256"]: reasons.append(f"incompatible {key} identity") for key in ("effective_template_sha256", "tool_schemas", "packages"): if reports[0]["provenance"].get(key) != reports[1]["provenance"].get(key): reasons.append(f"incompatible {key}") if reports[0]["source_kind"] != reports[1]["source_kind"]: reasons.append("incompatible source_kind") if reasons: return {"passed": False, "release_eligible": False, "reasons": reasons, "paired": None} outcomes = [ {(ep["id"], ep["trial"]): ep["outcome"] for ep in report["episodes"]} for report in reports ] if outcomes[0].keys() != outcomes[1].keys(): return { "passed": False, "release_eligible": False, "reasons": ["incompatible paired episode coverage"], "paired": None, } wins = sum( outcomes[1][key]["success"] and not value["success"] for key, value in outcomes[0].items() ) losses = sum( value["success"] and not outcomes[1][key]["success"] for key, value in outcomes[0].items() ) by_item = {} for key, value in outcomes[0].items(): by_item.setdefault(key[0], []).append( int(outcomes[1][key]["success"]) - int(value["success"]) ) differences = [mean(values) for values in by_item.values()] rng = random.Random(20261015) samples = sorted(mean(rng.choices(differences, k=len(differences))) for _ in range(10000)) for key in ("success_rate", "workflow_rate"): if reports[1]["metrics"][key] - reports[0]["metrics"][key] < -max_drop: reasons.append(f"{key} regression exceeds allowed drop") return { "passed": not reasons, "release_eligible": False, "reasons": reasons, "paired": { "items": len(differences), "trials": len(outcomes[0]), "wins": wins, "losses": losses, "accuracy_delta": mean(differences), "descriptive_bootstrap_95": [samples[249], samples[9749]], "seed": 20261015, "resamples": 10000, }, "limitations": "Paired item bootstrap is descriptive; small synthetic task families and one stochastic trial do not establish generalization. Training-seed and generation-seed variance are excluded.", } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("spec", type=Path) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() report = run(args.spec, args.output) print( json.dumps( {"status": report["status"], "metrics": report["metrics"], "release_eligible": False}, indent=2, ) ) return 0 if report["status"] == "complete" else 1 if __name__ == "__main__": raise SystemExit(main())