416 lines
20 KiB
Python
416 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate the deterministic K8 full-configuration regression report.
|
|
|
|
The LinuxCNC checkout used by this project is intentionally a source/config
|
|
fixture rather than a complete desktop installation. This report therefore
|
|
separates a deterministic static startup preflight from the native/WASM and
|
|
Worker runtime probes. Missing resources and unavailable official oracles
|
|
remain explicit blockers; they are never converted into passing goldens.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
CONTEXT_PATH = Path("context-lock.json")
|
|
GRAPH_PATH = Path("coverage/kinematics-config-graph.report.json")
|
|
RESOURCE_PATH = Path("coverage/config-resources.jsonl")
|
|
RESOURCE_REPORT_PATH = Path("coverage/config-resources.report.json")
|
|
RECOVERY_PATH = Path("coverage/workspace-recovery.report.json")
|
|
REGISTRY_PATH = Path("coverage/kinematics-registry.report.json")
|
|
|
|
RUNTIME_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
|
("kinematics", ("build-native/sim_kinematics_test",)),
|
|
("motion", ("build-native/sim_kinematics_motion_test",)),
|
|
("feedback", ("build-native/sim_kinematics_feedback_test",)),
|
|
("home_limit", ("build-native/sim_kinematics_home_limit_test",)),
|
|
("io", ("build-native/sim_io_test",)),
|
|
)
|
|
WASM_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
|
("kinematics", ("node", "build-wasm/sim_kinematics_wasm_test.js")),
|
|
("motion", ("node", "build-wasm/sim_kinematics_motion_wasm_test.js")),
|
|
("feedback", ("node", "build-wasm/sim_kinematics_feedback_wasm_test.js")),
|
|
("home_limit", ("node", "build-wasm/sim_kinematics_home_limit_wasm_test.js")),
|
|
("io", ("node", "build-wasm/sim_io_wasm_test.js")),
|
|
)
|
|
WORKER_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
|
("replay", ("node", "web/worker/test_replay.mjs")),
|
|
("workspace_recovery", ("node", "web/workspace/test_recovery.mjs")),
|
|
("sim_session", ("node", "web/worker/test_sim_session.mjs")),
|
|
)
|
|
UNRESOLVED_STATUSES = {"unresolved", "host_or_variable_path", "outside_repo"}
|
|
KINEMATICS_SAMPLE_KEYS = {
|
|
"corexykins": "corexy",
|
|
"rosekins": "rose",
|
|
"maxkins": "maxkins",
|
|
"rotatekins": "rotatekins",
|
|
"5axiskins": "fiveaxiskins",
|
|
"xyzac-trt-kins": "xyzac_trt",
|
|
"xyzbc-trt-kins": "xyzbc_trt",
|
|
"scarakins": "scara",
|
|
"pumakins": "puma",
|
|
"scorbot-kins": "scorbot",
|
|
"genserkins": "genser",
|
|
"lineardeltakins": "lineardelta",
|
|
"rotarydeltakins": "rotarydelta",
|
|
"tripodkins": "tripod",
|
|
"genhexkins": "genhex",
|
|
"pentakins": "pentapod",
|
|
}
|
|
|
|
|
|
def canonical(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def digest(value: Any) -> str:
|
|
return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
handle = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
|
handle.update(block)
|
|
return handle.hexdigest()
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path} must contain an object")
|
|
return value
|
|
|
|
|
|
def read_jsonl(path: Path) -> list[dict[str, Any]]:
|
|
rows: list[dict[str, Any]] = []
|
|
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
if not line.strip():
|
|
continue
|
|
value = json.loads(line)
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path}:{line_number} must contain an object")
|
|
rows.append(value)
|
|
return rows
|
|
|
|
|
|
def source_record(root: Path, path: Path) -> dict[str, Any]:
|
|
relative = path.relative_to(root).as_posix()
|
|
return {"path": relative, "sha256": sha256(path)} if path.is_file() else {"path": relative, "sha256": None, "status": "missing"}
|
|
|
|
|
|
def parse_ini_shape(path: Path) -> dict[str, Any]:
|
|
"""Check only syntax needed by the preflight; it is not an INI oracle."""
|
|
if not path.is_file():
|
|
return {"file": False, "sections": [], "balanced_sections": False}
|
|
sections: list[str] = []
|
|
balanced = True
|
|
for raw in path.read_text(encoding="utf-8", errors="strict").splitlines():
|
|
line = raw.split("#", 1)[0].split(";", 1)[0].strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith("["):
|
|
if not line.endswith("]"):
|
|
balanced = False
|
|
else:
|
|
sections.append(line[1:-1].strip().upper())
|
|
return {"file": True, "sections": sorted(set(sections)), "balanced_sections": balanced}
|
|
|
|
|
|
def registered_kinematics(root: Path) -> set[str]:
|
|
registry = read_json(REGISTRY_PATH if REGISTRY_PATH.is_absolute() else root / REGISTRY_PATH)
|
|
rows = registry.get("samples", {}).get("native", {}).get("entries", [])
|
|
return {str(row.get("name")) for row in rows if row.get("name")}
|
|
|
|
|
|
def graph_entries(root: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
graph = read_json(root / GRAPH_PATH)
|
|
entries = [row for row in graph.get("nodes", []) if row.get("kind") == "kinematics"]
|
|
entries.sort(key=lambda row: str(row.get("node_id", "")))
|
|
return graph, entries
|
|
|
|
|
|
def resource_summaries(root: Path, resource_report: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
|
by_entrypoint: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for row in rows:
|
|
by_entrypoint[str(row.get("entrypoint", ""))].append(row)
|
|
indexed = {str(path) for path in resource_report.get("entrypoint_paths", [])}
|
|
result: dict[str, list[dict[str, Any]]] = {}
|
|
for path in sorted(indexed):
|
|
result[path] = by_entrypoint.get(path, [])
|
|
return result
|
|
|
|
|
|
def preflight_entry(root: Path, row: dict[str, Any], resources: dict[str, list[dict[str, Any]]], indexed_paths: set[str], registered: set[str]) -> dict[str, Any]:
|
|
path = str(row["path"])
|
|
module = str(row.get("module", ""))
|
|
resource_rows = resources.get(path, [])
|
|
missing = sorted({str(item.get("path")) for item in resource_rows if item.get("parse_status") == "missing"})
|
|
unresolved = sorted({str(item.get("path")) for item in resource_rows if item.get("parse_status") in UNRESOLVED_STATUSES})
|
|
ini_shape = parse_ini_shape(root / path)
|
|
if module not in registered:
|
|
startup_status = "blocked_unregistered_kinematics"
|
|
elif path not in indexed_paths and path.startswith("linuxcnc/configs/sim/"):
|
|
startup_status = "blocked_unindexed_resource_graph"
|
|
elif missing or unresolved:
|
|
startup_status = "blocked_resource"
|
|
elif not ini_shape["file"] or not ini_shape["balanced_sections"]:
|
|
startup_status = "blocked_ini_syntax"
|
|
else:
|
|
startup_status = "static_ready"
|
|
if path not in indexed_paths:
|
|
resource_status = "unindexed"
|
|
elif missing and unresolved:
|
|
resource_status = "blocked_missing_and_unresolved"
|
|
elif missing:
|
|
resource_status = "blocked_missing"
|
|
elif unresolved:
|
|
resource_status = "blocked_unresolved"
|
|
else:
|
|
resource_status = "ready"
|
|
return {
|
|
"config_id": row.get("node_id"),
|
|
"path": path,
|
|
"line": row.get("line"),
|
|
"module": module,
|
|
"family": row.get("family"),
|
|
"resource_status": resource_status,
|
|
"missing_resources": missing,
|
|
"unresolved_resources": unresolved,
|
|
"resource_rows": len(resource_rows),
|
|
"ini": ini_shape,
|
|
"kinematics_registered": module in registered,
|
|
"startup_preflight": startup_status,
|
|
}
|
|
|
|
|
|
def run_json_command(root: Path, command: tuple[str, ...]) -> dict[str, Any]:
|
|
executable = root / command[0] if command[0] not in {"node"} else root / command[1]
|
|
if not executable.exists():
|
|
return {"status": "blocked", "reason": f"missing_runtime_artifact:{executable.relative_to(root).as_posix()}"}
|
|
completed = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
|
if completed.returncode != 0:
|
|
return {"status": "failed", "exit_code": completed.returncode, "stderr_sha256": digest(completed.stderr)}
|
|
json_lines = [line.strip() for line in completed.stdout.splitlines() if line.lstrip().startswith("{")]
|
|
if not json_lines:
|
|
return {"status": "failed", "reason": "runtime_did_not_emit_json", "stdout_sha256": digest(completed.stdout)}
|
|
try:
|
|
sample = json.loads(json_lines[-1])
|
|
except json.JSONDecodeError:
|
|
return {"status": "failed", "reason": "runtime_emitted_invalid_json", "stdout_sha256": digest(completed.stdout)}
|
|
return {"status": "passed", "schema": sample.get("schema"), "sample": sample, "sample_sha256": digest(sample)}
|
|
|
|
|
|
def run_runtime_group(root: Path, commands: Iterable[tuple[str, tuple[str, ...]]]) -> dict[str, Any]:
|
|
records: dict[str, Any] = {}
|
|
for name, command in commands:
|
|
first = run_json_command(root, command)
|
|
second = run_json_command(root, command)
|
|
first_key = first.get("sample_sha256")
|
|
second_key = second.get("sample_sha256")
|
|
records[name] = {
|
|
"command": list(command),
|
|
"first": first,
|
|
"second": second,
|
|
"repeat_stable": bool(first_key and first_key == second_key),
|
|
}
|
|
passed = all(row["first"].get("status") == "passed" and row["repeat_stable"] for row in records.values())
|
|
return {"status": "passed" if passed else "blocked", "records": records, "summary_sha256": digest(records)}
|
|
|
|
|
|
def run_worker_group(root: Path) -> dict[str, Any]:
|
|
records: dict[str, Any] = {}
|
|
for name, command in WORKER_COMMANDS:
|
|
target = root / command[-1]
|
|
if not target.is_file():
|
|
records[name] = {"command": list(command), "status": "blocked", "reason": "missing_worker_test"}
|
|
continue
|
|
first = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
|
second = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
|
first_value = {"exit_code": first.returncode, "stdout": first.stdout, "stderr": first.stderr}
|
|
second_value = {"exit_code": second.returncode, "stdout": second.stdout, "stderr": second.stderr}
|
|
records[name] = {
|
|
"command": list(command),
|
|
"status": "passed" if first.returncode == 0 else "failed",
|
|
"first_sha256": digest(first_value),
|
|
"second_sha256": digest(second_value),
|
|
"repeat_stable": digest(first_value) == digest(second_value),
|
|
"stdout": first.stdout.strip(),
|
|
}
|
|
passed = all(value.get("status") == "passed" and value.get("repeat_stable") for value in records.values())
|
|
return {"status": "passed" if passed else "blocked", "records": records, "summary_sha256": digest(records)}
|
|
|
|
|
|
def extract_state_hashes(value: Any, path: str = "") -> list[dict[str, Any]]:
|
|
found: list[dict[str, Any]] = []
|
|
if isinstance(value, dict):
|
|
for key, child in sorted(value.items()):
|
|
child_path = f"{path}.{key}" if path else key
|
|
if key in {"state_hash", "hash", "repeat_hash"} and isinstance(child, (int, str)):
|
|
found.append({"path": child_path, "value": child})
|
|
found.extend(extract_state_hashes(child, child_path))
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
found.extend(extract_state_hashes(child, f"{path}[{index}]"))
|
|
return found
|
|
|
|
|
|
def build_config_state_summary(entries: list[dict[str, Any]], runtime: dict[str, Any], registry: set[str]) -> dict[str, Any]:
|
|
samples: dict[str, Any] = {}
|
|
for name, record in runtime.get("records", {}).items():
|
|
sample = record.get("first", {}).get("sample")
|
|
if isinstance(sample, dict):
|
|
samples[name] = {
|
|
"schema": sample.get("schema"),
|
|
"sample_sha256": record["first"].get("sample_sha256"),
|
|
"state_hashes": extract_state_hashes(sample),
|
|
}
|
|
rows = []
|
|
for entry in entries:
|
|
module = str(entry.get("module", ""))
|
|
sample_key = KINEMATICS_SAMPLE_KEYS.get(module)
|
|
rows.append({
|
|
"config_id": entry.get("config_id"),
|
|
"module": module,
|
|
"sample": sample_key,
|
|
"sample_status": "verified_sample" if sample_key in samples else "registry_only" if module in registry else "blocked_unregistered",
|
|
"state_digest": digest({"module": module, "sample": samples.get(sample_key), "registry": module in registry}),
|
|
})
|
|
rows.sort(key=lambda row: str(row["config_id"]))
|
|
return {"tick_policy": "fixed logical tick; wall clock excluded", "entries": rows, "runtime_samples": samples, "summary_sha256": digest(rows)}
|
|
|
|
|
|
def generate(root: Path, output: Path) -> dict[str, Any]:
|
|
root = root.resolve()
|
|
context = read_json(root / CONTEXT_PATH)
|
|
graph, entries = graph_entries(root)
|
|
resource_report = read_json(root / RESOURCE_REPORT_PATH)
|
|
resources_rows = read_jsonl(root / RESOURCE_PATH)
|
|
resources = resource_summaries(root, resource_report, resources_rows)
|
|
indexed_paths = set(resources)
|
|
registered = registered_kinematics(root)
|
|
preflight = [preflight_entry(root, row, resources, indexed_paths, registered) for row in entries]
|
|
preflight.sort(key=lambda row: str(row["config_id"]))
|
|
runtime = run_runtime_group(root, RUNTIME_COMMANDS)
|
|
wasm = run_runtime_group(root, WASM_COMMANDS)
|
|
worker = run_worker_group(root)
|
|
native_wasm_equal = all(
|
|
runtime["records"].get(name, {}).get("first", {}).get("sample_sha256")
|
|
== wasm["records"].get(name, {}).get("first", {}).get("sample_sha256")
|
|
and runtime["records"].get(name, {}).get("first", {}).get("status") == "passed"
|
|
for name, _command in RUNTIME_COMMANDS
|
|
)
|
|
graph_orphans = list(graph.get("closure", {}).get("orphan_entry_ids", []))
|
|
unowned = list(graph.get("closure", {}).get("unowned_template_ids", []))
|
|
resource_orphans = sorted({str(row.get("entrypoint")) for row in resources_rows if str(row.get("entrypoint")) not in indexed_paths})
|
|
startup_counts: dict[str, int] = defaultdict(int)
|
|
resource_counts: dict[str, int] = defaultdict(int)
|
|
for row in preflight:
|
|
startup_counts[str(row["startup_preflight"])] += 1
|
|
resource_counts[str(row["resource_status"])] += 1
|
|
missing_resource_paths = sorted({path for row in preflight for path in row["missing_resources"]})
|
|
unresolved_resource_paths = sorted({path for row in preflight for path in row["unresolved_resources"]})
|
|
recovery = read_json(root / RECOVERY_PATH)
|
|
state_summary = build_config_state_summary(entries, runtime, registered)
|
|
report: dict[str, Any] = {
|
|
"schema": "configuration-regression-report.v1",
|
|
"context_id": context.get("context_id"),
|
|
"status": "partial",
|
|
"implementation_status": "verified",
|
|
"scope": {"configs_root": "linuxcnc/configs/sim", "tests_root": "linuxcnc/tests", "preflight_only": True},
|
|
"inputs": {
|
|
# The report is itself pinned by context-lock. Recording only the
|
|
# context ID here avoids a circular lock/report content hash.
|
|
"context_lock": {"path": CONTEXT_PATH.as_posix(), "context_id": context.get("context_id")},
|
|
"kinematics_graph": source_record(root, root / GRAPH_PATH),
|
|
"config_resources": source_record(root, root / RESOURCE_PATH),
|
|
"config_resources_report": source_record(root, root / RESOURCE_REPORT_PATH),
|
|
"workspace_recovery": source_record(root, root / RECOVERY_PATH),
|
|
"kinematics_registry": source_record(root, root / REGISTRY_PATH),
|
|
},
|
|
"counts": {
|
|
"registered_configuration_entries": len(entries),
|
|
"config_entrypoints": int(resource_report.get("entrypoints", 0)),
|
|
"resource_rows": len(resources_rows),
|
|
"resource_missing_rows": sum(1 for row in resources_rows if row.get("parse_status") == "missing"),
|
|
"resource_unresolved_rows": sum(1 for row in resources_rows if row.get("parse_status") in UNRESOLVED_STATUSES),
|
|
"missing_resource_paths": len(missing_resource_paths),
|
|
"unresolved_resource_paths": len(unresolved_resource_paths),
|
|
"resource_orphans": len(resource_orphans),
|
|
"graph_orphans": len(graph_orphans),
|
|
"unowned_templates": len(unowned),
|
|
"startup_preflight": dict(sorted(startup_counts.items())),
|
|
"resource_status": dict(sorted(resource_counts.items())),
|
|
},
|
|
"configuration_entries": preflight,
|
|
"closure": {
|
|
"graph_fingerprint_sha256": graph.get("graph_fingerprint_sha256"),
|
|
"graph_orphan_entry_ids": graph_orphans,
|
|
"unowned_template_ids": unowned,
|
|
"resource_orphan_entrypoints": resource_orphans,
|
|
"all_registered": len(entries) == int(graph.get("counts", {}).get("configuration_entries", -1)),
|
|
},
|
|
"runtime": {
|
|
"native": runtime,
|
|
"wasm": wasm,
|
|
"native_wasm": {"status": "passed" if native_wasm_equal else "blocked", "sample_hashes_equal": native_wasm_equal},
|
|
"worker": worker,
|
|
},
|
|
"state_summary": state_summary,
|
|
"recovery_summary": {
|
|
"status": recovery.get("status"),
|
|
"implementation_status": recovery.get("implementation_status"),
|
|
"checkpoint_schema": recovery.get("replay", {}).get("schema"),
|
|
"replay_state_hash": recovery.get("replay", {}).get("state_hash"),
|
|
"continuous_state_hash": recovery.get("replay", {}).get("continuous_state_hash"),
|
|
"endpoints_equal": recovery.get("replay", {}).get("endpoints_equal"),
|
|
"oracle_status": recovery.get("oracle", {}).get("status"),
|
|
},
|
|
"oracle": {
|
|
"status": "blocked",
|
|
"reason": "official LinuxCNC kinematics/recovery oracle is absent from the locked source",
|
|
"official_root": "linuxcnc/unit_tests/kinematics",
|
|
"derived_configuration_results_are_not_official": True,
|
|
},
|
|
"unclosed": {
|
|
"resource_missing": missing_resource_paths,
|
|
"resource_unresolved": unresolved_resource_paths,
|
|
"unregistered_kinematics": sorted({str(row["module"]) for row in preflight if not row["kinematics_registered"]}),
|
|
"startup_blocked_entries": [row["config_id"] for row in preflight if str(row["startup_preflight"]).startswith("blocked")],
|
|
},
|
|
"report_sha256": None,
|
|
}
|
|
report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"})
|
|
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("--output", type=Path, default=Path("coverage/configuration-regression.report.json"))
|
|
args = parser.parse_args()
|
|
report = generate(args.repo_root, args.repo_root / args.output)
|
|
print(
|
|
"configuration_regression="
|
|
f"entries={report['counts']['registered_configuration_entries']} "
|
|
f"startup_blocked={len(report['unclosed']['startup_blocked_entries'])} "
|
|
f"resource_orphans={report['counts']['resource_orphans']} "
|
|
f"native_wasm={report['runtime']['native_wasm']['status']} "
|
|
f"worker={report['runtime']['worker']['status']}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|