#!/usr/bin/env python3 """Compare the source LinuxCNC XYZAC oracle with the WASM worker trace.""" from __future__ import annotations import hashlib import json import subprocess import math from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] NATIVE_REPORT = ROOT / "coverage/xyzac-native-tick-trace.report.json" WASM_REPORT = ROOT / "coverage/xyzac-wasm-tick-trace.report.json" REPORT = ROOT / "coverage/xyzac-native-wasm-tick-parity.report.json" def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def digest(value: Any) -> str: return hashlib.sha256((value if isinstance(value, bytes) else canonical(value).encode())).hexdigest() def first_difference(left: Any, right: Any, path: str = "$") -> str | None: if type(left) is not type(right): return path if isinstance(left, dict): for key in sorted(set(left) | set(right)): if key not in left or key not in right: return f"{path}.{key}" difference = first_difference(left[key], right[key], f"{path}.{key}") if difference: return difference return None if isinstance(left, list): for index in range(max(len(left), len(right))): if index >= len(left) or index >= len(right): return f"{path}[{index}]" difference = first_difference(left[index], right[index], f"{path}[{index}]") if difference: return difference return None return None if left == right else path def canonical_row(row: dict[str, Any]) -> dict[str, Any]: result = dict(row) for key in ("joint_actual", "joint_commanded", "homed", "limits"): if isinstance(result.get(key), list): result[key] = result[key][:5] return result def values_equal(left: Any, right: Any, absolute: float = 1e-9, relative: float = 1e-9) -> bool: if isinstance(left, (int, float)) and isinstance(right, (int, float)) and not isinstance(left, bool) and not isinstance(right, bool): scale = max(abs(float(left)), abs(float(right)), 1.0) return abs(float(left) - float(right)) <= max(absolute, relative * scale) if isinstance(left, list) and isinstance(right, list): return len(left) == len(right) and all(values_equal(a, b, absolute, relative) for a, b in zip(left, right)) return left == right def compare_rows(native: list[dict[str, Any]], wasm: list[dict[str, Any]]) -> dict[str, Any]: authoritative_fields = ("task_state", "task_mode", "interp_state", "exec_state", "current_line", "motion_line", "commanded_pose", "joint_commanded", "velocity", "acceleration", "queue_depth", "active_queue", "queue_full", "motion_id", "active_segment_id", "position", "homed", "limits", "kinematics_type", "motion_type", "paused") feedback_fields = ("actual_pose", "joint_actual") def segments(rows: list[dict[str, Any]]) -> list[tuple[int, list[dict[str, Any]]]]: result: list[tuple[int, list[dict[str, Any]]]] = [] for row in rows: motion_id = int(row.get("motion_id") or 0) if motion_id > 0: if not result or result[-1][0] != motion_id: result.append((motion_id, [])) result[-1][1].append(row) return result native_segments = segments(native) wasm_segments = segments(wasm) if [motion_id for motion_id, _ in native_segments] != [motion_id for motion_id, _ in wasm_segments]: return {"status": "diverged", "category": "segment_schedule", "first_divergent_tick": 0, "difference_path": "$.segments", "native": [motion_id for motion_id, _ in native_segments], "wasm": [motion_id for motion_id, _ in wasm_segments]} sample_index = 0 duration_deltas: dict[str, int] = {} for segment_index, ((motion_id, native_segment), (_, wasm_segment)) in enumerate(zip(native_segments, wasm_segments)): if len(native_segment) != len(wasm_segment): duration_deltas[f"{segment_index}:{motion_id}"] = len(native_segment) - len(wasm_segment) if segment_index != len(native_segments) - 1: return {"status": "diverged", "category": "segment_duration", "first_divergent_tick": sample_index, "difference_path": f"$.segments[{segment_index}].length", "native": len(native_segment), "wasm": len(wasm_segment)} for native_row, wasm_row in zip(native_segment, wasm_segment): native_row = canonical_row(native_row) wasm_row = canonical_row(wasm_row) index = sample_index sample_index += 1 for category, fields in (("authoritative", authoritative_fields), ("feedback", feedback_fields)): for field in fields: if not values_equal(native_row.get(field), wasm_row.get(field)): return {"status": "diverged", "category": category, "motion_id": motion_id, "first_divergent_tick": index, "difference_path": f"$.segments[{motion_id}][{index}].{field}", "native": native_row.get(field), "wasm": wasm_row.get(field)} return {"status": "verified_partial" if duration_deltas else "verified", "category": "all", "first_divergent_tick": None, "difference_path": None, "duration_deltas": duration_deltas} def main() -> int: native = json.loads(NATIVE_REPORT.read_text(encoding="utf-8")) native_rows = native.get("rows", []) target_rows = max(1, min(5000, int(native.get("row_count") or 1))) elapsed_ms = max((float(row.get("native_elapsed_ms", 0.0)) for row in native_rows), default=0.0) wasm_ticks = max(1, math.ceil(elapsed_ms / 10.0)) # A full XYZAC remap run can contain tens of thousands of 1 ms servo # samples. The timeout is a liveness guard, not a parity boundary; keep # it well above the native elapsed window so a valid complete replay is # never replaced by a stale short-window report. timeout_seconds = max(1800, int(elapsed_ms / 1000.0 * 8 + 120)) try: build = subprocess.run(["node", "scripts/run_xyzac_wasm_tick_trace.mjs", "--ticks", str(wasm_ticks), "--output", str(WASM_REPORT)], cwd=ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False, timeout=timeout_seconds) except subprocess.TimeoutExpired as error: output = error.stdout or "" if isinstance(output, bytes): output = output.decode("utf-8", errors="replace") build = subprocess.CompletedProcess(error.cmd, 124, stdout=output + "\nwasm_replay_timeout") wasm = json.loads(WASM_REPORT.read_text(encoding="utf-8")) if WASM_REPORT.is_file() else {"status": "blocked", "rows": []} comparison = compare_rows(native.get("rows", [])[:target_rows], wasm.get("rows", [])[:target_rows]) if native.get("status") == "verified" and wasm.get("status") == "verified" else {"status": "blocked", "first_divergent_tick": None, "difference_path": None} report = { "schema": "xyzac-native-wasm-tick-parity.v1", "status": comparison["status"], "candidate_eligible": comparison["status"] == "verified", "case": native.get("case"), "native": {"report": str(NATIVE_REPORT.relative_to(ROOT)), "report_sha256": native.get("report_sha256"), "trace_sha256": native.get("trace_sha256"), "row_count": native.get("row_count")}, "wasm": {"report": str(WASM_REPORT.relative_to(ROOT)), "report_sha256": wasm.get("report_sha256"), "trace_sha256": wasm.get("trace_sha256"), "row_count": wasm.get("row_count"), "runner_returncode": build.returncode, "runner_output_sha256": digest(build.stdout)}, "comparison": comparison, "first_divergent_tick": comparison.get("first_divergent_tick"), "difference_path": comparison.get("difference_path"), "known_boundaries": ["native public stat does not expose TP segment geometry", "WASM public snapshot does not expose acceleration/active blend geometry", "verified oracle availability is not a full TP numerical parity claim"], "release_status": "STOP_RELEASE", } report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"}) REPORT.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") print(json.dumps({"schema": report["schema"], "status": report["status"], "first_divergent_tick": report["first_divergent_tick"], "difference_path": report["difference_path"], "report_sha256": report["report_sha256"]}, ensure_ascii=False)) return 0 if report["status"] == "verified" else 1 if __name__ == "__main__": raise SystemExit(main())