From 71d72c2393b4107ce73fc8cb83d7f8ee72978081 Mon Sep 17 00:00:00 2001
From: MCKero <ai@mckero.com>
Date: Fri, 28 Aug 2026 05:57:22 +0100
Subject: [PATCH] Add a supervisor that owns the QEMU process

Power on/off cannot live inside the QMP connection: QMP quit destroys the socket
a later power on would have to arrive through. So something outside it has to be
able to spawn the process again.

Off then On is a cold boot -- process replaced, guest from reset -- which is the
behaviour asked for: like cutting mains power and restoring it. system_reset is the
warm alternative and keeps the process.

adopt() is for attaching to a run.sh instance. power_off then refuses, because we
did not start that process. is_running() also reports False for a process that
exited on its own, rather than trusting our own bookkeeping.

power_off tolerates quit raising: the socket usually drops before the reply
arrives, so that is success rather than an error. The launcher clears a stale
socket first, since QEMU failing to bind presents as On doing nothing.

Verified against real QEMU: starts with 0 processes, On gives 1, Reset keeps the
same one, Off returns to 0, On again cold boots. 14 unit tests with a fake
launcher, so they need no emulator.
---
 tools/test_uvk5_supervisor.py | 158 ++++++++++++++++++++++++++++++++++
 tools/uvk5_supervisor.py      | 142 ++++++++++++++++++++++++++++++
 2 files changed, 300 insertions(+)
 create mode 100644 tools/test_uvk5_supervisor.py
 create mode 100644 tools/uvk5_supervisor.py

diff --git a/tools/test_uvk5_supervisor.py b/tools/test_uvk5_supervisor.py
new file mode 100644
index 0000000..2967d06
--- /dev/null
+++ b/tools/test_uvk5_supervisor.py
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Unit tests for the QEMU supervisor. Uses a fake launcher, not real QEMU."""
+import unittest
+
+from uvk5_supervisor import Supervisor
+
+
+class FakeProc:
+    def __init__(self):
+        self.terminated = False
+        self.killed = False
+        self._alive = True
+        self.stderr = None
+
+    def poll(self):
+        return None if self._alive else 0
+
+    def terminate(self):
+        self.terminated = True
+        self._alive = False
+
+    def wait(self, timeout=None):
+        self._alive = False
+        return 0
+
+    def kill(self):
+        self.killed = True
+        self._alive = False
+
+
+class FakeClient:
+    def __init__(self):
+        self.commands = []
+        self.closed = False
+
+    def command(self, name, **args):
+        self.commands.append(name)
+        if name == "query-status":
+            return {"status": "running", "running": True}
+        return {}
+
+    def close(self):
+        self.closed = True
+
+
+class TestSupervisor(unittest.TestCase):
+    def setUp(self):
+        self.procs = []
+        self.clients = []
+
+        def launch():
+            proc = FakeProc()
+            self.procs.append(proc)
+            return proc
+
+        def connect():
+            client = FakeClient()
+            self.clients.append(client)
+            return client
+
+        self.sup = Supervisor(launch=launch, connect=connect)
+
+    def test_starts_powered_off_when_nothing_is_running(self):
+        """The user presses On; the server does not boot it for them."""
+        self.assertFalse(self.sup.is_running())
+        self.assertIsNone(self.sup.client())
+        self.assertEqual(self.procs, [])
+
+    def test_power_on_launches_and_connects(self):
+        self.assertTrue(self.sup.power_on())
+        self.assertTrue(self.sup.is_running())
+        self.assertEqual(len(self.procs), 1)
+        self.assertIsNotNone(self.sup.client())
+
+    def test_power_on_twice_does_not_launch_twice(self):
+        self.sup.power_on()
+        self.assertFalse(self.sup.power_on())
+        self.assertEqual(len(self.procs), 1)
+
+    def test_power_off_quits_via_qmp_then_stops_the_process(self):
+        self.sup.power_on()
+        client = self.sup.client()
+        self.assertTrue(self.sup.power_off())
+        self.assertIn("quit", client.commands)
+        self.assertTrue(client.closed)
+        self.assertFalse(self.sup.is_running())
+        self.assertIsNone(self.sup.client())
+
+    def test_power_cycle_boots_a_fresh_process(self):
+        """Off then On is a cold boot, like mains power cut and restored."""
+        self.sup.power_on()
+        self.sup.power_off()
+        self.sup.power_on()
+        self.assertEqual(len(self.procs), 2)
+        self.assertTrue(self.sup.is_running())
+
+    def test_reset_uses_system_reset_when_running(self):
+        self.sup.power_on()
+        client = self.sup.client()
+        self.sup.reset()
+        self.assertIn("system_reset", client.commands)
+        self.assertEqual(len(self.procs), 1, "reset must not respawn QEMU")
+
+    def test_reset_powers_on_when_stopped(self):
+        self.sup.reset()
+        self.assertTrue(self.sup.is_running())
+        self.assertEqual(len(self.procs), 1)
+
+    def test_pause_and_resume(self):
+        self.sup.power_on()
+        client = self.sup.client()
+        self.assertTrue(self.sup.pause())
+        self.assertTrue(self.sup.resume())
+        self.assertIn("stop", client.commands)
+        self.assertIn("cont", client.commands)
+
+    def test_pause_when_off_is_refused(self):
+        self.assertFalse(self.sup.pause())
+        self.assertFalse(self.sup.resume())
+
+    def test_power_off_when_already_off_is_harmless(self):
+        self.assertFalse(self.sup.power_off())
+        self.assertFalse(self.sup.is_running())
+
+    def test_power_off_survives_a_quit_that_raises(self):
+        """The socket usually drops mid-quit; that is success, not failure."""
+        class Rude(FakeClient):
+            def command(self, name, **args):
+                if name == "quit":
+                    raise RuntimeError("connection reset")
+                return super().command(name, **args)
+
+        sup = Supervisor(launch=lambda: FakeProc(), connect=Rude)
+        sup.power_on()
+        self.assertTrue(sup.power_off())
+        self.assertFalse(sup.is_running())
+
+    def test_dead_process_is_reported_as_not_running(self):
+        self.sup.power_on()
+        self.procs[0]._alive = False       # QEMU crashed on its own
+        self.assertFalse(self.sup.is_running())
+
+    def test_adopting_an_external_emulator_does_not_own_the_process(self):
+        """Attaching to a run.sh instance must not let power_off kill it."""
+        client = FakeClient()
+        sup = Supervisor(launch=lambda: self.fail("must not launch"),
+                         connect=lambda: client)
+        sup.adopt(client)
+        self.assertTrue(sup.is_running())
+        self.assertFalse(sup.owns_process())
+
+    def test_owns_process_is_true_after_power_on(self):
+        self.sup.power_on()
+        self.assertTrue(self.sup.owns_process())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tools/uvk5_supervisor.py b/tools/uvk5_supervisor.py
new file mode 100644
index 0000000..8a16ff9
--- /dev/null
+++ b/tools/uvk5_supervisor.py
@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""Owns the QEMU process, so the web UI can power the emulator on and off.
+
+QMP `quit` stops the emulator but also destroys the socket, so nothing is left to
+receive a later "power on". Power control therefore needs something outside the
+QMP connection that can spawn the process again -- that is this.
+
+Off then On is a cold boot: the process is replaced and the guest starts from
+reset, the same as cutting mains power and restoring it. `system_reset` is the
+warm alternative and keeps the process.
+
+`adopt()` covers the other case: the server attached to an emulator someone else
+started with run.sh. Then `power_off` must refuse, because we did not start that
+process and killing it is not ours to do.
+"""
+import os
+import subprocess
+import threading
+import time
+
+DEFAULT_QMP = "/tmp/uvk5-qmp.sock"
+
+
+def default_launcher(qemu: str, flash: str, elf: str,
+                     qmp_path: str = DEFAULT_QMP, gdb_port: int = 1234,
+                     capture_stderr: bool = False):
+    """Reproduces the command line in tools/run.sh."""
+    def launch():
+        # A stale socket makes QEMU fail to bind, which looks like "power on did
+        # nothing". Clear it first.
+        if os.path.exists(qmp_path):
+            os.unlink(qmp_path)
+        return subprocess.Popen(
+            [qemu, "-M", f"uv-k5-v3,flash-image={flash}",
+             "-nographic", "-monitor", "none",
+             "-qmp", f"unix:{qmp_path},server=on,wait=off",
+             "-kernel", elf, "-gdb", f"tcp::{gdb_port}"],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE if capture_stderr else subprocess.DEVNULL)
+    return launch
+
+
+def wait_for_socket(path: str, timeout: float = 15.0) -> bool:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if os.path.exists(path):
+            return True
+        time.sleep(0.05)
+    return False
+
+
+class Supervisor:
+    def __init__(self, launch, connect):
+        self._launch = launch
+        self._connect = connect
+        self._lock = threading.Lock()
+        self._proc = None
+        self._client = None
+
+    def is_running(self) -> bool:
+        with self._lock:
+            if self._client is None:
+                return False
+            # A process that exited on its own is not running, whatever we think.
+            if self._proc is not None and self._proc.poll() is not None:
+                return False
+            return True
+
+    def owns_process(self) -> bool:
+        """True when we launched it, and may therefore stop it."""
+        with self._lock:
+            return self._proc is not None
+
+    def client(self):
+        with self._lock:
+            return self._client
+
+    def process(self):
+        with self._lock:
+            return self._proc
+
+    def adopt(self, client):
+        """Use an emulator we did not start. power_off will refuse to kill it."""
+        with self._lock:
+            self._client = client
+            self._proc = None
+
+    def power_on(self) -> bool:
+        with self._lock:
+            if self._client is not None:
+                return False
+            self._proc = self._launch()
+            self._client = self._connect()
+            return True
+
+    def power_off(self) -> bool:
+        with self._lock:
+            client, proc = self._client, self._proc
+            self._client, self._proc = None, None
+        if client is None:
+            return False
+        try:
+            client.command("quit")
+        except Exception:
+            # Expected: quit tears the socket down, often before the reply.
+            pass
+        try:
+            client.close()
+        except Exception:
+            pass
+        if proc is not None:
+            try:
+                proc.wait(timeout=10)
+            except Exception:
+                proc.terminate()
+                try:
+                    proc.wait(timeout=5)
+                except Exception:
+                    proc.kill()
+        return True
+
+    def reset(self) -> bool:
+        """Warm reboot, or a cold boot when the emulator is off."""
+        client = self.client()
+        if client is None:
+            return self.power_on()
+        client.command("system_reset")
+        return True
+
+    def pause(self) -> bool:
+        client = self.client()
+        if client is None:
+            return False
+        client.command("stop")
+        return True
+
+    def resume(self) -> bool:
+        client = self.client()
+        if client is None:
+            return False
+        client.command("cont")
+        return True