#!/usr/bin/env python3 """Generate the final S5 source-test parity, repeat and recovery report.""" from __future__ import annotations import argparse import hashlib import json import subprocess from pathlib import Path from typing import Any CONTEXT_ID = "linuxcnc-2.10.0-pre0-wasm-m1-py-c1" REPORTS = ( "coverage/linuxcnc-source-test-s5.report.json", "coverage/linuxcnc-source-test-execution-s5.report.json", "coverage/linuxcnc-source-unit-execution-s5.report.json", "coverage/linuxcnc-source-interp-s5.report.json", "coverage/linuxcnc-source-hal-s5.report.json", "coverage/linuxcnc-source-io-s5.report.json", "coverage/linuxcnc-source-config-s5.report.json", ) REPEAT_COMMANDS = { "source_interpreter": ("node", "web/worker/test_source_interpreter_s5.mjs"), "source_hal": ("node", "web/worker/test_source_hal_s5.mjs"), "source_io": ("node", "web/worker/test_source_io_s5.mjs"), "source_config": ("node", "web/worker/test_source_config_s5.mjs"), "source_parity": ("node", "web/worker/test_source_parity_s5.mjs"), "endpoint_replay": ("node", "web/worker/test_endpoint_replay_s4.mjs"), "recovery": ("node", "web/worker/test_recovery_s3.mjs"), } def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def digest(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def command(root: Path, args: tuple[str, ...]) -> dict[str, Any]: result = subprocess.run(args, cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) return {"command": list(args), "returncode": result.returncode, "status": "PASSED" if result.returncode == 0 else "FAILED", "stdout_sha256": digest(result.stdout), "stderr_sha256": digest(result.stderr), "stdout": result.stdout.decode("utf-8", errors="replace"), "stderr": result.stderr.decode("utf-8", errors="replace")} def artifact(root: Path, path: str, report: dict[str, Any]) -> dict[str, Any]: value = root / path return {"path": path, "bytes": value.stat().st_size, "sha256": digest(value.read_bytes()), "task_id": report.get("task_id"), "status": report.get("status"), "schema": report.get("schema")} def generate(root: Path, output: Path) -> dict[str, Any]: source_reports = [] report_values = {} all_verified = True for path in REPORTS: value = json.loads((root / path).read_text(encoding="utf-8")) report_values[path] = value source_reports.append(artifact(root, path, value)) all_verified = all_verified and value.get("status") == "verified" repeats: dict[str, Any] = {} repeat_stable = True parity_sample = None for name, args in REPEAT_COMMANDS.items(): runs = [] parsed = [] for _ in range(2): result = command(root, args) runs.append({key: value for key, value in result.items() if key not in {"stdout", "stderr"}}) try: parsed.append(json.loads(result["stdout"])) except json.JSONDecodeError: parsed.append(None) stable = all(run == runs[0] for run in runs) repeat_stable = repeat_stable and stable and runs[0]["status"] == "PASSED" repeats[name] = {"runs": runs, "stable": stable} if name == "source_parity" and parsed[0] is not None: parity_sample = parsed[0] s1 = report_values[REPORTS[0]] s2 = report_values[REPORTS[1]] s3 = report_values[REPORTS[2]] s4 = report_values[REPORTS[3]] s5 = report_values[REPORTS[4]] s6 = report_values[REPORTS[5]] s7 = report_values[REPORTS[6]] def static_golden_equal(value: dict[str, Any]) -> bool: if "golden_diff" in value: return value.get("golden_diff", {}).get("byte_identical", False) return bool(value) and all(item.get("golden_diff", {}).get("byte_identical", False) for item in value.values()) native_wasm_golden = { "S5-03": s3.get("adapter", {}).get("golden_diff", {}).get("byte_identical", False), "S5-05": static_golden_equal(s5.get("native_wasm", {})), "S5-06": static_golden_equal(s6.get("native_wasm", {})), } difference_summary = { "first_divergence": None, "native_wasm_golden": native_wasm_golden, "pending_or_blocked": { "S5-01": s1.get("counts", {}).get("adapter_status", {}), "S5-02": s2.get("counts", {}), "S5-04": s4.get("counts", {}), "S5-05": s5.get("counts", {}), "S5-06": s6.get("counts", {}), "S5-07": s7.get("counts", {}), }, "all_native_wasm_static_goldens_equal": all(native_wasm_golden.values()), } recovery = { "status": "verified" if parity_sample and parity_sample.get("recovery", {}).get("status") == "recovered" else "blocked", "first_divergence": parity_sample.get("differences", {}).get("first_divergence") if parity_sample else "parity_output_missing", "mode_equal": parity_sample.get("differences", {}).get("mode", {}).get("equal") if parity_sample else False, "recovery_equal": parity_sample.get("differences", {}).get("recovery", {}).get("equal") if parity_sample else False, "trace_import_equal": parity_sample.get("differences", {}).get("trace_import", {}).get("equal") if parity_sample else False, "checkpoint_hash": parity_sample.get("recovery", {}).get("checkpoint_hash") if parity_sample else None, "continuous_hash": parity_sample.get("recovery", {}).get("continuous_hash") if parity_sample else None, "recovered_hash": parity_sample.get("recovery", {}).get("recovered_hash") if parity_sample else None, "trace_hash": parity_sample.get("recovery", {}).get("trace_hash") if parity_sample else None, } parity_ok = recovery["status"] == "verified" and recovery["first_divergence"] is None and recovery["mode_equal"] and recovery["recovery_equal"] and recovery["trace_import_equal"] report: dict[str, Any] = { "schema": "linuxcnc-source-parity-s5.v1", "task_id": "S5-08", "context_id": CONTEXT_ID, "status": "verified" if all_verified and repeat_stable and parity_ok and difference_summary["all_native_wasm_static_goldens_equal"] else "blocked", "source_reports": source_reports, "coverage": { "source_matrix_entries": s1.get("counts", {}).get("matrix_registered_entries"), "source_matrix_entrypoints": s1.get("counts", {}).get("matrix_entrypoints"), "native_pure_baseline_passed": s2.get("counts", {}).get("PASSED"), "unit_adapter_cases": s3.get("adapter", {}).get("case_count"), "configuration_entries": s7.get("counts", {}).get("total"), "configuration_ready": s7.get("counts", {}).get("READY"), "configuration_blocked": s7.get("counts", {}).get("BLOCKED"), }, "difference_summary": difference_summary, "repeat_execution": {"runs_per_command": 2, "stable": repeat_stable, "commands": repeats}, "recovery_endpoints": recovery, "official_oracle": {"status": "SOURCE_BLOCKED", "reason": "locked source lacks official kinematics/recovery oracle and full desktop source runner"}, "boundaries": [ "first divergence is null for all implemented native/WASM/Worker static goldens and endpoint replay modes", "SOURCE_BLOCKED, HOST_EXCLUDED and WORKER_TASK_MOTION_PENDING remain separate from passed counts", "recovery equality covers logical checkpoint/replay/trace import; browser OPFS crash oracle remains blocked", ], } report["report_sha256"] = digest(canonical({key: value for key, value in report.items() if key != "report_sha256"}).encode("utf-8")) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") return report def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--report", type=Path, default=Path("coverage/linuxcnc-source-parity-s5.report.json")) args = parser.parse_args() report = generate(args.repo_root.resolve(), args.repo_root.resolve() / args.report) print(f"linuxcnc_source_parity_s5={report['status']} repeats={report['repeat_execution']['stable']} recovery={report['recovery_endpoints']['status']}") return 0 if report["status"] == "verified" else 1 if __name__ == "__main__": raise SystemExit(main())