| """Explicit startup backends for real hardware and isolated MuJoCo deployment. |
| |
| The real entry point never exposes the simulation backend. The simulation |
| entry point cannot attach the production SHM name and cannot manufacture the |
| real MotionSwitcher-attestation bit. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from pathlib import Path |
| import subprocess |
| import sys |
| import time |
| from typing import Protocol |
|
|
| from .h2_contract import H2_JOINT_MAP_REVISION |
| from .h2_shm_bridge import ( |
| H2SharedMemoryBridge, |
| SHM_NAME, |
| SIMULATION_ACK, |
| SIM_SHM_PREFIX, |
| ) |
| from .motion_attestation import ( |
| MOTION_SWITCHER_ACK, |
| consume_attestation, |
| require_interface_up, |
| ) |
|
|
|
|
| class RunnerStartupBackend(Protocol): |
| @property |
| def identity(self) -> str: ... |
|
|
| def validate_before_model_load(self, *, shm_name: str, config_revision: str) -> None: ... |
|
|
| def attach_after_model_load(self, *, shm_name: str) -> tuple[H2SharedMemoryBridge, str]: ... |
|
|
|
|
| @dataclass(frozen=True) |
| class RealRunnerStartup: |
| network_interface: str |
| joint_map_ack: str |
| motion_switcher_ack: str |
| check_motion_switcher_on_attach: bool = False |
|
|
| @property |
| def identity(self) -> str: |
| return str(self.network_interface) |
|
|
| def validate_before_model_load(self, *, shm_name: str, config_revision: str) -> None: |
| if str(shm_name).lstrip("/") != SHM_NAME: |
| raise RuntimeError( |
| f"real H2 control accepts only the production SHM name {SHM_NAME!r}" |
| ) |
| if config_revision != H2_JOINT_MAP_REVISION: |
| raise RuntimeError( |
| "config H2 joint-map revision does not match this executable: " |
| f"{config_revision!r} != {H2_JOINT_MAP_REVISION!r}" |
| ) |
| if str(self.joint_map_ack) != H2_JOINT_MAP_REVISION: |
| raise RuntimeError( |
| "real H2 control requires explicit --ack-joint-map-revision " |
| f"{H2_JOINT_MAP_REVISION} after suspended ID verification" |
| ) |
| if str(self.motion_switcher_ack) != MOTION_SWITCHER_ACK: |
| raise RuntimeError( |
| "real H2 control requires explicit --ack-motion-switcher-attestation " |
| f"{MOTION_SWITCHER_ACK} after a fresh successful CheckMode" |
| ) |
| require_interface_up(self.network_interface) |
|
|
| def attach_after_model_load(self, *, shm_name: str) -> tuple[H2SharedMemoryBridge, str]: |
| if self.check_motion_switcher_on_attach: |
| started = time.monotonic() |
| print("[STARTUP] CheckMode begin (models ready; read-only, no release)", flush=True) |
| guard = Path(__file__).resolve().parents[1] / "scripts/motion_switcher_guard.py" |
| try: |
| subprocess.run([sys.executable, str(guard), "--network-interface", |
| self.network_interface, "--write-attestation"], |
| check=True, timeout=15) |
| except (subprocess.SubprocessError, OSError) as error: |
| raise RuntimeError(f"startup CheckMode failed after {time.monotonic()-started:.2f}s: {error}") from error |
| print(f"[STARTUP] CheckMode complete elapsed_s={time.monotonic()-started:.3f}", flush=True) |
| bridge = H2SharedMemoryBridge(name=shm_name, claim_writer=True) |
| try: |
| age_s = consume_attestation(self.network_interface) |
| bridge.attest_motion_switcher_released(self.network_interface) |
| except BaseException: |
| bridge.close() |
| raise |
| return bridge, ( |
| f"consumed MotionSwitcher attestation age={age_s:.3f}s " |
| f"interface={self.network_interface}" |
| ) |
|
|
|
|
| @dataclass(frozen=True) |
| class SimulationRunnerStartup: |
| acknowledgement: str |
|
|
| @property |
| def identity(self) -> str: |
| return "MuJoCo simulation only" |
|
|
| def validate_before_model_load(self, *, shm_name: str, config_revision: str) -> None: |
| normalized = str(shm_name).lstrip("/") |
| if normalized == SHM_NAME or not normalized.startswith(SIM_SHM_PREFIX): |
| raise RuntimeError( |
| f"simulation runner requires a random {SIM_SHM_PREFIX}<run-id> SHM name" |
| ) |
| if config_revision != H2_JOINT_MAP_REVISION: |
| raise RuntimeError( |
| "simulation and real runner must use the same pinned H2 joint-map revision" |
| ) |
| if str(self.acknowledgement) != SIMULATION_ACK: |
| raise RuntimeError( |
| "simulation deployment requires explicit --ack-simulation-only " |
| f"{SIMULATION_ACK}" |
| ) |
|
|
| def attach_after_model_load(self, *, shm_name: str) -> tuple[H2SharedMemoryBridge, str]: |
| bridge = H2SharedMemoryBridge(name=shm_name, claim_writer=True) |
| try: |
| bridge.attest_simulation_runner() |
| except BaseException: |
| bridge.close() |
| raise |
| return bridge, "validated isolated MuJoCo H2BR owner and simulation-only command gate" |
|
|
|
|
| __all__ = [ |
| "RealRunnerStartup", |
| "RunnerStartupBackend", |
| "SimulationRunnerStartup", |
| ] |
|
|
