#!/usr/bin/env python3 """Compare selected XYZAC native, WASM and Worker traces by semantic level.""" from __future__ import annotations import copy import hashlib import json import subprocess from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] REPORT = ROOT / "coverage/xyzac-trt-trace-parity.report.json" CASE = "linuxcnc/configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzac-trt.ini" CONTEXT_ID = "linuxcnc-2.10.0-pre0-wasm-m1-py-c1" LEVELS = ("protocol", "control", "state", "canonical", "recovery") COMMANDS = { "native": ["./build-native/sim_xyzac_trace_test"], "wasm": ["node", "build-wasm/sim_xyzac_trace_wasm_test.js"], "worker": ["node", "web/worker/test_xyzac_trace_parity.mjs"], } def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def digest(value: str | bytes) -> str: data = value.encode() if isinstance(value, str) else value return hashlib.sha256(data).hexdigest() def first_path(left: Any, right: Any, path: str = "$") -> str: 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}" if left[key] != right[key]: return first_path(left[key], right[key], f"{path}.{key}") return path 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}]" if left[index] != right[index]: return first_path(left[index], right[index], f"{path}[{index}]") return path return path def row_tick(left: Any, right: Any, index: int) -> int: for value in (left, right): if isinstance(value, dict) and isinstance(value.get("tick"), int): return value["tick"] return index + 1 def compare_level(left: list[Any], right: list[Any]) -> dict[str, Any]: for index in range(max(len(left), len(right))): lhs = left[index] if index < len(left) else None rhs = right[index] if index < len(right) else None if lhs != rhs: return { "status": "diverged", "first_divergent_tick": row_tick(lhs, rhs, index), "difference_path": first_path(lhs, rhs, f"$[{index}]"), "left": lhs, "right": rhs, } return {"status": "verified", "first_divergent_tick": None, "difference_path": None} def compare_traces(left: dict[str, Any], right: dict[str, Any]) -> dict[str, Any]: comparisons = {level: compare_level(left["levels"][level], right["levels"][level]) for level in LEVELS} divergent = [value["first_divergent_tick"] for value in comparisons.values() if value["first_divergent_tick"] is not None] return { "status": "verified" if not divergent else "diverged", "levels": comparisons, "first_divergent_tick": min(divergent) if divergent else None, } def validate_trace(value: Any) -> dict[str, Any]: if not isinstance(value, dict) or value.get("schema") != "xyzac-structured-trace.v1" or value.get("case") != "xyzac-trt.ini": raise ValueError("xyzac_trace_schema") levels = value.get("levels") if not isinstance(levels, dict) or any(not isinstance(levels.get(level), list) for level in LEVELS): raise ValueError("xyzac_trace_levels") return value def run_trace(name: str) -> tuple[dict[str, Any], dict[str, Any]]: result = subprocess.run(COMMANDS[name], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, timeout=30) trace = None for line in reversed(result.stdout.splitlines()): try: candidate = json.loads(line) except json.JSONDecodeError: continue if isinstance(candidate, dict) and candidate.get("schema") == "xyzac-structured-trace.v1": trace = validate_trace(candidate) break if result.returncode != 0 or trace is None or trace.get("status") != "verified": raise RuntimeError(f"{name}_trace_failed:{result.returncode}") evidence = {"command": COMMANDS[name], "returncode": result.returncode, "output_sha256": digest(result.stdout), "trace_sha256": digest(canonical(trace["levels"]))} return trace, evidence def main() -> int: build_commands = [ ["cmake", "--build", "build-native", "--target", "sim_xyzac_trace_test"], ["cmake", "--build", "build-wasm", "--target", "sim_xyzac_trace_wasm_test"], ] build_evidence = [] failure_samples: list[dict[str, Any]] = [] traces: dict[str, dict[str, Any]] = {} trace_evidence: dict[str, dict[str, Any]] = {} status = "verified" try: for command in build_commands: result = subprocess.run(command, cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, timeout=60) build_evidence.append({"command": command, "returncode": result.returncode, "output_sha256": digest(result.stdout)}) if result.returncode != 0: raise RuntimeError(f"build_failed:{command[-1]}") for name in COMMANDS: traces[name], trace_evidence[name] = run_trace(name) native_wasm = compare_traces(traces["native"], traces["wasm"]) native_worker = compare_traces(traces["native"], traces["worker"]) mutated = copy.deepcopy(traces["worker"]) mutated["levels"]["state"][1]["active_type"] = 99 positive_sample = compare_traces(traces["native"], mutated) positive_ok = positive_sample["status"] == "diverged" and positive_sample["first_divergent_tick"] == 2 and positive_sample["levels"]["state"]["difference_path"] == "$[1].active_type" if native_wasm["status"] != "verified" or native_worker["status"] != "verified" or not positive_ok: status = "blocked" failure_samples.append({"code": "TRACE_PARITY_DIVERGED", "native_wasm": native_wasm, "native_worker": native_worker, "positive_sample_ok": positive_ok}) except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as error: status = "blocked" native_wasm = {"status": "blocked", "first_divergent_tick": None, "levels": {}} native_worker = {"status": "blocked", "first_divergent_tick": None, "levels": {}} positive_sample = {"status": "blocked", "first_divergent_tick": None, "levels": {}} positive_ok = False failure_samples.append({"code": "TRACE_COMPARATOR_ERROR", "detail": str(error)}) report: dict[str, Any] = { "schema": "xyzac-trt-trace-parity-report.v1", "task_id": "PARITY-00", "context_id": CONTEXT_ID, "case": {"path": CASE, "scope": "xyzac_only"}, "status": status, "candidate_eligible": status == "verified", "levels": list(LEVELS), "native_wasm": native_wasm, "native_worker": native_worker, "positive_divergence_sample": { "mutation": "state[1].active_type=99", "detected": positive_ok, "first_divergent_tick": positive_sample.get("first_divergent_tick"), "difference_path": positive_sample.get("levels", {}).get("state", {}).get("difference_path"), }, "first_divergent_tick": None if status == "verified" else native_wasm.get("first_divergent_tick") or native_worker.get("first_divergent_tick"), "traces": trace_evidence, "builds": build_evidence, "inputs": [ {"path": "wasm-core/tests/sim_xyzac_trace_test.c", "sha256": digest((ROOT / "wasm-core/tests/sim_xyzac_trace_test.c").read_bytes())}, {"path": "web/worker/test_xyzac_trace_parity.mjs", "sha256": digest((ROOT / "web/worker/test_xyzac_trace_parity.mjs").read_bytes())}, {"path": "coverage/xyzac-trt-rtcp.report.json", "sha256": digest((ROOT / "coverage/xyzac-trt-rtcp.report.json").read_bytes())}, ], "failure_samples": failure_samples, "known_blockers": [], "exclusions": ["xyzbc", "3axis", "other cases", "Safari", "physical power loss", "host hardware"], "release_status": "STOP_RELEASE", "report_sha256": None, } report["report_sha256"] = digest(canonical({key: value for key, value in report.items() if key != "report_sha256"})) REPORT.parent.mkdir(parents=True, exist_ok=True) 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": status, "levels": report["levels"], "first_divergent_tick": report["first_divergent_tick"], "positive_sample_tick": report["positive_divergence_sample"]["first_divergent_tick"], "report_sha256": report["report_sha256"]}, ensure_ascii=False)) return 0 if status == "verified" else 1 if __name__ == "__main__": raise SystemExit(main())