NODEDC_MISSION_CORE/simulation/s1/stock_rover_lifecycle.py

from __future__ import annotations

import argparse
import hashlib
import os
import re
import shutil
import socket
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)
    evidence_root = paths.d_root / "artifacts/s1/lifecycle" / args.run_id
    evidence_root.mkdir(mode=0o700, parents=True, exist_ok=False)
    lifecycle_profile_path = args.lifecycle_profile.expanduser().resolve()
    s0_profile_path = args.s0_profile.expanduser().resolve()
    try:
        lifecycle_profile = load_stock_rover_lifecycle_profile(lifecycle_profile_path)
        guard = S0WorkerGuard(s0_profile_path)
        _preflight(paths, guard.profile)
        interface_names = _loopback_only_interfaces()
    except Exception as exc:
        write_json_atomic(
            evidence_root / "result.json",
            {
                "schema_version": "missioncore.s1b-stock-rover-result/v1",
                "verdict": "fail",
                "phase": "preflight",
                "detail": f"{type(exc).__name__}: {exc}",
                "run_id": args.run_id,
                "mission_core_commit": args.mission_core_commit,
                "network_scope": "loopback-only-netns-required",
                "actuator_authority": False,
                "direct_actuator_setpoints_allowed": False,
                "finished_at_utc": _utc_now(),
            },
        )
        _write_digest_index(evidence_root)
        print("verdict=fail")
        print(f"evidence_root={evidence_root}")
        return 1

    artifact_runs_root = paths.d_root / "artifacts/s1/runs"
    runtime_runs_root = paths.d_root / "runtime/s1/runs"
    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(name for _, name in socket.if_nameindex()))
    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())