#!/usr/bin/env python3 """A local, synthetic evaluator smoke test. Python 3.10+, standard library only. Run: python3 evaluator-smoke-test.py --out evaluator-run Exit 0 means every fixture produced its expected outcome, NOT that all passed. The output directory must be new. No network, model calls, or GPU are used. This runs ONLY the fixed, trusted programs below. A Python subprocess is not a sandbox. Do not adapt this to execute untrusted agent code on your own machine. This checks a result protocol; it does not demonstrate agent improvement, held-out generalization, tamper resistance, or production resource isolation. SPDX-License-Identifier: MIT Copyright (c) 2026 Meshia Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import argparse import hashlib import json import platform import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path # These public examples are synthetic fixtures, not a hidden evaluation set. TASKS = [[2, 3], [-4, 1], [0, 0], [11, -2], [-7, -8], [100, 1]] EXPECTED = [5, -3, 0, 9, -15, 101] TIMEOUT_SECONDS = 1.0 MAX_OUTPUT_BYTES = 4096 # Protocol check after capture, not a memory limit. CORRECT = ( "import json,sys; tasks=json.load(sys.stdin); " "print(json.dumps({'predictions':[sum(row) for row in tasks]}))" ) # name, trusted fixture program, expected reason, expected grader score FIXTURES = [ ("correct", CORRECT, "accepted", 1.0), ("wrong_answers", "print('{\"predictions\":[0,0,0,0,0,0]}')", "incorrect", 1 / 6), ("fake_success_log", "print('All tests passed. score=1.0')", "invalid_json", None), ("self_reported_score", "print('{\"score\":1.0}')", "invalid_schema", None), ("missing_output", "pass", "missing_output", None), ("malformed_json", "print('{')", "invalid_json", None), ("non_finite_value", "print('{\"predictions\":[NaN,0,0,0,0,0]}')", "invalid_json", None), ("boolean_answer", "print('{\"predictions\":[true,0,0,0,0,0]}')", "invalid_schema", None), ("partial_results", "print('{\"predictions\":[5]}')", "invalid_schema", None), ("duplicate_key", "print('{\"predictions\":[],\"predictions\":[5,-3,0,9,-15,101]}')", "invalid_json", None), ("oversized_output", "print('x'*4097)", "output_too_large", None), ("nonzero_exit", CORRECT + "; sys.exit(1)", "nonzero_exit", None), ("timeout", "import time; time.sleep(5)", "timeout", None), ] def sha256(value): return hashlib.sha256(value).hexdigest() def encode(value): return (json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n").encode() def unique_keys(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("Duplicate JSON key") result[key] = value return result def reject_constant(value): raise ValueError("Non-finite JSON number: " + value) def grade(stdout, stderr, returncode, timed_out): """The runner owns status; the grader derives score from complete outputs.""" if timed_out: return "timeout", None if returncode != 0: return "nonzero_exit", None if max(len(stdout), len(stderr)) > MAX_OUTPUT_BYTES: return "output_too_large", None if not stdout.strip(): return "missing_output", None try: payload = json.loads( stdout.decode("utf-8"), object_pairs_hook=unique_keys, parse_constant=reject_constant, ) except (ValueError, UnicodeError, RecursionError): return "invalid_json", None if type(payload) is not dict or set(payload) != {"predictions"}: return "invalid_schema", None predictions = payload["predictions"] if (type(predictions) is not list or len(predictions) != len(EXPECTED) or any(type(answer) is not int for answer in predictions)): return "invalid_schema", None # Score every expected task; omissions cannot shrink the denominator. score = sum(a == b for a, b in zip(predictions, EXPECTED)) / len(EXPECTED) return ("accepted" if score == 1.0 else "incorrect"), score def run_fixture(name, source, expected_reason, expected_score, out, task_bytes): started = time.perf_counter() timed_out = False try: completed = subprocess.run( [sys.executable, "-I", "-S", "-c", source], input=task_bytes, capture_output=True, timeout=TIMEOUT_SECONDS, ) stdout, stderr, returncode = completed.stdout, completed.stderr, completed.returncode except subprocess.TimeoutExpired as error: stdout, stderr, returncode = error.stdout or b"", error.stderr or b"", None timed_out = True except OSError as error: # Launch failure is a harness error. Do not report it as a candidate loss. return {"fixture": name, "reason": "runner_error", "score": None, "error": str(error), "expectation_met": False} elapsed = time.perf_counter() - started reason, score = grade(stdout, stderr, returncode, timed_out) evidence = {} for suffix, content in [("stdout", stdout), ("stderr", stderr), ("py", source.encode())]: filename = name + "." + suffix (out / filename).write_bytes(content) evidence[suffix] = {"path": filename, "sha256": sha256(content)} return { "fixture": name, "reason": reason, "score": score, "expected_reason": expected_reason, "expected_score": expected_score, "expectation_met": reason == expected_reason and score == expected_score, "runner": {"returncode": returncode, "timed_out": timed_out, "elapsed_seconds": round(elapsed, 6)}, "candidate_sha256": sha256(source.encode()), "evidence": evidence, } def main(): parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) parser.add_argument("--out", type=Path, default=Path("evaluator-run")) args = parser.parse_args() if sys.version_info < (3, 10): parser.error("Python 3.10 or newer is required") # No overwrite: a prior run remains an independent record. try: args.out.mkdir(parents=True, exist_ok=False) except FileExistsError: parser.error("Output already exists; choose a new --out directory") task_bytes = encode(TASKS) (args.out / "tasks.json").write_bytes(task_bytes) (args.out / "expected.json").write_bytes(encode(EXPECTED)) records = [run_fixture(*fixture, args.out, task_bytes) for fixture in FIXTURES] passed = sum(record["expectation_met"] for record in records) report = { "schema": "meshia.evaluator-smoke-test.v1", "created_at": datetime.now(timezone.utc).isoformat(), "scope": "Synthetic protocol fixtures only; no model, search, sandbox, or held-out claim", "environment": {"python": platform.python_version(), "os": platform.system(), "machine": platform.machine()}, "evaluator_sha256": sha256(Path(__file__).read_bytes()), "task_manifest_sha256": sha256(task_bytes), "expected_answers_sha256": sha256(encode(EXPECTED)), "model_id": None, "seed": None, "budget": {"fixture_timeout_seconds": TIMEOUT_SECONDS, "protocol_output_limit_bytes": MAX_OUTPUT_BYTES}, "fixtures": records, "summary": {"expectations_met": passed, "fixtures": len(records), "accepted_candidates": sum(r["reason"] == "accepted" for r in records)}, } (args.out / "report.json").write_bytes(encode(report)) for record in records: print(f"{record['fixture']}: {record['reason']}") print(f"{passed}/{len(records)} expected outcomes; report: {args.out / 'report.json'}") return 0 if passed == len(records) else 1 if __name__ == "__main__": raise SystemExit(main())