fix(relay): reader clears the dead socket handle on unexpected exit

Two reader-exit paths arm NO reconnect supervisor: a terminal 4401
revocation (deliberately never re-dials) and reconnect=False
transports. On both, _ws kept pointing at the dead socket after the
reader unwound, so the 'is None' liveness guard reported connected and
a send registered a future nothing could ever resolve — the full
_outbound_timeout_s (~30s) wedge this PR exists to eliminate. The
revocation case is the sharpest: the fatal-error notification that
path emits is itself an outbound send, so it ate the stall.

Null the handle in the reader's finally, identity-guarded (only if _ws
still points at the socket THIS reader served) so a supervisor re-dial
that already installed a fresh socket is never clobbered, and gated on
not _closing so disconnect() keeps sole ownership of teardown.

This also makes _ws the single honest liveness signal for the redial
window itself — groundwork for retiring the supervisor-state send
guard, which misreads that window from both directions.

Regression tests cover both uncovered paths, asserting _ws is cleared
and a post-drop send fails fast; both wedge (fail) with the finally
reverted.

gateway/relay/ws_transport.py

@@ -833,6 +833,10 @@ class WebSocketRelayTransport:
 
     async def _read_loop(self) -> None:
         assert self._ws is not None
+        # Bind the socket this reader serves: the finally below must only
+        # clear _ws if it still points at THIS socket (a supervisor re-dial
+        # may have already installed a fresh one by the time we unwind).
+        ws = self._ws
         buf = ""
         try:
             try:
@@ -878,6 +882,19 @@ class WebSocketRelayTransport:
                     self._reconnect_loop(), name="relay-ws-reconnect"
                 )
         finally:
+            # The socket this reader served is dead. Drop the handle (identity-
+            # guarded: a re-dial that already installed a FRESH socket must not
+            # be clobbered) so every `self._ws is None` liveness check — send,
+            # _request_response, go_idle, go_dormant — reports "not connected"
+            # for the whole outage. Without this, _ws kept pointing at the dead
+            # socket on every reader exit that arms NO supervisor (terminal
+            # 4401 revocation, reconnect=False transports), and a send there
+            # registered a future nothing could resolve: a full
+            # _outbound_timeout_s (~30s) wedge — including the revocation
+            # path's own fatal-error notification. disconnect() owns the
+            # handle during deliberate teardown, so leave it alone then.
+            if self._ws is ws and not self._closing:
+                self._ws = None
             # The reader is the ONLY thing that can resolve a pending
             # outbound_result future — once it exits (socket dropped, error,
             # or cancellation cleanup) every in-flight _request_response waiter

tests/gateway/relay/test_ws_transport_hardening.py

@@ -36,11 +36,12 @@ class _DroppingWS:
     """Fake socket: accepts sends, then the read loop dies mid-iteration —
     the shape of an unexpected close (e.g. 1011 keepalive ping timeout)."""
 
-    def __init__(self):
+    def __init__(self, close_code: int | None = None):
         self.sent: list[str] = []
         # Reader blocks here until the test releases it, so the outbound
         # future is registered BEFORE the "socket" drops.
         self.drop = asyncio.Event()
+        self._close_code = close_code
 
     async def send(self, data):
         self.sent.append(data)
@@ -50,6 +51,10 @@ class _DroppingWS:
 
     async def __anext__(self):
         await self.drop.wait()
+        if self._close_code is not None:
+            from websockets.frames import Close
+
+            raise ConnectionClosedError(Close(self._close_code, ""), None)
         raise ConnectionClosedError(None, None)
 
     async def close(self):
@@ -152,3 +157,59 @@ async def test_connect_passes_wan_keepalive_tuning(monkeypatch):
     for kwargs in captured:
         assert kwargs.get("ping_interval") == 30
         assert kwargs.get("ping_timeout") == 60
+
+
+async def _run_reader_to_exit(t: WebSocketRelayTransport, fake: _DroppingWS) -> None:
+    """Start the reader on ``fake``, drop the socket, and wait for the reader
+    to fully unwind — the state every post-drop assertion depends on."""
+    t._reader = asyncio.create_task(t._read_loop())
+    await asyncio.sleep(0)
+    fake.drop.set()
+    await t._reader
+
+
+@pytest.mark.asyncio
+async def test_send_after_terminal_4401_revocation_fails_fast():
+    """A terminal 4401 revocation deliberately arms NO reconnect supervisor,
+    so the reader's exit is the LAST liveness transition this transport will
+    ever make. If _ws still points at the dead socket afterwards, the
+    revocation path's own fatal-error notification send wedges for the full
+    _outbound_timeout_s. The reader must leave _ws cleared so the
+    not-connected guard answers instantly."""
+    t = WebSocketRelayTransport(
+        "ws://unused", "discord", "bot1", reconnect=True, outbound_timeout_s=30.0
+    )
+    fake = _DroppingWS(close_code=4401)
+    t._ws = fake
+    t._handshake_succeeded = True  # prior handshake -> 4401 is a revocation
+    await _run_reader_to_exit(t, fake)
+
+    assert t._auth_revoked is True
+    assert t._supervisor is None  # revocation must not re-dial
+    assert t._ws is None, "dead socket handle must not survive the reader"
+
+    result = await asyncio.wait_for(
+        t.send_outbound({"op": "send_message", "text": "hi"}), timeout=2.0
+    )
+    assert result["success"] is False
+    assert t._pending == {}
+
+
+@pytest.mark.asyncio
+async def test_send_after_drop_with_reconnect_disabled_fails_fast():
+    """reconnect=False transports never arm a supervisor either — the same
+    stranded-_ws wedge as the revocation path, reachable by configuration."""
+    t = WebSocketRelayTransport(
+        "ws://unused", "discord", "bot1", reconnect=False, outbound_timeout_s=30.0
+    )
+    fake = _DroppingWS()
+    t._ws = fake
+    await _run_reader_to_exit(t, fake)
+
+    assert t._ws is None, "dead socket handle must not survive the reader"
+
+    result = await asyncio.wait_for(
+        t.send_outbound({"op": "send_message", "text": "hi"}), timeout=2.0
+    )
+    assert result["success"] is False
+    assert t._pending == {}