diff --git a/gateway/run.py b/gateway/run.py index 6c9556d10f..bd8507597c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1729,6 +1729,20 @@ def _handoff_watch_scopes(runner: object) -> list: return scopes +async def _resolve_handoff_watch_scopes(runner: object) -> list: + """``_handoff_watch_scopes`` for an on-loop watcher tick. Multiplex resolution walks the filesystem + (``profiles_to_serve``), so it hops to the executor; single-profile mode does no I/O and returns the + root poll directly — no per-tick thread spawn on the unbounded executor. A config-less stand-in + (tests) and a runner without the executor hop fall through to the plain resolver.""" + config = getattr(runner, "config", None) + if config is not None and not getattr(config, "multiplex_profiles", False): + return [(None, None)] + offload = getattr(runner, "_run_in_executor_with_context", None) + if callable(offload): + return await offload(_handoff_watch_scopes, runner) + return _handoff_watch_scopes(runner) + + async def _reclaim_stale(runner: object) -> None: """Fail handoffs left in ``running`` by a gateway that died mid-dispatch (once per store at startup). ``running`` is only set for one in-process dispatch, so a leftover row belongs to a dead process and diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index 416022c81d..c74fd18060 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -473,7 +473,7 @@ class GatewayAdapterLifecycleMixin: """Process pending CLI→gateway session handoffs from ``state.db``: claim atomically (pending → running), re-bind the home channel to the CLI session_id, dispatch a synthetic event, mark ``completed``/``failed``.""" - from gateway.run import _async_profile_runtime_scope, _handoff_watch_scopes, _reclaim_stale + from gateway.run import _async_profile_runtime_scope, _reclaim_stale, _resolve_handoff_watch_scopes from gateway.run_idle_gates import off_loop_gate, profile_has_pending_handoff await asyncio.sleep(5) # let platforms connect before dispatching through them # Does _process_handoff accept the profile argument? Test stand-ins bind a one-arg callable. @@ -534,22 +534,16 @@ class GatewayAdapterLifecycleMixin: def _scope(profile_home): # local: tests bind this watcher onto bare SimpleNamespace runners return GatewayAdapterLifecycleMixin._async_scope_or_null(_async_profile_runtime_scope, profile_home) - # Resolve watch scopes off the loop: the profiles_to_serve() filesystem walk can stall the - # loop past the liveness probe. Bare stand-ins without the executor hop resolve inline. - offload = getattr(self, "_run_in_executor_with_context", None) - - async def _resolve_scopes(): - return (await offload(_handoff_watch_scopes, self) if callable(offload) - else _handoff_watch_scopes(self)) - - for _pname, _phome in await _resolve_scopes(): + # Multiplex scope resolution walks the filesystem (profiles_to_serve) off the loop, so a + # stalled walk cannot trip the liveness probe — startup reclaim and every tick alike. + for _pname, _phome in await _resolve_handoff_watch_scopes(self): with _log_suppressed(logging.DEBUG, "Stale-handoff reclaim failed", exc_info=True): async with _scope(_phome): await _reclaim_stale(self) try: while self._running: try: - for profile_name, profile_home in await _resolve_scopes(): + for profile_name, profile_home in await _resolve_handoff_watch_scopes(self): # Idle gate (run_idle_gates): skip the scope entry when the profile's store # holds no pending handoff. The root poll (None) is unscoped and stays cheap. if profile_home is not None and not await off_loop_gate( diff --git a/gateway/run_goals.py b/gateway/run_goals.py index b916ed8334..4bb2024551 100644 --- a/gateway/run_goals.py +++ b/gateway/run_goals.py @@ -460,7 +460,7 @@ class GatewayGoalsMixin: store — a ``/loop`` set from a secondary profile's chat would never fire. Every served profile's store is scanned under its own runtime scope (same shape as ``_handoff_watcher``), and each hit is fired against that profile's adapters.""" - from gateway.run import _async_profile_runtime_scope, _handoff_watch_scopes + from gateway.run import _async_profile_runtime_scope, _resolve_handoff_watch_scopes from gateway.run_idle_gates import profile_has_active_loop await asyncio.sleep(5) # let platforms finish connecting warned_no_route: set = set() @@ -487,10 +487,9 @@ class GatewayGoalsMixin: while self._running: try: - # Off-loop: profiles_to_serve() walks the filesystem (realpath chains + profile-dir - # scans) every pass; a stalled walk trips the loop-liveness watchdog (exit 75). - scopes = await self._run_in_executor_with_context(_handoff_watch_scopes, self) - for profile_name, profile_home in scopes: + # Multiplex resolution walks the filesystem off-loop; a stalled walk on the loop + # trips the loop-liveness watchdog (exit 75). + for profile_name, profile_home in await _resolve_handoff_watch_scopes(self): # Idle gate (run_idle_gates): skip the scope entry when the profile's store holds # no active loop. The root scan (None) is unscoped and stays cheap. if profile_home is not None and not await self._run_in_executor_with_context( diff --git a/tests/gateway/test_loop_command.py b/tests/gateway/test_loop_command.py index c0d60bfa74..1921468492 100644 --- a/tests/gateway/test_loop_command.py +++ b/tests/gateway/test_loop_command.py @@ -330,7 +330,14 @@ async def test_loop_wakeup_watcher_gates_profile_scope_on_active_loops(loop_env, goals._DB_CACHE[str(work_home)] = work_db scopes = [(None, None), ("work", work_home)] - monkeypatch.setattr("gateway.run._handoff_watch_scopes", lambda _r: scopes) + on_main_thread = [] + + def _scopes(_runner): + # profiles_to_serve() walks the filesystem: it must resolve off the loop thread. + on_main_thread.append(threading.current_thread() is threading.main_thread()) + return scopes + + monkeypatch.setattr("gateway.run._handoff_watch_scopes", _scopes) entered = [] @@ -348,6 +355,7 @@ async def test_loop_wakeup_watcher_gates_profile_scope_on_active_loops(loop_env, monkeypatch.setattr("gateway.run._async_profile_runtime_scope", _SpyScope) runner = _make_runner() + runner.config.multiplex_profiles = True runner._running_agents = {} runner.adapters = {} @@ -372,6 +380,7 @@ async def test_loop_wakeup_watcher_gates_profile_scope_on_active_loops(loop_env, # Idle: no loop rows in the work profile's store → its scope is never entered. await _run_one_tick() assert entered == [], f"idle profile scope must not be entered; got {entered}" + assert on_main_thread == [False], f"scopes must resolve off the loop; got {on_main_thread}" # An ACTIVE loop row in the work profile's store opens the gate. await GatewayRunner._handle_loop_command(runner, _make_event("/loop 5m poll CI"))