1 files changed, 91 insertions, 0 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