From 01ec12263aaab6232f1732c77cf7c214d3f16573 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 24 Jul 2026 18:26:43 +0300 Subject: [PATCH] feat(simulation): qualify real S1B provider lifecycle --- simulation/s1/stock-rover-lifecycle.yaml | 19 + simulation/s1/stock_rover_lifecycle.py | 347 ++++++++++++++++++ simulation/s1/worker-stock-rover-lifecycle.sh | 53 +++ src/k1link/simulation/__init__.py | 16 + src/k1link/simulation/process_supervisor.py | 80 +++- src/k1link/simulation/stock_rover.py | 190 ++++++++++ tests/test_simulation_s1_orchestrator.py | 114 ++++++ 7 files changed, 813 insertions(+), 6 deletions(-) create mode 100644 simulation/s1/stock-rover-lifecycle.yaml create mode 100644 simulation/s1/stock_rover_lifecycle.py create mode 100755 simulation/s1/worker-stock-rover-lifecycle.sh create mode 100644 src/k1link/simulation/stock_rover.py diff --git a/simulation/s1/stock-rover-lifecycle.yaml b/simulation/s1/stock-rover-lifecycle.yaml new file mode 100644 index 0000000..c5e1ed4 --- /dev/null +++ b/simulation/s1/stock-rover-lifecycle.yaml @@ -0,0 +1,19 @@ +schema_version: missioncore.stock-rover-lifecycle/v1 +profile_id: s1b-stock-rover-lifecycle-v1 +speed_factor: 1 +dwell_seconds: 5 + +authority: + simulation_or_shadow_only: true + actuator_authority: false + navigation_or_safety_accepted: false + direct_actuator_setpoints_allowed: false + +agent_ready_log_patterns: + - running... + +px4_ready_log_patterns: + - 'world: rover, model: rover_ackermann_0' + - Startup script returned successfully + - successfully created rt/fmu/out/vehicle_status_v1 data writer + - Running, connected diff --git a/simulation/s1/stock_rover_lifecycle.py b/simulation/s1/stock_rover_lifecycle.py new file mode 100644 index 0000000..ae5d18c --- /dev/null +++ b/simulation/s1/stock_rover_lifecycle.py @@ -0,0 +1,347 @@ +from __future__ import annotations + +import argparse +import hashlib +import os +import re +import shutil +import sys +import time +from datetime import UTC, datetime +from pathlib import Path + +from k1link.artifacts import write_json_atomic +from k1link.simulation import ( + AuthorityProfile, + LocalProcessWorkerAdapter, + PosixProcessSupervisor, + ProviderPin, + QualificationRun, + QualificationRunStore, + ReproducibilityTier, + RunKind, + RunState, + S0WorkerGuard, + SimulationApplicationService, + StockRoverTargetPaths, + load_stock_rover_lifecycle_profile, + stock_rover_process_environment, + stock_rover_process_specs, +) +from k1link.simulation.s0 import ComponentPin, S0Profile +from k1link.simulation.worker import SimulationWorldControl + +EXPECTED_DISTRO = "MissionCore-Sim" +COMMIT_PATTERN = re.compile(r"^[a-f0-9]{40}$") +PROVIDER_IDS = ( + "px4-autopilot", + "gazebo", + "px4-gazebo-models", + "micro-xrce-dds-agent", +) + + +class _LifecycleOnlyWorldControl(SimulationWorldControl): + def pause(self, run_id: str) -> None: + raise RuntimeError(f"world pause is not admitted by S1B lifecycle run {run_id}") + + def resume(self, run_id: str) -> None: + raise RuntimeError(f"world resume is not admitted by S1B lifecycle run {run_id}") + + def step(self, run_id: str, step_count: int) -> int: + raise RuntimeError(f"world step is not admitted by S1B lifecycle run {run_id}:{step_count}") + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Run the accepted PX4/Gazebo stock rover under Mission Core S1B ownership." + ) + parser.add_argument("--run-id", required=True) + parser.add_argument("--mission-core-commit", required=True) + parser.add_argument("--lifecycle-profile", type=Path, required=True) + parser.add_argument( + "--s0-profile", + type=Path, + default=Path("/mnt/d/NDC_MISSIONCORE/simulation/source/qualification-profile.yaml"), + ) + parser.add_argument( + "--d-root", + type=Path, + default=Path("/mnt/d/NDC_MISSIONCORE/simulation"), + ) + args = parser.parse_args() + + if not COMMIT_PATTERN.fullmatch(args.mission_core_commit): + parser.error("--mission-core-commit must be an exact 40-character Git revision") + if os.geteuid() == 0: + parser.error("the lifecycle harness must run as the unprivileged missioncore user") + if os.environ.get("WSL_DISTRO_NAME") != EXPECTED_DISTRO: + parser.error(f"the lifecycle harness requires WSL distribution {EXPECTED_DISTRO}") + + paths = StockRoverTargetPaths.from_d_root(args.d_root) + lifecycle_profile_path = args.lifecycle_profile.expanduser().resolve() + s0_profile_path = args.s0_profile.expanduser().resolve() + lifecycle_profile = load_stock_rover_lifecycle_profile(lifecycle_profile_path) + guard = S0WorkerGuard(s0_profile_path) + _preflight(paths, guard.profile) + + artifact_runs_root = paths.d_root / "artifacts/s1/runs" + runtime_runs_root = paths.d_root / "runtime/s1/runs" + evidence_root = paths.d_root / "artifacts/s1/lifecycle" / args.run_id + evidence_root.mkdir(mode=0o700, parents=True, exist_ok=False) + + interface_names = _loopback_only_interfaces() + write_json_atomic( + evidence_root / "network.json", + { + "schema_version": "missioncore.s1b-network-evidence/v1", + "run_id": args.run_id, + "namespace": "product-owned-unshare-net", + "interfaces": list(interface_names), + "loopback_only": interface_names == ("lo",), + }, + ) + + run = QualificationRun( + run_id=args.run_id, + episode_id=f"episode-{args.run_id}", + kind=RunKind.SIMULATION_CLOSED_LOOP, + state=RunState.ADMITTED, + scenario_generation="px4-v1.17.0-stock-rover-ackermann", + scenario_sha256=_sha256(paths.scenario_path), + profile_generation=lifecycle_profile.profile_id, + profile_sha256=_sha256(lifecycle_profile_path), + mission_core_commit=args.mission_core_commit, + providers=_provider_pins(guard.profile), + host_profile_id=guard.profile.profile_id, + host_profile_sha256=guard.profile_sha256, + seed=42, + reproducibility_tier=ReproducibilityTier.R1, + authority=AuthorityProfile( + generation=1, + command_ttl_max_ns=250_000_000, + heartbeat_timeout_monotonic_ns=500_000_000, + ), + clock_domain="gazebo:/clock", + created_at_utc=_utc_now(), + ) + store = QualificationRunStore(artifact_runs_root) + store.create(run) + supervisor = PosixProcessSupervisor( + runtime_runs_root, + base_environment=stock_rover_process_environment(paths), + ) + adapter = LocalProcessWorkerAdapter( + guard, + supervisor, + stock_rover_process_specs(paths, lifecycle_profile), + _LifecycleOnlyWorldControl(), + ) + service = SimulationApplicationService(store, adapter) + + verdict = "fail" + detail = "lifecycle did not start" + observed_provider_ids: tuple[str, ...] = () + try: + running = service.start( + run.run_id, + idempotency_key=f"{run.run_id}:start", + observed_at_utc=_utc_now(), + host_monotonic_ns=time.monotonic_ns(), + ) + if running.state is not RunState.RUNNING: + raise RuntimeError( + f"Mission Core did not admit provider readiness: {running.state.value}" + ) + observed_provider_ids = tuple(record.process_id for record in supervisor.snapshot()) + time.sleep(lifecycle_profile.dwell_seconds) + supervisor.snapshot() + terminal = service.stop( + run.run_id, + idempotency_key=f"{run.run_id}:stop", + observed_at_utc=_utc_now(), + host_monotonic_ns=time.monotonic_ns(), + sim_time_ns=None, + ) + if terminal.state is not RunState.COMPLETED: + raise RuntimeError( + f"Mission Core provider shutdown was not clean: {terminal.state.value}" + ) + residue = _provider_residue() + if supervisor.residue() or residue: + raise RuntimeError("provider process residue remained after Mission Core stop") + if store.list_commands(run.run_id): + raise RuntimeError("S1B lifecycle unexpectedly contains control commands") + verdict = "pass" + detail = "real stock-rover providers reached declared readiness and stopped cleanly" + except BaseException as exc: + detail = f"{type(exc).__name__}: {exc}" + _abort_active_run(service, store, run.run_id) + supervisor.stop() + finally: + terminal_run = store.load(run.run_id) + residue = _provider_residue() + _capture_directory( + artifact_runs_root / run.run_id, + evidence_root / "qualification-run", + ) + runtime_run_root = runtime_runs_root / run.run_id + if runtime_run_root.is_dir(): + _capture_directory(runtime_run_root, evidence_root / "provider-runtime") + write_json_atomic( + evidence_root / "result.json", + { + "schema_version": "missioncore.s1b-stock-rover-result/v1", + "verdict": verdict, + "detail": detail, + "run_id": run.run_id, + "run_state": terminal_run.state.value, + "terminal_reason": terminal_run.terminal_reason, + "mission_core_commit": run.mission_core_commit, + "host_profile_id": run.host_profile_id, + "host_profile_sha256": run.host_profile_sha256, + "lifecycle_profile_id": lifecycle_profile.profile_id, + "lifecycle_profile_sha256": run.profile_sha256, + "scenario_sha256": run.scenario_sha256, + "provider_ids": list(observed_provider_ids), + "event_count": len(store.list_events(run.run_id)), + "command_count": len(store.list_commands(run.run_id)), + "process_residue": list(residue), + "network_interfaces": list(interface_names), + "network_scope": "loopback-only-netns", + "actuator_authority": False, + "direct_actuator_setpoints_allowed": False, + "finished_at_utc": _utc_now(), + }, + ) + _write_digest_index(evidence_root) + + print(f"verdict={verdict}") + print(f"evidence_root={evidence_root}") + return 0 if verdict == "pass" else 1 + + +def _preflight(paths: StockRoverTargetPaths, profile: S0Profile) -> None: + required = ( + paths.px4_root, + paths.agent_binary, + paths.scenario_path, + ) + missing = tuple(str(path) for path in required if not path.exists()) + if missing: + raise RuntimeError(f"S1B target input is missing: {', '.join(missing)}") + free_bytes = shutil.disk_usage(paths.d_root).free + stop_floor_bytes = profile.storage.stop_below_gib * 1024**3 + if free_bytes < stop_floor_bytes: + raise RuntimeError("D free space is below the accepted S0 stop floor") + if not all(component.qualification_state == "accepted" for component in profile.components): + raise RuntimeError("an S0 provider generation is no longer accepted") + + +def _loopback_only_interfaces() -> tuple[str, ...]: + interfaces = tuple(sorted(path.name for path in Path("/sys/class/net").iterdir())) + if interfaces != ("lo",): + raise RuntimeError(f"S1B must run in a loopback-only namespace; observed {interfaces}") + return interfaces + + +def _provider_pins(profile: S0Profile) -> tuple[ProviderPin, ...]: + components = {component.identifier: component for component in profile.components} + missing = tuple(identifier for identifier in PROVIDER_IDS if identifier not in components) + if missing: + raise RuntimeError(f"S0 profile is missing provider pins: {', '.join(missing)}") + return tuple(_provider_pin(components[identifier]) for identifier in PROVIDER_IDS) + + +def _provider_pin(component: ComponentPin) -> ProviderPin: + version = component.resolved_version or component.requested_ref + revision = component.resolved_commit or version + return ProviderPin( + identifier=component.identifier, + version=version, + revision=revision, + ) + + +def _abort_active_run( + service: SimulationApplicationService, + store: QualificationRunStore, + run_id: str, +) -> None: + current = store.load(run_id) + if current.state not in {RunState.RUNNING, RunState.PAUSED, RunState.STOPPING}: + return + try: + service.stop( + run_id, + idempotency_key=f"{run_id}:harness-abort", + observed_at_utc=_utc_now(), + host_monotonic_ns=time.monotonic_ns(), + sim_time_ns=None, + terminal_state=RunState.ABORTED, + reason="target-harness-failure", + ) + except Exception: + return + + +def _provider_residue() -> tuple[str, ...]: + tokens = ( + "MicroXRCEAgent", + "/build/px4_sitl_default/bin/px4", + "gz sim", + "rover.sdf", + ) + residue: list[str] = [] + for process_dir in Path("/proc").iterdir(): + if not process_dir.name.isdigit() or int(process_dir.name) == os.getpid(): + continue + try: + command = (process_dir / "cmdline").read_bytes().replace(b"\x00", b" ").decode() + except (FileNotFoundError, PermissionError, UnicodeDecodeError): + continue + if command and any(token in command for token in tokens): + residue.append(f"{process_dir.name}:{command.strip()}") + return tuple(sorted(residue)) + + +def _capture_directory(source: Path, destination: Path) -> None: + if destination.exists(): + raise RuntimeError(f"evidence destination already exists: {destination}") + shutil.copytree(source, destination) + + +def _write_digest_index(evidence_root: Path) -> None: + records = [] + for path in sorted(evidence_root.rglob("*")): + if path.is_file() and path.name != "sha256-index.json": + records.append( + { + "path": path.relative_to(evidence_root).as_posix(), + "sha256": _sha256(path), + "bytes": path.stat().st_size, + } + ) + write_json_atomic( + evidence_root / "sha256-index.json", + { + "schema_version": "missioncore.evidence-digest-index/v1", + "files": records, + }, + ) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/simulation/s1/worker-stock-rover-lifecycle.sh b/simulation/s1/worker-stock-rover-lifecycle.sh new file mode 100755 index 0000000..b2cab0a --- /dev/null +++ b/simulation/s1/worker-stock-rover-lifecycle.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# +# Enter a fresh loopback-only namespace and execute the S1B stock-rover +# lifecycle as the unprivileged missioncore worker identity. + +set -Eeuo pipefail + +readonly EXPECTED_DISTRO="MissionCore-Sim" +readonly D_ROOT="/mnt/d/NDC_MISSIONCORE/simulation" +readonly SOURCE_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd -P)" +readonly RUN_ID="${1:-}" +readonly MISSION_CORE_COMMIT="${2:-}" + +if [[ -z "${RUN_ID}" || -z "${MISSION_CORE_COMMIT}" ]]; then + echo "usage: $0 " >&2 + exit 2 +fi +if [[ "${SOURCE_ROOT}" != "${D_ROOT}/source/"* ]]; then + echo "error: S1B source generation must be staged below the reviewed D root" >&2 + exit 2 +fi +if ! [[ "${MISSION_CORE_COMMIT}" =~ ^[a-f0-9]{40}$ ]]; then + echo "error: Mission Core commit must contain exactly 40 lowercase hex characters" >&2 + exit 2 +fi + +if [[ "${MISSIONCORE_S1_NETNS:-0}" != "1" ]]; then + if [[ "${EUID}" -ne 0 ]]; then + echo "error: initial S1B invocation must run as root to create the network namespace" >&2 + exit 2 + fi + if [[ "${WSL_DISTRO_NAME:-}" != "${EXPECTED_DISTRO}" ]]; then + echo "error: expected WSL distro ${EXPECTED_DISTRO}" >&2 + exit 2 + fi + exec unshare --net -- bash -c ' + set -e + ip link set lo up + exec runuser -u missioncore -- env MISSIONCORE_S1_NETNS=1 "$@" + ' bash "$0" "$@" +fi + +if [[ "${EUID}" -eq 0 || "${WSL_DISTRO_NAME:-}" != "${EXPECTED_DISTRO}" ]]; then + echo "error: S1B inner lifecycle escaped the reviewed worker identity" >&2 + exit 2 +fi + +exec env \ + PYTHONPATH="${SOURCE_ROOT}/src" \ + python3 "${SOURCE_ROOT}/simulation/s1/stock_rover_lifecycle.py" \ + --run-id "${RUN_ID}" \ + --mission-core-commit "${MISSION_CORE_COMMIT}" \ + --lifecycle-profile "${SOURCE_ROOT}/simulation/s1/stock-rover-lifecycle.yaml" diff --git a/src/k1link/simulation/__init__.py b/src/k1link/simulation/__init__.py index 384296b..843ec18 100644 --- a/src/k1link/simulation/__init__.py +++ b/src/k1link/simulation/__init__.py @@ -49,6 +49,15 @@ from k1link.simulation.s0 import ( load_s0_profile, run_s0_doctor, ) +from k1link.simulation.stock_rover import ( + LIFECYCLE_PROFILE_SCHEMA, + StockRoverLifecycleProfile, + StockRoverProfileError, + StockRoverTargetPaths, + load_stock_rover_lifecycle_profile, + stock_rover_process_environment, + stock_rover_process_specs, +) from k1link.simulation.worker import ( LocalProcessWorkerAdapter, S0WorkerGuard, @@ -68,6 +77,7 @@ __all__ = [ "DifferentialControlSetpoint", "DoctorVerdict", "LocalProcessWorkerAdapter", + "LIFECYCLE_PROFILE_SCHEMA", "OwnedProcess", "PosixProcessSupervisor", "ProcessSpec", @@ -91,6 +101,9 @@ __all__ = [ "S0DoctorReport", "S0Profile", "S0ProfileError", + "StockRoverLifecycleProfile", + "StockRoverProfileError", + "StockRoverTargetPaths", "SimulationContractError", "SimulationApplicationService", "SimulationOrchestratorError", @@ -101,5 +114,8 @@ __all__ = [ "WorkerStartResult", "WorkerStopResult", "load_s0_profile", + "load_stock_rover_lifecycle_profile", "run_s0_doctor", + "stock_rover_process_environment", + "stock_rover_process_specs", ] diff --git a/src/k1link/simulation/process_supervisor.py b/src/k1link/simulation/process_supervisor.py index 6aa61f3..d07e05f 100644 --- a/src/k1link/simulation/process_supervisor.py +++ b/src/k1link/simulation/process_supervisor.py @@ -27,6 +27,9 @@ class ProcessSpec: shutdown_order: int environment: tuple[tuple[str, str], ...] = () startup_grace_seconds: float = 0.05 + ready_log_patterns: tuple[str, ...] = () + health_timeout_seconds: float = 0.0 + health_poll_interval_seconds: float = 0.05 interrupt_timeout_seconds: float = 2.0 terminate_timeout_seconds: float = 1.0 @@ -39,10 +42,18 @@ class ProcessSpec: raise ValueError("process order must not be negative") if ( self.startup_grace_seconds < 0 + or self.health_timeout_seconds < 0 + or self.health_poll_interval_seconds <= 0 or self.interrupt_timeout_seconds < 0 or self.terminate_timeout_seconds < 0 ): raise ValueError("process timeouts must not be negative") + if any(not pattern or "\x00" in pattern for pattern in self.ready_log_patterns): + raise ValueError("ready log patterns must be nonempty and NUL-free") + if len(self.ready_log_patterns) != len(set(self.ready_log_patterns)): + raise ValueError("ready log patterns must be unique") + if self.ready_log_patterns and self.health_timeout_seconds <= 0: + raise ValueError("ready log patterns require a positive health timeout") keys = [key for key, _ in self.environment] if len(keys) != len(set(keys)) or any( not key or "=" in key or "\x00" in key or "\x00" in value @@ -221,17 +232,73 @@ class PosixProcessSupervisor: close_fds=True, ) self._processes[spec.process_id] = (spec, process) - if spec.startup_grace_seconds: + return_code = process.poll() + if return_code is not None: + phase = "before readiness" if spec.ready_log_patterns else "during startup" + raise ProcessSupervisorError( + f"provider {spec.process_id} exited {phase} with {return_code}" + ) + try: + process_group_id = os.getpgid(process.pid) + except ProcessLookupError as exc: + return_code = process.poll() + phase = "before readiness" if spec.ready_log_patterns else "during startup" + raise ProcessSupervisorError( + f"provider {spec.process_id} exited {phase} with {return_code}" + ) from exc + if process_group_id != process.pid: + raise ProcessSupervisorError( + f"provider {spec.process_id} did not acquire a dedicated process group" + ) + if spec.ready_log_patterns: + self._wait_until_ready( + process, + spec, + stdout_path=stdout_path, + stderr_path=stderr_path, + ) + elif spec.startup_grace_seconds: time.sleep(spec.startup_grace_seconds) return_code = process.poll() if return_code is not None: raise ProcessSupervisorError( f"provider {spec.process_id} exited during startup with {return_code}" ) - if os.getpgid(process.pid) != process.pid: - raise ProcessSupervisorError( - f"provider {spec.process_id} did not acquire a dedicated process group" - ) + + def _wait_until_ready( + self, + process: subprocess.Popen[bytes], + spec: ProcessSpec, + *, + stdout_path: Path, + stderr_path: Path, + ) -> None: + pending = {pattern: pattern.encode("utf-8") for pattern in spec.ready_log_patterns} + deadline = time.monotonic() + spec.health_timeout_seconds + while pending: + return_code = process.poll() + if return_code is not None: + raise ProcessSupervisorError( + f"provider {spec.process_id} exited before readiness with {return_code}; " + f"missing markers: {', '.join(pending)}" + ) + for path in (stdout_path, stderr_path): + content = path.read_bytes() + pending = { + pattern: encoded + for pattern, encoded in pending.items() + if encoded not in content + } + if not pending: + return + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ProcessSupervisorError( + f"provider {spec.process_id} did not reach readiness within " + f"{spec.health_timeout_seconds:g} seconds; " + f"missing markers: {', '.join(pending)}" + ) + time.sleep(min(spec.health_poll_interval_seconds, remaining)) def _stop_group( self, @@ -266,7 +333,8 @@ class PosixProcessSupervisor: while time.monotonic() < deadline: if process.poll() is not None: process.wait(timeout=0) - return not group_signal_supported or not _group_exists(pgid) + if not group_signal_supported or not _group_exists(pgid): + return True time.sleep(0.01) return process.poll() is not None and ( not group_signal_supported or not _group_exists(pgid) diff --git a/src/k1link/simulation/stock_rover.py b/src/k1link/simulation/stock_rover.py new file mode 100644 index 0000000..9bf46d5 --- /dev/null +++ b/src/k1link/simulation/stock_rover.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import yaml + +from k1link.simulation.process_supervisor import ProcessSpec + +LIFECYCLE_PROFILE_SCHEMA: Final = "missioncore.stock-rover-lifecycle/v1" +EXPECTED_D_ROOT: Final = Path("/mnt/d/NDC_MISSIONCORE/simulation") + + +class StockRoverProfileError(ValueError): + """The reviewed stock-rover lifecycle profile is unsafe or incomplete.""" + + +@dataclass(frozen=True, slots=True) +class StockRoverLifecycleProfile: + profile_id: str + speed_factor: int + dwell_seconds: float + agent_ready_log_patterns: tuple[str, ...] + px4_ready_log_patterns: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class StockRoverTargetPaths: + d_root: Path + px4_root: Path + agent_prefix: Path + process_home: Path + xdg_cache_home: Path + + @classmethod + def from_d_root(cls, d_root: Path) -> StockRoverTargetPaths: + root = d_root.expanduser().resolve() + if root != EXPECTED_D_ROOT: + raise StockRoverProfileError( + f"stock-rover lifecycle requires exact D root {EXPECTED_D_ROOT}" + ) + return cls( + d_root=root, + px4_root=root / "source/PX4-Autopilot", + agent_prefix=root / "runtime/micro-xrce-dds-agent/2.4.3", + process_home=root / "runtime/s1/home", + xdg_cache_home=root / "cache/s1/xdg", + ) + + @property + def agent_binary(self) -> Path: + return self.agent_prefix / "bin/MicroXRCEAgent" + + @property + def scenario_path(self) -> Path: + return self.px4_root / "Tools/simulation/gz/worlds/rover.sdf" + + +def load_stock_rover_lifecycle_profile(path: Path) -> StockRoverLifecycleProfile: + try: + document = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, yaml.YAMLError) as exc: + raise StockRoverProfileError("stock-rover profile is not valid UTF-8 YAML") from exc + if not isinstance(document, dict) or any(not isinstance(key, str) for key in document): + raise StockRoverProfileError("stock-rover profile must be a YAML mapping") + expected_keys = { + "schema_version", + "profile_id", + "speed_factor", + "dwell_seconds", + "authority", + "agent_ready_log_patterns", + "px4_ready_log_patterns", + } + if set(document) != expected_keys: + raise StockRoverProfileError("stock-rover profile keys do not match the v1 schema") + if document["schema_version"] != LIFECYCLE_PROFILE_SCHEMA: + raise StockRoverProfileError("stock-rover profile schema is incompatible") + authority = document["authority"] + if authority != { + "simulation_or_shadow_only": True, + "actuator_authority": False, + "navigation_or_safety_accepted": False, + "direct_actuator_setpoints_allowed": False, + }: + raise StockRoverProfileError("stock-rover lifecycle cannot grant actuator authority") + profile_id = document["profile_id"] + speed_factor = document["speed_factor"] + dwell_seconds = document["dwell_seconds"] + if not isinstance(profile_id, str) or not profile_id: + raise StockRoverProfileError("stock-rover profile id must be nonempty") + if speed_factor != 1: + raise StockRoverProfileError("S1B lifecycle admits only speed factor 1") + if not isinstance(dwell_seconds, int | float) or isinstance(dwell_seconds, bool): + raise StockRoverProfileError("stock-rover dwell must be numeric") + if not 1 <= float(dwell_seconds) <= 30: + raise StockRoverProfileError("stock-rover dwell must be between 1 and 30 seconds") + return StockRoverLifecycleProfile( + profile_id=profile_id, + speed_factor=speed_factor, + dwell_seconds=float(dwell_seconds), + agent_ready_log_patterns=_patterns( + document["agent_ready_log_patterns"], + "agent", + ), + px4_ready_log_patterns=_patterns( + document["px4_ready_log_patterns"], + "PX4", + ), + ) + + +def stock_rover_process_environment( + paths: StockRoverTargetPaths, +) -> dict[str, str]: + paths.process_home.mkdir(mode=0o700, parents=True, exist_ok=True) + paths.xdg_cache_home.mkdir(mode=0o700, parents=True, exist_ok=True) + return { + "HOME": str(paths.process_home), + "LANG": os.environ.get("LANG", "C.UTF-8"), + "LOGNAME": "missioncore", + "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + "USER": "missioncore", + "XDG_CACHE_HOME": str(paths.xdg_cache_home), + } + + +def stock_rover_process_specs( + paths: StockRoverTargetPaths, + profile: StockRoverLifecycleProfile, +) -> tuple[ProcessSpec, ...]: + return ( + ProcessSpec( + process_id="micro-xrce-dds-agent", + argv=( + str(paths.agent_binary), + "udp4", + "-p", + "8888", + "-v", + "4", + ), + start_order=10, + shutdown_order=30, + environment=(("LD_LIBRARY_PATH", str(paths.agent_prefix / "lib")),), + startup_grace_seconds=0, + ready_log_patterns=profile.agent_ready_log_patterns, + health_timeout_seconds=15, + health_poll_interval_seconds=0.1, + interrupt_timeout_seconds=3, + terminate_timeout_seconds=2, + ), + ProcessSpec( + process_id="px4-gazebo-stock-rover", + argv=( + "/usr/bin/make", + "-C", + str(paths.px4_root), + "px4_sitl", + "gz_rover_ackermann", + ), + start_order=20, + shutdown_order=40, + environment=( + ("HEADLESS", "1"), + ("PX4_SIM_SPEED_FACTOR", str(profile.speed_factor)), + ), + startup_grace_seconds=0, + ready_log_patterns=profile.px4_ready_log_patterns, + health_timeout_seconds=120, + health_poll_interval_seconds=0.25, + interrupt_timeout_seconds=10, + terminate_timeout_seconds=5, + ), + ) + + +def _patterns(value: object, label: str) -> tuple[str, ...]: + if ( + not isinstance(value, list) + or not value + or any(not isinstance(item, str) or not item or "\x00" in item for item in value) + ): + raise StockRoverProfileError(f"{label} ready patterns must be nonempty strings") + patterns = tuple(value) + if len(patterns) != len(set(patterns)): + raise StockRoverProfileError(f"{label} ready patterns must be unique") + return patterns diff --git a/tests/test_simulation_s1_orchestrator.py b/tests/test_simulation_s1_orchestrator.py index b84b0dc..1685a59 100644 --- a/tests/test_simulation_s1_orchestrator.py +++ b/tests/test_simulation_s1_orchestrator.py @@ -22,12 +22,16 @@ from k1link.simulation import ( RunState, S0WorkerGuard, SimulationApplicationService, + StockRoverTargetPaths, WorkerAdmissionError, WorkerStartResult, WorkerStopResult, + load_stock_rover_lifecycle_profile, + stock_rover_process_specs, ) PROFILE = Path(__file__).resolve().parents[1] / "simulation/s0/qualification-profile.yaml" +S1B_PROFILE = Path(__file__).resolve().parents[1] / "simulation/s1/stock-rover-lifecycle.yaml" SHA_A = "a" * 64 SHA_B = "b" * 64 @@ -336,6 +340,116 @@ def test_process_supervisor_rolls_back_partial_start(tmp_path: Path) -> None: assert supervisor.residue() == () +@pytest.mark.skipif(os.name != "posix", reason="target supervisor is POSIX-only") +def test_process_supervisor_waits_for_all_declared_ready_markers(tmp_path: Path) -> None: + child = ( + "import signal,time;" + "signal.signal(signal.SIGINT,lambda *_:exit(0));" + "print('provider booting',flush=True);" + "time.sleep(0.05);" + "print('transport connected',flush=True);" + "time.sleep(60)" + ) + supervisor = PosixProcessSupervisor(tmp_path / "runtime") + + records = supervisor.start( + "run-ready-001", + ( + ProcessSpec( + process_id="ready-provider", + argv=(sys.executable, "-c", child), + start_order=1, + shutdown_order=1, + ready_log_patterns=("provider booting", "transport connected"), + health_timeout_seconds=1, + health_poll_interval_seconds=0.01, + ), + ), + ) + + assert tuple(record.process_id for record in records) == ("ready-provider",) + assert supervisor.stop().residue_process_ids == () + + +@pytest.mark.skipif(os.name != "posix", reason="target supervisor is POSIX-only") +def test_process_supervisor_readiness_timeout_rolls_back_without_residue( + tmp_path: Path, +) -> None: + supervisor = PosixProcessSupervisor(tmp_path / "runtime") + + with pytest.raises(ProcessSupervisorError, match="missing markers: never-ready"): + supervisor.start( + "run-ready-timeout-001", + ( + ProcessSpec( + process_id="not-ready", + argv=(sys.executable, "-c", "import time;time.sleep(60)"), + start_order=1, + shutdown_order=1, + ready_log_patterns=("never-ready",), + health_timeout_seconds=0.05, + health_poll_interval_seconds=0.01, + ), + ), + ) + + assert supervisor.residue() == () + + +@pytest.mark.skipif(os.name != "posix", reason="target supervisor is POSIX-only") +def test_process_supervisor_rejects_exit_before_readiness(tmp_path: Path) -> None: + supervisor = PosixProcessSupervisor(tmp_path / "runtime") + + with pytest.raises(ProcessSupervisorError, match="exited before readiness"): + supervisor.start( + "run-ready-exit-001", + ( + ProcessSpec( + process_id="early-exit", + argv=(sys.executable, "-c", "raise SystemExit(7)"), + start_order=1, + shutdown_order=1, + ready_log_patterns=("never-ready",), + health_timeout_seconds=1, + health_poll_interval_seconds=0.01, + ), + ), + ) + + assert supervisor.residue() == () + + +def test_stock_rover_profile_builds_exact_reviewed_provider_graph() -> None: + profile = load_stock_rover_lifecycle_profile(S1B_PROFILE) + paths = StockRoverTargetPaths.from_d_root(Path("/mnt/d/NDC_MISSIONCORE/simulation")) + + specs = stock_rover_process_specs(paths, profile) + + assert profile.profile_id == "s1b-stock-rover-lifecycle-v1" + assert tuple(spec.process_id for spec in specs) == ( + "micro-xrce-dds-agent", + "px4-gazebo-stock-rover", + ) + assert specs[0].ready_log_patterns == ("running...",) + assert specs[1].argv[-2:] == ("px4_sitl", "gz_rover_ackermann") + assert specs[1].ready_log_patterns[-1] == "Running, connected" + assert dict(specs[1].environment) == { + "HEADLESS": "1", + "PX4_SIM_SPEED_FACTOR": "1", + } + + +def test_ready_patterns_require_a_positive_timeout() -> None: + with pytest.raises(ValueError, match="positive health timeout"): + ProcessSpec( + process_id="unsafe-readiness", + argv=(sys.executable, "-c", "pass"), + start_order=1, + shutdown_order=1, + ready_log_patterns=("ready",), + ) + + def test_s0_worker_guard_binds_exact_digest_and_d_only_run_path() -> None: guard = S0WorkerGuard(PROFILE) run = _run(host_profile_sha256=hashlib.sha256(PROFILE.read_bytes()).hexdigest())