trycua--cua/libs/python/cua-sandbox/cua_sandbox/runtime/qemu.py

"""QEMU runtime — bare-metal or Docker-wrapped QEMU VMs.

Two modes:
  QEMURuntime(mode="docker")     — default, uses trycua/cua-qemu-* Docker images
  QEMURuntime(mode="bare-metal") — launches qemu-system-* directly on the host
"""

from __future__ import annotations

import asyncio
import json as _json
import logging
import platform as _plat
import shutil
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Optional

if TYPE_CHECKING:
    from cua_sandbox.image import Image

import httpx
from cua_sandbox.image import Image
from cua_sandbox.runtime.base import Runtime, RuntimeInfo
from cua_sandbox.runtime.docker import DockerRuntime, _has_kvm
from cua_sandbox.runtime.images import (
    DEFAULT_API_PORT,
    QEMU_VNC_PORT,
)

logger = logging.getLogger(__name__)

# ── Storage directory for QEMU disk images ──────────────────────────────────

QEMU_STORAGE_ROOT = Path.home() / ".cua" / "cua-sandbox" / "qemu-storage"


async def _qmp_command(
    host: str, port: int, command: str, arguments: Optional[dict] = None
) -> dict:
    """Send a single QMP command and return the response."""
    reader, writer = await asyncio.open_connection(host, port)
    try:
        # Read greeting
        await asyncio.wait_for(reader.readline(), timeout=5)
        # Negotiate capabilities
        writer.write(b'{"execute":"qmp_capabilities"}\n')
        await writer.drain()
        await asyncio.wait_for(reader.readline(), timeout=5)
        # Send command
        msg: dict = {"execute": command}
        if arguments:
            msg["arguments"] = arguments
        writer.write((_json.dumps(msg) + "\n").encode())
        await writer.drain()
        raw = await asyncio.wait_for(reader.readline(), timeout=30)
        return _json.loads(raw)
    finally:
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass


class QEMUDockerRuntime(DockerRuntime):
    """QEMU inside Docker — delegates to DockerRuntime with QEMU image tags.

    For Windows QEMU images, automatically:
    - Creates a /storage volume mount for the QEMU disk
    - Sets KVM=N if /dev/kvm is not available
    - Mounts an existing cached disk image if available
    """

    def __init__(
        self,
        *,
        api_port: int = DEFAULT_API_PORT,
        vnc_port: int = QEMU_VNC_PORT,
        ephemeral: bool = True,
        storage_dir: Optional[str | Path] = None,
        memory_mb: int = 8192,
        cpu_count: int = 4,
    ):
        self._storage_dir = Path(storage_dir) if storage_dir else None
        self._memory_mb = memory_mb
        self._cpu_count = cpu_count
        super().__init__(api_port=api_port, vnc_port=vnc_port, ephemeral=ephemeral)

    async def start(self, image: Image, name: str, **opts) -> RuntimeInfo:
        # Resolve storage directory
        storage = self._storage_dir or QEMU_STORAGE_ROOT / name
        storage.mkdir(parents=True, exist_ok=True)

        # Volume: host storage dir → /storage in container
        self.volumes = [f"{storage}:/storage"]

        # Environment
        self.environment = {
            "RAM_SIZE": f"{self._memory_mb // 1024}G",
            "CPU_CORES": str(self._cpu_count),
        }

        # KVM handling
        if _has_kvm():
            self.devices = ["/dev/kvm"]
        else:
            self.environment["KVM"] = "N"

        # Platform — QEMU Windows/Android images are linux/amd64
        if image.os_type in ("windows", "android"):
            self.platform = "linux/amd64"

        # Android: forward ADB port (5555) in addition to API
        if image.os_type == "android":
            self.environment["ADB_PORT"] = "5555"

        # Longer boot timeout for Windows/Android VMs
        opts.pop("boot_timeout", None)

        info = await super().start(image, name, **opts)
        return info

    async def is_ready(self, info: RuntimeInfo, timeout: float = 300) -> bool:
        """Wait for the QEMU VM's computer-server to come up.

        Windows VMs take longer to boot (3-5 min), so default timeout is 300s.
        """
        return await super().is_ready(info, timeout=timeout)

    async def suspend(self, name: str) -> None:
        """Pause the Docker container running this QEMU VM."""
        subprocess.run(["docker", "pause", name], capture_output=True)

    async def resume(self, image: "Image", name: str, **opts) -> RuntimeInfo:
        """Unpause the Docker container and return RuntimeInfo."""
        subprocess.run(["docker", "unpause", name], capture_output=True)
        result = subprocess.run(
            [
                "docker",
                "inspect",
                "--format",
                '{{(index (index .NetworkSettings.Ports "8000/tcp") 0).HostPort}}',
                name,
            ],
            capture_output=True,
            text=True,
        )
        api_port = int(result.stdout.strip()) if result.stdout.strip().isdigit() else self.api_port
        info = RuntimeInfo(host="localhost", api_port=api_port, vnc_port=self.vnc_port, name=name)
        await self.is_ready(info)
        return info


class QEMUBaremetalRuntime(Runtime):
    """Bare-metal QEMU — launches qemu-system-* directly on the host.

    Requires:
      - qemu-system-x86_64 (or qemu-system-aarch64) on PATH
      - A disk image (qcow2/vhdx/raw) with computer-server pre-installed,
        provided via Image.from_file('/path/to/disk.qcow2')
    """

    def __init__(
        self,
        *,
        api_port: int = DEFAULT_API_PORT,
        vnc_display: int = 0,
        memory_mb: int = 4096,
        cpu_count: int = 2,
        arch: str = "x86_64",
        qmp_port: int = 4444,
        use_qmp_transport: bool = False,
        extra_args: Optional[list[str]] = None,
    ):
        self.api_port = api_port
        self.vnc_display = vnc_display
        self.memory_mb = memory_mb
        self.cpu_count = cpu_count
        self.arch = arch
        self.qmp_port = qmp_port
        self.use_qmp_transport = use_qmp_transport
        self.extra_args = extra_args or []
        self._processes: dict[str, subprocess.Popen] = {}

    def _qemu_bin(self) -> str:
        from cua_sandbox.runtime.qemu_installer import qemu_bin

        return qemu_bin(self.arch)

    @staticmethod
    def _create_disk_for_iso(name: str, size_gb: int = 32) -> Path:
        """Create a qcow2 disk image for ISO-based installations.

        The disk is stored alongside other QEMU storage and reused across sessions.
        """
        disk_dir = QEMU_STORAGE_ROOT / name
        disk_dir.mkdir(parents=True, exist_ok=True)
        disk_path = disk_dir / "disk.qcow2"
        if not disk_path.exists():
            result = subprocess.run(
                ["qemu-img", "create", "-f", "qcow2", str(disk_path), f"{size_gb}G"],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise RuntimeError(f"qemu-img create failed: {result.stderr}")
        return disk_path

    async def start(self, image: Image, name: str, **opts) -> RuntimeInfo:
        ephemeral = opts.pop("ephemeral", True)

        # If image has layers or no direct disk path, use the builder to resolve
        if not opts.get("disk_path") and not image._disk_path and image.kind == "vm":
            from cua_sandbox.builder.build import create_session_disk

            disk_path = str(await create_session_disk(image, name))
        elif image._layers and (image._disk_path or opts.get("disk_path")):
            # Has a base disk AND user layers — build user image + session overlay
            from cua_sandbox.builder.build import create_session_disk

            base = Path(opts.get("disk_path") or image._disk_path)
            disk_path = str(await create_session_disk(image, name, base_disk=base))
        else:
            disk_path = opts.get("disk_path") or image._disk_path

        if not disk_path:
            raise ValueError(
                "Bare-metal QEMU requires a disk image path. "
                "Use Image.from_file('/path/to/disk.qcow2') or pass disk_path='...'"
            )

        # Handle ISO files — create a qcow2 disk and boot from ISO as CD-ROM
        self._iso_path: Optional[str] = None
        if Path(disk_path).suffix.lower() == ".iso":
            self._iso_path = disk_path
            disk_path = str(self._create_disk_for_iso(name, opts.get("disk_size_gb", 32)))
            logger.info(f"Created qcow2 disk for ISO install: {disk_path}")

        self._session_disk = Path(disk_path) if ephemeral else None

        from cua_sandbox.runtime.docker import _find_free_port

        memory = opts.get("memory_mb", self.memory_mb)
        cpus = opts.get("cpu_count", self.cpu_count)
        vnc_display = opts.get("vnc_display", self.vnc_display)
        hostfwd_port = opts.get("api_port") or _find_free_port(self.api_port)
        enable_kvm = opts.get("enable_kvm", True)

        # Detect guest server port from transport hint
        guest_port = 5000 if image._agent_type == "osworld" else 8000

        # Detect disk format from extension
        disk_ext = Path(disk_path).suffix.lower()
        disk_fmt = {".qcow2": "qcow2", ".vhdx": "vhdx", ".raw": "raw", ".img": "raw"}.get(
            disk_ext, "raw"
        )

        # Locate OVMF UEFI firmware for Windows VMs
        qemu_dir = Path(self._qemu_bin()).parent
        ovmf_code = None
        if image.os_type == "windows":
            for candidate in [
                qemu_dir / "share" / "edk2-x86_64-code.fd",
                Path("/usr/share/OVMF/OVMF_CODE.fd"),
                Path("/usr/share/qemu/edk2-x86_64-code.fd"),
            ]:
                if candidate.exists():
                    ovmf_code = candidate
                    break

        # EFI vars — look next to disk or copy template
        efivars = Path(disk_path).parent / "efivars.fd"
        if ovmf_code and not efivars.exists():
            import shutil as _shutil

            vars_template = qemu_dir / "share" / "edk2-i386-vars.fd"
            if vars_template.exists():
                _shutil.copy2(vars_template, efivars)
            else:
                efivars.write_bytes(b"\x00" * (256 * 1024))

        # Build QEMU command — Android gets different machine/device config
        is_android = image.os_type == "android"

        if is_android:
            cmd = self._build_android_cmd(
                name,
                disk_path,
                disk_fmt,
                memory,
                cpus,
                hostfwd_port,
                vnc_display,
                enable_kvm,
            )
        else:
            cmd = [
                self._qemu_bin(),
                "-name",
                name,
                "-machine",
                "q35,smm=off",
                "-m",
                str(memory),
                "-smp",
                str(cpus),
                "-cpu",
                "qemu64,+ssse3,+sse4.1,+sse4.2,+popcnt",
            ]

            # UEFI firmware (Windows requires this)
            if ovmf_code:
                cmd += [
                    "-drive",
                    f"if=pflash,format=raw,readonly=on,file={ovmf_code}",
                    "-drive",
                    f"if=pflash,format=raw,file={efivars}",
                ]

            cmd += [
                "-drive",
                f"file={disk_path},format={disk_fmt},if=virtio",
                "-netdev",
                f"user,id=net0,restrict=on,hostfwd=tcp:127.0.0.1:{hostfwd_port}-:{guest_port}",
                "-device",
                "virtio-net-pci,netdev=net0,mac=52:55:00:d1:55:01",
                "-vnc",
                f":{vnc_display}",
            ]

            # -daemonize not supported on Windows
            if _plat.system() != "Windows":
                cmd.append("-daemonize")

            if enable_kvm:
                if _plat.system() == "Darwin":
                    host_arm = _plat.machine() in ("arm64", "aarch64")
                    guest_x86 = self.arch == "x86_64"
                    if host_arm and guest_x86:
                        # Apple Silicon can't HVF-accelerate x86_64 guests
                        cmd += ["-accel", "tcg"]
                    else:
                        cmd += ["-accel", "hvf"]
                elif _plat.system() != "Windows":
                    cmd.append("-enable-kvm")
                elif ovmf_code:
                    # WHPX has MMIO bugs with OVMF pflash — use TCG for UEFI VMs
                    cmd += ["-accel", "tcg"]
                else:
                    cmd += ["-accel", "whpx"]

            # QMP socket — always enabled for VM management (suspend/resume)
            cmd += ["-qmp", f"tcp:127.0.0.1:{self.qmp_port},server,nowait"]

        # Attach ISO as CD-ROM if booting from an ISO file
        if self._iso_path:
            cmd += ["-cdrom", self._iso_path, "-boot", "d"]

        cmd.extend(self.extra_args)

        logger.info(f"Starting bare-metal QEMU: {' '.join(cmd)}")
        if "-daemonize" in cmd:
            # daemonize mode: QEMU forks to background and parent exits immediately
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode != 0:
                raise RuntimeError(f"QEMU launch failed: {result.stderr}")
        else:
            # Popen mode: keep reference to the process (Windows, or when -daemonize removed)
            proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
            self._processes[name] = proc
            import time

            time.sleep(2)
            if proc.poll() is not None:
                stderr = proc.stderr.read().decode() if proc.stderr else ""
                raise RuntimeError(f"QEMU launch failed (exit {proc.returncode}): {stderr}")

        use_qmp = is_android or self.use_qmp_transport
        info = RuntimeInfo(
            host="localhost",
            api_port=hostfwd_port,
            vnc_port=5900 + vnc_display,
            name=name,
            qmp_port=self.qmp_port if use_qmp else None,
            environment=image.os_type if use_qmp else None,
            agent_type=image._agent_type,
            guest_server_port=guest_port if not is_android else 8000,
        )

        # For ISO boots, send Enter key via QMP to skip GRUB countdown
        if self._iso_path and use_qmp:
            await self._send_boot_key(info)

        # Windows and Android need much longer to boot (3–10 min)
        boot_timeout = 600 if image.os_type in ("windows", "android") else 120
        await self.is_ready(info, timeout=boot_timeout)

        if not ephemeral:
            from cua_sandbox import sandbox_state

            sandbox_state.save(
                name,
                runtime_type="qemu-baremetal",
                image=image.to_dict(),
                host="localhost",
                api_port=hostfwd_port,
                vnc_port=5900 + vnc_display,
                qmp_port=self.qmp_port,
                disk_path=str(disk_path) if disk_path else None,
                os_type=image.os_type,
                vnc_display=vnc_display,
                memory_mb=memory,
                cpu_count=cpus,
                arch=self.arch,
                status="running",
            )

        return info

    async def _send_boot_key(self, info: RuntimeInfo) -> None:
        """Send Enter key via QMP repeatedly to skip GRUB/bootloader countdown.

        GRUB may not be ready immediately after QEMU starts (especially with TCG),
        so we send Enter multiple times over several seconds to ensure it lands.
        """

        for attempt in range(15):
            try:
                reader, writer = await asyncio.open_connection(info.host, info.qmp_port)
                await asyncio.wait_for(reader.readline(), timeout=3)
                writer.write(b'{"execute":"qmp_capabilities"}\n')
                await writer.drain()
                await asyncio.wait_for(reader.readline(), timeout=3)
                # Send Enter key several times with delays to catch GRUB at the right moment
                for i in range(5):
                    writer.write(
                        b'{"execute":"send-key","arguments":{"keys":[{"type":"qcode","data":"ret"}]}}\n'
                    )
                    await writer.drain()
                    await asyncio.wait_for(reader.readline(), timeout=3)
                    await asyncio.sleep(2)
                writer.close()
                await writer.wait_closed()
                logger.info(f"Sent boot Enter keys to {info.name}")
                return
            except (ConnectionRefusedError, OSError, asyncio.TimeoutError):
                await asyncio.sleep(2)
        logger.warning(f"Could not send boot keys to {info.name}")

    def _build_android_cmd(
        self,
        name: str,
        disk_path: str,
        disk_fmt: str,
        memory: int,
        cpus: int,
        hostfwd_port: int,
        vnc_display: int,
        enable_kvm: bool,
    ) -> list[str]:
        """Build QEMU command for Android x86_64 VM.

        Android-x86 boots from a disk image with GRUB, uses virtio for
        disk/net, and needs a GPU (virtio-gpu or std) for the display.
        Port-forwards host:hostfwd_port -> guest:8000 for the computer-server
        API, and host:5555 -> guest:5555 for ADB.
        """
        cmd = [
            self._qemu_bin(),
            "-name",
            name,
            "-machine",
            "q35,smm=off",
            "-m",
            str(memory),
            "-smp",
            str(cpus),
        ]

        # CPU config — use HVF on macOS (aarch64 only), KVM on Linux, TCG fallback
        if _plat.system() == "Darwin":
            # HVF only works when host and guest arch match.
            # Apple Silicon (arm64) can't HVF-accelerate x86_64 guests.
            host_arm = _plat.machine() in ("arm64", "aarch64")
            guest_x86 = self.arch == "x86_64"
            if enable_kvm and not (host_arm and guest_x86):
                cmd += ["-cpu", "host", "-accel", "hvf"]
            else:
                cmd += ["-cpu", "max", "-accel", "tcg"]
        elif enable_kvm and _plat.system() != "Windows":
            cmd += ["-cpu", "host", "-enable-kvm"]
        else:
            cmd += ["-cpu", "qemu64,+ssse3,+sse4.1,+sse4.2,+popcnt"]

        # Disk — Android-x86 image
        cmd += ["-drive", f"file={disk_path},format={disk_fmt},if=virtio"]

        # Networking — forward API (8000); ADB forwarded on api_port+1
        adb_port = hostfwd_port + 1
        cmd += [
            "-netdev",
            (
                f"user,id=net0,"
                f"hostfwd=tcp:127.0.0.1:{hostfwd_port}-:8000,"
                f"hostfwd=tcp:127.0.0.1:{adb_port}-:5555"
            ),
            "-device",
            "virtio-net-pci,netdev=net0",
        ]

        # Display — std VGA for broad compatibility (virtio-gpu hangs on Apple Silicon TCG)
        cmd += [
            "-vga",
            "std",
            "-vnc",
            f":{vnc_display}",
            "-display",
            "none",
        ]

        # USB tablet for absolute pointer (touch input)
        cmd += ["-usb", "-device", "usb-tablet"]

        # QMP socket for direct VM control (mouse/keyboard/screenshot without guest agent)
        cmd += ["-qmp", f"tcp:127.0.0.1:{self.qmp_port},server,nowait"]

        # Daemonize on Unix
        if _plat.system() != "Windows":
            cmd.append("-daemonize")

        return cmd

    async def stop(self, name: str) -> None:
        # Try to kill tracked process first
        proc = self._processes.pop(name, None)
        if proc and proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=10)
            except subprocess.TimeoutExpired:
                proc.kill()
        else:
            # Fallback: find by name
            try:
                if shutil.which("pkill"):
                    # Match both standard QEMU VMs (-name {name}) and Android emulators (-avd {name})
                    subprocess.run(["pkill", "-f", f"qemu.*-name {name}"], capture_output=True)
                    subprocess.run(["pkill", "-f", f"qemu.*-avd {name}"], capture_output=True)
                else:
                    subprocess.run(
                        ["taskkill", "/F", "/FI", f"WINDOWTITLE eq {name}*"],
                        capture_output=True,
                    )
            except Exception as e:
                logger.warning(f"Failed to stop QEMU VM {name}: {e}")

        # Clean up ephemeral session disk
        session_disk = getattr(self, "_session_disk", None)
        if session_disk and session_disk.exists() and "sessions" in str(session_disk):
            try:
                session_disk.unlink()
                logger.info(f"Removed session disk: {session_disk}")
            except OSError:
                pass

        from cua_sandbox import sandbox_state

        sandbox_state.delete(name)

    async def is_ready(self, info: RuntimeInfo, timeout: float = 120) -> bool:
        if info.qmp_port and not info.agent_type:
            return await self._is_ready_qmp(info, timeout)
        # For OSWorld, check the Flask server; for computer-server, check /status
        endpoint = "/screenshot" if info.agent_type == "osworld" else "/status"
        url = f"http://{info.host}:{info.api_port}{endpoint}"
        deadline = asyncio.get_event_loop().time() + timeout
        async with httpx.AsyncClient(timeout=10) as client:
            while asyncio.get_event_loop().time() < deadline:
                try:
                    resp = await client.get(url)
                    if resp.status_code == 200:
                        logger.info(f"Bare-metal QEMU VM {info.name} is ready")
                        return True
                except (
                    httpx.ConnectError,
                    httpx.ReadTimeout,
                    httpx.ReadError,
                    httpx.RemoteProtocolError,
                    httpx.ConnectTimeout,
                ):
                    pass
                await asyncio.sleep(3)
        raise TimeoutError(f"Bare-metal QEMU VM {info.name} not ready after {timeout}s")

    async def suspend(self, name: str) -> None:
        """Save VM state via QMP savevm then quit QEMU."""
        from cua_sandbox import sandbox_state

        state = sandbox_state.load(name)
        qmp_port = state["qmp_port"] if state else self.qmp_port
        try:
            await _qmp_command("localhost", qmp_port, "stop")
            await _qmp_command("localhost", qmp_port, "savevm", {"name": "cua-snapshot"})
            await _qmp_command("localhost", qmp_port, "quit")
        except Exception as e:
            logger.warning(f"QMP savevm failed for {name}: {e}")
            raise
        sandbox_state.update(name, status="suspended")

    async def resume(self, image: "Image", name: str, **opts) -> RuntimeInfo:
        """Relaunch QEMU with -loadvm to restore saved state."""
        from cua_sandbox import sandbox_state

        state = sandbox_state.load(name)
        if state is None:
            raise ValueError(f"No state file found for sandbox '{name}'. Cannot resume.")

        # Reconstruct runtime params from state
        disk_path = state.get("disk_path")
        if not disk_path:
            raise ValueError(
                f"State for '{name}' has no disk_path — cannot resume bare-metal QEMU."
            )

        api_port = state.get("api_port", self.api_port)
        vnc_display = state.get("vnc_display", self.vnc_display)
        memory = state.get("memory_mb", self.memory_mb)
        cpus = state.get("cpu_count", self.cpu_count)
        qmp_port = state.get("qmp_port", self.qmp_port)

        # Rebuild minimal QEMU command with -loadvm
        cmd = [
            self._qemu_bin(),
            "-name",
            name,
            "-machine",
            "q35,smm=off",
            "-m",
            str(memory),
            "-smp",
            str(cpus),
            "-cpu",
            "qemu64,+ssse3,+sse4.1,+sse4.2,+popcnt",
        ]

        import platform as _platform

        disk_ext = Path(disk_path).suffix.lower()
        disk_fmt = {".qcow2": "qcow2", ".vhdx": "vhdx", ".raw": "raw", ".img": "raw"}.get(
            disk_ext, "raw"
        )
        guest_port = 8000

        cmd += [
            "-drive",
            f"file={disk_path},format={disk_fmt},if=virtio",
            "-netdev",
            f"user,id=net0,restrict=on,hostfwd=tcp:127.0.0.1:{api_port}-:{guest_port}",
            "-device",
            "virtio-net-pci,netdev=net0,mac=52:55:00:d1:55:01",
            "-vnc",
            f":{vnc_display}",
            "-qmp",
            f"tcp:127.0.0.1:{qmp_port},server,nowait",
            "-loadvm",
            "cua-snapshot",
        ]

        if _platform.system() != "Windows":
            cmd.append("-daemonize")

        # Acceleration
        if _platform.system() == "Darwin":
            host_arm = _platform.machine() in ("arm64", "aarch64")
            if host_arm and self.arch == "x86_64":
                cmd += ["-accel", "tcg"]
            else:
                cmd += ["-accel", "hvf"]
        elif _platform.system() != "Windows":
            cmd.append("-enable-kvm")

        logger.info(f"Resuming bare-metal QEMU from snapshot: {' '.join(cmd)}")
        if "-daemonize" in cmd:
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode != 0:
                raise RuntimeError(f"QEMU resume failed: {result.stderr}")
        else:
            proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
            self._processes[name] = proc
            import time

            time.sleep(2)
            if proc.poll() is not None:
                stderr = proc.stderr.read().decode() if proc.stderr else ""
                raise RuntimeError(f"QEMU resume failed (exit {proc.returncode}): {stderr}")

        info = RuntimeInfo(
            host="localhost",
            api_port=api_port,
            vnc_port=5900 + vnc_display,
            name=name,
        )
        await self.is_ready(info)
        sandbox_state.update(name, status="running")
        return info

    async def list(self) -> list[dict]:
        """List known bare-metal QEMU sandboxes from state files, checking if alive."""
        from cua_sandbox import sandbox_state

        states = [s for s in sandbox_state.list_all() if s.get("runtime_type") == "qemu-baremetal"]
        result = []
        for s in states:
            name = s["name"]
            status = s.get("status", "unknown")
            # Verify QEMU process is still running (standard VMs use -name, Android uses -avd)
            if status == "running":
                try:
                    alive = (
                        subprocess.run(
                            ["pgrep", "-f", f"qemu.*-name {name}"], capture_output=True
                        ).returncode
                        == 0
                        or subprocess.run(
                            ["pgrep", "-f", f"qemu.*-avd {name}"], capture_output=True
                        ).returncode
                        == 0
                    )
                    if not alive:
                        status = "stopped"
                        sandbox_state.update(name, status="stopped")
                except FileNotFoundError:
                    pass
            result.append(
                {
                    "name": name,
                    "status": status,
                    "runtime_type": "qemu-baremetal",
                    "os_type": s.get("os_type"),
                    "host": s.get("host"),
                    "api_port": s.get("api_port"),
                }
            )
        return result

    async def _is_ready_qmp(self, info: RuntimeInfo, timeout: float = 120) -> bool:
        """Wait until QMP socket is responsive (for VMs without computer-server)."""
        import json as _json

        deadline = asyncio.get_event_loop().time() + timeout
        while asyncio.get_event_loop().time() < deadline:
            try:
                reader, writer = await asyncio.open_connection(info.host, info.qmp_port)
                # Read QMP greeting
                greeting = await asyncio.wait_for(reader.readline(), timeout=3)
                if greeting:
                    data = _json.loads(greeting)
                    if "QMP" in data:
                        writer.close()
                        await writer.wait_closed()
                        logger.info(f"Bare-metal QEMU VM {info.name} QMP ready")
                        return True
                writer.close()
                await writer.wait_closed()
            except (ConnectionRefusedError, OSError, asyncio.TimeoutError):
                pass
            await asyncio.sleep(2)
        raise TimeoutError(f"Bare-metal QEMU VM {info.name} QMP not ready after {timeout}s")


def _win_to_wsl(p: Path | str) -> str:
    """Convert a Windows path to WSL /mnt/... path."""
    s = str(p).replace("\\", "/")
    # C:/foo → /mnt/c/foo
    if len(s) >= 2 and s[1] == ":":
        return f"/mnt/{s[0].lower()}{s[2:]}"
    return s


class QEMUWSL2Runtime(Runtime):
    """QEMU via WSL2 with KVM hardware acceleration.

    Runs qemu-system-x86_64 inside WSL2 where /dev/kvm is available,
    while keeping disk images on the Windows filesystem. Windows paths
    are automatically converted to /mnt/c/... paths for WSL access.

    This is the fastest way to run QEMU on Windows — native KVM speeds
    vs TCG software emulation (bare-metal Windows) or Hyper-V (needs admin).
    """

    def __init__(
        self,
        *,
        api_port: int = DEFAULT_API_PORT,
        vnc_display: int = 0,
        memory_mb: int = 4096,
        cpu_count: int = 4,
        arch: str = "x86_64",
        extra_args: Optional[list[str]] = None,
    ):
        self.api_port = api_port
        self.vnc_display = vnc_display
        self.memory_mb = memory_mb
        self.cpu_count = cpu_count
        self.arch = arch
        self.extra_args = extra_args or []
        self._processes: dict[str, subprocess.Popen] = {}

    @staticmethod
    def _wsl(cmd: str, timeout: float = 30) -> str:
        """Run a command inside WSL2 and return stdout."""
        r = subprocess.run(
            ["wsl", "-e", "bash", "-c", cmd],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        if r.returncode != 0:
            raise RuntimeError(f"WSL command failed: {r.stderr.strip()}")
        return r.stdout.strip()

    @staticmethod
    def available() -> bool:
        """Check if WSL2 + QEMU + KVM are available."""
        try:
            r = subprocess.run(
                ["wsl", "-e", "bash", "-c", "test -e /dev/kvm && which qemu-system-x86_64"],
                capture_output=True,
                timeout=10,
            )
            return r.returncode == 0
        except (subprocess.SubprocessError, FileNotFoundError):
            return False

    async def start(self, image: Image, name: str, **opts) -> RuntimeInfo:
        ephemeral = opts.pop("ephemeral", True)

        # Resolve disk path
        if not opts.get("disk_path") and not image._disk_path and image.kind == "vm":
            from cua_sandbox.builder.build import create_session_disk

            disk_path = str(await create_session_disk(image, name))
        elif image._layers and (image._disk_path or opts.get("disk_path")):
            from cua_sandbox.builder.build import create_session_disk

            base = Path(opts.get("disk_path") or image._disk_path)
            disk_path = str(await create_session_disk(image, name, base_disk=base))
        else:
            disk_path = opts.get("disk_path") or image._disk_path

        if not disk_path:
            raise ValueError(
                "WSL2 QEMU requires a disk image path. "
                "Use Image.from_file('/path/to/disk.qcow2') or pass disk_path='...'"
            )

        self._session_disk = Path(disk_path) if ephemeral else None

        memory = opts.get("memory_mb", self.memory_mb)
        cpus = opts.get("cpu_count", self.cpu_count)
        vnc_display = opts.get("vnc_display", self.vnc_display)
        hostfwd_port = opts.get("api_port", self.api_port)

        # Convert Windows paths to WSL paths
        wsl_disk = _win_to_wsl(disk_path)
        disk_ext = Path(disk_path).suffix.lower()
        disk_fmt = {".qcow2": "qcow2", ".vhdx": "vhdx", ".raw": "raw", ".img": "raw"}.get(
            disk_ext, "raw"
        )

        # Locate OVMF inside WSL
        ovmf_code = None
        if image.os_type == "windows":
            for candidate in [
                "/usr/share/OVMF/OVMF_CODE_4M.fd",
                "/usr/share/OVMF/OVMF_CODE.fd",
                "/usr/share/qemu/edk2-x86_64-code.fd",
            ]:
                try:
                    self._wsl(f"test -f {candidate}")
                    ovmf_code = candidate
                    break
                except RuntimeError:
                    pass

        # EFI vars — create in same dir as disk (WSL path)
        efivars_win = Path(disk_path).parent / "efivars.fd"
        wsl_efivars = _win_to_wsl(efivars_win)
        if ovmf_code and not efivars_win.exists():
            # Copy OVMF vars template via WSL
            for vars_candidate in [
                "/usr/share/OVMF/OVMF_VARS_4M.fd",
                "/usr/share/OVMF/OVMF_VARS.fd",
                "/usr/share/qemu/edk2-i386-vars.fd",
            ]:
                try:
                    self._wsl(f"test -f {vars_candidate} && cp {vars_candidate} '{wsl_efivars}'")
                    break
                except RuntimeError:
                    continue
            else:
                # Create empty vars file
                efivars_win.write_bytes(b"\x00" * (256 * 1024))

        # Build QEMU command (runs inside WSL)
        parts = [
            f"qemu-system-{self.arch}",
            f"-name {name}",
            "-machine q35,smm=off",
            f"-m {memory}",
            f"-smp {cpus}",
            "-cpu host",
            "-enable-kvm",
        ]

        if ovmf_code:
            parts += [
                f"-drive if=pflash,format=raw,readonly=on,file={ovmf_code}",
                f"-drive if=pflash,format=raw,file={wsl_efivars}",
            ]

        parts += [
            f"-drive file={wsl_disk},format={disk_fmt},if=virtio",
            f"-netdev user,id=net0,hostfwd=tcp:0.0.0.0:{hostfwd_port}-:8000",
            "-device virtio-net-pci,netdev=net0,mac=52:55:00:d1:55:01",
            f"-vnc :{vnc_display}",
            "-daemonize",
        ]

        for arg in self.extra_args:
            parts.append(arg)

        qemu_cmd = " ".join(parts)
        logger.info(f"Starting QEMU in WSL2: {qemu_cmd}")

        # Launch QEMU inside WSL2 (daemonize means it returns immediately)
        result = subprocess.run(
            ["wsl", "-e", "bash", "-c", qemu_cmd],
            capture_output=True,
            text=True,
            timeout=30,
        )
        if result.returncode != 0:
            raise RuntimeError(f"QEMU WSL2 launch failed: {result.stderr}")

        info = RuntimeInfo(
            host="localhost",
            api_port=hostfwd_port,
            vnc_port=5900 + vnc_display,
            name=name,
        )
        await self.is_ready(info)
        return info

    async def stop(self, name: str) -> None:
        try:
            subprocess.run(
                ["wsl", "-e", "bash", "-c", f"pkill -f 'qemu.*-name {name}'"],
                capture_output=True,
                timeout=10,
            )
        except Exception as e:
            logger.warning(f"Failed to stop QEMU VM {name} in WSL2: {e}")

        session_disk = getattr(self, "_session_disk", None)
        if session_disk and session_disk.exists() and "sessions" in str(session_disk):
            try:
                session_disk.unlink()
                logger.info(f"Removed session disk: {session_disk}")
            except OSError:
                pass

    async def is_ready(self, info: RuntimeInfo, timeout: float = 120) -> bool:
        url = f"http://{info.host}:{info.api_port}/status"
        deadline = asyncio.get_event_loop().time() + timeout
        async with httpx.AsyncClient(timeout=5) as client:
            while asyncio.get_event_loop().time() < deadline:
                try:
                    resp = await client.get(url)
                    if resp.status_code == 200:
                        logger.info(f"WSL2 QEMU VM {info.name} is ready")
                        return True
                except (
                    httpx.ConnectError,
                    httpx.ReadTimeout,
                    httpx.RemoteProtocolError,
                    httpx.ConnectTimeout,
                ):
                    pass
                await asyncio.sleep(2)
        raise TimeoutError(f"WSL2 QEMU VM {info.name} not ready after {timeout}s")


def QEMURuntime(mode: str = "docker", **kwargs) -> Runtime:
    """Factory that returns the appropriate QEMU runtime.

    Args:
        mode: "docker" (default), "bare-metal", or "wsl2"
    """
    if mode == "wsl2":
        return QEMUWSL2Runtime(**kwargs)
    if mode == "bare-metal":
        return QEMUBaremetalRuntime(**kwargs)
    return QEMUDockerRuntime(**kwargs)