fix(tui-gateway): keep the WebSocket open when a frame has a lone surrogate

Starlette encodes send_text as UTF-8. A lone UTF-16 surrogate in a
status payload used to UnicodeEncodeError and latch the whole Desktop
connection closed. Sanitize the frame and skip only that send.

tests/tui_gateway/test_ws_surrogate_send.py

@@ -0,0 +1,63 @@
+"""Lone UTF-16 surrogates must not tear down the Desktop WebSocket (#97288)."""
+
+from __future__ import annotations
+
+import asyncio
+
+from tui_gateway.ws import WSTransport, _sanitize_ws_text
+
+
+LONE_SURROGATE = "\ud83d"
+
+
+def test_sanitize_ws_text_makes_utf8_encodable() -> None:
+    dirty = f"gateway.ready {LONE_SURROGATE} payload"
+    out = _sanitize_ws_text(dirty)
+    out.encode("utf-8")
+    assert LONE_SURROGATE not in out
+
+
+def test_sanitize_ws_text_leaves_valid_text_unchanged() -> None:
+    clean = '{"type":"gateway.ready","ok":true}'
+    assert _sanitize_ws_text(clean) is clean or _sanitize_ws_text(clean) == clean
+
+
+class _FakeWS:
+    def __init__(self) -> None:
+        self.sent: list[str] = []
+        self.raise_on: str | None = None
+
+    async def send_text(self, line: str) -> None:
+        line.encode("utf-8")
+        if self.raise_on is not None and self.raise_on in line:
+            raise UnicodeEncodeError("utf-8", line, 0, 1, "surrogates not allowed")
+        self.sent.append(line)
+
+
+def test_safe_send_sanitizes_surrogate_and_keeps_connection() -> None:
+    async def _run() -> None:
+        loop = asyncio.get_running_loop()
+        ws = _FakeWS()
+        transport = WSTransport(ws, loop, peer="127.0.0.1:1")
+        dirty = f'{{"type":"gateway.ready","x":"{LONE_SURROGATE}"}}'
+        await transport._safe_send_many(["first", dirty, "third"])
+        assert transport.closed is False
+        assert ws.sent[0] == "first"
+        assert ws.sent[-1] == "third"
+        assert LONE_SURROGATE not in "".join(ws.sent)
+        assert len(ws.sent) == 3
+
+    asyncio.run(_run())
+
+
+def test_unicode_encode_error_does_not_close_socket() -> None:
+    async def _run() -> None:
+        loop = asyncio.get_running_loop()
+        ws = _FakeWS()
+        ws.raise_on = "BOOM"
+        transport = WSTransport(ws, loop, peer="127.0.0.1:1")
+        await transport._safe_send_many(["ok-a", "BOOM-frame", "ok-b"])
+        assert transport.closed is False
+        assert ws.sent == ["ok-a", "ok-b"]
+
+    asyncio.run(_run())

tui_gateway/ws.py

@@ -61,6 +61,24 @@ def _note_dashboard_client_activity(*, force: bool = False) -> None:
     except Exception:  # noqa: BLE001 - liveness garnish must never break the WS
         _log.debug("dashboard client heartbeat touch failed", exc_info=True)
 
+
+def _sanitize_ws_text(text: str) -> str:
+    """Return *text* that can be UTF-8 encoded for a WebSocket frame.
+
+    Python ``str`` may contain lone UTF-16 surrogates (``\\ud800``-``\\udfff``)
+    that ``json.dumps(..., ensure_ascii=False)`` will happily emit. Starlette
+    then encodes the frame as UTF-8 and raises ``UnicodeEncodeError``, which
+    used to latch the whole connection closed (#97288). Replace those
+    code points rather than dropping the connection.
+    """
+    if not text:
+        return text
+    try:
+        text.encode("utf-8")
+    except UnicodeEncodeError:
+        return text.encode("utf-8", "replace").decode("utf-8")
+    return text
+
 # Max seconds a pool-dispatched handler will block waiting for the event loop
 # to flush a WS frame before we mark the transport dead. Protects handler
 # threads from a wedged socket.
@@ -277,20 +295,30 @@ class WSTransport:
         async with self._send_lock:
             if self._closed:
                 return
-            try:
-                for line in lines:
-                    if self._closed:
-                        return
-                    await self._ws.send_text(line)
-            except Exception as exc:
-                # Latch while still holding the writer lock so queued batches
-                # observe the failure before they get a chance to touch the
-                # socket.
-                self._closed = True
-                _log.warning(
-                    "ws send failed peer=%s error_type=%s error=%s",
-                    self._peer, type(exc).__name__, exc,
-                )
+            for line in lines:
+                if self._closed:
+                    return
+                payload = _sanitize_ws_text(line)
+                try:
+                    await self._ws.send_text(payload)
+                except UnicodeEncodeError as exc:
+                    # A single illegal UTF-8 frame (lone surrogate in a
+                    # status/ready payload) must not tear down the socket.
+                    # Fresh Desktop installs looped on this (#97288).
+                    _log.warning(
+                        "ws send skipped invalid utf-8 frame peer=%s error=%s",
+                        self._peer, exc,
+                    )
+                    continue
+                except Exception as exc:
+                    # Latch while still holding the writer lock so queued
+                    # batches observe the failure before they touch the socket.
+                    self._closed = True
+                    _log.warning(
+                        "ws send failed peer=%s error_type=%s error=%s",
+                        self._peer, type(exc).__name__, exc,
+                    )
+                    return
 
     def close(self) -> None:
         self._closed = True