diff --git a/gateway/run.py b/gateway/run.py index fa5c655b6325f..e75d11b9404e9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -76,6 +76,10 @@ _PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0 # wall deadlines plus readiness; other platforms retain the 30s isolation bound. _TELEGRAM_CONNECT_TIMEOUT_SECS_DEFAULT = 180.0 _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0 +# Round-2 #2: upper bound on a single stall-notify adapter.send so a wedged +# transport cannot block the session-stall watcher pass (notify-only path; +# on timeout the latch stays clear and the next tick retries). +_STALL_NOTIFY_SEND_TIMEOUT_SECONDS = 15.0 _GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS = 16 * 1024 * 1024 _TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(? 0 # enabled by default; 0 would disable the watchdog + + +class _NeverResolvingAdapter: + """Adapter whose send() hangs forever (wedged transport).""" + + def __init__(self): + self._pending_messages = {} + self.send_attempts = 0 + + async def send(self, chat_id, content, metadata=None): + self.send_attempts += 1 + await asyncio.Event().wait() # never resolves + + +@pytest.mark.asyncio +async def test_check_session_stalls_bounds_wedged_send(monkeypatch): + """Round-2 #2: a never-resolving adapter.send must not wedge the watcher. + + The bounded send times out, does NOT latch (retry next tick), the pass + completes so the watcher keeps ticking, and a healthy sibling candidate + still receives its notification in the same pass. + """ + import gateway.run as gateway_run + + monkeypatch.setattr( + gateway_run, "_STALL_NOTIFY_SEND_TIMEOUT_SECONDS", 0.1 + ) + wedged = _NeverResolvingAdapter() + healthy = _FakeAdapter() + runner = _runner_for_stall(wedged) + runner.adapters = {"wedged": wedged, "healthy": healthy} + + wedged_key = "agent:main:telegram:dm:wedged" + healthy_key = "agent:main:discord:dm:healthy" + wedged._pending_messages[wedged_key] = _pending_event(chat_id="chat-w") + healthy._pending_messages[healthy_key] = _pending_event(chat_id="chat-h") + runner._running_agents[wedged_key] = _FakeAgent(time.time() - 120) + runner._running_agents[healthy_key] = _FakeAgent(time.time() - 120) + + # Pass must complete despite the wedged transport (bounded by wait_for). + sent = await asyncio.wait_for(runner._check_session_stalls(60), timeout=5) + + # Healthy sibling was notified in the SAME pass. + assert sent == 1 + assert len(healthy.sent) == 1 + assert healthy_key in runner._session_stall_notified + # Wedged session: send attempted, timed out, NOT latched. + assert wedged.send_attempts == 1 + assert wedged_key not in runner._session_stall_notified + + # Watcher ticks again: the wedged candidate is retried next pass. + sent2 = await asyncio.wait_for(runner._check_session_stalls(60), timeout=5) + assert sent2 == 0 # healthy already latched; wedged timed out again + assert wedged.send_attempts == 2 + assert wedged_key not in runner._session_stall_notified