fix(tui): a WS peer gets one 1011 close on overflow, not two under load

A slow WebSocket viewer could be closed from two independent paths at once — the per-send
deadline in _safe_send_many and abort() on fanout overflow — neither aware of the other. Under
load both fired and the client received two 1011 close frames (main-red flake in
test_overflow_closes_only_the_slow_peer). The socket close is now one-shot: the first scheduler
to run on the loop closes, the rest return.

tests/tui_gateway/test_multi_client_fanout.py

@@ -308,6 +308,27 @@ def test_overflow_closes_only_the_slow_peer_and_healthy_keeps_streaming():
         loop.close()
 
 
+def test_overflow_and_send_deadline_close_the_socket_once():
+    """abort() (fanout overflow) and the send-deadline path both reach for a 1011 close; under load both
+    fire. The peer must still get exactly one close frame, not two."""
+    loop = asyncio.new_event_loop()
+    loop_thread = threading.Thread(target=loop.run_forever, daemon=True)
+    loop_thread.start()
+    ws = _SocketWS()
+    t = WSTransport(ws, loop, peer="slow")
+    try:
+        # The two independent real triggers, scheduled on the loop as production does.
+        f1 = asyncio.run_coroutine_threadsafe(t._close_socket(1011, "send deadline"), loop)
+        f2 = asyncio.run_coroutine_threadsafe(t._close_socket(1011, "fanout overflow"), loop)
+        f1.result(2)
+        f2.result(2)
+        assert ws.close_codes == [1011]
+    finally:
+        loop.call_soon_threadsafe(loop.stop)
+        loop_thread.join(timeout=2)
+        loop.close()
+
+
 def test_fanout_close_and_detach_leave_peer_sockets_open():
     kept, detached = RecordingTransport(), RecordingTransport()
     fan = FanoutTransport(kept, detached)

tui_gateway/ws.py

@@ -163,6 +163,9 @@ class WSTransport:
         # Socket writes need an async boundary: several batches can queue on the loop during a stall.
         self._send_lock = asyncio.Lock()
         self._abort_requested = False
+        # One socket close per transport: the send-deadline path and abort() both reach for a 1011,
+        # and under load both fire — latched on the loop thread so the peer gets a single close frame.
+        self._close_scheduled = False
 
     def _on_loop(self) -> bool:
         try:
@@ -323,7 +326,12 @@ class WSTransport:
 
     async def _close_socket(self, code: int, reason: str) -> None:
         """Close the peer socket so ``handle_ws``'s ``receive_text`` unblocks and its disconnect teardown
-        runs. The server library bounds this (websockets ``close_timeout`` → abort)."""
+        runs. The server library bounds this (websockets ``close_timeout`` → abort). One-shot: the
+        send-deadline path and ``abort()`` can both schedule this; the first to run on the loop closes,
+        the rest return, so the peer never gets a second 1011."""
+        if self._close_scheduled:
+            return
+        self._close_scheduled = True
         try:
             await self._ws.close(code=code)
         except Exception as exc: