Stop the OTG state machines running blind in either role

Widens the #160 fix, which was scoped too narrowly to host role, and fixes
flare-edge#168.

Device role was deliberately left alone on the reasoning that the state
machine is only meaningless when the port is a host. That was wrong. On a
board with no extcon phandle the machine has no external role signal in
EITHER direction, and with bvalid pinned high by the VBUS bypass its charger
detection cannot be meaningful at all. In device role its verdict suspends
the phy underneath a bound gadget, so the USB recovery link dies.

Reproduced deterministically on warden-c8a3 with a host attached:

  device role, host attached   ccf=1  phy_sus=0x0    udc=configured  works
  -> host role                 ccf=1  phy_sus=0x0    udc=(none)
  -> back to gadget            ccf=0  phy_sus=0x1d1  udc=configured  SUSPENDED
  -> +15s / +30s               ccf=0  phy_sus=0x1d1  udc=configured  never recovers

and the 2207:0019 device disappears from the host for good.

Nothing notices, because every obvious signal lies: udc state reads
"configured", usb0 carrier reads 1, and DCTL RUN_STOP reads 1, so the gadget
really is asserting its pullup. It asserts into a phy whose analog front end
and 480M clock are off, so the SoC sees SE0 and the host sees no device at
all, not even an enumeration error.

It cannot self-heal either. bvalid is pinned high, so a host attaching later
produces no VBUS edge for the machine to trigger on: it ran once, powered the
phy off, and nothing re-evaluates. That is why a reboot appears to fix it and
why this read as a cable fault for weeks. It is not, and the cables were fine.

Gated on the extcon being self-allocated rather than on the board compatible,
so it describes the actual precondition: no external role signal exists.

Renamed from 29-usb2phy-no-otg-sm-in-host.patch, whose name now misstated the
scope.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MVGTC78dgCGfRANNKjoPea

kernel/rv1106-enablement/usb/rv1106_usb2phy_cfg.c.frag

@@ -1,12 +1,14 @@
  * RV1106 USB2 PHY: single OTG port at 0xff3e0000 (ported from the vendor driver;
  * fields map 1:1 to 6.18 except utmi_iddig -> utmi_id, and the 5.10-only
- * iddig_output/iddig_en/bvalid_grf_sel are dropped). Signal-quality phy_tuning is
- * left off for now (guarded, so NULL is safe): the phy is functional without it.
+ * iddig_output/iddig_en/bvalid_grf_sel are dropped). Signal-quality phy_tuning
+ * was originally left off here; it is now implemented (patches/27-usb-phy-tuning.patch)
+ * because "functional without it" turned out not to mean reliable -- see #160.
  */
 static const struct rockchip_usb2phy_cfg rv1106_phy_cfgs[] = {
 	{
 		.reg = 0xff3e0000,
 		.num_ports	= 1,
+		.phy_tuning	= rv1106_usb2phy_tuning,
 		.clkout_ctl	= { 0x0058, 4, 4, 1, 0 },
 		.port_cfgs	= {
 			[USB2PHY_PORT_OTG] = {

patches/27-usb-phy-tuning.patch

@@ -0,0 +1,128 @@
+diff --git a/drivers/phy/rockchip/phy-rockchip-inno-usb2.c b/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
+--- a/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
++++ b/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
+@@ -247,6 +247,7 @@
+ 	struct device	*dev;
+ 	struct regmap	*grf;
+ 	struct regmap	*usbgrf;
++	void __iomem	*phy_base;
+ 	struct clk_bulk_data	*clks;
+ 	struct clk	*clk480m;
+ 	struct clk_hw	clk480m_hw;
+@@ -1417,6 +1418,19 @@
+ 		return -EINVAL;
+ 	}
+ 
++	/*
++	 * RV1106 tunes signal quality through the PHY's own register block, which
++	 * is this node's "reg" resource and is distinct from the GRF regmap above.
++	 * Map it only where a tuning callback needs it; every other SoC here tunes
++	 * through the GRF and must keep working with phy_base left NULL.
++	 */
++	if (of_device_is_compatible(np, "rockchip,rv1106-usb2phy")) {
++		rphy->phy_base = devm_platform_ioremap_resource(pdev, 0);
++		if (IS_ERR(rphy->phy_base))
++			return dev_err_probe(dev, PTR_ERR(rphy->phy_base),
++					     "failed to map phy registers\n");
++	}
++
+ 	rphy->phy_reset = devm_reset_control_get_optional(dev, "phy");
+ 	if (IS_ERR(rphy->phy_reset))
+ 		return PTR_ERR(rphy->phy_reset);
+@@ -1506,6 +1520,30 @@
+ 	return ret;
+ }
+ 
++/*
++ * Direct PHY-register accessors. Most SoCs here tune through the GRF regmap,
++ * but the RV1106's signal-quality registers live in the PHY's own block, which
++ * is mapped separately into rphy->phy_base.
++ */
++static inline void phy_update_bits(void __iomem *reg, u32 mask, u32 val)
++{
++	u32 tmp = readl(reg);
++
++	tmp &= ~mask;
++	tmp |= val & mask;
++	writel(tmp, reg);
++}
++
++static inline void phy_set_bits(void __iomem *reg, u32 bits)
++{
++	writel(readl(reg) | bits, reg);
++}
++
++static inline void phy_clear_bits(void __iomem *reg, u32 bits)
++{
++	writel(readl(reg) & ~bits, reg);
++}
++
+ static int rk3128_usb2phy_tuning(struct rockchip_usb2phy *rphy)
+ {
+ 	/* Turn off differential receiver in suspend mode */
+@@ -2291,10 +2329,65 @@
+  * iddig_output/iddig_en/bvalid_grf_sel are dropped). Signal-quality phy_tuning is
+  * left off for now (guarded, so NULL is safe) — the phy is functional without it.
+  */
++/*
++ * RV1106 USB2 PHY signal-quality tuning, ported from the vendor 5.10 driver
++ * (drivers/phy/rockchip/phy-rockchip-inno-usb2.c, rv1106_usb2phy_tuning).
++ *
++ * The 6.18 forward-port originally omitted this ("the phy is functional without
++ * it"). It is functional, but not reliable: with these registers left at reset
++ * defaults the HS link is electrically marginal under sustained traffic. On a
++ * bench panel a USB-Ethernet dongle would stall a bulk transfer after a few
++ * minutes; the r8152 Tx-timeout recovery then issues an xHCI Stop Endpoint
++ * command which the controller never answers, and xhci-hcd declares the
++ * controller dead and tears the bus down (flare-edge issue #160). Measured
++ * carrier-on to death across four runs: 14s, 519s, 521s, 584s -- a stochastic
++ * spread, not a timer, which is what marginal signalling looks like.
++ *
++ * Pre-emphasis strength is silicon-revision dependent. The vendor driver picks
++ * it from rockchip_get_cpu_version(), which does not exist in mainline; it
++ * derives from RV1106_OS_REG1 (0xff020204) & 0x7. Our panels read 1, i.e. the
++ * non-zero branch, so 3'b001 is pinned here. If a board ever reports version 0
++ * it wants 3'b011 instead -- see the vendor driver.
++ */
++static int rv1106_usb2phy_tuning(struct rockchip_usb2phy *rphy)
++{
++	/* Mapped only for this compatible; refuse to poke a NULL base. */
++	if (!rphy->phy_base)
++		return -ENODEV;
++
++	/* Always enable pre-emphasis in SOF & EOP & chirp & non-chirp state */
++	phy_update_bits(rphy->phy_base + 0x30, GENMASK(2, 0), 0x07);
++
++	/* Set Tx HS pre_emphasize strength to 3'b001 (cpu_version != 0) */
++	phy_update_bits(rphy->phy_base + 0x40, GENMASK(5, 3), (0x01 << 3));
++
++	/* Set RX Squelch trigger point configure to 4'b0000(112.5 mV) */
++	phy_update_bits(rphy->phy_base + 0x64, GENMASK(6, 3), (0x00 << 3));
++
++	/* Turn off differential receiver by default to save power */
++	phy_clear_bits(rphy->phy_base + 0x100, BIT(6));
++
++	/* Set 45ohm HS ODT value to 5'b10111 to increase driver strength */
++	phy_update_bits(rphy->phy_base + 0x11c, GENMASK(4, 0), 0x17);
++
++	/* Set Tx HS eye height tuning to 3'b011(462 mV) */
++	phy_update_bits(rphy->phy_base + 0x124, GENMASK(4, 2), (0x03 << 2));
++
++	/* Bypass Squelch detector calibration */
++	phy_update_bits(rphy->phy_base + 0x1a4, GENMASK(7, 4), (0x01 << 4));
++	phy_update_bits(rphy->phy_base + 0x1b4, GENMASK(7, 4), (0x01 << 4));
++
++	/* Set HS disconnect detect mode to single ended detect mode */
++	phy_set_bits(rphy->phy_base + 0x70, BIT(2));
++
++	return 0;
++}
++
+ static const struct rockchip_usb2phy_cfg rv1106_phy_cfgs[] = {
+ 	{
+ 		.reg = 0xff3e0000,
+ 		.num_ports	= 1,
++		.phy_tuning	= rv1106_usb2phy_tuning,
+ 		.clkout_ctl	= { 0x0058, 4, 4, 1, 0 },
+ 		.port_cfgs	= {
+ 			[USB2PHY_PORT_OTG] = {

patches/29-usb2phy-no-blind-otg-sm.patch

@@ -0,0 +1,136 @@
+diff --git a/drivers/phy/rockchip/phy-rockchip-inno-usb2.c b/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
+--- a/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
++++ b/drivers/phy/rockchip/phy-rockchip-inno-usb2.c
+@@ -203,6 +203,7 @@
+  * @event_nb: hold event notification callback.
+  * @state: define OTG enumeration states before device reset.
+  * @mode: the dr_mode of the controller.
++ * @host_mode: the port is currently acting as a USB host.
+  */
+ struct rockchip_usb2phy_port {
+ 	struct phy	*phy;
+@@ -222,6 +223,7 @@
+ 	struct notifier_block	event_nb;
+ 	enum usb_otg_state	state;
+ 	enum usb_dr_mode	mode;
++	bool		host_mode;
+ };
+ 
+ /**
+@@ -239,6 +241,8 @@
+  * @dcd_retries: The retry count used to track Data contact
+  *		 detection process.
+  * @edev: extcon device for notification registration
++ * @edev_self_allocated: no extcon phandle in DT, so the OTG state machine
++ *			has no external role signal at all
+  * @irq: muxed interrupt for single irq configuration
+  * @phy_cfg: phy register configuration, assigned by driver data.
+  * @ports: phy port instance.
+@@ -257,6 +261,7 @@
+ 	enum power_supply_type	chg_type;
+ 	u8			dcd_retries;
+ 	struct extcon_dev	*edev;
++	bool			edev_self_allocated;
+ 	int			irq;
+ 	const struct rockchip_usb2phy_cfg	*phy_cfg;
+ 	struct rockchip_usb2phy_port	ports[USB2PHY_NUM_PORTS];
+@@ -459,6 +464,14 @@
+ 		if (ret)
+ 			return dev_err_probe(rphy->dev, ret,
+ 					     "failed to register extcon device\n");
++
++		/*
++		 * Nothing external drives the role on this board. The extcon
++		 * below is ours, and no code ever sets a cable state on it, so
++		 * the OTG state machine's view of the world is permanently
++		 * blank rather than merely stale.
++		 */
++		rphy->edev_self_allocated = true;
+ 	}
+ 
+ 	rphy->edev = edev;
+@@ -649,11 +662,52 @@
+ 	return 0;
+ }
+ 
++/*
++ * The OTG port's state machines only make sense while the port is a
++ * peripheral. In host role their inputs are meaningless on a board with no
++ * ID source: charger detection drives the analog front end of a port that is
++ * already enumerating a device, and its DCP verdict ends in
++ * rockchip_usb2phy_power_off(), which suspends the phy and drops the 480M
++ * clkout that the host controller's core runs on. The controller then stops
++ * clocking with its registers still readable, so it reports itself healthy
++ * while no transfer ever completes.
++ *
++ * dwc3 already calls phy_set_mode() on every role change, but this driver
++ * implemented no .set_mode, so the phy never learned the role. Track it and
++ * keep both state machines out of host role.
++ */
++static int rockchip_usb2phy_set_mode(struct phy *phy, enum phy_mode mode,
++				     int submode)
++{
++	struct rockchip_usb2phy_port *rport = phy_get_drvdata(phy);
++
++	if (rport->port_id != USB2PHY_PORT_OTG)
++		return 0;
++
++	switch (mode) {
++	case PHY_MODE_USB_HOST:
++		if (!rport->host_mode) {
++			rport->host_mode = true;
++			cancel_delayed_work_sync(&rport->otg_sm_work);
++			cancel_delayed_work_sync(&rport->chg_work);
++		}
++		break;
++	case PHY_MODE_USB_DEVICE:
++		rport->host_mode = false;
++		break;
++	default:
++		break;
++	}
++
++	return 0;
++}
++
+ static const struct phy_ops rockchip_usb2phy_ops = {
+ 	.init		= rockchip_usb2phy_init,
+ 	.exit		= rockchip_usb2phy_exit,
+ 	.power_on	= rockchip_usb2phy_power_on,
+ 	.power_off	= rockchip_usb2phy_power_off,
++	.set_mode	= rockchip_usb2phy_set_mode,
+ 	.owner		= THIS_MODULE,
+ };
+ 
+@@ -667,6 +721,15 @@
+ 	unsigned long delay;
+ 	bool vbus_attach, sch_work, notify_charger;
+ 
++	/*
++	 * This machine can only get the answer wrong here. In host role its
++	 * verdict suspends a phy that is enumerating a device (#160); with a
++	 * self-allocated extcon it has no role signal in EITHER direction, and
++	 * in device role it strands the gadget the same way (#168).
++	 */
++	if (rport->host_mode || rphy->edev_self_allocated)
++		return;
++
+ 	vbus_attach = property_enabled(rphy->grf,
+ 				       &rport->port_cfg->utmi_bvalid);
+ 
+@@ -825,6 +888,15 @@
+ 	bool is_dcd, tmout, vout, vbus_attach;
+ 	unsigned long delay;
+ 
++	/*
++	 * Never probe the front end of a port that is hosting a device, and
++	 * never at all on a board whose bvalid is pinned high by the VBUS
++	 * bypass: charger detection cannot be meaningful there, and its DCP
++	 * verdict powers the phy off underneath a working link (#160, #168).
++	 */
++	if (rport->host_mode || rphy->edev_self_allocated)
++		return;
++
+ 	vbus_attach = property_enabled(rphy->grf, &rport->port_cfg->utmi_bvalid);
+ 
+ 	dev_dbg(&rport->phy->dev, "chg detection work state = %d\n",

patches/README.md

@@ -18,6 +18,8 @@ with `../build/warden_defconfig` -> `zImage` + `rv1106-warden.dtb`).
 | `15-pinctrl.patch` | `pinctrl-rockchip` RV1106 iomux |
 | `20-arch-dts-bindings.patch` | `arch/arm` DTs (`rv1106.dtsi`, `rv1106-warden.dts`, pinctrl), mach, configs, `include/dt-bindings` |
 | `25-usb-phy.patch` | inno-usb2 phy |
+| `27-usb-phy-tuning.patch` | inno-usb2 phy: RV1106 signal-quality register tuning |
+| `29-usb2phy-no-blind-otg-sm.patch` | inno-usb2 phy: stop the OTG/charger state machines running blind |
 | `30-drm-vop-panel-rgb.patch` | VOP (`rgb_dclk_pol`=0 + WIN1 scanout), panel-simple bus-format, rockchip_rgb |
 | `35-mailbox-hpmcu.patch` | rockchip-mailbox (rv1106 num_chans=1) |
 | `40-pvtm-soc.patch` | PVTM (core+pmu) + `include/soc/rockchip/pvtm.h` |

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