From 7d3ce42040b5f6a400b3981ad49b70832cf55d90 Mon Sep 17 00:00:00 2001 From: BruceChen Date: Tue, 14 Apr 2026 11:08:01 +0000 Subject: [PATCH] feat: add theory-to-live pathing report --- tools/pathing_theory/report.py | 56 +++++++++++++++++++++ tools/pathing_theory_report.py | 29 +++++++++++ tools/tests/test_pathing_theory_report.py | 60 +++++++++++++++++++++++ 3 files changed, 145 insertions(+) create mode 100644 tools/pathing_theory/report.py create mode 100644 tools/pathing_theory_report.py create mode 100644 tools/tests/test_pathing_theory_report.py diff --git a/tools/pathing_theory/report.py b/tools/pathing_theory/report.py new file mode 100644 index 00000000..b9c50047 --- /dev/null +++ b/tools/pathing_theory/report.py @@ -0,0 +1,56 @@ +import json +from pathlib import Path + + +def classify_live_result(expected_result: str, live_result: str) -> str: + if live_result == "invalid_live_case": + return "invalid_live_case" + if expected_result == "pass" and live_result == "pass": + return "expected_pass/live_pass" + if expected_result == "pass" and live_result == "fail": + return "expected_pass/live_fail" + if expected_result == "reject" and live_result == "reject": + return "expected_reject/live_reject" + if expected_result == "reject" and live_result == "pass": + return "expected_reject/live_unexpected_pass" + return "invalid_live_case" + + +def summarize_results(rows: list[dict]) -> dict[str, int]: + summary: dict[str, int] = {} + for row in rows: + key = classify_live_result(row["expected_result"], row["live_result"]) + summary[key] = summary.get(key, 0) + 1 + return summary + + +def build_report(manifest_path: Path, results_path: Path) -> dict: + manifest_rows = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest_by_case = {row["case_id"]: row for row in manifest_rows} + result_rows = [ + json.loads(line) + for line in results_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + joined_rows: list[dict] = [] + for row in result_rows: + manifest = manifest_by_case.get(row["case_id"]) + if manifest is None: + joined_rows.append({**row, "classification": "invalid_live_case"}) + continue + joined_rows.append( + { + **manifest, + **row, + "classification": classify_live_result( + manifest["expected_result"], + row["live_result"], + ), + } + ) + + return { + "rows": joined_rows, + "summary": summarize_results(joined_rows), + } diff --git a/tools/pathing_theory_report.py b/tools/pathing_theory_report.py new file mode 100644 index 00000000..3b9fd65f --- /dev/null +++ b/tools/pathing_theory_report.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +import argparse +import json +from pathlib import Path +import sys + +REPO_ROOT = Path(__file__).resolve().parent.parent +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from tools.pathing_theory.report import build_report + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Join theory-aligned live results back to canonical cases.", + ) + parser.add_argument("--manifest", required=True) + parser.add_argument("--results", required=True) + parser.add_argument("--json-out", required=True) + args = parser.parse_args() + + report = build_report(Path(args.manifest), Path(args.results)) + Path(args.json_out).write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + print(f"Wrote report to {args.json_out}") + + +if __name__ == "__main__": + main() diff --git a/tools/tests/test_pathing_theory_report.py b/tools/tests/test_pathing_theory_report.py new file mode 100644 index 00000000..b13d32db --- /dev/null +++ b/tools/tests/test_pathing_theory_report.py @@ -0,0 +1,60 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from tools.pathing_theory.report import build_report, classify_live_result, summarize_results + + +class PathingTheoryReportTests(unittest.TestCase): + def test_classify_live_result_distinguishes_expected_pass_and_reject(self) -> None: + self.assertEqual(classify_live_result("pass", "pass"), "expected_pass/live_pass") + self.assertEqual(classify_live_result("pass", "fail"), "expected_pass/live_fail") + self.assertEqual(classify_live_result("reject", "reject"), "expected_reject/live_reject") + self.assertEqual(classify_live_result("reject", "pass"), "expected_reject/live_unexpected_pass") + + def test_summarize_results_counts_each_status(self) -> None: + rows = [ + {"case_id": "a", "expected_result": "pass", "live_result": "pass"}, + {"case_id": "b", "expected_result": "pass", "live_result": "fail"}, + {"case_id": "c", "expected_result": "reject", "live_result": "reject"}, + ] + + summary = summarize_results(rows) + + self.assertEqual(summary["expected_pass/live_pass"], 1) + self.assertEqual(summary["expected_pass/live_fail"], 1) + self.assertEqual(summary["expected_reject/live_reject"], 1) + + def test_build_report_keeps_case_traceability_fields(self) -> None: + manifest_rows = [ + { + "case_id": "linear-flat-sprint-mm12-gap5-dy0p0", + "bucket_id": "linear:flat:sprint:boundary", + "world_recipe_id": "linear-flat", + "expected_result": "pass", + } + ] + result_row = { + "case_id": "linear-flat-sprint-mm12-gap5-dy0p0", + "live_result": "pass", + "log_path": "/tmp/mcc-debug/mcc-debug.log", + } + + with tempfile.TemporaryDirectory() as temp_dir: + manifest_path = Path(temp_dir) / "manifest.json" + results_path = Path(temp_dir) / "results.jsonl" + manifest_path.write_text(json.dumps(manifest_rows), encoding="utf-8") + results_path.write_text(json.dumps(result_row) + "\n", encoding="utf-8") + + report = build_report(manifest_path, results_path) + + row = report["rows"][0] + self.assertEqual(row["bucket_id"], "linear:flat:sprint:boundary") + self.assertEqual(row["world_recipe_id"], "linear-flat") + self.assertEqual(row["log_path"], "/tmp/mcc-debug/mcc-debug.log") + self.assertEqual(row["classification"], "expected_pass/live_pass") + + +if __name__ == "__main__": + unittest.main()