fix(rtc): auto-reconnect signaling WebSocket on unexpected loss

`WebSocketClient` had no reconnect logic of its own — `_on_close`,
`_on_error`, and the health-check timeout in `_ping_loop` only logged
and stopped the connection. When the signaling WS dropped (transient
TCP reset, missed health check, server-side close without an SFU error
event), the session sat hanging until the frontend's own timeout fired
a DELETE.

Under concurrent load this is the dominant remaining cause of "the
agent disconnected randomly" reports: even with the late-offer fix in
place, a single brief WS blip is fatal because nothing kicks
reconnection.

- `signaling.py`: emit a `connection_lost` event with a reason string
  on (a) unexpected `_on_close`, (b) `_on_error` after the initial
  handshake completed, and (c) `_ping_loop` health-check timeout
  (notify *before* `self.close()` so the user-initiated guard does not
  suppress it). Idempotent via `_connection_lost_sent`.
- `connection_manager.py`: subscribe to `connection_lost` on the WS
  client and route it into the existing `ReconnectionManager`
  (`ReconnectionStrategy.FAST`). The manager already handles strategy
  escalation, locking, and the disconnection-timeout deadline.

Author: aliev

getstream/video/rtc/connection_manager.py

@@ -36,7 +36,7 @@
 from getstream.video.rtc.recording import RecordingManager
 from getstream.video.rtc.participants import ParticipantsState
 from getstream.video.rtc.tracks import SubscriptionConfig, SubscriptionManager
-from getstream.video.rtc.reconnection import ReconnectionManager
+from getstream.video.rtc.reconnection import ReconnectionManager, ReconnectionStrategy
 from getstream.video.rtc.peer_connection import PeerConnectionManager
 from getstream.video.rtc.models import JoinCallResponse
 from getstream.video.rtc.tracer import Tracer
@@ -277,6 +277,25 @@ async def _on_subscriber_offer(self, event: events_pb2.SubscriberOffer):
             finally:
                 self.subscriber_negotiation_lock.release()
 
+    async def _on_signaling_connection_lost(self, reason):
+        """Reconnect when the signaling WebSocket drops unexpectedly.
+
+        The WebSocketClient itself only logs the error and stops; it has
+        no reconnect of its own. This handler bridges that gap by routing
+        the loss into the existing `ReconnectionManager`, so a transient
+        TCP reset or a missed health check no longer means a dead session.
+        """
+        if not self.running:
+            return
+        logger.warning(f"Signaling WS lost; triggering reconnect: {reason}")
+        try:
+            await self._reconnector.reconnect(
+                strategy=ReconnectionStrategy.FAST,
+                reason=f"signaling ws lost: {reason}",
+            )
+        except Exception:
+            logger.exception("Reconnect after signaling WS loss failed")
+
     async def _connect_coordinator_ws(self):
         """
         Connects to the coordinator websocket and subscribes to events.
@@ -414,6 +433,15 @@ async def _connect_internal(
             # Connect subscriber offer event to handle SDP negotiation
             self._ws_client.on_event("subscriber_offer", self._on_subscriber_offer)
 
+            # Drive reconnection when the signaling WS drops outside of an
+            # SFU-level error event (raw socket close, health-check timeout,
+            # transport-level exceptions). Without this handler the
+            # WebSocketClient just logs and stops; the session sits hanging
+            # until the frontend times out and tears it down.
+            self._ws_client.on_event(
+                "connection_lost", self._on_signaling_connection_lost
+            )
+
             # Re-emit the events so they can be subscribed to on the ConnectionManager
             self._ws_client.on_wildcard("*", self.emit)
 

getstream/video/rtc/signaling.py

@@ -214,11 +214,43 @@ def _on_error(self, ws, error):
             error_event.error.error.message = str(error)
             self.first_message = error_event
             self.first_message_event.set()
+        elif not self.closed:
+            # Established connection failed unexpectedly — signal the owner
+            # so it can drive reconnection.
+            self._notify_connection_lost(f"error: {error}")
 
     def _on_close(self, ws, close_status_code, close_msg):
         """Handle WebSocket close event."""
         logger.debug(f"WebSocket connection closed: {close_status_code} {close_msg}")
+        was_unexpected = not self.closed
         self.running = False
+        if was_unexpected:
+            self._notify_connection_lost(
+                f"closed by remote (code={close_status_code} msg={close_msg})"
+            )
+
+    def _notify_connection_lost(self, reason: str) -> None:
+        """Schedule a ``connection_lost`` emit on the main event loop.
+
+        Called from the worker thread (websocket-client callbacks run there,
+        as does ``_ping_loop``). Owners that wire ``on_event("connection_lost",
+        ...)`` get a single notification per unexpected disconnect and can
+        drive reconnection. Idempotent guard via ``self._connection_lost_sent``
+        so a sequence like ``_on_error`` → ``_on_close`` only fires once.
+        """
+        if getattr(self, "_connection_lost_sent", False):
+            return
+        self._connection_lost_sent = True
+        try:
+            asyncio.run_coroutine_threadsafe(
+                self._emit_connection_lost(reason),
+                self.main_loop,
+            )
+        except Exception:
+            logger.exception("Failed to schedule connection_lost emit")
+
+    async def _emit_connection_lost(self, reason: str) -> None:
+        self.emit("connection_lost", reason)
 
     def _start_ping_handler(self):
         """Start the ping mechanism in a background thread."""
@@ -242,6 +274,10 @@ def _ping_loop(self):
             current_time = time.time()
             if current_time - self.last_health_check_time > self.ping_interval * 2:
                 logger.warning("Health check failed, closing connection")
+                # Notify before close() so the owner can reconnect; close()
+                # itself sets `self.closed=True` and would suppress the
+                # notification in `_on_close`.
+                self._notify_connection_lost("health check timeout")
                 self.close()
                 return