pyro-mcp/src/pyro_mcp/runtime_boot_check.py

"""Direct Firecracker boot validation for a bundled runtime profile."""

from __future__ import annotations

import argparse
import shutil
import subprocess
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace

from pyro_mcp.runtime import resolve_runtime_paths
from pyro_mcp.vm_firecracker import build_launch_plan
from pyro_mcp.vm_profiles import get_profile


@dataclass(frozen=True)
class BootCheckResult:
    profile: str
    workdir: Path
    firecracker_started: bool
    vm_alive_after_wait: bool
    process_returncode: int | None
    kernel_panic: bool
    failure_reason: str | None
    serial_log: str
    firecracker_log: str


def _read_text(path: Path) -> str:  # pragma: no cover - thin file helper
    if not path.exists():
        return ""
    return path.read_text(encoding="utf-8", errors="ignore")


def _classify_result(*, firecracker_log: str, serial_log: str, vm_alive: bool) -> str | None:
    if vm_alive:
        return None
    if "Could not initialize logger" in serial_log:
        return "firecracker logger initialization failed"
    if "Kernel panic" in serial_log:
        return "guest kernel panic during boot"
    if "Successfully started microvm" not in firecracker_log:
        return "firecracker did not fully start the microVM"
    return "microVM exited before boot validation window elapsed"


def run_boot_check(
    *,
    profile: str = "debian-base",
    vcpu_count: int = 1,
    mem_mib: int = 1024,
    wait_seconds: int = 8,
    keep_workdir: bool = False,
) -> BootCheckResult:  # pragma: no cover - integration helper
    get_profile(profile)
    if wait_seconds <= 0:
        raise ValueError("wait_seconds must be positive")

    runtime_paths = resolve_runtime_paths()
    profile_dir = runtime_paths.artifacts_dir / profile

    workdir = Path(tempfile.mkdtemp(prefix="pyro-boot-check-"))
    try:
        rootfs_copy = workdir / "rootfs.ext4"
        shutil.copy2(profile_dir / "rootfs.ext4", rootfs_copy)
        instance = SimpleNamespace(
            vm_id="abcd00000001",
            vcpu_count=vcpu_count,
            mem_mib=mem_mib,
            workdir=workdir,
            metadata={
                "kernel_image": str(profile_dir / "vmlinux"),
                "rootfs_image": str(rootfs_copy),
            },
            network=None,
        )
        launch_plan = build_launch_plan(instance)
        serial_path = workdir / "serial.log"
        firecracker_log_path = workdir / "firecracker.log"
        firecracker_log_path.touch()

        with serial_path.open("w", encoding="utf-8") as serial_fp:
            proc = subprocess.Popen(  # noqa: S603
                [
                    str(runtime_paths.firecracker_bin),
                    "--no-api",
                    "--config-file",
                    str(launch_plan.config_path),
                    "--log-path",
                    str(firecracker_log_path),
                    "--level",
                    "Info",
                ],
                stdout=serial_fp,
                stderr=subprocess.STDOUT,
                text=True,
            )
            time.sleep(wait_seconds)
            vm_alive = proc.poll() is None
            process_returncode = proc.poll()
            if vm_alive:
                proc.kill()
                proc.wait(timeout=10)
                process_returncode = proc.returncode

        serial_log = _read_text(serial_path)
        firecracker_log = _read_text(firecracker_log_path)
        failure_reason = _classify_result(
            firecracker_log=firecracker_log,
            serial_log=serial_log,
            vm_alive=vm_alive,
        )
        return BootCheckResult(
            profile=profile,
            workdir=workdir,
            firecracker_started="Successfully started microvm" in firecracker_log,
            vm_alive_after_wait=vm_alive,
            process_returncode=process_returncode,
            kernel_panic="Kernel panic" in serial_log,
            failure_reason=failure_reason,
            serial_log=serial_log,
            firecracker_log=firecracker_log,
        )
    finally:
        if not keep_workdir:
            shutil.rmtree(workdir, ignore_errors=True)


def main() -> None:  # pragma: no cover - CLI wiring
    parser = argparse.ArgumentParser(description="Run a direct Firecracker boot check.")
    parser.add_argument("--profile", default="debian-base")
    parser.add_argument("--vcpu-count", type=int, default=1)
    parser.add_argument("--mem-mib", type=int, default=1024)
    parser.add_argument("--wait-seconds", type=int, default=8)
    parser.add_argument("--keep-workdir", action="store_true")
    parser.add_argument("-v", "--verbose", action="store_true")
    args = parser.parse_args()

    result = run_boot_check(
        profile=args.profile,
        vcpu_count=args.vcpu_count,
        mem_mib=args.mem_mib,
        wait_seconds=args.wait_seconds,
        keep_workdir=args.keep_workdir,
    )
    print(f"[boot] profile={result.profile}")
    print(f"[boot] firecracker_started={result.firecracker_started}")
    print(f"[boot] vm_alive_after_wait={result.vm_alive_after_wait}")
    print(f"[boot] process_returncode={result.process_returncode}")
    if result.failure_reason is None:
        print("[boot] result=success")
        return
    print(f"[boot] result=failure reason={result.failure_reason}")
    if args.keep_workdir:
        print(f"[boot] workdir={result.workdir}")
    if args.verbose:
        print("[serial]")
        print(result.serial_log.rstrip())
        print("[firecracker]")
        print(result.firecracker_log.rstrip())
    raise SystemExit(1)