NODEDC_MISSION_CORE/simulation/s1/stock_rover_lifecycle.py

348 lines
13 KiB
Python

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())