Harden failure paths found in the reliability review

The success paths were fine; several failure paths drew a confident
conclusion without checking the thing they waited for.

Registry (was: any parse error killed the whole server, since load() runs
in the lifespan):
- quarantine an unreadable file and start empty instead of raising, so the
  tools that stop runaway VMs keep working when bookkeeping is damaged
- skip malformed or invalidly-named records rather than failing the load;
  report both through list_vms as registry_warnings
- read-modify-write under an exclusive flock so a second instance merges
  instead of clobbering, with a PID-unique temp file
- drop the lifespan shutdown save, which could resurrect deleted records
- version the schema and round-trip unknown record fields

sandbox_destroy (the only tool that deletes files):
- verify the process actually died, escalating to SIGKILL, and refuse to
  delete an overlay QEMU still holds open
- assert the target is inside the VM state tree before rmtree
- report cleanup errors instead of swallowing them; destroyed now reflects
  what happened

Launch races:
- reserve the name before the first await so two concurrent launches cannot
  race over one set of sockets
- refuse to unlink a QMP socket that is still accepting connections
- register the VM with a warning rather than orphaning it when the pidfile
  is unreadable but QEMU is up

Guest agent and QMP:
- bound every guest-agent call, not just the handshake; cap max_bytes and
  stop guest_file_read spinning on a zero-progress agent
- serialize QMP sessions per VM (the monitor is single-client) and say
  "another operation holds it" instead of "the VM has likely exited"
- poll liveness while waiting for SHUTDOWN so a crashed VM is reported as
  exited rather than as a guest ignoring ACPI
- default command timeout, with a longer bound for savevm/loadvm
- stricter snapshot tags; log destructive operations to stderr

Adds tests/test_reliability.py covering the conditions above.

src/mcqemu/errors.py

@@ -10,7 +10,23 @@ def vm_not_found(name: str, known: list[str]) -> ToolError:
     return ToolError(f"No VM named {name!r}. {hint} Use list_vms to see current state.")
 
 
-def qmp_unreachable(name: str, socket_path: str, detail: str, qemu_log: str | None) -> ToolError:
+def qmp_unreachable(
+    name: str,
+    socket_path: str,
+    detail: str,
+    qemu_log: str | None,
+    process_alive: bool = False,
+) -> ToolError:
+    """Connect failure. The advice differs sharply depending on whether the
+    process is still there, and guessing 'it exited' at a VM that is merely
+    busy sends the caller off to relaunch a VM that is still running."""
+    if process_alive:
+        return ToolError(
+            f"Could not open a QMP session to VM {name!r} at {socket_path} ({detail}), "
+            "but its process is still alive. QEMU's monitor accepts one client at a "
+            "time, so another tool call or an external qmp-shell may be holding it. "
+            "Do NOT relaunch the VM; retry in a moment."
+        )
     log_hint = f" Check the QEMU log at {qemu_log} for crash details." if qemu_log else ""
     return ToolError(
         f"Cannot reach QMP socket for VM {name!r} at {socket_path} ({detail}). "

src/mcqemu/launcher.py

@@ -50,11 +50,33 @@ def pick_accel(arch: str) -> str:
 
 
 def pick_free_port() -> int:
+    # The bind test closes before QEMU binds, so a racing process could still
+    # take the port; QEMU then fails to launch with a clear error. Accepted
+    # deliberately — holding the port open until exec is not possible here.
     with socket.socket() as s:
         s.bind(("", 0))
         return s.getsockname()[1]
 
 
+def socket_is_live(path: Path) -> bool:
+    """True if something is still accepting connections on this unix socket.
+
+    A leftover socket *file* is normal after a crash; a socket that still
+    accepts means a process owns this VM's name and must not be trampled.
+    """
+    if not path.exists():
+        return False
+    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        probe.settimeout(0.2)
+        probe.connect(str(path))
+        return True
+    except OSError:
+        return False
+    finally:
+        probe.close()
+
+
 def resolve_port_forwards(forwards: list[str]) -> list[str]:
     """Turn user forward specs into concrete 'HOSTPORT:GUESTPORT' entries.
 
@@ -196,7 +218,16 @@ async def spawn_vm(cfg: VMConfig, config: Config) -> VMRecord:
     config.vm_runtime_dir(cfg.name).mkdir(parents=True, exist_ok=True)
     config.vm_state_dir(cfg.name).mkdir(parents=True, exist_ok=True)
 
-    # Leftover sockets from a previous life of this name confuse liveness checks.
+    # Leftover sockets from a previous life of this name confuse liveness
+    # checks — but only delete them once we know nothing is listening. An
+    # unregistered-but-running QEMU (e.g. after forget_vm force=True) would
+    # otherwise lose its monitor socket and become unreachable forever.
+    if socket_is_live(paths.qmp_socket):
+        raise ToolError(
+            f"{paths.qmp_socket} is a LIVE QMP socket: an unregistered QEMU process is "
+            f"still using the name {cfg.name!r}. Re-register it with attach_vm and stop "
+            "it, or launch under a different name."
+        )
     for stale in (paths.qmp_socket, paths.qga_socket, paths.pidfile):
         stale.unlink(missing_ok=True)
 
@@ -228,13 +259,23 @@ async def spawn_vm(cfg: VMConfig, config: Config) -> VMRecord:
                 detail = f"exit code {proc.returncode}"
         raise ToolError(f"QEMU failed to launch VM {cfg.name!r}: {detail}")
 
-    try:
-        pid = int(paths.pidfile.read_text().strip())
-    except (OSError, ValueError) as e:
+    # QEMU exited 0, so the VM is up. If the pidfile is somehow unreadable we
+    # must still return a record: raising here would leave a running,
+    # unregistered VM that no tool could find or stop.
+    pid: int | None = None
+    for attempt in range(5):
+        try:
+            pid = int(paths.pidfile.read_text().strip())
+            break
+        except (OSError, ValueError):
+            if attempt < 4:
+                await asyncio.sleep(0.1)
+    if pid is None and not socket_is_live(paths.qmp_socket):
         raise ToolError(
-            f"QEMU launched but pidfile {paths.pidfile} is unreadable ({e}) — "
-            "the VM state is unknown; check the QEMU log."
-        ) from e
+            f"QEMU reported success for VM {cfg.name!r} but left neither a readable "
+            f"pidfile ({paths.pidfile}) nor a live QMP socket. Check {paths.qemu_log}; "
+            f'if a process is running, find it with: pgrep -af "qemu-system.*-name {cfg.name}"'
+        )
 
     accel = pick_accel(cfg.arch)
     return VMRecord(

src/mcqemu/models.py

@@ -41,11 +41,19 @@ class VMRecord:
     qemu_log: str | None = None
     serial_log: str | None = None
     created_at: str | None = None
+    # Fields written by a newer mcqemu that this build doesn't know about.
+    # Carried through unchanged so an older instance can't strip them on save.
+    extra: dict[str, Any] = field(default_factory=dict)
 
     def to_dict(self) -> dict[str, Any]:
-        return asdict(self)
+        data = asdict(self)
+        data.pop("extra", None)
+        data.update(self.extra)
+        return data
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> VMRecord:
-        known = {f for f in cls.__dataclass_fields__}
-        return cls(**{k: v for k, v in data.items() if k in known})
+        known = set(cls.__dataclass_fields__) - {"extra"}
+        fields = {k: v for k, v in data.items() if k in known}
+        extra = {k: v for k, v in data.items() if k not in known}
+        return cls(**fields, extra=extra)

src/mcqemu/qmp.py

@@ -4,65 +4,144 @@ QEMU's QMP unix socket accepts one client at a time, so every tool call
 opens a short-lived session and disconnects — the socket stays free for
 qmp-shell or any other tooling between calls, and there is no stale
 connection state to reconcile after either side restarts.
+
+Because the socket is single-client, concurrent tool calls against the same
+VM must not both try to hold it: the second would block until its connect
+timeout and then be told the VM had exited. `_session_locks` serializes them
+per VM name instead (a mutex table keyed by resource, not shared state).
 """
 
 from __future__ import annotations
 
 import asyncio
 import contextlib
+from collections import defaultdict
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
+from pathlib import Path
 from typing import Any
 
+from fastmcp.exceptions import ToolError
 from qemu.qmp import ExecuteError, QMPClient
 
 from .errors import qmp_command_failed, qmp_unreachable
 from .models import VMRecord
+from .registry import pid_alive
 
 CONNECT_TIMEOUT = 5.0
+COMMAND_TIMEOUT = 30.0
+
+_session_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
+
+
+def _looks_alive(record: VMRecord) -> bool:
+    """Best-effort liveness without the registry, for error messages only."""
+    if record.pid is not None:
+        return pid_alive(record.pid)
+    with contextlib.suppress(OSError):
+        return Path(record.qmp_socket).exists()
+    return False
 
 
 @asynccontextmanager
 async def qmp_session(record: VMRecord) -> AsyncIterator[QMPClient]:
-    client = QMPClient(record.name)
-    try:
+    async with _session_locks[record.name]:
+        client = QMPClient(record.name)
         try:
-            await asyncio.wait_for(client.connect(record.qmp_socket), timeout=CONNECT_TIMEOUT)
-        except TimeoutError:
-            raise qmp_unreachable(
-                record.name, record.qmp_socket, "connect timed out", record.qemu_log
-            ) from None
-        except OSError as e:
-            raise qmp_unreachable(record.name, record.qmp_socket, str(e), record.qemu_log) from e
-        except Exception as e:  # qemu.qmp wraps failures in its own ConnectError
-            raise qmp_unreachable(record.name, record.qmp_socket, str(e), record.qemu_log) from e
-        yield client
-    finally:
-        # After a 'quit' the peer hangs up first; EOF on teardown is normal.
-        with contextlib.suppress(Exception):
-            await client.disconnect()
+            try:
+                await asyncio.wait_for(client.connect(record.qmp_socket), timeout=CONNECT_TIMEOUT)
+            except TimeoutError:
+                raise qmp_unreachable(
+                    record.name,
+                    record.qmp_socket,
+                    "connect timed out",
+                    record.qemu_log,
+                    process_alive=_looks_alive(record),
+                ) from None
+            except OSError as e:
+                raise qmp_unreachable(
+                    record.name, record.qmp_socket, str(e), record.qemu_log
+                ) from e
+            except Exception as e:  # qemu.qmp wraps failures in its own ConnectError
+                raise qmp_unreachable(
+                    record.name,
+                    record.qmp_socket,
+                    str(e),
+                    record.qemu_log,
+                    process_alive=_looks_alive(record),
+                ) from e
+            yield client
+        finally:
+            # After a 'quit' the peer hangs up first; EOF on teardown is normal.
+            with contextlib.suppress(Exception):
+                await client.disconnect()
 
 
-async def execute(client: QMPClient, command: str, arguments: dict[str, Any] | None = None) -> Any:
-    """Run one QMP command, translating QMP-level errors to ToolError."""
+async def execute(
+    client: QMPClient,
+    command: str,
+    arguments: dict[str, Any] | None = None,
+    timeout: float | None = COMMAND_TIMEOUT,
+) -> Any:
+    """Run one QMP command, translating QMP-level errors to ToolError.
+
+    Commands are bounded by default: a QEMU wedged inside a device-model
+    operation would otherwise hang the tool call (and this VM's session lock)
+    forever. Pass timeout=None for genuinely long operations like savevm.
+    """
     try:
-        return await client.execute(command, arguments or {})
+        call = client.execute(command, arguments or {})
+        return await (asyncio.wait_for(call, timeout) if timeout else call)
     except ExecuteError as e:
         error_class = getattr(e, "error_class", None) or "GenericError"
         raise qmp_command_failed(command, error_class, str(e)) from e
-
-
-async def wait_for_event(client: QMPClient, event_name: str, timeout: float) -> bool:
-    """Drain the session's event stream until event_name arrives or time runs out."""
-
-    async def _drain() -> None:
-        while True:
-            event = await client.events.get()
-            if event.get("event") == event_name:
-                return
-
-    try:
-        await asyncio.wait_for(_drain(), timeout=timeout)
-        return True
     except TimeoutError:
-        return False
+        raise ToolError(
+            f"QMP command {command!r} did not complete within {timeout}s. The VM may "
+            "be wedged or under heavy load; check list_vms and the QEMU log."
+        ) from None
+    except Exception as e:
+        # 'quit' legitimately races the connection teardown: QEMU can drop the
+        # socket before acking. Anything else is a real transport failure.
+        if command == "quit" and type(e).__name__ in (
+            "ExecInterruptedError",
+            "StateError",
+            "EOFError",
+        ):
+            return {}
+        raise
+
+
+async def wait_for_event(
+    client: QMPClient,
+    event_name: str,
+    timeout: float,
+    record: VMRecord | None = None,
+) -> str:
+    """Wait for `event_name`, returning why the wait ended.
+
+    Returns "event", "process_exited", or "timeout". The liveness poll matters:
+    qemu.qmp's event queue never wakes on a dropped connection, so a VM that
+    crashes mid-wait would otherwise burn the caller's entire timeout and then
+    be misreported as a guest that ignores ACPI.
+    """
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        try:
+            event = await asyncio.wait_for(client.events.get(), timeout=1.0)
+        except TimeoutError:
+            if record is not None and record.pid is not None and not pid_alive(record.pid):
+                return "process_exited"
+            continue
+        if event.get("event") == event_name:
+            return "event"
+    return "timeout"
+
+
+def recent_events(client: QMPClient) -> list[str]:
+    """Event names seen this session — context for a timeout error message."""
+    try:
+        return [e.get("event") for e in client.events.history if e.get("event")]
+    except Exception:
+        return []

src/mcqemu/registry.py

@@ -1,15 +1,29 @@
-"""Persistent VM registry with PID-liveness staleness detection."""
+"""Persistent VM registry with PID-liveness staleness detection.
+
+Two properties matter more than speed here. First, a damaged registry file
+must never take the server down: VMs outlive this process, so the tools that
+stop them have to keep working even when their bookkeeping is unreadable.
+Second, several mcqemu instances may share one registry file (a second MCP
+client, a stray `uvx mcqemu`), so every write is a read-modify-write under an
+exclusive file lock — otherwise the last writer silently orphans the other's
+VMs.
+"""
 
 from __future__ import annotations
 
 import contextlib
+import fcntl
 import json
 import os
+import time
+from collections.abc import Callable, Iterator
 from pathlib import Path
 
-from .config import Config
+from .config import Config, valid_vm_name
 from .models import VMRecord
 
+SCHEMA_VERSION = 1
+
 
 def pid_alive(pid: int | None) -> bool:
     """True iff the PID exists AND is actually a qemu-system process.
@@ -31,32 +45,132 @@ class VMRegistry:
     def __init__(self, config: Config) -> None:
         self._config = config
         self._vms: dict[str, VMRecord] = {}
+        self._reserved: set[str] = set()
+        self.degraded: str | None = None
+        self.skipped: list[dict] = []
 
-    def load(self) -> None:
-        path = self._config.registry_path
-        if not path.exists():
-            self._vms = {}
-            return
-        data = json.loads(path.read_text())
-        self._vms = {name: VMRecord.from_dict(rec) for name, rec in data.items()}
+    # --- file primitives -------------------------------------------------
 
-    def save(self) -> None:
+    @contextlib.contextmanager
+    def _locked(self) -> Iterator[None]:
+        """Exclusive lock over the registry for one read-modify-write cycle.
+
+        Blocking, but only ever held for a small local JSON read plus rename.
+        """
         path = self._config.registry_path
         path.parent.mkdir(parents=True, exist_ok=True)
-        tmp = path.with_suffix(".tmp")
-        tmp.write_text(json.dumps({n: r.to_dict() for n, r in self._vms.items()}, indent=2))
+        fd = os.open(str(path) + ".lock", os.O_CREAT | os.O_RDWR, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX)
+            yield
+        finally:
+            os.close(fd)  # closing releases the flock
+
+    def _read_disk(self) -> dict[str, VMRecord]:
+        """Parse the registry, degrading instead of raising.
+
+        An unreadable file is quarantined and we continue with an empty
+        registry; individual malformed records are skipped. Both outcomes are
+        reported through `degraded` / `skipped` so list_vms can surface them.
+        """
+        path = self._config.registry_path
+        if not path.exists():
+            return {}
+        try:
+            raw = json.loads(path.read_text())
+        except (OSError, ValueError) as e:
+            quarantine = path.with_name(f"vms.corrupt.{int(time.time())}.json")
+            with contextlib.suppress(OSError):
+                os.replace(path, quarantine)
+            self.degraded = (
+                f"Registry file was unreadable ({e}); it has been quarantined at "
+                f"{quarantine} and an empty registry was started. Any running VMs "
+                "are unaffected but no longer tracked — re-register them with "
+                "attach_vm (their QMP sockets are under the runtime directory)."
+            )
+            return {}
+
+        if isinstance(raw, dict) and "vms" in raw and "version" in raw:
+            version = raw.get("version")
+            if not isinstance(version, int) or version > SCHEMA_VERSION:
+                self.degraded = (
+                    f"Registry schema version {version!r} is newer than this build "
+                    f"understands (max {SCHEMA_VERSION}); refusing to load it so a "
+                    "newer mcqemu's records are not damaged. Upgrade mcqemu."
+                )
+                return {}
+            entries = raw.get("vms") or {}
+        else:
+            entries = raw  # legacy layout: a bare {name: record} mapping
+
+        if not isinstance(entries, dict):
+            self.degraded = "Registry contents were not an object; starting empty."
+            return {}
+
+        vms: dict[str, VMRecord] = {}
+        for key, rec in entries.items():
+            # Keys become filesystem paths and are what destructive tools look
+            # up by, so an invalid one is dropped rather than trusted.
+            if not isinstance(key, str) or not valid_vm_name(key):
+                self.skipped.append({"key": str(key)[:64], "reason": "invalid VM name"})
+                continue
+            if not isinstance(rec, dict):
+                self.skipped.append({"key": key, "reason": "record was not an object"})
+                continue
+            try:
+                record = VMRecord.from_dict(rec)
+            except TypeError as e:
+                self.skipped.append({"key": key, "reason": f"malformed record: {e}"})
+                continue
+            record.name = key  # the key is authoritative; keep the record consistent
+            vms[key] = record
+        return vms
+
+    def _write_disk(self, vms: dict[str, VMRecord]) -> None:
+        path = self._config.registry_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            "version": SCHEMA_VERSION,
+            "vms": {name: record.to_dict() for name, record in vms.items()},
+        }
+        # PID-unique temp name: a shared one can interleave between instances
+        # and os.replace would then publish the mixture.
+        tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}")
+        tmp.write_text(json.dumps(payload, indent=2))
         os.replace(tmp, path)
 
+    def _mutate(self, apply: Callable[[dict[str, VMRecord]], None]) -> None:
+        """Re-read, apply, write — all under the lock, so a concurrent
+        instance's records are merged rather than clobbered."""
+        with self._locked():
+            vms = self._read_disk()
+            apply(vms)
+            self._write_disk(vms)
+            self._vms = vms
+
+    # --- public API ------------------------------------------------------
+
+    def load(self) -> None:
+        with self._locked():
+            self._vms = self._read_disk()
+
+    def refresh(self) -> None:
+        """Re-read the registry so records written by another instance (or by
+        a tool call that raced this one) are visible."""
+        self.load()
+
     def get(self, name: str) -> VMRecord | None:
         return self._vms.get(name)
 
     def add(self, record: VMRecord) -> None:
-        self._vms[record.name] = record
-        self.save()
+        self._mutate(lambda vms: vms.__setitem__(record.name, record))
+
+    def update(self, record: VMRecord) -> None:
+        """Persist in-place changes to an existing record."""
+        self.add(record)
 
     def remove(self, name: str) -> None:
-        self._vms.pop(name, None)
-        self.save()
+        self._mutate(lambda vms: vms.pop(name, None))
 
     def all(self) -> list[VMRecord]:
         return list(self._vms.values())
@@ -64,6 +178,26 @@ class VMRegistry:
     def names(self) -> list[str]:
         return list(self._vms)
 
+    # --- in-process name reservations ------------------------------------
+    # launch_vm awaits (disk probing, spawning) between checking a name and
+    # registering it. Without a reservation two concurrent calls both pass the
+    # duplicate check and race over the same sockets and pidfile.
+
+    def reserve(self, name: str) -> bool:
+        """Claim a name for an in-flight launch. False if already claimed."""
+        if name in self._reserved:
+            return False
+        self._reserved.add(name)
+        return True
+
+    def release(self, name: str) -> None:
+        self._reserved.discard(name)
+
+    def reserved(self) -> set[str]:
+        return set(self._reserved)
+
+    # --- liveness --------------------------------------------------------
+
     def refresh_pid(self, record: VMRecord) -> int | None:
         """Re-read the pidfile for a spawned VM; returns the PID or None."""
         if record.pidfile:
@@ -80,7 +214,12 @@ class VMRegistry:
         return pid_alive(record.pid)
 
     def disks_in_use(self) -> dict[str, str]:
-        """Map of absolute disk path -> VM name, for every live registered VM."""
+        """Map of absolute disk path -> VM name, for every live registered VM.
+
+        Advisory only: attached VMs report no disks, and backing files are not
+        walked. QEMU's own image locks are the authoritative guard (qemu_img
+        translates them into an actionable error).
+        """
         used: dict[str, str] = {}
         for rec in self._vms.values():
             if not self.is_process_alive(rec):

src/mcqemu/server.py

@@ -1,5 +1,7 @@
 """mcqemu server — composition root and entry point."""
 
+import logging
+import os
 import sys
 from contextlib import asynccontextmanager
 
@@ -11,6 +13,8 @@ from .registry import VMRegistry
 from .state import AppContext
 from .tools import register_all
 
+log = logging.getLogger(__name__)
+
 INSTRUCTIONS = """\
 mcqemu manages QEMU virtual machines on this host.
 
@@ -26,7 +30,10 @@ Typical flows:
   screenshot -> act -> screenshot; guests need time to react, so re-check
   rather than assuming.
 - Inspect: list_vms / vm_info. VMs survive MCP server restarts (they are
-  daemonized QEMU processes).
+  daemonized QEMU processes). list_vms returns {"vms": [...],
+  "registry_warnings": [...]}; a non-empty registry_warnings means bookkeeping
+  was damaged and some VMs may be running untracked — surface it to the user
+  rather than treating the list as complete.
 - Inside the guest: guest_exec / guest_file_read / guest_file_write — these
   need qemu-guest-agent installed in the guest OS (guest_ping to check).
 - Existing QEMU processes started outside this server can be managed after
@@ -47,8 +54,14 @@ async def lifespan(mcp: FastMCP):
     config.runtime_dir.mkdir(parents=True, exist_ok=True)
     registry = VMRegistry(config)
     registry.load()
+    if registry.degraded:
+        log.warning("registry degraded: %s", registry.degraded)
+    for entry in registry.skipped:
+        log.warning("skipped registry entry %r: %s", entry["key"], entry["reason"])
     yield AppContext(config=config, registry=registry)
-    registry.save()  # VMs keep running by design — they are daemonized.
+    # No save on shutdown: every mutation already persisted itself, and writing
+    # this instance's in-memory view here would clobber records another
+    # instance created while we were running. VMs keep running by design.
 
 
 mcp = FastMCP("mcqemu", lifespan=lifespan, instructions=INSTRUCTIONS)
@@ -58,6 +71,13 @@ prompts.register(mcp)
 
 
 def main() -> None:
+    # stderr only: stdout is the JSON-RPC transport. Destructive operations log
+    # here so there is a record when something goes wrong unattended.
+    logging.basicConfig(
+        stream=sys.stderr,
+        level=os.environ.get("MCQEMU_LOG_LEVEL", "INFO").upper(),
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
     print(f"mcqemu v{__version__}", file=sys.stderr)
     mcp.run()
 

src/mcqemu/tools/guest.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import base64
+import contextlib
 
 from fastmcp import Context, FastMCP
 from fastmcp.exceptions import ToolError
@@ -10,6 +11,39 @@ from ..qga import qga_session
 from ._common import require_vm
 
 _READ_CHUNK = 48 * 1024  # QGA caps single reads at 48 MiB of b64; stay modest.
+MAX_READ_BYTES = 8 * 1024 * 1024
+CALL_TIMEOUT = 10.0
+
+# Guest agents on RHEL-family distros ship with these RPCs disabled by default.
+_BLOCKED_HINT = (
+    "The guest agent refused the command ({desc}). On RHEL/CentOS/Fedora guests "
+    "these RPCs are disabled by default — remove them from BLOCK_RPCS in "
+    "/etc/sysconfig/qemu-ga and restart qemu-guest-agent."
+)
+
+
+async def _qga(client, name: str, command: str, args: dict | None = None, timeout=CALL_TIMEOUT):
+    """One guest-agent call, bounded and with failures translated.
+
+    The timeout is the whole point: qga_session only bounds the handshake, so
+    a guest that freezes (panic, OOM, or another tool pausing its vCPUs) after
+    the sync would otherwise hang this call forever.
+    """
+    try:
+        return await asyncio.wait_for(client.execute(command, args or {}), timeout)
+    except TimeoutError:
+        raise ToolError(
+            f"The guest agent in VM {name!r} stopped responding to {command!r} after "
+            f"{timeout}s. The guest may be frozen or paused — check vm_screenshot "
+            "and list_vms (a paused VM cannot run its agent)."
+        ) from None
+    except ToolError:
+        raise
+    except Exception as e:
+        desc = str(e) or type(e).__name__
+        if "CommandNotFound" in type(e).__name__ or "not found" in desc.lower():
+            raise ToolError(_BLOCKED_HINT.format(desc=desc)) from e
+        raise ToolError(f"Guest agent command {command!r} failed: {desc}") from e
 
 
 async def guest_ping(name: str, ctx: Context = None) -> dict:
@@ -18,7 +52,7 @@ async def guest_ping(name: str, ctx: Context = None) -> dict:
     or running — the other guest_* tools won't work until it does."""
     _, record = require_vm(ctx, name)
     async with qga_session(record) as client:
-        await asyncio.wait_for(client.execute("guest-ping"), timeout=3.0)
+        await _qga(client, name, "guest-ping", timeout=3.0)
     return {"name": name, "guest_agent": "responding"}
 
 
@@ -27,11 +61,10 @@ async def guest_info(name: str, ctx: Context = None) -> dict:
     agent's version and supported commands."""
     _, record = require_vm(ctx, name)
     async with qga_session(record) as client:
-        agent = await client.execute("guest-info")
-        try:
-            osinfo = await client.execute("guest-get-osinfo")
-        except Exception:
-            osinfo = None
+        agent = await _qga(client, name, "guest-info")
+        osinfo = None
+        with contextlib.suppress(ToolError):
+            osinfo = await _qga(client, name, "guest-get-osinfo")
     supported = sorted(c["name"] for c in agent.get("supported_commands", []) if c.get("enabled"))
     return {
         "name": name,
@@ -55,27 +88,46 @@ async def guest_exec(
     command="/bin/sh", args=["-c", "your | pipeline"]). Requires
     qemu-guest-agent in the guest."""
     _, record = require_vm(ctx, name)
+    if timeout < 1 or timeout > 3600:
+        raise ToolError("timeout must be between 1 and 3600 seconds.")
     exec_args: dict = {"path": command, "capture-output": True}
     if args:
         exec_args["arg"] = args
     if stdin is not None:
         exec_args["input-data"] = base64.b64encode(stdin.encode()).decode()
 
-    async with qga_session(record) as client:
-        reply = await client.execute("guest-exec", exec_args)
-        pid = reply["pid"]
-        deadline = asyncio.get_event_loop().time() + timeout
+    async def _run(client):
+        reply = await _qga(client, name, "guest-exec", exec_args)
+        try:
+            pid = reply["pid"]
+        except (KeyError, TypeError) as e:
+            raise ToolError(
+                f"Guest agent returned an unexpected guest-exec reply: {reply!r}"
+            ) from e
+        loop = asyncio.get_event_loop()
+        deadline = loop.time() + timeout
         while True:
-            status = await client.execute("guest-exec-status", {"pid": pid})
+            status = await _qga(client, name, "guest-exec-status", {"pid": pid})
             if status.get("exited"):
-                break
-            if asyncio.get_event_loop().time() > deadline:
+                return status
+            if loop.time() > deadline:
                 raise ToolError(
                     f"Command still running in guest after {timeout}s (guest pid {pid}). "
                     "It continues in the background; raise timeout for long commands."
                 )
             await asyncio.sleep(0.2)
 
+    async with qga_session(record) as client:
+        # Outer bound as well: the poll loop checks the clock between awaits,
+        # so without this a single hung call could outlive the deadline.
+        try:
+            status = await asyncio.wait_for(_run(client), timeout + CALL_TIMEOUT + 5)
+        except TimeoutError:
+            raise ToolError(
+                f"guest_exec on VM {name!r} exceeded its overall deadline; the guest "
+                "agent may have stopped responding mid-command."
+            ) from None
+
     def _decode(field: str) -> str:
         data = status.get(field)
         return base64.b64decode(data).decode(errors="replace") if data else ""
@@ -92,28 +144,42 @@ async def guest_exec(
 async def guest_file_read(
     name: str, path: str, max_bytes: int = 1_048_576, ctx: Context = None
 ) -> dict:
-    """Read a text file from inside the guest (up to max_bytes, default 1 MiB).
-    Requires qemu-guest-agent in the guest."""
+    """Read a text file from inside the guest (up to max_bytes, default 1 MiB,
+    8 MiB ceiling). Requires qemu-guest-agent in the guest."""
     _, record = require_vm(ctx, name)
+    if max_bytes < 1 or max_bytes > MAX_READ_BYTES:
+        raise ToolError(f"max_bytes must be between 1 and {MAX_READ_BYTES}.")
     async with qga_session(record) as client:
-        handle = await client.execute("guest-file-open", {"path": path, "mode": "r"})
+        handle = await _qga(client, name, "guest-file-open", {"path": path, "mode": "r"})
+        chunks: list[bytes] = []
+        total = 0
+        truncated = False
+        empty_reads = 0
         try:
-            chunks: list[bytes] = []
-            total = 0
-            truncated = False
             while total < max_bytes:
                 count = min(_READ_CHUNK, max_bytes - total)
-                reply = await client.execute("guest-file-read", {"handle": handle, "count": count})
-                if reply.get("count", 0) > 0:
+                reply = await _qga(
+                    client, name, "guest-file-read", {"handle": handle, "count": count}
+                )
+                got = reply.get("count", 0)
+                if got > 0:
                     chunk = base64.b64decode(reply["buf-b64"])
                     chunks.append(chunk)
                     total += len(chunk)
+                    empty_reads = 0
+                else:
+                    # FIFOs, ttys and some /proc files return 0 without EOF
+                    # forever; give up rather than spin on the agent.
+                    empty_reads += 1
+                    if empty_reads >= 3 and not reply.get("eof"):
+                        break
                 if reply.get("eof"):
                     break
             else:
                 truncated = True
         finally:
-            await client.execute("guest-file-close", {"handle": handle})
+            with contextlib.suppress(Exception):
+                await _qga(client, name, "guest-file-close", {"handle": handle})
     return {
         "path": path,
         "content": b"".join(chunks).decode(errors="replace"),
@@ -130,13 +196,16 @@ async def guest_file_write(
     _, record = require_vm(ctx, name)
     payload = base64.b64encode(content.encode()).decode()
     async with qga_session(record) as client:
-        handle = await client.execute(
-            "guest-file-open", {"path": path, "mode": "a" if append else "w"}
+        handle = await _qga(
+            client, name, "guest-file-open", {"path": path, "mode": "a" if append else "w"}
         )
         try:
-            reply = await client.execute("guest-file-write", {"handle": handle, "buf-b64": payload})
+            reply = await _qga(
+                client, name, "guest-file-write", {"handle": handle, "buf-b64": payload}
+            )
         finally:
-            await client.execute("guest-file-close", {"handle": handle})
+            with contextlib.suppress(Exception):
+                await _qga(client, name, "guest-file-close", {"handle": handle})
     return {"path": path, "bytes_written": reply.get("count", 0), "appended": append}
 
 

src/mcqemu/tools/lifecycle.py

@@ -10,7 +10,7 @@ from fastmcp.exceptions import ToolError
 from ..config import resolve_image_path, valid_vm_name
 from ..launcher import resolve_port_forwards, spawn_vm
 from ..models import VMConfig
-from ..qmp import execute, qmp_session, wait_for_event
+from ..qmp import execute, qmp_session, recent_events, wait_for_event
 from ..registry import pid_alive
 from ._common import app, require_vm
 
@@ -46,6 +46,9 @@ async def launch_vm(
             f"Invalid VM name {name!r}: use letters, digits, '.', '_', '-' (max 48 chars)."
         )
 
+    # Pick up records written by another instance (or another in-flight call)
+    # before deciding this name is free.
+    state.registry.refresh()
     existing = state.registry.get(name)
     if existing is not None:
         if state.registry.is_process_alive(existing):
@@ -54,38 +57,46 @@ async def launch_vm(
             )
         state.registry.remove(name)  # dead leftover — reuse the name
 
-    resolved_disks: list[str] = []
-    in_use = state.registry.disks_in_use()
-    for disk in disks or []:
-        try:
-            p = resolve_image_path(disk)
-        except ValueError as e:
-            raise ToolError(f"Bad disk path: {e}. Create images with image_create.") from e
-        if str(p) in in_use:
-            raise ToolError(f"Disk {p} is already attached to running VM {in_use[str(p)]!r}.")
-        resolved_disks.append(str(p))
+    # Claim the name before the first await. Everything below suspends, and
+    # two concurrent launches that both passed the check above would otherwise
+    # race over the same sockets and pidfile.
+    if not state.registry.reserve(name):
+        raise ToolError(f"A launch of VM {name!r} is already in progress.")
+    try:
+        resolved_disks: list[str] = []
+        in_use = state.registry.disks_in_use()
+        for disk in disks or []:
+            try:
+                p = resolve_image_path(disk)
+            except ValueError as e:
+                raise ToolError(f"Bad disk path: {e}. Create images with image_create.") from e
+            if str(p) in in_use:
+                raise ToolError(f"Disk {p} is already attached to running VM {in_use[str(p)]!r}.")
+            resolved_disks.append(str(p))
 
-    resolved_iso: str | None = None
-    if iso:
-        try:
-            resolved_iso = str(resolve_image_path(iso))
-        except ValueError as e:
-            raise ToolError(f"Bad ISO path: {e}") from e
+        resolved_iso: str | None = None
+        if iso:
+            try:
+                resolved_iso = str(resolve_image_path(iso))
+            except ValueError as e:
+                raise ToolError(f"Bad ISO path: {e}") from e
 
-    cfg = VMConfig(
-        name=name,
-        arch=arch,
-        machine=machine,
-        memory_mb=memory_mb,
-        cpus=cpus,
-        disks=resolved_disks,
-        iso=resolved_iso,
-        firmware=firmware,
-        port_forwards=resolve_port_forwards(port_forwards or []),
-        no_net=no_net,
-        extra_args=extra_args or [],
-    )
-    record = await spawn_vm(cfg, state.config)
+        cfg = VMConfig(
+            name=name,
+            arch=arch,
+            machine=machine,
+            memory_mb=memory_mb,
+            cpus=cpus,
+            disks=resolved_disks,
+            iso=resolved_iso,
+            firmware=firmware,
+            port_forwards=resolve_port_forwards(port_forwards or []),
+            no_net=no_net,
+            extra_args=extra_args or [],
+        )
+        record = await spawn_vm(cfg, state.config)
+    finally:
+        state.registry.release(name)
     state.registry.add(record)
     return {
         "name": name,
@@ -122,12 +133,21 @@ async def stop_vm(name: str, force: bool = False, timeout: int = 30, ctx: Contex
 
     async with qmp_session(record) as client:
         await execute(client, "system_powerdown")
-        if not await wait_for_event(client, "SHUTDOWN", timeout=timeout):
+        outcome = await wait_for_event(client, "SHUTDOWN", timeout=timeout, record=record)
+        if outcome == "timeout":
+            seen = recent_events(client)
+            context = f" Events seen meanwhile: {', '.join(seen)}." if seen else ""
             raise ToolError(
                 f"VM {name!r} did not shut down within {timeout}s after the ACPI "
                 "power signal — the guest may have no OS or ignores ACPI. "
-                "Retry with force=True to terminate it immediately."
+                f"Retry with force=True to terminate it immediately.{context}"
             )
+        if outcome == "process_exited":
+            return {
+                "name": name,
+                "status": "stopped",
+                "method": "process exited during shutdown",
+            }
     deadline = asyncio.get_event_loop().time() + 10
     while pid_alive(record.pid) and asyncio.get_event_loop().time() < deadline:
         await asyncio.sleep(0.1)

src/mcqemu/tools/query.py

@@ -18,15 +18,22 @@ async def _live_status(state, record) -> str:
         return "unreachable"
 
 
-async def list_vms(ctx: Context = None) -> list[dict]:
+async def list_vms(ctx: Context = None) -> dict:
     """List every registered VM with its live status (running, paused,
     shutdown, stopped, or unreachable). Includes both VMs launched by this
-    server and externally attached ones."""
+    server and externally attached ones.
+
+    Returns {"vms": [...], "registry_warnings": [...]}. A non-empty
+    registry_warnings means bookkeeping was damaged and some VMs may be
+    running but untracked — report it rather than assuming the list is
+    complete."""
     state = app(ctx)
-    out = []
+    # Pick up anything another mcqemu instance registered since we loaded.
+    state.registry.refresh()
+    vms = []
     for record in state.registry.all():
         status = await _live_status(state, record)
-        out.append(
+        vms.append(
             {
                 "name": record.name,
                 "status": status,
@@ -37,7 +44,13 @@ async def list_vms(ctx: Context = None) -> list[dict]:
                 "created_at": record.created_at,
             }
         )
-    return out
+    warnings = []
+    if state.registry.degraded:
+        warnings.append(state.registry.degraded)
+    warnings += [
+        f"skipped registry entry {e['key']!r}: {e['reason']}" for e in state.registry.skipped
+    ]
+    return {"vms": vms, "registry_warnings": warnings}
 
 
 async def vm_info(name: str, ctx: Context = None) -> dict:

src/mcqemu/tools/sandbox.py

@@ -2,7 +2,11 @@
 the matching destroy that cleans up everything the sandbox created."""
 
 import asyncio
+import contextlib
+import logging
+import os
 import shutil
+import signal
 from pathlib import Path
 
 from fastmcp import Context, FastMCP
@@ -16,6 +20,8 @@ from ..registry import pid_alive
 from ._common import app, require_vm
 from .lifecycle import launch_vm
 
+log = logging.getLogger(__name__)
+
 
 async def _agent_responding(record) -> bool:
     try:
@@ -25,6 +31,26 @@ async def _agent_responding(record) -> bool:
         return False
 
 
+async def _await_exit(pid: int | None, seconds: float) -> bool:
+    """Poll until the PID is gone. Returns True if it exited in time."""
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + seconds
+    while loop.time() < deadline:
+        if not pid_alive(pid):
+            return True
+        await asyncio.sleep(0.1)
+    return not pid_alive(pid)
+
+
+def _tail_log(path: str | None, lines: int = 3) -> str:
+    if not path:
+        return "(no log)"
+    try:
+        return " | ".join(Path(path).read_text(errors="replace").splitlines()[-lines:])
+    except OSError:
+        return "(log unreadable)"
+
+
 def _auto_name(registry) -> str:
     """First 'sandbox[-N]' name that is unregistered or belongs to a dead VM
     (launch_vm reclaims dead names, so reusing them is safe and keeps names
@@ -80,19 +106,26 @@ async def sandbox_vm(
     overlay.unlink(missing_ok=True)
     await run_qemu_img("create", "-f", "qcow2", "-b", str(base), "-F", base_fmt, str(overlay))
 
-    result = await launch_vm(
-        name=name,
-        disks=[str(overlay)],
-        memory_mb=memory_mb,
-        cpus=cpus,
-        port_forwards=["auto:22"] if port_forwards is None else port_forwards,
-        ctx=ctx,
-    )
+    try:
+        result = await launch_vm(
+            name=name,
+            disks=[str(overlay)],
+            memory_mb=memory_mb,
+            cpus=cpus,
+            port_forwards=["auto:22"] if port_forwards is None else port_forwards,
+            ctx=ctx,
+        )
+    except Exception:
+        # Nothing was registered, so sandbox_destroy could never reach this
+        # overlay — clean it up here or it leaks with no owner.
+        with contextlib.suppress(OSError):
+            overlay.unlink(missing_ok=True)
+        raise
 
     record = state.registry.get(name)
     record.config["sandbox_overlay"] = str(overlay)
     record.config["sandbox_base"] = str(base)
-    state.registry.save()
+    state.registry.update(record)
 
     agent = "unavailable"
     loop = asyncio.get_event_loop()
@@ -101,6 +134,14 @@ async def sandbox_vm(
         if await _agent_responding(record):
             agent = "responding"
             break
+        # A VM that died during boot will never answer; say so now rather than
+        # blaming a missing guest agent after the full timeout.
+        if not state.registry.is_process_alive(record):
+            raise ToolError(
+                f"VM {name!r} exited while booting. Last QEMU log lines: "
+                f"{_tail_log(record.qemu_log)}. The overlay was kept at {overlay}; "
+                "remove it with sandbox_destroy."
+            )
         await asyncio.sleep(2)
 
     result.update(
@@ -124,7 +165,8 @@ async def sandbox_destroy(name: str, ctx: Context = None) -> dict:
     """Destroy a sandbox created with sandbox_vm: force-stop the VM, remove it
     from the registry, and delete its overlay disk and logs. The base image is
     untouched. Refuses to operate on VMs that were not created by sandbox_vm —
-    use stop_vm / forget_vm for those (they never delete disks)."""
+    use stop_vm / forget_vm for those (they never delete disks). If the VM
+    cannot be killed, nothing is deleted and the call fails."""
     state, record = require_vm(ctx, name)
     overlay = record.config.get("sandbox_overlay")
     if not overlay:
@@ -132,23 +174,58 @@ async def sandbox_destroy(name: str, ctx: Context = None) -> dict:
             f"VM {name!r} was not created by sandbox_vm; refusing to delete its "
             "disks. Use stop_vm and forget_vm instead."
         )
+    # This is the only tool that deletes files, so it re-validates the name it
+    # was handed rather than trusting the registry key it came from.
+    if not valid_vm_name(name):
+        raise ToolError(f"Refusing to destroy {name!r}: not a valid VM name.")
 
+    targets = [state.config.vm_state_dir(name), state.config.vm_runtime_dir(name)]
+    roots = [state.config.state_dir / "vms", state.config.runtime_dir]
+    for target, root in zip(targets, roots, strict=True):
+        resolved, root_resolved = target.resolve(), root.resolve()
+        if resolved == root_resolved or not resolved.is_relative_to(root_resolved):
+            raise ToolError(
+                f"Refusing to delete {resolved}: it is not inside {root_resolved}. "
+                "The registry entry looks corrupt; inspect it before retrying."
+            )
+
+    # Verify death before deleting: an overlay unlinked while QEMU still holds
+    # it open keeps consuming space and the guest keeps writing into a file
+    # nobody can find.
     if state.registry.is_process_alive(record):
-        async with qmp_session(record) as client:
-            await execute(client, "quit")
-        deadline = asyncio.get_event_loop().time() + 10
-        while pid_alive(record.pid) and asyncio.get_event_loop().time() < deadline:
-            await asyncio.sleep(0.1)
+        log.info("sandbox_destroy: quitting VM %s (pid %s)", name, record.pid)
+        with contextlib.suppress(ToolError):
+            async with qmp_session(record) as client:
+                await execute(client, "quit")
+        if not await _await_exit(record.pid, 10):
+            log.warning("sandbox_destroy: %s ignored quit, sending SIGKILL", name)
+            with contextlib.suppress(OSError):
+                os.kill(record.pid, signal.SIGKILL)
+            await _await_exit(record.pid, 5)
+        if pid_alive(record.pid):
+            raise ToolError(
+                f"VM {name!r} (pid {record.pid}) survived both 'quit' and SIGKILL. "
+                "Nothing was deleted — its overlay is still open by that process. "
+                "Investigate the process before retrying."
+            )
 
     state.registry.remove(name)
-    shutil.rmtree(state.config.vm_state_dir(name), ignore_errors=True)
-    shutil.rmtree(state.config.vm_runtime_dir(name), ignore_errors=True)
-    return {
+    errors: list[str] = []
+    for target in targets:
+        if not target.exists():
+            continue  # nothing to remove is success, not a failure
+        shutil.rmtree(target, onexc=lambda _f, path, exc: errors.append(f"{path}: {exc}"))
+    overlay_gone = not Path(overlay).exists()
+    log.info("sandbox_destroy: removed %s (overlay_deleted=%s)", name, overlay_gone)
+    result = {
         "name": name,
-        "destroyed": True,
-        "overlay_deleted": not Path(overlay).exists(),
+        "destroyed": overlay_gone and not errors,
+        "overlay_deleted": overlay_gone,
         "base_image_untouched": record.config.get("sandbox_base"),
     }
+    if errors or not overlay_gone:
+        result["cleanup_errors"] = errors or [f"{overlay} still exists after rmtree"]
+    return result
 
 
 def register(mcp: FastMCP) -> None:

src/mcqemu/tools/snapshots.py

@@ -6,15 +6,26 @@ job polling for the same v1 outcome. Internal snapshots require all writable
 disks to be qcow2.
 """
 
+import re
+
 from fastmcp import Context, FastMCP
 from fastmcp.exceptions import ToolError
 
 from ..qmp import execute, qmp_session
 from ._common import require_vm
 
+# savevm/loadvm write or read the whole guest RAM and can legitimately take
+# minutes on a large VM, so they get a far longer bound than ordinary QMP
+# commands — but still a bound, so a wedged one cannot hang the tool forever.
+SNAPSHOT_TIMEOUT = 900.0
 
-async def _hmp(client, command_line: str) -> str:
-    return await execute(client, "human-monitor-command", {"command-line": command_line})
+_TAG_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")
+
+
+async def _hmp(client, command_line: str, timeout: float = SNAPSHOT_TIMEOUT) -> str:
+    return await execute(
+        client, "human-monitor-command", {"command-line": command_line}, timeout=timeout
+    )
 
 
 async def _hmp_expect_silence(client, command_line: str, what: str) -> None:
@@ -24,8 +35,13 @@ async def _hmp_expect_silence(client, command_line: str, what: str) -> None:
 
 
 def _check_tag(tag: str) -> None:
-    if not tag or any(c.isspace() for c in tag) or tag.startswith("-"):
-        raise ToolError(f"Invalid snapshot tag {tag!r}: no whitespace, must not start with '-'.")
+    """HMP has no statement separator, but quotes and spaces still confuse its
+    tokenizer, so a tag that survives here means what it says."""
+    if not _TAG_RE.match(tag or ""):
+        raise ToolError(
+            f"Invalid snapshot tag {tag!r}: use letters, digits, '.', '_', '-' "
+            "(must start alphanumeric, max 64 chars)."
+        )
 
 
 async def vm_snapshot_create(name: str, tag: str, ctx: Context = None) -> dict:

tests/conftest.py

@@ -1,6 +1,7 @@
 """Shared fixtures: temp XDG dirs, scriptable fake QMP client, in-memory MCP client."""
 
 import asyncio
+import inspect
 import json
 from types import SimpleNamespace
 
@@ -81,7 +82,10 @@ class FakeQMPClient:
         if isinstance(reply, Exception):
             raise reply
         if callable(reply):
-            return reply(arguments)
+            reply = reply(arguments)
+        # Awaitable replies let a test simulate an agent that stops answering.
+        if inspect.isawaitable(reply):
+            reply = await reply
         return reply
 
     @property
@@ -97,13 +101,23 @@ def fake_qmp(monkeypatch):
     return FakeQMPClient
 
 
+# Every module that imported pid_alive into its own namespace needs patching,
+# or a "dead" fixture leaves one module still seeing live processes.
+_PID_ALIVE_REFS = (
+    "mcqemu.registry.pid_alive",
+    "mcqemu.qmp.pid_alive",
+    "mcqemu.tools.lifecycle.pid_alive",
+    "mcqemu.tools.sandbox.pid_alive",
+)
+
+
 @pytest.fixture
 def all_pids_dead(monkeypatch):
-    monkeypatch.setattr("mcqemu.registry.pid_alive", lambda pid: False)
-    monkeypatch.setattr("mcqemu.tools.lifecycle.pid_alive", lambda pid: False)
+    for ref in _PID_ALIVE_REFS:
+        monkeypatch.setattr(ref, lambda pid: False)
 
 
 @pytest.fixture
 def all_pids_alive(monkeypatch):
-    monkeypatch.setattr("mcqemu.registry.pid_alive", lambda pid: bool(pid))
-    monkeypatch.setattr("mcqemu.tools.lifecycle.pid_alive", lambda pid: bool(pid))
+    for ref in _PID_ALIVE_REFS:
+        monkeypatch.setattr(ref, lambda pid: bool(pid))

tests/test_integration.py

@@ -34,16 +34,16 @@ async def test_real_vm_lifecycle(dirs):
             assert data["status"] == "running"
             assert pid_alive(pid)
 
-            vms = result_data(await client.call_tool("list_vms", {}))
-            assert [(v["name"], v["status"]) for v in vms] == [("itest", "running")]
+            listing = result_data(await client.call_tool("list_vms", {}))
+            assert [(v["name"], v["status"]) for v in listing["vms"]] == [("itest", "running")]
 
             info = result_data(await client.call_tool("vm_info", {"name": "itest"}))
             assert info["vcpus"] == 1
 
             data = result_data(await client.call_tool("pause_vm", {"name": "itest"}))
             assert data["status"] == "paused"
-            vms = result_data(await client.call_tool("list_vms", {}))
-            assert vms[0]["status"] == "paused"
+            listing = result_data(await client.call_tool("list_vms", {}))
+            assert listing["vms"][0]["status"] == "paused"
 
             await client.call_tool("resume_vm", {"name": "itest"})
 

tests/test_lifecycle.py

@@ -59,8 +59,8 @@ async def test_launch_happy_path_registers(dirs, monkeypatch):
         data = result_data(await client.call_tool("launch_vm", {"name": "fresh", "no_net": True}))
         assert data["status"] == "running"
         assert data["accel"] == "kvm"
-        vms = result_data(await client.call_tool("list_vms", {}))
-        assert [v["name"] for v in vms] == ["fresh"]
+        result = result_data(await client.call_tool("list_vms", {}))
+        assert [v["name"] for v in result["vms"]] == ["fresh"]
 
 
 async def test_stop_force_sends_quit(dirs, fake_qmp, all_pids_dead, monkeypatch):
@@ -84,15 +84,27 @@ async def test_stop_graceful_waits_for_shutdown_event(dirs, fake_qmp, all_pids_d
     assert ("system_powerdown", {}) in FakeQMPClient.calls
 
 
-async def test_stop_graceful_timeout_advises_force(dirs, fake_qmp, all_pids_dead, monkeypatch):
+async def test_stop_graceful_timeout_advises_force(dirs, fake_qmp, all_pids_alive, monkeypatch):
+    # Guest ignores ACPI: no SHUTDOWN event AND the process stays alive.
     monkeypatch.setattr("mcqemu.tools.lifecycle.Path.exists", lambda self: True)
-    FakeQMPClient.events_to_emit = []  # guest ignores ACPI
+    FakeQMPClient.events_to_emit = []
     write_registry(dirs, seeded_record(dirs, "vm1"))
     async with Client(mcp) as client:
         with pytest.raises(ToolError, match="force=True"):
             await client.call_tool("stop_vm", {"name": "vm1", "timeout": 1})
 
 
+async def test_stop_graceful_reports_process_exit(dirs, fake_qmp, all_pids_dead, monkeypatch):
+    """A VM that dies mid-shutdown must not be misreported as ignoring ACPI."""
+    monkeypatch.setattr("mcqemu.tools.lifecycle.Path.exists", lambda self: True)
+    FakeQMPClient.events_to_emit = []  # crashed before emitting SHUTDOWN
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        data = result_data(await client.call_tool("stop_vm", {"name": "vm1", "timeout": 30}))
+    assert data["status"] == "stopped"
+    assert "exited" in data["method"]
+
+
 async def test_stop_already_stopped(dirs, all_pids_dead):
     write_registry(dirs, seeded_record(dirs, "vm1"))
     async with Client(mcp) as client:
@@ -139,8 +151,8 @@ async def test_attach_and_forget(dirs, fake_qmp, tmp_path):
         assert data["status"] == "running"
         data = result_data(await client.call_tool("forget_vm", {"name": "ext"}))
         assert data["forgotten"] is True
-        vms = result_data(await client.call_tool("list_vms", {}))
-        assert vms == []
+        result = result_data(await client.call_tool("list_vms", {}))
+        assert result["vms"] == []
 
 
 async def test_forget_refuses_running_spawned_vm(dirs, all_pids_alive):

tests/test_query.py

@@ -10,24 +10,25 @@ from test_lifecycle import seeded_record
 async def test_list_vms_stopped(dirs, all_pids_dead):
     write_registry(dirs, seeded_record(dirs, "vm1"))
     async with Client(mcp) as client:
-        vms = result_data(await client.call_tool("list_vms", {}))
-    assert vms[0]["status"] == "stopped"
+        result = result_data(await client.call_tool("list_vms", {}))
+    assert result["vms"][0]["status"] == "stopped"
+    assert result["registry_warnings"] == []
 
 
 async def test_list_vms_running(dirs, fake_qmp, all_pids_alive):
     FakeQMPClient.responses["query-status"] = {"status": "running"}
     write_registry(dirs, seeded_record(dirs, "vm1"))
     async with Client(mcp) as client:
-        vms = result_data(await client.call_tool("list_vms", {}))
-    assert vms[0]["status"] == "running"
+        result = result_data(await client.call_tool("list_vms", {}))
+    assert result["vms"][0]["status"] == "running"
 
 
 async def test_list_vms_unreachable_when_qmp_fails(dirs, fake_qmp, all_pids_alive):
     FakeQMPClient.connect_error = OSError("connection refused")
     write_registry(dirs, seeded_record(dirs, "vm1"))
     async with Client(mcp) as client:
-        vms = result_data(await client.call_tool("list_vms", {}))
-    assert vms[0]["status"] == "unreachable"
+        result = result_data(await client.call_tool("list_vms", {}))
+    assert result["vms"][0]["status"] == "unreachable"
 
 
 async def test_vm_info_running_includes_block_devices(dirs, fake_qmp, all_pids_alive):

tests/test_reliability.py

@@ -0,0 +1,261 @@
+"""Failure-path tests for the conditions found in the reliability review.
+
+Every test here corresponds to a way the server used to fail badly: a corrupt
+state file taking down every tool, a destructive tool deleting outside its
+tree or while QEMU still held the file, a guest agent that stops answering
+mid-call, two instances clobbering each other's records.
+"""
+
+import asyncio
+import json
+import socket
+
+import pytest
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+
+from conftest import FakeQMPClient, result_data, write_registry
+from mcqemu.config import Config
+from mcqemu.models import VMRecord
+from mcqemu.registry import VMRegistry
+from mcqemu.server import mcp
+from test_lifecycle import seeded_record
+
+
+def config_for(dirs) -> Config:
+    return Config(state_dir=dirs.state, runtime_dir=dirs.run)
+
+
+# --- registry survives damage ----------------------------------------------
+
+
+async def test_corrupt_registry_does_not_kill_the_server(dirs):
+    """A truncated state file must not make every tool unreachable."""
+    (dirs.state / "vms.json").write_text('{"broken": ')
+    async with Client(mcp) as client:
+        listing = result_data(await client.call_tool("list_vms", {}))
+    assert listing["vms"] == []
+    assert any("quarantined" in w for w in listing["registry_warnings"])
+    assert list(dirs.state.glob("vms.corrupt.*.json")), "bad file should be preserved"
+
+
+async def test_malformed_record_is_skipped_not_fatal(dirs):
+    good = seeded_record(dirs, "good")
+    (dirs.state / "vms.json").write_text(
+        json.dumps({"good": good.to_dict(), "broken": {"no": "required fields"}})
+    )
+    async with Client(mcp) as client:
+        listing = result_data(await client.call_tool("list_vms", {}))
+    assert [v["name"] for v in listing["vms"]] == ["good"]
+    assert any("broken" in w for w in listing["registry_warnings"])
+
+
+async def test_traversal_key_is_dropped_on_load(dirs):
+    """A registry key that is not a valid VM name never reaches the tools."""
+    rec = seeded_record(dirs, "ok")
+    payload = {"../../escape": rec.to_dict(), "ok": rec.to_dict()}
+    (dirs.state / "vms.json").write_text(json.dumps(payload))
+    async with Client(mcp) as client:
+        listing = result_data(await client.call_tool("list_vms", {}))
+    assert [v["name"] for v in listing["vms"]] == ["ok"]
+    assert any("invalid VM name" in w for w in listing["registry_warnings"])
+
+
+def test_unknown_schema_version_refuses_rather_than_stripping(dirs):
+    (dirs.state / "vms.json").write_text(json.dumps({"version": 99, "vms": {}}))
+    registry = VMRegistry(config_for(dirs))
+    registry.load()
+    assert registry.all() == []
+    assert "newer than this build" in registry.degraded
+
+
+def test_unknown_record_fields_round_trip(dirs):
+    """An older build must not strip a newer one's fields when it saves."""
+    registry = VMRegistry(config_for(dirs))
+    registry.load()
+    raw = seeded_record(dirs, "vm1").to_dict() | {"future_field": {"keep": "me"}}
+    (dirs.state / "vms.json").write_text(json.dumps({"vm1": raw}))
+
+    registry2 = VMRegistry(config_for(dirs))
+    registry2.load()
+    registry2.add(seeded_record(dirs, "vm2"))  # triggers a full rewrite
+
+    on_disk = json.loads((dirs.state / "vms.json").read_text())["vms"]
+    assert on_disk["vm1"]["future_field"] == {"keep": "me"}
+
+
+def test_concurrent_instances_merge_instead_of_clobbering(dirs):
+    """Two mcqemu processes sharing a registry must not orphan each other."""
+    a, b = VMRegistry(config_for(dirs)), VMRegistry(config_for(dirs))
+    a.load()
+    b.load()  # both start from an empty view
+    a.add(seeded_record(dirs, "from-a"))
+    b.add(seeded_record(dirs, "from-b"))  # stale in-memory view, must still merge
+
+    fresh = VMRegistry(config_for(dirs))
+    fresh.load()
+    assert sorted(fresh.names()) == ["from-a", "from-b"]
+
+
+def test_name_reservation_blocks_a_second_in_flight_launch(dirs):
+    registry = VMRegistry(config_for(dirs))
+    registry.load()
+    assert registry.reserve("vm1") is True
+    assert registry.reserve("vm1") is False
+    registry.release("vm1")
+    assert registry.reserve("vm1") is True
+
+
+# --- destructive tool refuses to guess -------------------------------------
+
+
+async def test_sandbox_destroy_refuses_path_outside_state_dir(dirs, all_pids_dead, monkeypatch):
+    """Even with a hostile record, deletion stays inside the VM state tree."""
+    victim = dirs.state / "IMPORTANT"
+    victim.mkdir()
+    (victim / "precious.txt").write_text("keep me")
+
+    record = seeded_record(dirs, "sb", config={"disks": [], "sandbox_overlay": "/tmp/x.qcow2"})
+    write_registry(dirs, record)
+    # Simulate a corrupted config whose state dir escapes the vms/ root.
+    monkeypatch.setattr(Config, "vm_state_dir", lambda self, name: victim)
+
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="not inside"):
+            await client.call_tool("sandbox_destroy", {"name": "sb"})
+    assert (victim / "precious.txt").exists()
+
+
+async def test_sandbox_destroy_refuses_when_vm_survives_kill(dirs, all_pids_alive, monkeypatch):
+    """If the process will not die, its overlay must not be deleted."""
+    overlay = dirs.state / "vms" / "sb" / "overlay.qcow2"
+    overlay.parent.mkdir(parents=True)
+    overlay.write_bytes(b"disk")
+    write_registry(
+        dirs,
+        seeded_record(dirs, "sb", config={"disks": [], "sandbox_overlay": str(overlay)}),
+    )
+    monkeypatch.setattr("mcqemu.tools.sandbox.os.kill", lambda pid, sig: None)
+    monkeypatch.setattr("mcqemu.tools.sandbox._await_exit", _never_exits)
+
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="survived both"):
+            await client.call_tool("sandbox_destroy", {"name": "sb"})
+    assert overlay.exists(), "overlay must survive a failed kill"
+    async with Client(mcp) as client:
+        listing = result_data(await client.call_tool("list_vms", {}))
+    assert [v["name"] for v in listing["vms"]] == ["sb"], "record must not be dropped"
+
+
+async def _never_exits(pid, seconds):
+    return False
+
+
+async def test_sandbox_destroy_reports_cleanup_failure_honestly(dirs, all_pids_dead, monkeypatch):
+    overlay = dirs.state / "vms" / "sb" / "overlay.qcow2"
+    overlay.parent.mkdir(parents=True)
+    overlay.write_bytes(b"disk")
+    write_registry(
+        dirs,
+        seeded_record(dirs, "sb", config={"disks": [], "sandbox_overlay": str(overlay)}),
+    )
+
+    def boom(target, onexc=None):
+        onexc(None, str(target), PermissionError("read-only filesystem"))
+
+    monkeypatch.setattr("mcqemu.tools.sandbox.shutil.rmtree", boom)
+    async with Client(mcp) as client:
+        data = result_data(await client.call_tool("sandbox_destroy", {"name": "sb"}))
+    assert data["destroyed"] is False
+    assert data["cleanup_errors"], "a failed delete must be reported, not swallowed"
+
+
+# --- launch does not trample a live VM -------------------------------------
+
+
+async def test_launch_refuses_to_reuse_a_live_qmp_socket(dirs):
+    """An unregistered but running QEMU must not lose its monitor socket."""
+    run_dir = dirs.run / "ghost"
+    run_dir.mkdir(parents=True)
+    sock_path = run_dir / "qmp.sock"
+    listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    listener.bind(str(sock_path))
+    listener.listen(1)
+    try:
+        async with Client(mcp) as client:
+            with pytest.raises(ToolError, match="LIVE QMP socket"):
+                await client.call_tool("launch_vm", {"name": "ghost", "no_net": True})
+        assert sock_path.exists(), "the live socket must not be unlinked"
+    finally:
+        listener.close()
+
+
+# --- guest agent cannot hang the server ------------------------------------
+
+
+async def test_guest_exec_times_out_when_agent_stops_answering(dirs, fake_qmp, monkeypatch):
+    monkeypatch.setattr("mcqemu.tools.guest.CALL_TIMEOUT", 1.0)
+    FakeQMPClient.responses["guest-sync"] = lambda args: args["id"]
+    FakeQMPClient.responses["guest-exec"] = {"pid": 7}
+    FakeQMPClient.responses["guest-exec-status"] = lambda args: asyncio.sleep(3600)
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="stopped responding"):
+            await asyncio.wait_for(
+                client.call_tool("guest_exec", {"name": "vm1", "command": "sleep", "timeout": 2}),
+                timeout=30,
+            )
+
+
+async def test_guest_file_read_gives_up_on_a_zero_progress_agent(dirs, fake_qmp):
+    """A FIFO or tty returns count=0 without EOF forever; don't spin on it."""
+    FakeQMPClient.responses["guest-sync"] = lambda args: args["id"]
+    FakeQMPClient.responses["guest-file-open"] = 3
+    FakeQMPClient.responses["guest-file-read"] = {"count": 0, "eof": False}
+    FakeQMPClient.responses["guest-file-close"] = {}
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        data = result_data(
+            await asyncio.wait_for(
+                client.call_tool("guest_file_read", {"name": "vm1", "path": "/dev/fifo"}),
+                timeout=20,
+            )
+        )
+    assert data["bytes_read"] == 0
+    reads = [c for c, _ in FakeQMPClient.calls if c == "guest-file-read"]
+    assert len(reads) < 10, "should stop after a few empty reads, not loop"
+
+
+async def test_guest_file_read_rejects_absurd_max_bytes(dirs, fake_qmp):
+    FakeQMPClient.responses["guest-sync"] = lambda args: args["id"]
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="max_bytes"):
+            await client.call_tool(
+                "guest_file_read", {"name": "vm1", "path": "/x", "max_bytes": 10**12}
+            )
+
+
+# --- untrusted bytes --------------------------------------------------------
+
+
+async def test_screenshot_rejects_non_png_output(dirs, fake_qmp):
+    def write_junk(args):
+        with open(args["filename"], "wb") as f:
+            f.write(b"not a png at all")
+        return {}
+
+    FakeQMPClient.responses["screendump"] = write_junk
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="valid PNG"):
+            await client.call_tool("vm_click", {"name": "vm1", "x": 1, "y": 1})
+
+
+def test_record_name_follows_the_registry_key(dirs):
+    """Tools look records up by key, so a mismatched inner name is corrected."""
+    rec = VMRecord(name="lies", source="spawned", qmp_socket="/tmp/q.sock")
+    (dirs.state / "vms.json").write_text(json.dumps({"truth": rec.to_dict()}))
+    registry = VMRegistry(config_for(dirs))
+    registry.load()
+    assert registry.get("truth").name == "truth"

tests/test_sandbox.py

@@ -145,8 +145,8 @@ async def test_sandbox_destroy_removes_overlay(dirs, all_pids_dead):
         assert data["destroyed"] is True
         assert data["overlay_deleted"] is True
         assert not overlay.exists()
-        vms = result_data(await client.call_tool("list_vms", {}))
-        assert vms == []
+        result = result_data(await client.call_tool("list_vms", {}))
+        assert result["vms"] == []
 
 
 async def test_sandbox_destroy_refuses_non_sandbox(dirs, all_pids_dead):

tests/test_snapshots.py

@@ -62,3 +62,38 @@ async def test_snapshot_list_raw(dirs, fake_qmp):
     async with Client(mcp) as client:
         data = result_data(await client.call_tool("vm_snapshot_list", {"name": "vm1"}))
     assert "clean" in data["snapshots_raw"]
+
+
+async def test_snapshot_tag_rejects_quotes_and_specials(dirs, fake_qmp):
+    """HMP's tokenizer strips quotes, so 'x' and '\"x\"' would silently be the
+    same snapshot — reject anything that isn't literal."""
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        for tag in ['"quoted"', "semi;colon", "tab\tsep", "a" * 80]:
+            with pytest.raises(ToolError, match="Invalid snapshot tag"):
+                await client.call_tool("vm_snapshot_create", {"name": "vm1", "tag": tag})
+
+
+async def test_savevm_gets_a_long_but_finite_timeout(dirs, fake_qmp):
+    """A RAM-sized savevm must not be killed by the ordinary command timeout."""
+    from mcqemu.qmp import COMMAND_TIMEOUT
+    from mcqemu.tools.snapshots import SNAPSHOT_TIMEOUT
+
+    assert SNAPSHOT_TIMEOUT > COMMAND_TIMEOUT * 10
+    seen = {}
+
+    async def spy(client, command, arguments=None, timeout=None):
+        seen["timeout"] = timeout
+        return ""
+
+    import mcqemu.tools.snapshots as snapshots_mod
+
+    original = snapshots_mod.execute
+    snapshots_mod.execute = spy
+    try:
+        write_registry(dirs, seeded_record(dirs, "vm1"))
+        async with Client(mcp) as client:
+            await client.call_tool("vm_snapshot_create", {"name": "vm1", "tag": "clean"})
+    finally:
+        snapshots_mod.execute = original
+    assert seen["timeout"] == SNAPSHOT_TIMEOUT