#!/usr/bin/env python3 """Evaluate completion state, GSM8K answers, and HumanEval code safely. The input is a raw JSON emitted by v2_lite_chat_special_token_behavior_probe.py. Generated code is executed only inside one fresh, networkless, read-only Docker container per candidate. The result preserves fixed-budget and strict-completion metrics separately. """ from __future__ import annotations import argparse import ast import gzip import hashlib import json import re import subprocess import time from collections import Counter from decimal import Decimal, InvalidOperation from pathlib import Path from statistics import mean from typing import Any NUMBER = r"[-+]?(?:\d[\d,]*\.?\d*|\.\d+)" NUMBER_PATTERN = re.compile(NUMBER) BOXED_PATTERN = re.compile( rf"\\boxed\s*\{{\s*({NUMBER})\s*\}}", flags=re.IGNORECASE, ) HASH_PATTERN = re.compile( rf"####\s*({NUMBER})", flags=re.IGNORECASE, ) ANSWER_PATTERN = re.compile( rf"(?:final\s+answer|answer|result|total|profit)" rf"(?:\s+(?:is|equals|will\s+be))?\s*[:=]?\s*" rf"(?:\$|USD\s*)?({NUMBER})", flags=re.IGNORECASE, ) FENCE_PATTERN = re.compile( r"```(?P[A-Za-z0-9_+-]*)[ \t]*\n?" r"(?P.*?)(?:```|$)", flags=re.DOTALL, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--behavior-json", type=Path, required=True) parser.add_argument("--baseline-json", type=Path) parser.add_argument("--human-eval", type=Path, required=True) parser.add_argument("--gsm8k", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument( "--sandbox-image", required=True, help="Pinned image reference including @sha256 digest.", ) parser.add_argument("--timeout-seconds", type=float, default=5.0) parser.add_argument("--skip-code-execution", action="store_true") return parser.parse_args() def sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while chunk := handle.read(16 * 1024 * 1024): digest.update(chunk) return digest.hexdigest() def canonical_hash(value: Any) -> str: return sha256_bytes( json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode() ) def load_jsonl(path: Path) -> list[dict[str, Any]]: with path.open(encoding="utf-8") as handle: return [json.loads(line) for line in handle if line.strip()] def load_tasks( human_eval_path: Path, gsm8k_path: Path, ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]: with gzip.open(human_eval_path, "rt", encoding="utf-8") as handle: human_eval = { row["task_id"]: row for line in handle if line.strip() for row in [json.loads(line)] } gsm8k = { f"gsm8k/test/{index:04d}": row for index, row in enumerate(load_jsonl(gsm8k_path)) } return human_eval, gsm8k def normalize_number(value: str | None) -> str | None: if value is None: return None cleaned = value.replace(",", "").strip().rstrip(".") try: number = Decimal(cleaned) except InvalidOperation: return None normalized = format(number.normalize(), "f") if "." in normalized: normalized = normalized.rstrip("0").rstrip(".") return "0" if normalized in {"-0", "+0"} else normalized def last_match( pattern: re.Pattern[str], text: str, ) -> str | None: matches = list(pattern.finditer(text)) return matches[-1].group(1) if matches else None def evaluate_math( text: str, hit_eos: bool, gold_answer: str, ) -> dict[str, Any]: candidates = [ ("boxed", last_match(BOXED_PATTERN, text)), ("hash_marker", last_match(HASH_PATTERN, text)), ("answer_phrase", last_match(ANSWER_PATTERN, text)), ] method = "none" extracted = None for name, value in candidates: if value is not None: method = name extracted = value break explicit_terminal = extracted is not None if extracted is None: values = NUMBER_PATTERN.findall(text) if values: method = "last_number_fallback" extracted = values[-1] gold_values = NUMBER_PATTERN.findall( gold_answer.rsplit("####", 1)[-1] ) gold = normalize_number(gold_values[-1] if gold_values else None) predicted = normalize_number(extracted) fixed_budget_exact = ( gold is not None and predicted is not None and gold == predicted ) semantic_terminal = bool(hit_eos or explicit_terminal) return { "gold_final": gold, "predicted_final": predicted, "extraction_method": method, "explicit_final_marker": explicit_terminal, "semantic_terminal": semantic_terminal, "evaluator_covered": predicted is not None, "fixed_budget_numeric_exact": fixed_budget_exact, "strict_complete_numeric_exact": ( fixed_budget_exact and semantic_terminal ), } def code_fences(text: str) -> list[dict[str, str | bool]]: rows = [] for match in FENCE_PATTERN.finditer(text): language = match.group("lang").lower() body = match.group("body").strip() rows.append( { "language": language, "body": body, "closed": text[match.start():match.end()].rstrip().endswith( "```" ), } ) return rows def extract_code( text: str, prompt: str, entry_point: str, ) -> dict[str, Any]: target = re.compile( rf"(?m)^\s*(?:async\s+)?def\s+{re.escape(entry_point)}\s*\(" ) fences = code_fences(text) chosen = next( ( row for row in fences if target.search(str(row["body"])) and row["language"] in {"", "py", "python", "python3"} ), None, ) if chosen is None: chosen = next( ( row for row in fences if row["language"] in {"", "py", "python", "python3"} ), None, ) if chosen is not None: raw = str(chosen["body"]) mode = "fenced" fence_closed = bool(chosen["closed"]) else: match = target.search(text) if match: raw = text[match.start():].strip() mode = "direct_definition" else: raw = text.strip() mode = "prompt_completion" fence_closed = False contains_entry_point = target.search(raw) is not None candidate = raw if contains_entry_point else prompt + raw try: ast.parse(candidate) ast_ok = True ast_error = None except SyntaxError as error: ast_ok = False ast_error = { "line": error.lineno, "offset": error.offset, "type": type(error).__name__, } return { "candidate": candidate, "candidate_sha256": sha256_bytes(candidate.encode()), "extraction_mode": mode, "contains_entry_point_definition": contains_entry_point, "closed_code_fence": fence_closed, "python_ast_parse": ast_ok, "parse_error": ast_error, } def sandbox_harness( candidate: str, test: str, entry_point: str, ) -> str: return ( "import sys\n" "try:\n" + "\n".join( f" {line}" if line else "" for line in candidate.splitlines() ) + "\n" + "\n".join( f" {line}" if line else "" for line in test.splitlines() ) + f"\n check({entry_point})\n" + "except AssertionError:\n" + " raise SystemExit(10)\n" + "except BaseException:\n" + " raise SystemExit(11)\n" + "raise SystemExit(0)\n" ) def execute_code( candidate: str, task: dict[str, Any], image: str, timeout_seconds: float, ) -> dict[str, Any]: harness = sandbox_harness( candidate, task["test"], task["entry_point"], ) command = [ "docker", "run", "--rm", "-i", "--network", "none", "--read-only", "--tmpfs", "/tmp:rw,noexec,nosuid,size=16m", "--memory", "256m", "--memory-swap", "256m", "--pids-limit", "64", "--cpus", "0.5", "--cap-drop", "ALL", "--security-opt", "no-new-privileges", "--user", "65534:65534", image, "python", "-I", "-", ] started = time.perf_counter() try: completed = subprocess.run( command, input=harness.encode(), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout_seconds, check=False, ) except subprocess.TimeoutExpired: return { "status": "timeout", "return_code": None, "runtime_ms": (time.perf_counter() - started) * 1000, "harness_sha256": sha256_bytes(harness.encode()), } runtime_ms = (time.perf_counter() - started) * 1000 statuses = { 0: "passed", 10: "assertion_failed", 11: "runtime_error", } return { "status": statuses.get( completed.returncode, "sandbox_error", ), "return_code": completed.returncode, "runtime_ms": runtime_ms, "harness_sha256": sha256_bytes(harness.encode()), } def evaluate_code( text: str, hit_eos: bool, task: dict[str, Any], image: str, timeout_seconds: float, skip_execution: bool, ) -> dict[str, Any]: extracted = extract_code( text, task["prompt"], task["entry_point"], ) execution = { "status": "not_run", "return_code": None, "runtime_ms": None, "harness_sha256": None, } if extracted["python_ast_parse"] and not skip_execution: execution = execute_code( extracted["candidate"], task, image, timeout_seconds, ) passed = execution["status"] == "passed" semantic_terminal = bool( hit_eos or extracted["closed_code_fence"] or passed ) return { key: value for key, value in extracted.items() if key != "candidate" } | { "semantic_terminal": semantic_terminal, "evaluator_covered": ( extracted["python_ast_parse"] and execution["status"] != "not_run" ), "execution": execution, "fixed_budget_tests_pass": passed, "strict_complete_tests_pass": passed and semantic_terminal, } def completion_class( output: dict[str, Any], task_evaluation: dict[str, Any] | None, ) -> str: if output["hit_eos"]: return "NATURAL_EOS" if ( task_evaluation is not None and task_evaluation.get("semantic_terminal") ): return "TASK_TERMINAL_BEFORE_EOS" if output["stopped_at_max_new_tokens"]: if ( task_evaluation is not None and task_evaluation.get("extraction_method") == "last_number_fallback" ): return "BUDGET_TRUNCATED_WITH_FALLBACK_ONLY" return "BUDGET_TRUNCATED_UNRESOLVED" return "OTHER_STOP" def baseline_index( baseline: dict[str, Any] | None, ) -> dict[tuple[str, str], dict[str, Any]]: if baseline is None: return {} return { (source["id"], output["condition"]): output for source in baseline["sources"] for output in source["outputs"] } def prefix_audit( source_id: str, output: dict[str, Any], baseline_rows: dict[tuple[str, str], dict[str, Any]], ) -> dict[str, Any] | None: prior = baseline_rows.get((source_id, output["condition"])) if prior is None: return None reference = prior["generated_token_ids"] observed = output["generated_token_ids"][: len(reference)] return { "baseline_generated_tokens": len(reference), "compared_tokens": min( len(reference), len(output["generated_token_ids"]), ), "prompt_hash_exact": ( prior["prompt_token_ids_sha256"] == output["prompt_token_ids_sha256"] ), "generated_prefix_exact": observed == reference, "baseline_hit_eos": prior["hit_eos"], } def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: by_condition: dict[str, dict[str, Any]] = {} conditions = sorted({row["condition"] for row in rows}) for condition in conditions: subset = [ row for row in rows if row["condition"] == condition ] math_rows = [ row for row in subset if row["domain"] == "math" ] code_rows = [ row for row in subset if row["domain"] == "code" ] by_condition[condition] = { "outputs": len(subset), "natural_eos": sum(row["hit_eos"] for row in subset), "budget_truncated": sum( row["stopped_at_max_new_tokens"] for row in subset ), "mean_generated_tokens": mean( row["generated_tokens"] for row in subset ), "completion_classes": dict( Counter(row["completion_class"] for row in subset) ), "math_fixed_budget_exact": sum( row["task_evaluation"][ "fixed_budget_numeric_exact" ] for row in math_rows ), "math_strict_complete_exact": sum( row["task_evaluation"][ "strict_complete_numeric_exact" ] for row in math_rows ), "math_sources": len(math_rows), "code_ast_parse": sum( row["task_evaluation"]["python_ast_parse"] for row in code_rows ), "code_executed": sum( row["task_evaluation"]["execution"]["status"] != "not_run" for row in code_rows ), "code_tests_pass": sum( row["task_evaluation"]["fixed_budget_tests_pass"] for row in code_rows ), "code_sources": len(code_rows), } prefix_rows = [ row["baseline_prefix_audit"] for row in rows if row["baseline_prefix_audit"] is not None ] return { "outputs": len(rows), "natural_eos": sum(row["hit_eos"] for row in rows), "budget_truncated": sum( row["stopped_at_max_new_tokens"] for row in rows ), "completion_classes": dict( Counter(row["completion_class"] for row in rows) ), "baseline_prefix": { "cells": len(prefix_rows), "prompt_hash_exact": sum( row["prompt_hash_exact"] for row in prefix_rows ), "generated_prefix_exact": sum( row["generated_prefix_exact"] for row in prefix_rows ), }, "by_condition": by_condition, } def main() -> None: args = parse_args() for path in ( args.behavior_json, args.human_eval, args.gsm8k, ): if not path.is_file(): raise FileNotFoundError(path) if args.baseline_json is not None and not args.baseline_json.is_file(): raise FileNotFoundError(args.baseline_json) if "@sha256:" not in args.sandbox_image: raise ValueError("--sandbox-image must include an immutable digest") if args.timeout_seconds <= 0: raise ValueError("--timeout-seconds must be positive") behavior = json.loads( args.behavior_json.read_text(encoding="utf-8") ) baseline = ( json.loads(args.baseline_json.read_text(encoding="utf-8")) if args.baseline_json is not None else None ) baseline_rows = baseline_index(baseline) human_eval, gsm8k = load_tasks(args.human_eval, args.gsm8k) rows = [] for source in behavior["sources"]: for output in source["outputs"]: evaluation = None if source["domain"] == "math": evaluation = evaluate_math( output["text"], output["hit_eos"], gsm8k[source["id"]]["answer"], ) elif source["domain"] == "code": evaluation = evaluate_code( output["text"], output["hit_eos"], human_eval[source["id"]], args.sandbox_image, args.timeout_seconds, args.skip_code_execution, ) rows.append( { "source_id": source["id"], "domain": source["domain"], "condition": output["condition"], "generated_tokens": output["generated_tokens"], "hit_eos": output["hit_eos"], "stopped_at_max_new_tokens": ( output["stopped_at_max_new_tokens"] ), "prompt_token_ids_sha256": ( output["prompt_token_ids_sha256"] ), "generated_token_ids_sha256": ( output["generated_token_ids_sha256"] ), "text_sha256": output["text_sha256"], "task_evaluation": evaluation, "completion_class": completion_class( output, evaluation, ), "baseline_prefix_audit": prefix_audit( source["id"], output, baseline_rows, ), } ) result = { "schema_version": 1, "input": { "behavior_path": str(args.behavior_json), "behavior_sha256": sha256_file(args.behavior_json), "baseline_path": ( str(args.baseline_json) if args.baseline_json is not None else None ), "baseline_sha256": ( sha256_file(args.baseline_json) if args.baseline_json is not None else None ), "human_eval_sha256": sha256_file(args.human_eval), "gsm8k_sha256": sha256_file(args.gsm8k), "model_revision": behavior["model"]["revision"], "max_new_tokens": behavior[ "generation_contract" ]["max_new_tokens"], }, "sandbox": { "image": args.sandbox_image, "timeout_seconds": args.timeout_seconds, "code_execution_skipped": args.skip_code_execution, "network": "none", "filesystem": "read-only", "user": "65534:65534", "capabilities": "ALL dropped", "memory": "256m", "memory_swap": "256m", "pids_limit": 64, "cpus": 0.5, "tmpfs": "/tmp:rw,noexec,nosuid,size=16m", "host_mounts": 0, }, "rows": rows, "summary": summarize(rows), "claim_boundary": [ "Four math and four code sources are not benchmark estimates.", "Completion-conditioned metrics are selection-biased diagnostics.", "A passing HumanEval test is functional evidence, not code-safety evidence.", "Fallback last-number extraction is not strict completion.", "Counterfactual token sequences are not official-valid chats.", ], "content_hash": canonical_hash(rows), } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) payload = args.output.read_bytes() print( json.dumps( { "output": str(args.output), "sha256": sha256_bytes(payload), "bytes": len(payload), "summary": result["summary"], }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()