solvecraft/assets/render/selene/tools/validate_native.py

#!/usr/bin/env python3
"""Check sampled SELENE mechanism behavior through a disposable native MCP process.

This tool only reads the input file and drives the in-memory copy. It never saves,
exports, patches native JSON, or changes the source. A report lists actual native
observations, tolerances and scope. Missing parts, unchecked pairs and timeouts
cannot become passes. Python 3.11 standard library; SolidCraft is the CAD runtime.
"""
from __future__ import annotations

import argparse
from datetime import datetime, timezone
import hashlib
import json
import math
import os
from pathlib import Path
import queue
import subprocess
import sys
import threading
import time
from typing import Callable

# Local helper imports must not leave generated bytecode inside production assets.
sys.dont_write_bytecode = True

from bearings import bearing_specs
from prepare import inspect_native

PACKAGE = Path(__file__).resolve().parents[1]
SOURCE_SHA256 = "1b0df486f30da157db27a33e0b5ed67f13d527d4dfdc6c334717a1df4e5b8677"
WHEELS = [f"Wheel axis {s}:{x}" for s in (-1, 1) for x in (-90, 0, 90)]
HINGES = [f"SELENE suspension hinge {s}:{x}" for s in (-1, 1) for x in (-90, 0, 90)]
PRIMARY = WHEELS[:1] + [
    "Port solar tilt articulation", "Turret azimuth articulation", "Turret elevation articulation",
    "Arm azimuth articulation", "Shoulder articulation", "Elbow articulation",
    "Wrist articulation", "Tool tilt articulation", "Dish azimuth articulation",
    "Dish elevation articulation", "Vial cap thread", "SELENE gripper jaw -1 articulation",
] + HINGES
READ_OR_DRIVE = {"joint.drive", "joint.list", "occurrence.properties", "inspect.interference"}
MAX_REPLY = 32 * 1024 * 1024
ANGLE_TOLERANCE_DEG = 1e-6
DISTANCE_TOLERANCE_MM = 1e-5
COM_TOLERANCE_MM = 2e-3
NATIVE_ENVIRONMENT = {
    "RAYON_NUM_THREADS": "2",
    "MALLOC_ARENA_MAX": "1",
    "MALLOC_MMAP_THRESHOLD_": "65536",
    "MALLOC_TRIM_THRESHOLD_": "65536",
    "SOLIDCRAFT_GPU": "cpu",
}


def digest(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(block)
    return h.hexdigest()


def require(condition: bool, message: str) -> None:
    if not condition:
        raise ValueError(message)


def vector(value: object, label: str) -> list[float]:
    require(isinstance(value, list) and len(value) == 3, f"{label}: expected a three-vector")
    require(all(type(x) in (int, float) and math.isfinite(x) for x in value), f"{label}: non-finite coordinate")
    return [float(x) for x in value]


def rotated(point: list[float], center: list[float], axis: list[float], angle: float) -> list[float]:
    """Independent Rodrigues rotation, with command angles in degrees."""
    length = math.sqrt(sum(x*x for x in axis))
    require(length > 0, "Zero rotation axis")
    u = [x/length for x in axis]
    v = [point[i]-center[i] for i in range(3)]
    c, s = math.cos(math.radians(angle)), math.sin(math.radians(angle))
    dot = sum(u[i]*v[i] for i in range(3))
    cross = [u[1]*v[2]-u[2]*v[1], u[2]*v[0]-u[0]*v[2], u[0]*v[1]-u[1]*v[0]]
    return [center[i]+v[i]*c+cross[i]*s+u[i]*dot*(1-c) for i in range(3)]


def damper_stroke(angle: float) -> float:
    t = math.radians(angle)
    return math.hypot(35*math.cos(t)-20*math.sin(t)-35,
                      -35*math.sin(t)-20*math.cos(t)-2)-22


class NativeMcp:
    """Bounded JSON-lines client; protocol and command errors are raised, never hidden."""
    def __init__(self, cli: Path, model: Path, directory: Path, timeout: float, startup_timeout: float):
        self.timeout = timeout
        self.sequence = 0
        self.messages: queue.Queue = queue.Queue(maxsize=32)
        self.stderr = (directory / "native.stderr.log").open("xb")
        self.transcript = (directory / "native.rpc.jsonl").open("x", encoding="utf-8")
        env = os.environ.copy()
        env.update(NATIVE_ENVIRONMENT)
        self.process = subprocess.Popen([str(cli), "mcp", "--in", str(model)],
                                        stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                        stderr=self.stderr, env=env)
        self.reader = threading.Thread(target=self._read, daemon=True)
        self.reader.start()
        try:
            self.server = self.request("initialize", {
                "protocolVersion": "2025-06-18", "capabilities": {},
                "clientInfo": {"name": "selene-native-validation", "version": "1.0"},
            }, timeout=startup_timeout)
            self._send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        except BaseException:
            self.close()
            raise

    def _read(self) -> None:
        try:
            while True:
                data = self.process.stdout.readline(MAX_REPLY+1)
                if not data:
                    self.messages.put(RuntimeError("Native MCP process closed stdout"))
                    return
                if len(data) > MAX_REPLY or not data.endswith(b"\n"):
                    self.messages.put(ValueError("Native response exceeded 32 MiB or lacked newline"))
                    return
                self.messages.put(json.loads(data))
        except (OSError, ValueError) as error:
            self.messages.put(error)

    def _send(self, value: dict) -> None:
        encoded = json.dumps(value, allow_nan=False, separators=(",", ":"))
        self.transcript.write(json.dumps({"direction": "request", "value": value})+"\n")
        self.transcript.flush()
        self.process.stdin.write((encoded+"\n").encode())
        self.process.stdin.flush()

    def request(self, method: str, params: dict, timeout: float | None = None) -> dict:
        self.sequence += 1
        require(self.sequence <= 4000, "Validation request budget exceeded")
        identity = self.sequence
        self._send({"jsonrpc": "2.0", "id": identity, "method": method, "params": params})
        deadline = time.monotonic() + (self.timeout if timeout is None else timeout)
        for _ in range(100):
            remaining = deadline-time.monotonic()
            if remaining <= 0:
                raise TimeoutError(f"Native request {identity} timed out")
            try:
                value = self.messages.get(timeout=remaining)
            except queue.Empty as error:
                raise TimeoutError(f"Native request {identity} timed out") from error
            if isinstance(value, Exception):
                raise value
            self.transcript.write(json.dumps({"direction": "response", "value": value})+"\n")
            self.transcript.flush()
            if "id" not in value:
                continue
            require(value.get("id") == identity, "Native response ID mismatch")
            require("error" not in value, f"Native protocol error: {value.get('error')}")
            result = value.get("result")
            require(isinstance(result, dict), "Native response lacks result object")
            return result
        raise ValueError("Excessive native notifications")

    def tool(self, name: str, arguments: dict) -> dict:
        require(name in {"execute", "inspect_design"}, "Validation tool not allowed")
        result = self.request("tools/call", {"name": name, "arguments": arguments})
        texts = [item.get("text", "") for item in result.get("content", []) if item.get("type") == "text"]
        require(result.get("isError") is not True, "Native tool error: "+" ".join(texts))
        require(len(texts) == 1, "Expected exactly one native JSON result")
        value = json.loads(texts[0])
        require(isinstance(value, dict), "Expected native result object")
        return value

    def execute(self, command: str, **params) -> dict:
        require(command in READ_OR_DRIVE, "Validation command not allowed: "+command)
        return self.tool("execute", {"command": command, "params": params, "verbose": False})

    def close(self) -> None:
        if self.process.poll() is None:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait(timeout=5)
        self.reader.join(timeout=1)
        if self.process.stdin is not None:
            self.process.stdin.close()
        if self.process.stdout is not None:
            self.process.stdout.close()
        self.stderr.close()
        self.transcript.close()


class Verification:
    def __init__(self, native: NativeMcp, areas: set[str], bearing_names: set[str], inventory: dict):
        self.native, self.areas, self.bearing_names = native, areas, bearing_names
        self.occurrences = {}
        for component in inventory["components"]:
            placements = [o for o in inventory["occurrences"] if o["component"] == component["id"]]
            if len(placements) == 1:
                require(component["name"] not in self.occurrences, "Ambiguous component name")
                self.occurrences[component["name"]] = placements[0]["id"]
        self.cases: list[dict] = []
        self.current: dict = {}
        self.initial = self.graph()
        self.drivers = [name for name in PRIMARY if name in self.initial]

    def graph(self) -> dict[str, dict]:
        value = self.native.execute("joint.list")
        for field in ("errors", "conflicts", "floating"):
            require(value.get(field) == [], f"Joint graph {field}: {value.get(field)}")
        require(not any(link.get("error") for link in value.get("links", [])), "Motion coupling error")
        joints = value.get("joints")
        require(isinstance(joints, list), "Joint list missing")
        names = [j.get("name") for j in joints]
        require(len(names) == len(set(names)), "Ambiguous duplicate joint names")
        return dict(zip(names, joints))

    def near(self, label: str, actual: object, expected: float, tolerance: float) -> None:
        require(type(actual) in (int, float) and math.isfinite(actual), label+": invalid numeric result")
        error = abs(actual-expected)
        self.current.setdefault("checks", []).append({"quantity": label, "actual": actual,
                                                      "expected": expected, "absolute_error": error,
                                                      "tolerance": tolerance, "pass": error <= tolerance})
        require(error <= tolerance, f"{label}: {actual} != {expected} within {tolerance}")

    def point(self, label: str, actual: object, expected: list[float]) -> None:
        got = vector(actual, label)
        for i, axis in enumerate("xyz"):
            self.near(label+"."+axis, got[i], expected[i], COM_TOLERANCE_MM)

    def angle(self, graph: dict, name: str, expected: float) -> None:
        self.value(graph, name, 0, "deg", expected, ANGLE_TOLERANCE_DEG)

    def distance(self, graph: dict, name: str, expected: float, index: int = 0) -> None:
        self.value(graph, name, index, "mm", expected, DISTANCE_TOLERANCE_MM)

    def value(self, graph: dict, name: str, index: int, unit: str, expected: float, tolerance: float) -> None:
        require(name in graph, "Missing required joint: "+name)
        values = graph[name].get("values", [])
        require(index < len(values) and values[index].get("unit") == unit, f"{name}: missing {unit} value {index}")
        self.near(f"{name}[{index}] ({unit})", values[index].get("value"), expected, tolerance)

    def drive(self, name: str, angle: float) -> None:
        value = self.native.execute("joint.drive", joint=name, value=angle, quiet=True)
        require(value.get("moved") is True, f"{name} did not move: {value.get('reason')}")
        require(not value.get("errors") and not value.get("conflicts"), f"{name}: invalid drive graph")
        require(not value.get("stopped_by_loop") and not value.get("stopped_by_contact"), f"{name}: drive stopped early")
        self.current.setdefault("drives", []).append({"joint": name, "requested_degrees": angle,
                                                       "moved": value.get("moved"), "values": value.get("values")})

    def neutral(self) -> None:
        # Read every independent coordinate, reset the nonzero ones, then verify all.
        # Avoid re-solving hundreds of zero-to-zero drives on the complete assembly.
        graph = self.graph()
        for name in self.drivers:
            values = graph[name].get("values", [])
            require(values and values[0].get("unit") == "deg", "Missing neutral driver angle: "+name)
            actual = values[0].get("value")
            require(type(actual) in (int, float) and math.isfinite(actual), "Invalid neutral driver angle: "+name)
            if abs(actual) <= 1e-12:
                continue
            value = self.native.execute("joint.drive", joint=name, value=0, quiet=True)
            require(not value.get("errors") and not value.get("conflicts"), f"Neutral reset failed: {name}")
        graph = self.graph()
        for name in self.drivers:
            self.angle(graph, name, 0)

    def properties(self, name: str) -> dict:
        require(name in self.occurrences, "Component requires exactly one occurrence: "+name)
        p = self.native.execute("occurrence.properties", occurrence=self.occurrences[name])
        require(p.get("coordinate_frame") == "world" and p.get("body_count", 0) > 0,
                "Missing measured occurrence: "+name)
        return p

    def com(self, name: str) -> list[float]:
        return vector(self.properties(name).get("total", {}).get("center_of_mass"), name)

    def moved_com(self, component: str, before: list[float], center: list[float], axis: list[float], angle: float) -> None:
        after = self.com(component)
        require(math.dist(before, after) > COM_TOLERANCE_MM, component+": center did not move")
        self.point(component+" COM (mm)", after, rotated(before, center, axis, angle))

    def case(self, name: str, action: Callable[[], None]) -> None:
        self.current = {"name": name, "status": "running", "checks": []}
        self.cases.append(self.current)
        start = time.monotonic()
        try:
            self.neutral()
            action()
            self.graph()
            self.current["status"] = "passed"
        except (ValueError, KeyError, TypeError) as error:
            self.current.update(status="failed", error=str(error))
        finally:
            self.current["seconds"] = round(time.monotonic()-start, 3)
            print(json.dumps({"case": name, "status": self.current["status"],
                              "checks": len(self.current["checks"]),
                              "error": self.current.get("error")}), flush=True)

    def wheels(self) -> None:
        # An off-axis grouser/fastener COM proves rotation even when the wheel's total COM lies on its axis.
        before = self.properties("Wheel -1:-90")
        center, axis = [-90, -86, 40], [0, 1, 0]
        bodies = before["bodies"]
        body = max(bodies, key=lambda b: math.hypot(b["center_of_mass"][0]+90, b["center_of_mass"][2]-40))
        point = vector(body["center_of_mass"], body["name"])
        require(math.hypot(point[0]+90, point[2]-40) > 1, "No off-axis wheel witness")
        track = self.com("Regolith track")
        self.drive(WHEELS[0], 90)
        graph = self.graph()
        for name in WHEELS:
            self.angle(graph, name, 90)
        travel = -40*math.pi/2
        self.distance(graph, "Track slide", travel)
        after = {b["name"]: b for b in self.properties("Wheel -1:-90")["bodies"]}
        self.point("Wheel witness "+body["name"], after[body["name"]]["center_of_mass"], rotated(point, center, axis, 90))
        self.point("Track COM (mm)", self.com("Regolith track"), [track[0]+travel, track[1], track[2]])

    def solar(self) -> None:
        before = self.com("Solar wing -1")
        self.drive("Port solar tilt articulation", 20)
        graph = self.graph()
        self.angle(graph, "Port solar tilt articulation", 20)
        self.angle(graph, "Starboard solar tilt articulation", -20)
        for s in (-1, 1):
            for x in (-62, 62):
                self.distance(graph, f"Actuator stroke ({s}, {x})", -2.8662098186467375)
        self.moved_com("Solar wing -1", before, [0, -140, 158], [1, 0, 0], 20)

    def pointing(self, sensor: bool) -> None:
        driver = "Turret azimuth articulation" if sensor else "Dish azimuth articulation"
        elevation = "Turret elevation articulation" if sensor else "Dish elevation articulation"
        moving = "Turret yaw" if sensor else "Dish yaw mount"
        still = "Dish yaw mount" if sensor else "Turret yaw"
        origin = [-45, 0, 157] if sensor else [-177, 0, 135]
        before, stationary = self.com(moving), self.com(still)
        angle = 30 if sensor else -25
        self.drive(driver, angle)
        self.drive(elevation, 15 if sensor else 20)
        graph = self.graph()
        self.angle(graph, driver, angle)
        self.angle(graph, elevation, 15 if sensor else 20)
        self.angle(graph, "Dish azimuth articulation" if sensor else "Turret azimuth articulation", 0)
        self.angle(graph, "Dish elevation articulation" if sensor else "Turret elevation articulation", 0)
        self.point(still+" stationary COM (mm)", self.com(still), stationary)
        # The yaw bracket can be centered on its axis; numeric pose is still compared and head motion has separate bearing checks.
        self.point(moving+" COM (mm)", self.com(moving), rotated(before, origin, [0, 0, 1], angle))

    def gripper(self, angle: float) -> None:
        before = {s: self.com(f"Gripper jaw {s}") for s in (-1, 1)}
        self.drive("SELENE gripper jaw -1 articulation", angle)
        graph = self.graph()
        for s in (-1, 1):
            actual_angle = angle if s == -1 else -angle
            self.angle(graph, f"SELENE gripper jaw {s} articulation", actual_angle)
            self.moved_com(f"Gripper jaw {s}", before[s], [443, s*9, 249.5], [0, 0, 1], actual_angle)

    def cap(self, angle: float) -> None:
        component = "Vial cap (-160, -22)"
        before = self.com(component)
        self.drive("Vial cap thread", angle)
        graph = self.graph()
        self.angle(graph, "Vial cap thread", angle)
        travel = 1.5*angle/360
        self.distance(graph, "Vial cap thread", travel, index=1)
        expected = rotated(before, [53, -22, 179], [0, 0, 1], angle)
        expected[2] += travel
        self.point(component+" COM (mm)", self.com(component), expected)

    def suspension(self, side: int, x: int, angle: float) -> None:
        component = f"Drive module {side}:{x}"
        before = self.com(component)
        hinge = f"SELENE suspension hinge {side}:{x}"
        self.drive(hinge, angle)
        graph = self.graph()
        self.angle(graph, hinge, angle)
        self.distance(graph, f"SELENE suspension stroke {side}:{x}", damper_stroke(angle))
        self.moved_com(component, before, [x-20, side*54, 96], [0, 1, 0], angle)

    def bearings(self, driver: str, specs: list[dict]) -> None:
        self.drive(driver, 10)
        graph = self.graph()
        self.angle(graph, driver, 10)
        for spec in specs:
            radius, ball = spec["pitch_radius"], spec["ball_radius"]
            cage = .5*(1-ball/radius)*10
            spin = -.5*(radius/ball-ball/radius)*10
            self.angle(graph, spec["cage_joint"], cage)
            for n in range(1, spec["balls"]+1):
                self.angle(graph, f"{spec['name']} ball {n:02d} spin", spin)
            center = spec["center"]
            phi = math.radians(cage)
            expected = [center[0]+radius*math.cos(phi), center[1], center[2]-radius*math.sin(phi)]
            self.point(spec["name"]+" ball 01 orbit COM (mm)", self.com(spec["name"]+" ball 01"), expected)

    def run(self) -> None:
        if "primary" in self.areas:
            self.case("Wheel train and 40 mm nominal track coupling", self.wheels)
            self.case("Solar closed loops at 20 degrees", self.solar)
            for sensor in (True, False):
                self.case("Independent sensor pointing" if sensor else "Independent dish pointing", lambda s=sensor: self.pointing(s))
            for angle in (-15, 18):
                self.case(f"Gripper jaw driver at {angle} degrees", lambda a=angle: self.gripper(a))
            for angle in (360, 1440):
                self.case(f"Vial screw cap at {angle} degrees", lambda a=angle: self.cap(a))
        if "suspension" in self.areas:
            for side in (-1, 1):
                for x in (-90, 0, 90):
                    for angle in (-4, 1.5):
                        self.case(f"Suspension {side}:{x} at {angle} degrees", lambda s=side, p=x, a=angle: self.suspension(s, p, a))
        if "bearings" in self.areas:
            specs = bearing_specs()
            if self.bearing_names:
                require(self.bearing_names <= {s["name"] for s in specs}, "Unknown bearing selection")
                specs = [s for s in specs if s["name"] in self.bearing_names]
            for driver in dict.fromkeys(s["driver"] for s in specs):
                selected = [s for s in specs if s["driver"] == driver]
                self.case("Bearing kinematics: "+driver, lambda d=driver, ss=selected: self.bearings(d, ss))
        self.case("Return independent drivers to neutral", lambda: None)

    def interference(self, pairs: list[list[str]], budget: float, body_names: set[str]) -> None:
        """Optional exact-name pairs only; incomplete work is recorded and ends this phase."""
        for a, b in pairs:
            def action(left=a, right=b):
                require(left in body_names and right in body_names, "Interference target is absent")
                result = self.native.execute("inspect.interference", bodies=[left, right], budget_s=budget)
                self.current["native_interference"] = result
                require(result.get("checked") == 2, "Native interference did not select both bodies")
                require(result.get("unchecked") == [] and result.get("pairs_not_checked") == 0
                        and result.get("skipped_pairs") == 0 and result.get("truncated") is False,
                        "Interference pair was not completely checked")
                require(result.get("count") == 0 and result.get("interferences") == [], "Interference found")
            self.case("Scoped neutral interference: "+a+" / "+b, action)
            if self.current.get("native_interference", {}).get("unchecked"):
                # A budget-limited native worker may still be running. Do not accumulate them.
                break


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--cli", type=Path, required=True)
    parser.add_argument("--model", type=Path, required=True)
    parser.add_argument("--source", type=Path, default=PACKAGE / "source/SELENE-showcase.solidcraft")
    parser.add_argument("--out", type=Path, required=True, help="New evidence directory, normally below ignored target/")
    parser.add_argument("--areas", default="primary,suspension,bearings", help="Comma-separated subset; default all three")
    parser.add_argument("--bearing-name", action="append", default=[], help="Explicit fixture subset; full validation omits this")
    parser.add_argument("--interference-pairs", type=Path, help="Optional JSON array of up to 20 exact body-name pairs")
    parser.add_argument("--interference-budget", type=float, default=8, help="Seconds per optional native pair, at most 60")
    parser.add_argument("--timeout", type=float, default=180)
    parser.add_argument("--startup-timeout", type=float, default=900)
    args = parser.parse_args()
    cli, model, source, output = [p.resolve() for p in (args.cli, args.model, args.source, args.out)]
    areas = set(args.areas.split(","))
    require(areas and areas <= {"primary", "suspension", "bearings"}, "Unknown validation area")
    require(not args.bearing_name or "bearings" in areas, "Bearing selection requires the bearings area")
    require(math.isfinite(args.timeout) and 0 < args.timeout <= 3600, "Invalid timeout")
    require(math.isfinite(args.startup_timeout) and 0 < args.startup_timeout <= 3600, "Invalid startup timeout")
    require(math.isfinite(args.interference_budget) and 0 < args.interference_budget <= 60, "Invalid pair budget")
    pairs = []
    if args.interference_pairs is not None:
        require(args.interference_pairs.stat().st_size <= 65536, "Interference selection exceeds 64 KiB")
        pairs = json.loads(args.interference_pairs.read_text(encoding="utf-8"))
        require(isinstance(pairs, list) and 0 < len(pairs) <= 20, "Expected 1–20 interference pairs")
        require(all(isinstance(p, list) and len(p) == 2 and p[0] != p[1]
                    and all(isinstance(n, str) and 0 < len(n) <= 256 for n in p) for p in pairs),
                "Expected distinct exact body names per pair")
    require(cli.is_file() and model.is_file() and source.is_file(), "CLI, model and pinned source must exist")
    require(not output.exists(), "Evidence directory exists; choose a new output to preserve earlier evidence")
    before, source_before = digest(model), digest(source)
    require(source_before == SOURCE_SHA256, "Pinned original source digest mismatch")
    output.mkdir(parents=True)
    report = {
        "schema_version": "1.0.0", "language": "en", "status": "incomplete",
        "started_utc": datetime.now(timezone.utc).isoformat(),
        "model_sha256": before, "source_sha256": source_before, "cli_sha256": digest(cli),
        "validator_sha256": digest(Path(__file__)), "bearing_spec_sha256": digest(Path(__file__).with_name("bearings.py")),
        "native_process_environment": NATIVE_ENVIRONMENT.copy(),
        "memory_policy": "Native subprocess only: two Rayon workers, one glibc arena and 64 KiB mmap/trim thresholds. Full-model native processes must run serially.",
        "areas": sorted(areas), "bearing_subset": args.bearing_name,
        "full_kinematic_suite_requested": areas == {"primary", "suspension", "bearings"} and not args.bearing_name,
        "interference_pairs_requested": pairs,
        "scope": "Sampled native kinematic checks and world-coordinate COM checks. No manufacturing, full-motion collision, load, electrical or dynamics certification.",
        "tolerances": {"angle_degrees": ANGLE_TOLERANCE_DEG, "distance_mm": DISTANCE_TOLERANCE_MM, "com_mm": COM_TOLERANCE_MM},
        "cases": [], "errors": [],
    }
    native = None
    try:
        inventory = inspect_native(model)
        report["native_container_counts"] = inventory["counts"]
        report["native_container_joint_graph"] = inventory["joint_graph"]
        if report["full_kinematic_suite_requested"]:
            rocks = [o for o in inventory["occurrences"] if o["name"] == "Rock sample:1"]
            report["retained_scenery"] = [
                {"occurrence_id": o["id"], "name": o["name"], "grounded": o.get("grounded", False)} for o in rocks
            ]
            require(len(rocks) == 1 and rocks[0].get("grounded") is True,
                    "The retained Rock sample:1 helper must be explicitly grounded scenery")
        native = NativeMcp(cli, model, output, args.timeout, args.startup_timeout)
        report["server"] = native.server.get("serverInfo")
        inspection = native.tool("inspect_design", {"measure": False, "sketches": False, "joints": False})
        with (output / "initial-inspection.json").open("x", encoding="utf-8") as stream:
            json.dump(inspection, stream, indent=2, allow_nan=False)
            stream.write("\n")
        require(isinstance(inspection.get("timeline"), list) and inspection["timeline"], "Missing native timeline")
        issues = [{k: f.get(k) for k in ("id", "name", "error", "warning")}
                  for f in inspection["timeline"] if f.get("error") or f.get("warning")]
        report["timeline_issues"] = issues
        report["evaluated_bodies"] = len(inspection.get("bodies", []))
        require(not issues, "Native timeline has unreviewed errors or warnings")
        check = Verification(native, areas, set(args.bearing_name), inventory)
        if report["full_kinematic_suite_requested"]:
            saved = []
            report["saved_independent_driver_coordinates"] = saved
            for name in PRIMARY:
                require(name in check.initial, "Saved master is missing an independent driver: "+name)
                values = check.initial[name].get("values", [])
                require(values and values[0].get("unit") == "deg", "Missing saved driver angle: "+name)
                actual = values[0].get("value")
                require(type(actual) in (int, float) and math.isfinite(actual), "Invalid saved driver angle: "+name)
                saved.append({"joint": name, "degrees": actual, "neutral": abs(actual) <= ANGLE_TOLERANCE_DEG})
            require(all(x["neutral"] for x in saved), "Saved master is not at the documented neutral pose")
        report["cases"] = check.cases
        check.run()
        if pairs:
            check.interference(pairs, args.interference_budget, {b["name"] for b in inspection["bodies"]})
            checked_pairs = [c for c in check.cases if "native_interference" in c]
            require(len(checked_pairs) == len(pairs), "Some requested interference pairs were not checked")
        report["status"] = "passed" if all(c["status"] == "passed" for c in check.cases) else "failed"
    except (OSError, ValueError, KeyError, TypeError, RuntimeError, TimeoutError) as error:
        report["errors"].append(str(error))
    finally:
        if native is not None:
            native.close()
        report["model_unchanged"] = digest(model) == before
        report["source_unchanged"] = digest(source) == source_before
        if not report["model_unchanged"] or not report["source_unchanged"]:
            report["status"] = "failed"
            report["errors"].append("Input file changed during validation")
        report["numeric_checks"] = sum(len(c.get("checks", [])) for c in report["cases"])
        report["finished_utc"] = datetime.now(timezone.utc).isoformat()
        with (output / "validation.json").open("x", encoding="utf-8") as stream:
            json.dump(report, stream, indent=2, allow_nan=False)
            stream.write("\n")
    print(json.dumps({"status": report["status"], "cases": len(report["cases"]), "numeric_checks": report["numeric_checks"],
                      "errors": report["errors"], "report": str(output / "validation.json")}), flush=True)
    return 0 if report["status"] == "passed" else 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError) as error:
        print("Validation setup failed: "+str(error), file=sys.stderr)
        raise SystemExit(2)