The harness grows a QMP channel (M21.0)

Every case now gets a -qmp unix socket (additive; no case notices). A
minimal client does the capabilities handshake and executes one command;
the per-case qmp_after hook sends it N seconds after boot, retrying until
the guest's socket is up. A case with a hook configured cannot pass until
the hook delivered — and the smoke case now carries a harmless
query-status hook, so the channel is proven end to end on every run.
This is how the power scenarios inject the real ACPI power-button event
(system_powerdown) in M21.1 and M21.3.

test/qemu_test.py

@@ -18,9 +18,11 @@ Usage:
 """
 
 import argparse
+import json
 import os
 import re
 import shutil
+import socket
 import subprocess
 import sys
 import time
@@ -80,7 +82,10 @@ ARCHES = {
 # `expect`: a regex that must appear in serial output => pass.
 # `fail`:   optional regex whose appearance => immediate fail.
 CASES = [
+    # smoke also proves the QMP channel: the harmless query must be delivered
+    # (handshake + command) before the case may pass — see run_case.
     {"name": "smoke",
+     "qmp_after": {"delay": 2, "command": "query-status"},
      "expect": r"DANOS-TEST-RESULT: PASS",
      "fail":   r"DANOS-TEST-RESULT: FAIL"},
     {"name": "discovery",
@@ -418,6 +423,27 @@ def resolve_firmware(arch):
                      + "\nInstall OVMF (edk2-ovmf / ovmf) or add its path above.")
 
 
+def qmp_send(path, command):
+    """One QMP command: connect, capabilities handshake, execute. Raises on any
+    failure — the caller retries until the guest's socket is ready. This is how
+    a case injects a host-side event (system_powerdown = the ACPI power button)
+    into the running guest (docs/m21-plan.md)."""
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.settimeout(5)
+    try:
+        sock.connect(path)
+        stream = sock.makefile("rw")
+        stream.readline()  # the QMP greeting
+        stream.write(json.dumps({"execute": "qmp_capabilities"}) + "\n")
+        stream.flush()
+        stream.readline()  # {"return": {}}
+        stream.write(json.dumps({"execute": command}) + "\n")
+        stream.flush()
+        stream.readline()
+    finally:
+        sock.close()
+
+
 def run_case(arch, case):
     err = build(arch, case["name"])
     if err:
@@ -439,12 +465,27 @@ def run_case(arch, case):
         cmd += ["-smp", str(case["smp"])]
     if case.get("qemu_extra"):  # extra qemu args, e.g. -device intel-iommu for the IOMMU case
         cmd += case["qemu_extra"]
+    # A QMP control socket, always present (additive): how a case's `qmp_after`
+    # hook injects host-side events into the guest mid-run.
+    qmp_path = os.path.join(WORK, "qmp.sock")
+    if os.path.exists(qmp_path):
+        os.remove(qmp_path)
+    cmd += ["-qmp", f"unix:{qmp_path},server,nowait"]
+    qmp_after = case.get("qmp_after")  # {"delay": seconds, "command": "..."}
+    qmp_sent = False
+    started = time.monotonic()
     qemu = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
     try:
         timeout = case.get("timeout", TIMEOUT)
         deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             time.sleep(0.2)
+            if qmp_after and not qmp_sent and time.monotonic() - started >= qmp_after["delay"]:
+                try:
+                    qmp_send(qmp_path, qmp_after["command"])
+                    qmp_sent = True
+                except OSError:
+                    pass  # socket not up yet; retry next tick
             text = ""
             if os.path.exists(serial):
                 with open(serial, "r", errors="replace") as f:
@@ -452,6 +493,8 @@ def run_case(arch, case):
             if fail and fail.search(text):
                 return False, "hit failure marker"
             if expect.search(text):
+                if qmp_after and not qmp_sent:
+                    continue  # the hook must deliver before the case may pass
                 return True, "matched " + repr(case["expect"])
             if qemu.poll() is not None:  # QEMU exited on its own
                 if expect.search(text):