#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re import shlex import sys from pathlib import Path, PurePosixPath from typing import Any EXAMPLE_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(EXAMPLE_DIR / "src")) from euler_common import ( # noqa: E402 answer_is_correct, extract_final_answer, final_assistant_text, load_jsonl, normalize_answer, ) RUNTIME_EXECUTABLES = { "python": {"python", "python3"}, "javascript": {"node", "bun"}, "julia": {"julia"}, "ruby": {"ruby"}, } def _tool_result_text(result: Any) -> str: if not isinstance(result, dict): return "" content = result.get("content") if not isinstance(content, list): return "" parts: list[str] = [] for part in content: if isinstance(part, dict) and isinstance(part.get("text"), str): parts.append(part["text"]) return "\n".join(parts) def _text_contains_expected(text: str, expected: str) -> bool: normalized = expected.strip().replace(",", "") if re.fullmatch(r"[-+]?\d+(?:[.]\d+)?", normalized): compact = text.replace(",", "") pattern = rf"(? str | None: path = PurePosixPath(value) if not value or path.is_absolute() or ".." in path.parts: return None normalized = str(path) return normalized if normalized not in {"", "."} else None def _direct_program(command: str, language: str) -> str | None: try: argv = shlex.split(command) except ValueError: return None if len(argv) != 2: return None executables = RUNTIME_EXECUTABLES.get(language.lower(), {language.lower()}) if PurePosixPath(argv[0]).name not in executables: return None return _normalized_relative_path(argv[1]) def _successful_exit(event: dict[str, Any]) -> bool: if event.get("isError") is True: return False result = event.get("result") if isinstance(result, dict): details = result.get("details") if isinstance(details, dict): for key in ("exitCode", "exit_code", "returnCode", "return_code", "code"): value = details.get(key) if isinstance(value, int): return value == 0 text = _tool_result_text(result) match = re.search(r"(?im)command exited with code\s+(-?\d+)", text) return match is None or int(match.group(1)) == 0 def _output_is_exact(text: str, expected: str) -> bool: lines = [line.strip() for line in text.splitlines() if line.strip()] if not lines: return False return normalize_answer(lines[-1]) == normalize_answer(expected) def execution_metrics( events: list[dict[str, Any]], language: str, expected: str ) -> dict[str, Any]: program_paths: dict[str, str] = {} mutation_starts: dict[str, tuple[str, str]] = {} successful_ids: set[str] = set() matched_execution_ids: set[str] = set() correct_output_ids: set[str] = set() successful_write_ids: set[str] = set() successful_edit_ids: set[str] = set() successful_mutation_paths: set[str] = set() observed_commands: list[str] = [] for event in events: event_type = event.get("type") tool_name = str(event.get("toolName", "")) call_id = str(event.get("toolCallId", "")) if event_type == "tool_execution_start" and tool_name in {"write", "edit"}: args = event.get("args") value = args.get("path", "") if isinstance(args, dict) else "" path = _normalized_relative_path(value) if isinstance(value, str) else None if path is not None: mutation_starts[call_id] = (tool_name, path) elif event_type == "tool_execution_start" and tool_name == "bash": args = event.get("args") command = args.get("command", "") if isinstance(args, dict) else "" if not isinstance(command, str): continue observed_commands.append(command) program = _direct_program(command, language) if program is not None: program_paths[call_id] = program elif event_type == "tool_execution_end": mutation = mutation_starts.get(call_id) if mutation is not None and event.get("isError") is not True: mutation_tool, mutation_path = mutation successful_mutation_paths.add(mutation_path) if mutation_tool == "write": successful_write_ids.add(call_id) else: successful_edit_ids.add(call_id) if call_id in program_paths and _successful_exit(event): successful_ids.add(call_id) if program_paths[call_id] in successful_mutation_paths: matched_execution_ids.add(call_id) if call_id in matched_execution_ids and _output_is_exact( _tool_result_text(event.get("result")), expected ): correct_output_ids.add(call_id) return { "attempted_execution": bool(program_paths), "successful_execution": bool(successful_ids), "executed_mutated_file": bool(matched_execution_ids), "execution_output_correct": bool(correct_output_ids), "wrote_file": bool(successful_write_ids), "edited_file": bool(successful_edit_ids), "file_mutation_succeeded": bool(successful_write_ids or successful_edit_ids), "commands": observed_commands, "executed_program_paths": sorted( {program_paths[call_id] for call_id in successful_ids} ), "mutated_file_paths": sorted(successful_mutation_paths), } def score_events(events: list[dict[str, Any]], expected: str, language: str) -> dict[str, Any]: response = final_assistant_text(events) answer, answer_format = extract_final_answer(response) execution = execution_metrics(events, language, expected) exact_answer_correct = answer_is_correct(answer, expected) response_answer_correct = exact_answer_correct or _text_contains_expected(response, expected) functional_correct = ( execution["file_mutation_succeeded"] and execution["executed_mutated_file"] and execution["execution_output_correct"] and response_answer_correct ) return { "response": response, "extracted_answer": answer, "answer_format": answer_format, "format_valid": answer_format == "boxed", "correct": exact_answer_correct, "response_answer_correct": response_answer_correct, "functional_correct": functional_correct, **execution, } def main() -> None: parser = argparse.ArgumentParser(description="Score one Pi JSON event stream.") parser.add_argument("events", type=Path) parser.add_argument("--answer", required=True) parser.add_argument("--language", required=True) args = parser.parse_args() print(json.dumps(score_events(load_jsonl(args.events), args.answer, args.language), indent=2)) if __name__ == "__main__": main()