fix(gateway): bound the stall-notify adapter.send (re-review #2)

A wedged adapter transport (network hang, dead websocket) previously
blocked _check_session_stalls forever: sibling candidates in the same
pass were never evaluated and the watcher stopped ticking. Wrap the
send in asyncio.wait_for (15s); on timeout log a WARNING and do NOT
latch, so the next tick retries. Regression uses a never-resolving fake
adapter and proves the pass completes, a healthy sibling candidate is
still notified in the same pass, and the watcher ticks again
(sabotage-verified against the unbounded send).

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"(?<![\w:/])/([A-Za-z0-9][A-Za-z0-9_-]*)")
 _GATEWAY_HYGIENE_PLATFORM = "gateway_hygiene"
@@ -12023,11 +12027,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     if source is not None and hasattr(self, "_thread_metadata_for_source")
                     else None
                 )
-                result = await adapter.send(
-                    str(chat_id),
-                    format_session_stall_notification(idle_seconds),
-                    metadata=metadata,
-                )
+                # Round-2 #2: bound the send. A wedged adapter transport
+                # (network hang, dead websocket) must not block the whole
+                # watcher pass — sibling candidates in this loop would never
+                # be evaluated and the watcher itself would stop ticking.
+                try:
+                    result = await asyncio.wait_for(
+                        adapter.send(
+                            str(chat_id),
+                            format_session_stall_notification(idle_seconds),
+                            metadata=metadata,
+                        ),
+                        timeout=_STALL_NOTIFY_SEND_TIMEOUT_SECONDS,
+                    )
+                except asyncio.TimeoutError:
+                    logger.warning(
+                        "Session stall notify send timed out after %.0fs "
+                        "for %s; will retry next tick",
+                        _STALL_NOTIFY_SEND_TIMEOUT_SECONDS,
+                        session_key,
+                    )
+                    continue  # do not latch; retry next tick
                 # Adapters often return SendResult(success=False) instead of raising.
                 if result is not None and getattr(result, "success", True) is False:
                     logger.warning(

tests/gateway/test_session_stall_watchdog.py

@@ -460,3 +460,58 @@ def test_session_stall_timeout_in_default_config():
     timeout = DEFAULT_CONFIG["agent"]["session_stall_timeout"]
     assert isinstance(timeout, (int, float))
     assert timeout > 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