test(ipc): wait for the state subscription before publishing in the end-to-end test (#800)

The first current-status read can be answered by a one-shot state.get
while the subscription's own snapshot (taken before the publish) is still
in flight. The poll loop then saw 'stocks' from a one-shot, broke, and the
next read got the subscription's older 'weather' snapshot, before its
event for the change arrived (CI run 38774723656, 3.13, failed in 0.7 s).

Wait until the subscription holds a snapshot before publishing, then
until the change reaches it, and only then assert on the route.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

test/test_state_stream_readers.py

@@ -604,6 +604,15 @@ class TestFreshnessOverTicks:
 
 # --- end to end over a real socket -------------------------------------------------------
 
+def _until(predicate, timeout=5.0):
+    end = time.monotonic() + timeout
+    while time.monotonic() < end:
+        if predicate():
+            return True
+        time.sleep(0.005)
+    return False
+
+
 @needs_unix_sockets
 class TestEndToEnd:
     def test_routes_read_the_socket_then_fall_back_when_it_goes(self, web, tmp_path,
@@ -624,14 +633,17 @@ class TestEndToEnd:
         try:
             data = _data(client, '/api/v3/display/current-status')
             assert (data['mode'], data['source']) == ('weather', 'socket')
-            # The subscription picks up a change the display publishes.
+            # The subscription picks up a change the display publishes. That
+            # first answer can be a one-shot state.get with the subscription's
+            # own snapshot still on its way: a change published in between
+            # would read new (one-shot), then old (that snapshot), then new
+            # again (its event). So wait for the subscription first.
+            feed = display_state._subscription()
+            assert _until(lambda: feed.latest() is not None)
             hub.publish('display', dict(hub.snapshot()['state']['display'], mode='stocks'),
                         volatile=('last_updated',))
-            deadline = time.monotonic() + 5
+            assert _until(lambda: (display_state.current_status(feed.latest()) or {})
-            while time.monotonic() < deadline:
+                          .get('mode') == 'stocks')
-                if _data(client, '/api/v3/display/current-status')['mode'] == 'stocks':
-                    break
-                time.sleep(0.05)
             assert _data(client, '/api/v3/display/current-status')['mode'] == 'stocks'
             assert hub.subscribers == 1 and hub.readers_active()
         finally: