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/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
@@ -588,6 +593,29 @@ 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*$'),
@@ -908,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", "presses", "termbusy", "termintr", "termfg", "termsig"):
-        return "fail", (f"unknown stat field {field!r} "
-                        f"(known: cpu, fps, render, idle, presses, termbusy, termintr, termfg, termsig)")
-    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):
@@ -1057,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,
@@ -1112,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

@@ -387,6 +387,24 @@ class PureHelpers(unittest.TestCase):
                                "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
         # socket timeout) must stop poll_until() and come back with the
@@ -425,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"
@@ -435,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):
@@ -447,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-
@@ -484,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"
@@ -500,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]:
@@ -534,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",
@@ -565,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