sync: extract shared settle/wait helpers (idatui/_sync.py)

Factor the pilot suite's ad-hoc 'poll until the UI reacted' loop into one
reusable module so the upcoming RPC driver and the tests share a single
source of truth for quiescence. wait_for() takes the yield strategy
(pilot.pause vs asyncio.sleep); drain()/workers_idle()/settle() give the
live driver a 'block until quiescent, then until pred holds' primitive
built on Textual's own _wait_for_screen + the worker manager.

Ctx.wait now delegates to wait_for; suite unchanged (101 green).

2 files changed, 93 insertions, 7 deletions

diff --git a/idatui/_sync.py b/idatui/_sync.py
new file mode 100644
index 0000000..b7da338
--- /dev/null
+++ b/idatui/_sync.py
@@ -0,0 +1,91 @@
+"""Settle/wait helpers shared by the pilot tests and the live RPC driver.
+
+The single source of truth for "the UI has finished reacting to what I just did".
+Both the headless test harness and the in-process RPC server need the exact same
+guarantee before they read state back, so it lives here once.
+
+Two yield strategies feed the same poll loop: under a Pilot (tests) we yield with
+``pilot.pause`` (which also drains the screen); live (RPC) we yield with
+``asyncio.sleep`` and drain explicitly via a throwaway ``Pilot(app)``.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Awaitable, Callable, Optional
+
+from textual.pilot import Pilot
+
+
+async def wait_for(
+    pred: Callable[[], bool],
+    tick: Callable[[float], Awaitable[None]],
+    timeout: float = 20.0,
+    step: float = 0.02,
+) -> bool:
+    """Poll ``pred`` until true or ``timeout`` elapses.
+
+    ``tick(step)`` yields control between polls: ``pilot.pause`` in tests,
+    ``asyncio.sleep`` in the live app. Returns whether ``pred`` became true.
+    """
+    waited = 0.0
+    while waited < timeout:
+        if pred():
+            return True
+        await tick(step)
+        waited += step
+    return False
+
+
+async def drain(app, timeout: float = 15.0) -> None:
+    """Wait for the message pump and every widget to process queued events.
+
+    Reuses Textual's own ``Pilot._wait_for_screen`` (it only needs ``app``), so
+    it works against a live app with no real Pilot attached.
+    """
+    try:
+        await Pilot(app)._wait_for_screen(timeout=timeout)
+    except Exception:  # noqa: BLE001 — best-effort; never let settle explode
+        pass
+
+
+async def workers_idle(app, timeout: float = 20.0) -> None:
+    """Wait (bounded) for all threaded ``@work`` workers to finish.
+
+    Safe because the app has no perpetual Textual workers — the keepalive is a
+    client-side thread, not a worker. A timeout guards against a wedged worker.
+    """
+    try:
+        await asyncio.wait_for(app.workers.wait_for_complete(), timeout)
+    except Exception:  # noqa: BLE001 — timeout or manager churn; fall through
+        pass
+
+
+async def settle(
+    app,
+    pred: Optional[Callable[[], bool]] = None,
+    *,
+    timeout: float = 20.0,
+    step: float = 0.02,
+    rounds: int = 3,
+) -> bool:
+    """Block until the app is quiescent, then optionally until ``pred`` holds.
+
+    A round = drain the pump, wait for workers, repeat — because a completing
+    worker can post a message that spawns the next worker (nav -> decompile).
+    With a ``pred`` (the reliable signal, e.g. ``dec.loaded_ea == ea``) we return
+    as soon as it holds; without one we return once no workers remain.
+    """
+    for _ in range(max(1, rounds)):
+        await drain(app, min(timeout, 15.0))
+        await workers_idle(app, timeout)
+        if pred is not None:
+            if pred():
+                await drain(app, min(timeout, 15.0))
+                return True
+        elif len(app.workers) == 0:
+            await drain(app, min(timeout, 15.0))
+            return True
+    await drain(app, min(timeout, 15.0))
+    if pred is not None:
+        return await wait_for(pred, asyncio.sleep, timeout, step)
+    return True
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py
index c19bfa0..0b22f90 100644
--- a/tests/test_scenarios.py
+++ b/tests/test_scenarios.py
@@ -31,6 +31,7 @@ from textual.widgets import (  # noqa: E402
     DataTable, Footer, Input, OptionList, Static, TextArea,
 )
 from rich.text import Text  # noqa: E402
+from idatui._sync import wait_for  # noqa: E402
 
 PASS = FAIL = 0
 STOP_AFTER = None
@@ -73,13 +74,7 @@ class Ctx:
             raise _StopSuite
 
     async def wait(self, pred, t=20.0, step=0.02):
-        waited = 0.0
-        while waited < t:
-            if pred():
-                return True
-            await self.pilot.pause(step)
-            waited += step
-        return False
+        return await wait_for(pred, self.pilot.pause, t, step)
 
     async def press(self, *keys):
         for k in keys: