#!/usr/bin/env python3 """Reproduce a retrospective, function-only rover voting comparison. Standard library only; Python 3.9+. Examples: python3 -B rover-voting-evidence.py --output rover-voting-evidence.json python3 -B rover-voting-evidence.py --repo /path/to/clone --output result.json Local reads use git cat-file at full pinned SHAs, never the working tree. Missing local objects fall back to public raw URLs. No git fetch, rover imports, main blocks, device access, or test-runner plugins are executed. Only --output is written. Without --output, JSON goes to stdout. Exit 0 means the specified comparison expectations held; exit 1 means they did not; exit 2 means evidence could not be retrieved or checked. """ import argparse import ast import builtins import copy import datetime import hashlib import json from pathlib import Path import platform import subprocess import sys import urllib.error import urllib.request PUBLIC_REPO = "https://github.com/Robiswell/Euler_Rover_2026" RAW_ROOT = "https://raw.githubusercontent.com/Robiswell/Euler_Rover_2026" API_ROOT = "https://api.github.com/repos/Robiswell/Euler_Rover_2026" PARENT = "ffdb479d11a01b13ed928eb431a24c10827289f1" FIX = "484a18b64e121635a9161a07736a2883fb9d478c" REVIEWED_HEAD = "bba2a57fcd80a41d3f30ef6e64f3bfd26b412f95" PRODUCTION_FILE = "final_full_gait_test.py" TEST_FILE = "test_nav_logic.py" CONSTANT_NAMES = {"DIST_DANGER", "DIST_NEAR", "DIST_CAUTION", "DIST_CLEAR", "DIST_UNKNOWN"} EXPECTED_PARENT_FAILURES = { "test_voting_front_1of3_danger_rejected", "test_voting_front_multipath_bounce_rejected", "test_voting_left_2sensor_disagree_downgrade", "test_voting_right_2sensor_disagree_downgrade", "test_voting_fault_sensor_treated_as_caution_not_danger", "test_voting_left_near_vs_clear_downgrade", } class EvidenceError(Exception): pass def require(condition, message): if not condition: raise EvidenceError(message) def digest(data): return hashlib.sha256(data).hexdigest() def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False) def normalize_numbers(value): """Treat e.g. an explicit 100 and a default 100.0 as the same input.""" if isinstance(value, dict): return {key: normalize_numbers(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [normalize_numbers(item) for item in value] if isinstance(value, float) and value.is_integer(): return int(value) return value def public_get(url): request = urllib.request.Request(url, headers={"User-Agent": "rover-voting-evidence/1", "Accept": "application/vnd.github+json"}) # Preserve the host's normal TLS and proxy configuration. try: with urllib.request.urlopen(request, timeout=30) as response: data = response.read(2_000_001) except urllib.error.HTTPError as error: raise EvidenceError("Public source returned HTTP %s: %s" % (error.code, url)) from None except (urllib.error.URLError, TimeoutError, OSError): raise EvidenceError("Public source unavailable: " + url) from None require(len(data) <= 2_000_000, "Public source exceeds the bounded read limit") return data def load_source(revision, filename, repo): raw_url = "%s/%s/%s" % (RAW_ROOT, revision, filename) data = None retrieval = "public_raw" if repo: try: result = subprocess.run( ["git", "-C", repo, "cat-file", "blob", revision + ":" + filename], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30, check=False, ) if result.returncode == 0: data = result.stdout retrieval = "local_committed_git_blob" except (OSError, subprocess.TimeoutExpired): pass if data is None: retrieval = "public_raw_fallback_local_object_unavailable" if data is None: data = public_get(raw_url) text = data.decode("utf-8-sig") record = { "revision": revision, "path": filename, "public_url": "%s/blob/%s/%s" % (PUBLIC_REPO, revision, filename), "raw_url": raw_url, "retrieval": retrieval, "byte_count": len(data), "raw_sha256": digest(data), "computed_git_blob_sha1": hashlib.sha1(b"blob " + str(len(data)).encode("ascii") + b"\0" + data).hexdigest(), } return text, record def ast_digest(nodes): module = ast.Module(body=nodes, type_ignores=[]) return digest(ast.dump(module, annotate_fields=True, include_attributes=False).encode("utf-8")) def describe_node(node, text, record): segment = ast.get_source_segment(text, node) require(segment is not None, "Missing source segment for selected AST node") return { "name": node.name if isinstance(node, ast.FunctionDef) else node.targets[0].id, "start_line": node.lineno, "end_line": node.end_lineno, "source_url": "%s#L%d-L%d" % (record["public_url"], node.lineno, node.end_lineno), "source_segment_utf8_sha256": digest(segment.encode("utf-8")), "ast_sha256": ast_digest([node]), } def select_production(text, classifier): tree = ast.parse(text) constants = [node for node in ast.walk(tree) if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name) and node.targets[0].id in CONSTANT_NAMES] functions = [node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef) and node.name in {"classify_distance", classifier}] require(len(constants) == 5 and {node.targets[0].id for node in constants} == CONSTANT_NAMES, "Expected exactly five unique production distance constants") require(len(functions) == 2 and {node.name for node in functions} == {"classify_distance", classifier}, "Expected exactly two unique production classifier functions") return sorted(constants, key=lambda node: node.lineno) + sorted(functions, key=lambda node: node.lineno) def validate_selected_code(nodes): """Reject unneeded capabilities before compiling the small reviewed subset. This is a narrow extraction check, not a general Python security sandbox. No source-module imports or top-level runtime statements are executed. """ permitted_nodes = { ast.Module, ast.FunctionDef, ast.arguments, ast.arg, ast.Assign, ast.Name, ast.Load, ast.Store, ast.Constant, ast.Expr, ast.Return, ast.If, ast.Compare, ast.Is, ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.GtE, ast.Call, ast.keyword, ast.Subscript, ast.Tuple, ast.List, ast.Dict, ast.GeneratorExp, ast.comprehension, ast.BinOp, ast.Sub, ast.UnaryOp, ast.USub, ast.Assert, } permitted_calls = {"any", "max", "min", "sorted", "classify_distance", "classify_sectors", "make_frame"} for node in ast.walk(ast.Module(body=nodes, type_ignores=[])): require(type(node) in permitted_nodes, "Unexpected selected AST node: " + type(node).__name__) if isinstance(node, ast.FunctionDef): require(not node.decorator_list and node.returns is None, "Unexpected function decoration or return annotation") if isinstance(node, ast.Name): require(not node.id.startswith("__"), "Dunder access is outside the selected code scope") if isinstance(node, ast.Call): require(isinstance(node.func, ast.Name) and node.func.id in permitted_calls, "Unexpected call in selected code") def public_attribution(): url = API_ROOT + "/commits/" + FIX raw = public_get(url) metadata = json.loads(raw) require(metadata.get("sha") == FIX, "Public commit metadata SHA does not match the pinned fix") require([entry["sha"] for entry in metadata["parents"]] == [PARENT], "Pinned parent does not match public commit metadata") commit = metadata["commit"] return { "source_url": url, "response_bytes_sha256": digest(raw), "commit_url": metadata["html_url"], "revision": FIX, "parent_revisions": [entry["sha"] for entry in metadata["parents"]], "author": {key: commit["author"][key] for key in ("name", "date")}, "committer": {key: commit["committer"][key] for key in ("name", "date")}, "github_author_login": (metadata.get("author") or {}).get("login"), "github_committer_login": (metadata.get("committer") or {}).get("login"), "subject": commit["message"].splitlines()[0], "signature_verification": {key: commit["verification"].get(key) for key in ("verified", "reason")}, "limits": "Public attribution metadata is not proof of exclusive authorship, tool use, deployment authority, or human acceptance. Emails and the full API response are not stored.", } def compare(repo): started = datetime.datetime.now(datetime.timezone.utc).isoformat() test_text, fixture_source = load_source(FIX, TEST_FILE, repo) test_tree = ast.parse(test_text) fixtures = [node for node in test_tree.body if isinstance(node, ast.FunctionDef) and (node.name == "make_frame" or node.name.startswith("test_voting_"))] tests = [node for node in fixtures if node.name.startswith("test_voting_")] require(len(fixtures) == 16 and len(tests) == 15 and len({node.name for node in fixtures}) == 16, "Expected make_frame plus exactly 15 unique committed voting assertion functions") fixture_catalog = {} runs = [] source_records = [fixture_source] attribution = public_attribution() for label, revision, classifier in ( ("parent", PARENT, "classify_sectors"), ("fix", FIX, "classify_sectors_voted"), ("reviewed_head", REVIEWED_HEAD, "classify_sectors_voted"), ): text, source = load_source(revision, PRODUCTION_FILE, repo) source_records.append(source) production_nodes = select_production(text, classifier) selected = production_nodes + fixtures validate_selected_code(selected) module = ast.Module(body=copy.deepcopy(selected), type_ignores=[]) code = compile(module, "public-rover:" + revision, "exec", dont_inherit=True, optimize=0) per_test = [] for test in tests: # Fresh globals for each assertion prevent cross-test state leakage. namespace = {"__builtins__": {key: getattr(builtins, key) for key in ("any", "max", "min", "sorted", "AssertionError")}} exec(code, namespace) implementation = namespace[classifier] observations = [] def observed_classifier(frame): normalized = normalize_numbers(frame) input_id = digest(canonical(normalized).encode("ascii")) entry = fixture_catalog.setdefault(input_id, {"normalized_input_sha256": input_id, "frame": normalized, "test_names": []}) if test.name not in entry["test_names"]: entry["test_names"].append(test.name) result = implementation(frame) observations.append({"input_sha256": input_id, "actual_front_left_right": list(result)}) return result namespace["classify_sectors"] = observed_classifier try: namespace[test.name]() status = "pass" except AssertionError: status = "fail" require(len(observations) == 1, "Expected exactly one classifier call per committed voting assertion") per_test.append({"test_name": test.name, "status": status, "observations": observations}) passed = sum(item["status"] == "pass" for item in per_test) runs.append({ "label": label, "revision": revision, "source": source, "classifier": classifier, "production_nodes": [describe_node(node, text, source) for node in production_nodes], "production_ast_sha256": ast_digest(production_nodes), "executed_module_ast_sha256": ast_digest(selected), "passed": passed, "failed": len(per_test) - passed, "per_test": per_test, }) expected_tests = [node.name for node in tests] same_inputs = all( [item["observations"][0]["input_sha256"] for item in run["per_test"]] == [item["observations"][0]["input_sha256"] for item in runs[0]["per_test"]] for run in runs[1:] ) expectations = { "parent_has_exact_six_previously_observed_failures": {item["test_name"] for item in runs[0]["per_test"] if item["status"] == "fail"} == EXPECTED_PARENT_FAILURES, "fix_passes_all_15_assertion_functions": runs[1]["passed"] == 15, "reviewed_head_passes_all_15_assertion_functions": runs[2]["passed"] == 15, "identical_ordered_fixture_inputs_across_revisions": same_inputs, } return { "schema_version": 1, "public_repository": PUBLIC_REPO, "started_at_utc": started, "finished_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "runtime": {"implementation": platform.python_implementation(), "python_version": platform.python_version()}, "script": {"name": "rover-voting-evidence.py", "sha256": digest(Path(__file__).read_bytes())}, "scope": { "retrospective": True, "function_only": True, "field_test": False, "description": "Replay the fix commit's 15 voting assertion functions against extracted production distance constants and classifier functions from three pinned revisions.", "not_established": ["raw telemetry authenticity or physical multipath cause", "full test-suite or CI success", "serial parsing, smoothing, navigation integration, or device behavior", "deployment, field retest, or measured operating improvement", "stakeholder acceptance or decision rights", "exclusive authorship or Claude Code/MCP use"], }, "method": { "retrieval": "Read committed blobs with git cat-file if --repo is supplied; otherwise or when objects are missing, GET exact pinned public raw files. Never fetch into the repository or read worktree source files. Public commit attribution always uses unauthenticated GitHub REST.", "execution": "Parse source with ast; select only five distance constants, classify_distance, the revision's sector classifier, make_frame, and 15 test_voting functions. Check a narrow AST allowlist, compile with assertions enabled, execute in fresh globals per test with only any/max/min/sorted/AssertionError builtins, and instrument classifier calls. No imports or main blocks from rover code execute.", "hashes": "raw_sha256 hashes exact downloaded/git blob bytes. computed_git_blob_sha1 hashes the Git blob header plus those bytes. Source-segment hashes use UTF-8 ast.get_source_segment. AST hashes use ast.dump without location attributes; these may vary with Python versions. Input hashes use canonical sorted ASCII JSON with integral floats normalized to integers. Hashes document observed content; they do not authenticate identity or approvals.", "fixture_scope": "15 assertion functions are not 15 independent new scenarios. Full normalized frames are grouped across tests; repeated runs across revisions are paired comparisons, not additional observations. Distinct synthetic inputs are not independent field samples.", "expectations": "Parent failures are expected when applying the fix's new assertions retrospectively; they are not a claim that the historical parent suite failed. A successful script exit means the explicit comparison expectations matched, not that every test on every revision passed.", "scope_limits": "Direct function extraction bypasses enclosing main blocks, imports, hardware, and upstream processing. The 119-test historical pass claim is not verified here. The reviewed head is pinned, not resolved from today's moving main branch.", }, "sources": source_records, "fix_attribution": attribution, "fixtures": { "revision": FIX, "source": fixture_source, "selected_nodes": [describe_node(node, test_text, fixture_source) for node in fixtures], "selected_ast_sha256": ast_digest(fixtures), "test_names": expected_tests, "assertion_function_count": len(tests), "distinct_normalized_input_count": len(fixture_catalog), "inputs": list(fixture_catalog.values()), "shared_input_groups": [{"input_sha256": key, "test_names": value["test_names"]} for key, value in fixture_catalog.items() if len(value["test_names"]) > 1], }, "severity_encoding": {"CLEAR": 0, "CAUTION_OR_UNKNOWN": 1, "NEAR": 2, "DANGER": 3}, "runs": runs, "comparison_expectations": expectations, "comparison_expectations_met": all(expectations.values()), } def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--repo", help="Optional local git repository; missing committed objects use public raw files") parser.add_argument("--output", help="JSON destination; omit to print JSON to stdout") args = parser.parse_args() try: report = compare(args.repo) serialized = json.dumps(report, indent=2, ensure_ascii=True, allow_nan=False) + "\n" if args.output: Path(args.output).write_text(serialized, encoding="ascii") print(canonical({"comparison_expectations_met": report["comparison_expectations_met"], "distinct_normalized_inputs": report["fixtures"]["distinct_normalized_input_count"], "runs": [{key: run[key] for key in ("label", "passed", "failed")} for run in report["runs"]]})) else: sys.stdout.write(serialized) return 0 if report["comparison_expectations_met"] else 1 except EvidenceError as error: print(canonical({"error": str(error), "report_written": False}), file=sys.stderr) except Exception as error: # Avoid exposing a local path through exceptions or tracebacks. print(canonical({"error": type(error).__name__, "report_written": False}), file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())