fix: verify work during continuously hidden guest intervals

dev/smoke.py

@@ -35,6 +35,51 @@ def counter(manager, name):
     return values[0]
 
 
+def focus_events(log):
+    """Only keyboard focus, excluding pointer-enter and old window titles."""
+    return re.findall(r'wl_keyboard#\d+\.(enter|leave)\(', log.read_text())
+
+
+def fresh_focus_transition(source_before, target_before, source_after, target_after):
+    return (len(source_after) > len(source_before) and source_after[-1] == 'leave'
+            and len(target_after) > len(target_before) and target_after[-1] == 'enter')
+
+
+def measure_hidden_interval(read_counter, hidden_log, visible_log, duration=.5):
+    hidden_focus, visible_focus = focus_events(hidden_log), focus_events(visible_log)
+    assert hidden_focus and hidden_focus[-1] == 'leave', 'sampled guest is not hidden'
+    assert visible_focus and visible_focus[-1] == 'enter', 'other guest is not focused'
+
+    def unchanged():
+        assert (focus_events(hidden_log) == hidden_focus and
+                focus_events(visible_log) == visible_focus), 'focus changed during hidden interval'
+
+    start = time.monotonic()
+    before = read_counter()
+    unchanged()
+    while time.monotonic() - start < duration:
+        time.sleep(min(.05, duration))
+        unchanged()
+    after = read_counter()
+    unchanged()
+    assert after != before, 'hidden guest stopped working during continuously hidden interval'
+    return {'before': before, 'after': after,
+            'elapsed_seconds': time.monotonic() - start, 'focus_unchanged': True}
+
+
+def classify_compositor_errors(log):
+    expected = []
+    for line in log.splitlines():
+        if '[ERROR]' not in line:
+            continue
+        if re.search(r'\[xwayland/sockets\.c:\d+\] Failed to bind socket '
+                     r'@?/tmp/\.X11-unix/X\d+: Address already in use$', line):
+            expected.append(line)
+        else:
+            raise AssertionError('Unexpected compositor error: ' + line)
+    return expected
+
+
 def main():
     cli = argparse.ArgumentParser(description=__doc__)
     cli.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
@@ -74,7 +119,7 @@ def main():
     (config / 'environment').touch()
     env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), WLR_BACKENDS='headless',
                WLR_RENDERER='pixman', WLR_XWAYLAND='/bin/false', XDG_DATA_DIRS='/nonexistent',
-               GSETTINGS_BACKEND='memory')
+               GSETTINGS_BACKEND='memory', LABWC_UPDATE_ACTIVATION_ENV='0')
     for key in ('DISPLAY', 'WAYLAND_DISPLAY'):
         env.pop(key, None)
     manager = Runtime(root / 'config', root / 'state', root / 'run')
@@ -111,6 +156,15 @@ def main():
         def chord(code):
             keys(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29')
 
+        def switch(key, source_log, target_log):
+            source_before, target_before = focus_events(source_log), focus_events(target_log)
+            chord(key)
+            wait_for(lambda: fresh_focus_transition(source_before, target_before,
+                                                   focus_events(source_log), focus_events(target_log)),
+                     'reserved switch did not produce fresh keyboard leave/enter')
+            return {'from_events': len(focus_events(source_log)),
+                    'to_events': len(focus_events(target_log)), 'keycode': key}
+
         def attach(name, number):
             vl = (root / f'{name}-{number}.log').open('w'); streams.append(vl)
             previous = dict(os.environ)
@@ -150,11 +204,10 @@ def main():
         chord(23)  # I: test-only ToggleKeybinds.
         keys('down 88\nup 88')
         assert not unreserved.exists(), 'keybinding inhibition did not suppress ordinary F12'
-        a_leave = alog.read_text().count('.leave(')
+        switch(106, alog, blog)  # Right: beta focused, alpha continuously hidden.
-        chord(106)  # Right: production reserved host switch.
+        hidden_intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
-        wait_for(lambda: alog.read_text().count('.leave(') > a_leave and '.enter(' in blog.read_text(),
+        switch(105, blog, alog)  # Left: alpha focused, beta continuously hidden.
-                 'reserved switch failed with inhibition')
+        hidden_intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
-        chord(105)  # Left: return to inhibited alpha.
         chord(1)  # Escape: production recovery binding still works.
         wait_for(recovery.exists, 'reserved recovery failed with inhibition')
         dialog = 'set_title("Select USB devices for redirection")'
@@ -164,8 +217,7 @@ def main():
         wait_for(lambda: dialog in alog.read_text(), 'manual USB selector did not open', timeout=3)
         keys('up 22\nup 42\nup 56\nup 29')
         keys('down 1\nup 1')  # Close selector.
-        chord(106)
+        post_usb_switches = [switch(106, alog, blog), switch(105, blog, alog)]
-        chord(105)
         assert alpha.poll() is None and beta.poll() is None
         assert identities == {name: manager.paths(name).pid.read_text().strip() for name in names}, 'switch restarted a guest'
         assert alog.read_text().count('set_title("HydraOS: alpha")') == 1, 'switch reconnected alpha'
@@ -191,7 +243,10 @@ def main():
         time.sleep(1)
         disk_boot_counter = counter(manager, 'alpha')
         assert disk_boot_counter > 0
-        result.update(manual_usb_dialog=True, standard_us_keymap=True,
+        expected_compositor_noise = classify_compositor_errors((root / 'labwc.log').read_text())
+        result.update(hidden_guest_intervals=hidden_intervals, post_usb_switches=post_usb_switches,
+                      expected_compositor_noise=expected_compositor_noise,
+                      manual_usb_dialog=True, standard_us_keymap=True,
                       native_labwc_inhibition=True, reserved_switch_and_recovery=True,
                       qemu_survives_viewer_and_compositor=True, independent_lifecycle=True,
                       guest_counters_before=before, guest_counters_after=after,

docs/verification.md

@@ -18,7 +18,10 @@ The smoke test builds a private input helper with an ordinary evdev/pc105 US
 keyboard map. It boots two tiny counter workloads from separate scratch qcow2
 disks, connects two fullscreen viewers, and checks reserved switching/recovery,
 manual USB selection, explicit reconnect and independent QMP lifecycles.
-Guest memory counters must advance and QEMU PIDs must remain unchanged through
+After each confirmed workspace switch, the hidden guest is sampled across a
+half-second interval with no focus changes, switches or reconnects; the check is
+then repeated with the other guest hidden. `result.json` records both intervals.
+Guest memory counters must advance and QEMU PIDs remain unchanged through
 switching. Viewer protocol logs must show no reconnection during switching.
 A test-only `ToggleKeybinds` action and F12 marker establish native labwc
 keybinding inhibition before checking the production reserved bindings. Those
@@ -28,7 +31,8 @@ extra controls are absent from the production session configuration. See the
 for native `ToggleKeybinds` behavior.
 
 USB acceptance checks the actual selector window title, holds the modifiers
-through opening, releases them inside the modal dialog, then switches again.
+through opening, releases them inside the modal dialog, then asserts fresh
+keyboard leave/enter transitions for both Right and Left switches.
 QEMU's USB device list must stay unchanged without a selection. Both viewer
 USB filters deny automatic hotplug and connection redirection. This check does
 not attach a physical USB device or prove guest driver behavior.
@@ -38,6 +42,17 @@ forwarding an event to the guest. Upstream keyboard grabs remove the GTK
 application accelerator group; without this adaptation, the configured USB
 shortcut reaches the viewer but does not open its selector.
 
+The pinned labwc 0.20.2 runtime has no documented runtime XWayland-disable
+option. This native-Wayland probe sets `WLR_XWAYLAND=/bin/false` so no X server
+is launched, and explicitly disables activation-environment updates. It can
+still log an occupied X socket while wlroots reserves a display number. Only
+`[xwayland/sockets.c] Failed to bind socket ... Address already in use` is
+classified as expected noise in `result.json`; any other compositor `[ERROR]`
+fails the probe. Full logs remain available. This classification does not
+suppress viewer errors or compiler/test diagnostics. The viewer build's upstream
+Meson boolean-check and compiler-warning-option guidance are build noise; the four-test
+result, compiler errors and process exit status remain the build acceptance gate.
+
 ## Installed guest workflow
 
 `--iso` is optional. Attach or detach media only while the VM is stopped:
@@ -78,8 +93,11 @@ Only `add` reads distribution defaults. Explicit `--network`, `--bridge`,
 not change. `--defaults PATH` selects another preferences file for development.
 Missing defaults use developer preferences; unknown or invalid fields fail
 before creating a profile. Dimensions are integers from 1 to 16384, refresh is
-an integer from 0 to 1000000 mHz. Nonzero refresh requires the tracked patched
+zero for automatic refresh or an integer from 50000 to 255000 mHz, matching
-QEMU provided by `dev/tools.nix` and distribution packaging.
+the vertical range descriptor limits. This bound does not establish every
+resolution/refresh combination; pixel-clock and horizontal timing limits are
+mode-dependent and the verified target remains 1920×1080/144000 mHz. Nonzero
+refresh requires the tracked patched QEMU provided by `dev/tools.nix` and distribution packaging.
 
 The patch keeps the requested refresh through frontend resizing and extends
 EDID monitor ranges to include the generated preferred mode. To verify the

hydra/profiles.py

@@ -40,8 +40,9 @@ class Profile:
         for key in ('width', 'height'):
             if type(getattr(self, key)) is not int or not 1 <= getattr(self, key) <= 16384:
                 raise ValueError(f'{key} must be an integer between 1 and 16384')
-        if type(self.refresh_millihz) is not int or not 0 <= self.refresh_millihz <= 1000000:
+        if type(self.refresh_millihz) is not int or not (self.refresh_millihz == 0 or
-            raise ValueError('refresh_millihz must be an integer between 0 and 1000000')
+                                                       50000 <= self.refresh_millihz <= 255000):
+            raise ValueError('refresh_millihz must be 0 (automatic) or an integer between 50000 and 255000')
         if not isinstance(self.disk_size, str) or not re.fullmatch(r'[1-9][0-9]{0,5}[MGT]', self.disk_size):
             raise ValueError('disk_size must be a positive size with M, G or T suffix')
         if self.graphics not in ('auto', 'software'):

tests/test_profiles.py

@@ -65,6 +65,16 @@ class ProfilesTests(unittest.TestCase):
                 with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                     Profile('guest', self.iso, **{field: value})
 
+    def test_refresh_vertical_descriptor_boundaries(self):
+        for value in (0, 50000, 255000):
+            path = self.root / 'guest.toml'
+            save_profile(path, Profile('guest', refresh_millihz=value))
+            self.assertEqual(load_profile(path).refresh_millihz, value)
+            path.unlink()
+        for value in (1, 30000, 49999, 255001, 1000000):
+            with self.subTest(refresh=value), self.assertRaises(ValueError):
+                Profile('guest', refresh_millihz=value)
+
     def test_profile_name_must_match_filename(self):
         path = self.root / 'guest.toml'
         path.write_text(f'name = "other"\niso = "{self.iso}"\n')

tests/test_smoke.py

@@ -0,0 +1,48 @@
+"""Evidence gates used by the real private-display integration probe."""
+import tempfile
+import unittest
+from pathlib import Path
+
+from dev import smoke
+
+
+class SmokeEvidenceTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix='hydra-smoke-test-')
+        self.addCleanup(self.tmp.cleanup)
+        root = Path(self.tmp.name)
+        self.hidden = root / 'hidden.log'
+        self.visible = root / 'visible.log'
+        self.hidden.write_text('wl_keyboard#21.enter(1)\nwl_keyboard#21.leave(2)\n')
+        self.visible.write_text('wl_keyboard#21.enter(3)\n')
+
+    def test_focus_parser_ignores_pointer_events(self):
+        self.hidden.write_text(self.hidden.read_text() + 'wl_pointer#22.enter(4)\n')
+        self.assertEqual(smoke.focus_events(self.hidden), ['enter', 'leave'])
+
+    def test_old_focus_events_cannot_prove_a_new_switch(self):
+        self.assertFalse(smoke.fresh_focus_transition(['enter', 'leave'], ['enter'],
+                                                     ['enter', 'leave'], ['enter']))
+        self.assertTrue(smoke.fresh_focus_transition(['enter'], ['leave'],
+                                                    ['enter', 'leave'], ['leave', 'enter']))
+
+    def test_hidden_guest_must_advance_during_hidden_interval(self):
+        with self.assertRaisesRegex(AssertionError, 'hidden guest stopped'):
+            smoke.measure_hidden_interval(lambda: 10, self.hidden, self.visible, duration=.01)
+        readings = iter((10, 20))
+        result = smoke.measure_hidden_interval(lambda: next(readings), self.hidden, self.visible, duration=.01)
+        self.assertEqual((result['before'], result['after']), (10, 20))
+        self.assertGreaterEqual(result['elapsed_seconds'], .01)
+
+    def test_focus_change_during_sampling_invalidates_hidden_evidence(self):
+        def reading():
+            self.hidden.write_text(self.hidden.read_text() + 'wl_keyboard#21.enter(4)\n')
+            return 10
+        with self.assertRaisesRegex(AssertionError, 'focus changed'):
+            smoke.measure_hidden_interval(reading, self.hidden, self.visible, duration=.01)
+
+    def test_compositor_classifier_does_not_hide_other_errors(self):
+        occupied = '[ERROR] [xwayland/sockets.c:47] Failed to bind socket @/tmp/.X11-unix/X2: Address already in use'
+        self.assertEqual(smoke.classify_compositor_errors(occupied), [occupied])
+        with self.assertRaisesRegex(AssertionError, 'Unexpected compositor'):
+            smoke.classify_compositor_errors(occupied + '\n[ERROR] [config.c:1] invalid configuration')