qemu: add wait_stat verb for polling stat checks

assert_stat samples a stat field once. warden-ui's fps counter is a
rolling one-second window, so a single sample right after a page opens
can read 0 even though the UI is live (flare-edge #44). Add wait_stat
FIELD OP VALUE TIMEOUT, the same polling pattern as wait_hit/wait_json,
built on a shared eval_stat() that assert_stat now uses too.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SMRwnkPp1upR6QFWCZouE7

qemu/lib.sh

@@ -22,15 +22,34 @@ qemu_get_busybox() {
     echo "FATAL: no pinned sha256 for busybox (expected $sha_file): refusing to build from an unverified binary" >&2
     exit 1
   }
-  if [ ! -f "$BB" ]; then
-    qemu_log "downloading $BB_URL"
-    curl --retry 3 --retry-delay 5 --retry-connrefused -fSL "$BB_URL" -o "$BB"
-  fi
   local want got
   want="$(cat "$sha_file")"
-  got="$(sha256sum "$BB" | awk '{print $1}')"
-  [ "$want" = "$got" ] || { echo "busybox sha256 mismatch: want $want got $got" >&2; exit 1; }
-  qemu_log "busybox sha256 verified"
+  if [ -f "$BB" ]; then
+    got="$(sha256sum "$BB" | awk '{print $1}')"
+    if [ "$want" = "$got" ]; then
+      qemu_log "busybox sha256 verified"
+      return
+    fi
+    qemu_log "cached busybox failed verification; fetching a clean copy"
+  fi
+
+  local attempt tmp="${BB}.download.$$"
+  rm -f "$tmp"
+  for attempt in 1 2 3; do
+    qemu_log "downloading $BB_URL (attempt $attempt/3)"
+    if curl --retry 3 --retry-delay 5 --retry-connrefused -fSL "$BB_URL" -o "$tmp"; then
+      got="$(sha256sum "$tmp" | awk '{print $1}')"
+      if [ "$want" = "$got" ]; then
+        mv -f "$tmp" "$BB"
+        qemu_log "busybox sha256 verified"
+        return
+      fi
+      echo "busybox sha256 mismatch on attempt $attempt: want $want got $got" >&2
+    fi
+    rm -f "$tmp"
+  done
+  echo "FATAL: could not fetch pinned busybox after 3 verified attempts" >&2
+  exit 1
 }
 
 # Stage the shared rootfs skeleton (qemu/rootfs/ + busybox) into $1.

qemu/rootfs/sbin/init

@@ -10,6 +10,17 @@
 exec </dev/console >/dev/console 2>&1
 /bin/busybox --install -s /bin
 
+# mkfs.ext4 -d preserves the checkout owner's uid. Production owns shadow as
+# root, and the UI intentionally rejects any other owner before verifying it.
+chown 0:0 /etc/shadow
+chmod 0600 /etc/shadow
+root_hash="$(awk -F: '$1 == "root" { print $2 }' /etc/shadow)"
+test_hash="$(printf '%s' root | /usr/bin/mkpasswd -m md5 -S wardenrs 2>/dev/null)"
+if [ -z "$root_hash" ] || [ "$test_hash" != "$root_hash" ]; then
+    echo "FATAL: QEMU root credential verifier is unavailable"
+    poweroff -f
+fi
+
 mount -t proc proc /proc
 mount -t sysfs sysfs /sys
 mount -t tmpfs tmpfs /tmp
@@ -84,6 +95,9 @@ export WARDEN_HPMCU=0
 # post-apply "reboot" surfaces as a clean flared error; the scenario harness
 # performs the actual reboot into the applied slot.
 export WARDEN_HARD_RESET=0
+# Production init scripts create this volatile socket directory before their
+# daemons start. The compact QEMU rootfs must provide the same contract.
+mkdir -p /run/warden
 # OTA apply is opt-in per boot (run.sh --allow-apply): writing rootfs_b is
 # safe inside disk.img but must never be the default posture.
 if grep -qw warden.fwapply /proc/cmdline; then

qemu/tests/busybox-fetch.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+TEST_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+QEMU_DIR="$(cd "$TEST_DIR/.." && pwd)"
+SCRATCH="$(mktemp -d "${TMPDIR:-/tmp}/busybox-fetch.XXXXXX")"
+trap 'rm -rf "$SCRATCH"' EXIT
+
+FAKE_BIN="$SCRATCH/bin"
+FAKE_CURL_COUNT="$SCRATCH/curl-count"
+mkdir -p "$FAKE_BIN" "$SCRATCH/qemu"
+printf 'good payload' > "$SCRATCH/good"
+sha256sum "$SCRATCH/good" | awk '{print $1}' > "$SCRATCH/qemu/busybox.sha256"
+
+FAKE_CURL="$FAKE_BIN/curl"
+apply_fake_curl() {
+  chmod 0755 "$FAKE_CURL"
+  export PATH="$FAKE_BIN:$PATH"
+  export FAKE_CURL_COUNT FAKE_CURL_MODE
+}
+
+# The path is resolved at runtime from this file's location.
+# shellcheck disable=SC1091
+source "$QEMU_DIR/lib.sh"
+QEMU_DIR="$SCRATCH/qemu"
+# Read by qemu_get_busybox from the sourced helper.
+# shellcheck disable=SC2034
+BB_URL="https://invalid.example/busybox"
+
+# These single-quoted lines are the literal source of the fake curl program.
+# shellcheck disable=SC2016
+printf '%s\n' '#!/bin/sh' \
+  'count=0' \
+  '[ ! -f "$FAKE_CURL_COUNT" ] || count=$(cat "$FAKE_CURL_COUNT")' \
+  'count=$((count + 1))' \
+  'printf "%s\n" "$count" > "$FAKE_CURL_COUNT"' \
+  'out=' \
+  'while [ "$#" -gt 0 ]; do' \
+  '  if [ "$1" = "-o" ]; then shift; out="$1"; fi' \
+  '  shift' \
+  'done' \
+  '[ -n "$out" ] || exit 2' \
+  'if [ "$FAKE_CURL_MODE" = "flaky" ] && [ "$count" -gt 1 ]; then' \
+  '  printf "good payload" > "$out"' \
+  'else' \
+  '  printf "bad payload" > "$out"' \
+  'fi' > "$FAKE_CURL"
+
+FAKE_CURL_MODE=flaky
+apply_fake_curl
+OUT="$SCRATCH/flaky" qemu_get_busybox
+test "$(cat "$FAKE_CURL_COUNT")" = 2
+test "$(sha256sum "$BB" | awk '{print $1}')" = "$(cat "$QEMU_DIR/busybox.sha256")"
+
+printf '0\n' > "$FAKE_CURL_COUNT"
+FAKE_CURL_MODE=bad
+apply_fake_curl
+if (OUT="$SCRATCH/always-bad" qemu_get_busybox) >"$SCRATCH/failure.log" 2>&1; then
+  echo "FAIL: an unverified download was accepted" >&2
+  exit 1
+fi
+test "$(cat "$FAKE_CURL_COUNT")" = 3
+test ! -e "$SCRATCH/always-bad/busybox-armv7l"
+grep -q 'after 3 verified attempts' "$SCRATCH/failure.log"
+
+printf '0\n' > "$FAKE_CURL_COUNT"
+mkdir -p "$SCRATCH/cached"
+cp "$SCRATCH/good" "$SCRATCH/cached/busybox-armv7l"
+OUT="$SCRATCH/cached" qemu_get_busybox
+test "$(cat "$FAKE_CURL_COUNT")" = 0
+
+echo "busybox fetch tests passed"

qemu/tests/qmp.py

@@ -78,6 +78,11 @@ math, this file only drives it.
                            1 dimmed, 2 asleep), read off a fresh `stats`
                            reply (warden_debug.c). Same OP vocabulary as
                            assert_json.
+    wait_stat FIELD OP VALUE TIMEOUT_S
+                           assert_stat, polled every 0.5s (poll_until) until
+                           it holds or TIMEOUT_S elapses -- for a field like
+                           fps that a rolling window can still report as 0
+                           right after the page holding it opens.
     capture_region NAME X Y W H TOLERANCE
                            screendump now, crop to X,Y,WxH, and (over)write
                            NAME in the refs file with both a phash and a
@@ -148,6 +153,8 @@ def _load_imgtools():
     return imgtools
 
 AXIS_MAX = 32767
+TAP_HOLD_S = 0.35
+TAP_OBSERVE_TIMEOUT_S = 2.0
 
 # Bounds every blocking read on the QMP socket -- the greeting banner, the
 # qmp_capabilities handshake, and every screendump/tap/quit round trip --
@@ -206,15 +213,57 @@ def to_axis(px, size):
     return min(AXIS_MAX, math.ceil(px * AXIS_MAX / (size - 1)))
 
 
-def do_tap(s, f, ax, ay, hold=0.2):
+def do_tap(s, f, ax, ay, hold=TAP_HOLD_S):
     send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
-    # Hold the press across several LVGL indev poll periods (33 ms each): an
-    # instantaneous press+release lands inside one poll and no click is ever
-    # registered.
+    # Hold the press across several LVGL indev poll periods (33 ms each), but
+    # stay below LVGL's 400 ms long-press threshold. A longer hold repeats
+    # controls such as Backspace and no longer represents a tap.
     time.sleep(hold)
     send_events(s, f, [btn_ev(False)])
 
 
+def wait_input_count(ctx, field, before):
+    """Wait until one guest input counter advances."""
+    if before is None:
+        time.sleep(0.05)
+        return True
+    deadline = time.monotonic() + TAP_OBSERVE_TIMEOUT_S
+    while time.monotonic() < deadline:
+        now = parse_stats(ctx.ctl.send("stats")).get(field)
+        if now is not None and now != before:
+            return True
+        time.sleep(0.02)
+    return False
+
+
+def do_observed_tap(ctx, ax, ay):
+    """Wait until the guest consumes both halves of a tap."""
+    before = parse_stats(ctx.ctl.send("stats"))
+    if before.get("presses") is None:
+        do_tap(ctx.s, ctx.f, ax, ay)
+        return True
+
+    send_events(ctx.s, ctx.f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
+    pressed = False
+    try:
+        pressed = wait_input_count(ctx, "presses", before.get("presses"))
+    finally:
+        send_events(ctx.s, ctx.f, [btn_ev(False)])
+    if not pressed:
+        return False
+    return wait_input_count(ctx, "releases", before.get("releases"))
+
+
+def _move_swipe(s, f, ax1, ay1, ax2, ay2, ms, steps):
+    for i in range(1, steps + 1):
+        t = i / steps
+        send_events(s, f, [
+            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
+            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
+        ])
+        time.sleep(ms / 1000.0 / steps)
+
+
 def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
     """Drag with interpolated motion.
 
@@ -230,15 +279,32 @@ def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
     if steps is None:
         steps = max(6, int(ms / 25))
     send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
-    time.sleep(0.05)
-    for i in range(1, steps + 1):
-        t = i / steps
-        send_events(s, f, [
-            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
-            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
-        ])
-        time.sleep(ms / 1000.0 / steps)
-    send_events(s, f, [btn_ev(False)])
+    try:
+        time.sleep(0.05)
+        _move_swipe(s, f, ax1, ay1, ax2, ay2, ms, steps)
+    finally:
+        send_events(s, f, [btn_ev(False)])
+
+
+def do_observed_swipe(ctx, x1, y1, x2, y2, ms=400, steps=None):
+    """Start a drag only after the guest has consumed its press."""
+    ax1, ay1 = to_axis(x1, ctx.size), to_axis(y1, ctx.size)
+    ax2, ay2 = to_axis(x2, ctx.size), to_axis(y2, ctx.size)
+    if steps is None:
+        steps = max(6, int(ms / 25))
+
+    before = parse_stats(ctx.ctl.send("stats"))
+    send_events(ctx.s, ctx.f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
+    pressed = False
+    try:
+        pressed = wait_input_count(ctx, "presses", before.get("presses"))
+        if not pressed:
+            return False
+
+        _move_swipe(ctx.s, ctx.f, ax1, ay1, ax2, ay2, ms, steps)
+    finally:
+        send_events(ctx.s, ctx.f, [btn_ev(False)])
+    return wait_input_count(ctx, "releases", before.get("releases"))
 
 
 class Ctl:
@@ -527,11 +593,40 @@ def eval_json(ctl, path, op, value):
     return ok, "" if ok else f"{path} is {actual!r}"
 
 
+# The fields assert_stat/wait_stat expose. parse_stats() itself recognizes a
+# couple more (STATS_FIELD_RE has 'releases') that these verbs don't -- this
+# is the whitelist both verbs enforce, not everything a `stats` reply carries.
+STAT_FIELDS = ("cpu", "fps", "render", "idle", "presses", "termbusy", "termintr", "termfg", "termsig")
+
+
+def eval_stat(ctl, field, op, value):
+    """One `stats` fetch + field-lookup + op-apply round for
+    assert_stat/wait_stat. -> (ok, detail); detail is empty on success,
+    otherwise the reason."""
+    if field not in STAT_FIELDS:
+        return False, (f"unknown stat field {field!r} "
+                        f"(known: {', '.join(STAT_FIELDS)})")
+    stats = parse_stats(ctl.send("stats"))
+    if field not in stats:
+        return False, "no such field"
+    try:
+        ok = apply_op(op, stats[field], value)
+    except (TypeError, ValueError) as e:
+        return False, str(e)
+    return ok, "" if ok else f"{field} is {stats[field]}"
+
+
 STATS_FIELD_RE = {
     "cpu": re.compile(r'^cpu:\s*(-?\d+(?:\.\d+)?)%?\s*$'),
     "fps": re.compile(r'^fps:\s*(-?\d+(?:\.\d+)?)\s*$'),
     "render": re.compile(r'^render:\s*(-?\d+(?:\.\d+)?)\s*ms/frame\s*$'),
     "idle": re.compile(r'^idle:\s*(\d+)\s*$'),
+    "presses": re.compile(r'^presses:\s*(\d+)\s*$'),
+    "releases": re.compile(r'^releases:\s*(\d+)\s*$'),
+    "termbusy": re.compile(r'^termbusy:\s*(\d+)\s*$'),
+    "termintr": re.compile(r'^termintr:\s*(\d+)\s*$'),
+    "termfg": re.compile(r'^termfg:\s*(-?\d+)\s*$'),
+    "termsig": re.compile(r'^termsig:\s*(\d+)\s*$'),
 }
 
 
@@ -689,20 +784,24 @@ def verb_shot(ctx, lineno, cmd, args, line):
 
 def verb_tap(ctx, lineno, cmd, args, line):
     x, y = int(args[0]), int(args[1])
-    do_tap(ctx.s, ctx.f, to_axis(x, ctx.size), to_axis(y, ctx.size))
-    return "ok", ""
+    observed = do_observed_tap(ctx, to_axis(x, ctx.size), to_axis(y, ctx.size))
+    return (("ok", "") if observed else
+            ("fail", "input press or release was not consumed within 2 seconds"))
 
 
 def verb_swipe(ctx, lineno, cmd, args, line):
     ms = int(args[4]) if len(args) > 4 else 400
-    do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ctx.size, ms)
-    return "ok", ""
+    observed = do_observed_swipe(
+        ctx, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ms)
+    return (("ok", "") if observed else
+            ("fail", "input press or release was not consumed within 2 seconds"))
 
 
 def verb_fling(ctx, lineno, cmd, args, line):
-    do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
-             ctx.size, ms=120)
-    return "ok", ""
+    observed = do_observed_swipe(
+        ctx, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ms=120)
+    return (("ok", "") if observed else
+            ("fail", "input press or release was not consumed within 2 seconds"))
 
 
 def verb_sleep(ctx, lineno, cmd, args, line):
@@ -837,18 +936,26 @@ def verb_wait_json(ctx, lineno, cmd, args, line):
 def verb_assert_stat(ctx, lineno, cmd, args, line):
     # assert_stat FIELD OP VALUE: FIELD off a fresh `stats` reply.
     ctx.need_ctl(lineno, cmd)
-    field, op, value = args[0], args[1], args[2]
-    if field not in ("cpu", "fps", "render", "idle"):
-        return "fail", (f"unknown stat field {field!r} "
-                         f"(known: cpu, fps, render, idle)")
-    stats = parse_stats(ctx.ctl.send("stats"))
-    if field not in stats:
-        return "fail", "no such field"
-    try:
-        ok = apply_op(op, stats[field], value)
-    except (TypeError, ValueError) as e:
-        return "fail", str(e)
-    return ("ok" if ok else "fail"), ("" if ok else f"{field} is {stats[field]}")
+    ok, detail = eval_stat(ctx.ctl, args[0], args[1], args[2])
+    return ("ok" if ok else "fail"), detail
+
+
+def verb_wait_stat(ctx, lineno, cmd, args, line):
+    # wait_stat FIELD OP VALUE TIMEOUT_S: eval_stat polled every 0.5s
+    # (poll_until) until it holds or the deadline passes. warden-ui's fps
+    # counter is a rolling one-second window (warden_debug.c): a single
+    # sample taken right after a page opens can read 0 even though the UI
+    # is live and about to report a real rate, on a runner slow enough that
+    # window hasn't filled yet (flare-edge #44). This exists for
+    # exactly that shape of check, the same reason wait_json exists for the
+    # asynchronously-written status file.
+    ctx.need_ctl(lineno, cmd)
+    field, op, value, timeout_s = args[0], args[1], args[2], float(args[3])
+    ok, detail, waited, fatal = poll_until(
+        lambda: eval_stat(ctx.ctl, field, op, value), timeout_s)
+    if fatal:
+        return "fatal", detail
+    return ("ok" if ok else "fail"), (f"waited {waited:.1f}s" if ok else detail)
 
 
 def verb_capture_region(ctx, lineno, cmd, args, line):
@@ -986,6 +1093,7 @@ VERBS = {
     "assert_json": verb_assert_json,
     "wait_json": verb_wait_json,
     "assert_stat": verb_assert_stat,
+    "wait_stat": verb_wait_stat,
     "capture_region": verb_capture_region,
     "assert_region": verb_assert_region,
     "wait_region": verb_wait_region,
@@ -1041,12 +1149,12 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref
             # reason: several verbs parse their own arguments with bare
             # int()/float()/positional indexing before any handler-local
             # guard (tap, swipe, fling, sleep, wait_hit, wait_json,
-            # capture_region, wait_region), and a malformed or missing
-            # argument -- a typo'd coordinate, a hand-edited *.txt script, a
-            # future flowc.py bug -- used to raise straight out of drive()
-            # and silently drop every row from that line onward, the exact
-            # truncated-run failure mode this file exists to rule out
-            # (flare-edge #244).
+            # wait_stat, capture_region, wait_region), and a malformed or
+            # missing argument -- a typo'd coordinate, a hand-edited *.txt
+            # script, a future flowc.py bug -- used to raise straight out
+            # of drive() and silently drop every row from that line
+            # onward, the exact truncated-run failure mode this file
+            # exists to rule out (flare-edge #244).
             # Recorded the same as any other fatal row: the run keeps going
             # past it.
             result = ("fatal", str(e))

qemu/tests/test_qmp_drive.py

@@ -22,6 +22,7 @@ import tempfile
 import threading
 import time
 import unittest
+from types import SimpleNamespace
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, HERE)
@@ -39,12 +40,17 @@ class FakeCtl:
 
     def __init__(self, path, timeout=15.0):
         self.path = path
+        self.stats_calls = 0
 
     def send(self, cmd):
         if cmd == "page":
             return "Demo/Rows"
         if cmd == "stats":
-            return "page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\nrga: 0%\nidle: 0"
+            self.stats_calls += 1
+            return ("page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\n"
+                    f"rga: 0%\nidle: 0\npresses: {self.stats_calls}\n"
+                    f"releases: {self.stats_calls}\ntermbusy: 1\n"
+                    "termintr: 2\ntermfg: -1\ntermsig: 3")
         if cmd == "wake":
             return "wake: ok"
         if cmd == "home":
@@ -212,6 +218,127 @@ def run_script(text, refs=None, rs485_control=None, ctl_cls=None, rpc_fn=None, c
 
 
 class PureHelpers(unittest.TestCase):
+    def test_tap_holds_press_across_loaded_guest_polls(self):
+        calls = []
+        saved_send, saved_sleep = qmp.send_events, qmp.time.sleep
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda seconds: calls.append(seconds)
+        try:
+            qmp.do_tap(None, None, 123, 456)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertEqual(calls[0], [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)])
+        self.assertGreaterEqual(calls[1], 0.25)
+        self.assertLess(calls[1], 0.4)
+        self.assertEqual(calls[2], [qmp.btn_ev(False)])
+
+    def test_observed_tap_releases_after_guest_consumes_press(self):
+        calls = []
+
+        class ObservingCtl:
+            def __init__(self):
+                self.calls = 0
+
+            def send(self, cmd):
+                self.calls += 1
+                presses = 0 if self.calls < 3 else 1
+                releases = 0 if self.calls < 4 else 1
+                return f"presses: {presses}\nreleases: {releases}"
+
+        saved_send, saved_sleep = qmp.send_events, qmp.time.sleep
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        try:
+            observed = qmp.do_observed_tap(
+                SimpleNamespace(s=None, f=None, ctl=ObservingCtl()), 123, 456)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertTrue(observed)
+        self.assertEqual(calls, [
+            [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)],
+            [qmp.btn_ev(False)],
+        ])
+
+    def test_observed_tap_timeout_still_releases_press(self):
+        calls = []
+        clock = iter((0.0, 0.0, qmp.TAP_OBSERVE_TIMEOUT_S + 0.1))
+
+        class UnobservingCtl:
+            def send(self, cmd):
+                return "presses: 0\nreleases: 0"
+
+        saved_send, saved_sleep, saved_monotonic = (
+            qmp.send_events, qmp.time.sleep, qmp.time.monotonic)
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        qmp.time.monotonic = lambda: next(clock)
+        try:
+            observed = qmp.do_observed_tap(
+                SimpleNamespace(s=None, f=None, ctl=UnobservingCtl()), 123, 456)
+        finally:
+            qmp.send_events = saved_send
+            qmp.time.sleep = saved_sleep
+            qmp.time.monotonic = saved_monotonic
+
+        self.assertFalse(observed)
+        self.assertEqual(calls[-1], [qmp.btn_ev(False)])
+
+    def test_observed_swipe_moves_after_guest_consumes_press(self):
+        calls = []
+
+        class ObservingCtl:
+            def __init__(self):
+                self.calls = 0
+
+            def send(self, cmd):
+                self.calls += 1
+                presses = 0 if self.calls < 3 else 1
+                releases = 0 if self.calls < 4 else 1
+                return f"presses: {presses}\nreleases: {releases}"
+
+        saved_send, saved_sleep = qmp.send_events, qmp.time.sleep
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        try:
+            observed = qmp.do_observed_swipe(
+                SimpleNamespace(s=None, f=None, ctl=ObservingCtl(), size=720),
+                10, 20, 110, 120, ms=50, steps=2)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertTrue(observed)
+        self.assertEqual(calls[0][-1], qmp.btn_ev(True))
+        self.assertEqual(calls[-1], [qmp.btn_ev(False)])
+        self.assertEqual(len(calls), 4)
+
+    def test_observed_swipe_timeout_still_releases_press(self):
+        calls = []
+        clock = iter((0.0, 0.0, qmp.TAP_OBSERVE_TIMEOUT_S + 0.1))
+
+        class UnobservingCtl:
+            def send(self, cmd):
+                return "presses: 0\nreleases: 0"
+
+        saved_send, saved_sleep, saved_monotonic = (
+            qmp.send_events, qmp.time.sleep, qmp.time.monotonic)
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        qmp.time.monotonic = lambda: next(clock)
+        try:
+            observed = qmp.do_observed_swipe(
+                SimpleNamespace(s=None, f=None, ctl=UnobservingCtl(), size=720),
+                10, 20, 110, 120, ms=50, steps=2)
+        finally:
+            qmp.send_events = saved_send
+            qmp.time.sleep = saved_sleep
+            qmp.time.monotonic = saved_monotonic
+
+        self.assertFalse(observed)
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(calls[-1], [qmp.btn_ev(False)])
+
     def test_every_pixel_round_trips_through_lvgl_calibration(self):
         # lv_evdev.c _evdev_calibrate: px = axis * (width - 1) / AXIS_MAX,
         # integer division. A tap requested at px must land at px, for every
@@ -255,7 +382,28 @@ class PureHelpers(unittest.TestCase):
 
     def test_parse_stats(self):
         got = qmp.parse_stats(FakeCtl("x").send("stats"))
-        self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2, "idle": 0.0})
+        self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2,
+                               "idle": 0.0, "presses": 1.0, "releases": 1.0,
+                               "termbusy": 1.0, "termintr": 2.0,
+                               "termfg": -1.0, "termsig": 3.0})
+
+    def test_eval_stat_rejects_unknown_fields_and_catches_bad_ops(self):
+        # eval_stat's own whitelist is narrower than parse_stats(): 'releases'
+        # is a real field in a `stats` reply (see test_parse_stats) but not
+        # one assert_stat/wait_stat expose. And it must catch apply_op's
+        # (TypeError, ValueError) the same way eval_json does, so a malformed
+        # OP reads as an ordinary fail, not a driver crash, for both the
+        # single-read and the polling verb built on top of it.
+        ctl = FakeCtl("x")
+        ok, detail = qmp.eval_stat(ctl, "releases", "eq", "1")
+        self.assertFalse(ok)
+        self.assertIn("unknown stat field", detail)
+        ok, detail = qmp.eval_stat(ctl, "fps", "bogus", "1")
+        self.assertFalse(ok)
+        self.assertIn("bogus", detail)
+        ok, detail = qmp.eval_stat(ctl, "fps", "eq", "10")
+        self.assertTrue(ok)
+        self.assertEqual(detail, "")
 
     def test_poll_until_turns_a_channel_fault_into_fatal_not_a_raise(self):
         # A CHECK that raises RuntimeError or OSError (Ctl.send on EOF or a
@@ -295,6 +443,7 @@ class DriveVerbs(unittest.TestCase):
             "wait_json list len_ge 2 2\n"
             "assert_json name eq warden\n"
             "assert_stat fps gt 0\n"
+            "wait_stat fps gt 0 2\n"
             "nav Demo/Rows\n"
             "capture_region r1 0 0 8 8 exact\n"
             "assert_region r1 exact\n"
@@ -305,8 +454,9 @@ class DriveVerbs(unittest.TestCase):
             "home\n"
         )
         self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"])
-        self.assertEqual(len(rows), 16)
+        self.assertEqual(len(rows), 17)
         self.assertIn("waited", by["wait_hit 47 676 obj 2 box=12,640,72x72"]["detail"])
+        self.assertIn("waited", by["wait_stat fps gt 0 2"]["detail"])
         self.assertTrue(all(r["status"] == "ok" for r in rows))
 
     def test_mismatches_are_fails_not_stops(self):
@@ -317,17 +467,19 @@ class DriveVerbs(unittest.TestCase):
             "wait_json a.b eq 2 1\n"
             "assert_json a.zz eq 1\n"
             "assert_stat fps lt 0\n"
+            "wait_stat fps lt 0 1\n"
             "scroll 360 400 0\n"
             "wait_hit 47 676 obj 1 box=0,0,1x1\n"
             "assert_page Demo/Rows\n"
         )
         self.assertEqual(rc, 1)
         self.assertEqual([r["status"] for r in rows],
-                         ["fail", "fail", "fail", "fail", "fail", "fail", "fail", "ok"])
+                         ["fail", "fail", "fail", "fail", "fail", "fail", "fail", "fail", "ok"])
         self.assertIn("moved", by["assert_hit 47 676 obj box=0,0,1x1"]["detail"])
         self.assertIn("moved", by["wait_hit 47 676 obj 1 box=0,0,1x1"]["detail"])
-        self.assertGreaterEqual(time.monotonic() - t0, 2.0,
-                                "wait_json and wait_hit must each honour their timeout")
+        self.assertIn("fps is 10.0", by["wait_stat fps lt 0 1"]["detail"])
+        self.assertGreaterEqual(time.monotonic() - t0, 3.0,
+                                "wait_json, wait_stat and wait_hit must each honour their timeout")
 
     def test_dying_channel_is_fatal_for_the_step_not_a_crash(self):
         # A page still building when the control channel itself drops mid-
@@ -354,13 +506,23 @@ class DriveVerbs(unittest.TestCase):
         self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
         self.assertIn("control channel closed", by["wait_hit 47 676 obj 2"]["detail"])
 
+        rc, by, rows = run_script(
+            "wait_stat fps gt 0 2\n"
+            "sleep 0\n",
+            ctl_cls=DyingCtl,
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
+        self.assertIn("control channel closed", by["wait_stat fps gt 0 2"]["detail"])
+
     def test_dying_channel_is_fatal_for_every_verb_not_just_wait(self):
-        # wait_hit/wait_json route the channel through poll_until, which has
-        # always caught this. Every other verb that reaches ctl.send()
-        # directly did not, and used to crash the whole run instead of
-        # recording one fatal row and moving on. One representative of each
-        # family, back to back: every one must read as its own `fatal` row
-        # and the script must still reach the last line.
+        # wait_hit/wait_json/wait_stat route the channel through poll_until,
+        # which has always caught this. Every other verb that reaches
+        # ctl.send() directly did not, and used to crash the whole run
+        # instead of recording one fatal row and moving on. One
+        # representative of each family, back to back: every one must read
+        # as its own `fatal` row and the script must still reach the last
+        # line.
         rc, by, rows = run_script(
             "nav Demo/Rows\n"
             "wake\n"
@@ -370,12 +532,13 @@ class DriveVerbs(unittest.TestCase):
             "assert_hit 47 676 obj\n"
             "assert_json a.b eq 1\n"
             "assert_stat idle eq 0\n"
+            "wait_stat fps gt 0 1\n"
             "sleep 0\n",
             ctl_cls=ImmediateDyingCtl,
         )
         self.assertEqual(rc, 1)
         self.assertEqual([r["status"] for r in rows],
-                         ["fatal"] * 8 + ["ok"],
+                         ["fatal"] * 9 + ["ok"],
                          "a dead channel must not crash the run: every ctl "
                          "verb gets its own fatal row and sleep still runs")
         for row in rows[:-1]:
@@ -404,12 +567,12 @@ class DriveVerbs(unittest.TestCase):
     def test_malformed_numeric_argument_is_fatal_for_the_step_not_a_crash(self):
         # Several verbs parse their own arguments with bare
         # int()/float()/positional indexing before any handler-local guard
-        # (tap, swipe, fling, sleep, wait_hit, wait_json, capture_region,
-        # wait_region). A typo'd coordinate or a missing argument -- exactly
-        # what a hand-edited *.txt script or a flowc.py bug can produce --
-        # used to raise ValueError/IndexError straight out of drive(),
-        # losing every row from that line onward instead of reading as its
-        # own fatal row (flare-edge #244).
+        # (tap, swipe, fling, sleep, wait_hit, wait_json, wait_stat,
+        # capture_region, wait_region). A typo'd coordinate or a missing
+        # argument -- exactly what a hand-edited *.txt script or a flowc.py
+        # bug can produce -- used to raise ValueError/IndexError straight
+        # out of drive(), losing every row from that line onward instead of
+        # reading as its own fatal row (flare-edge #244).
         rc, by, rows = run_script(
             "tap 10 abc\n"
             "sleep 0\n",
@@ -435,6 +598,15 @@ class DriveVerbs(unittest.TestCase):
         self.assertIn("invalid literal",
                       by["capture_region r1 0 0 8 notanumber exact"]["detail"])
 
+        rc, by, rows = run_script(
+            "wait_stat fps gt 0 notanumber\n"
+            "sleep 0\n",
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
+        self.assertIn("could not convert string to float",
+                      by["wait_stat fps gt 0 notanumber"]["detail"])
+
     def test_bad_op_reads_as_a_fail_row_not_a_crash(self):
         # apply_op's error paths (unknown OP -> ValueError, 'contains'
         # against the wrong type -> TypeError) are caught by both callers