feat: add theory-to-live pathing report

This commit is contained in:
BruceChen 2026-04-14 11:08:01 +00:00
parent ef3cd64475
commit 7d3ce42040
3 changed files with 145 additions and 0 deletions

View file

@ -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),
}

View file

@ -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()

View file

@ -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()