feat: Workspace-Umbau Schritt 7 — VNC-Live-Desktop durch RVS getunnelt

QEMU-VNC erscheint live in der App-Desktop-Kachel, komplett ueber RVS:
- Bridge: TCP↔RVS-VNC-Bruecke. vnc_open oeffnet asyncio-Verbindung zu
  host.docker.internal:5901, Reader-Loop streamt RFB-Bytes als vnc_data;
  vnc_input schreibt zurueck; vnc_close raeumt auf. check_desktop probt den
  Port und meldet desktop_status. Base64-in-JSON, kein websockify/noVNC auf
  dem Host noetig.
- App: novncHtml.ts laedt noVNC (CDN) und ersetzt window.WebSocket durch einen
  Shim, der RFB-Bytes per postMessage ueber RVS brueckt (server-speaks-first →
  timing-robust). VncTile mountet die WebView nur im Fokus, oeffnet bei 'ready'
  den Tunnel (desktop.openVnc), speist Server-Bytes ein und schickt Eingaben
  als vnc_input; beim Verlassen wird der Tunnel geschlossen (VM laeuft weiter).

tsc/py clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

bridge/aria_bridge.py

@@ -734,6 +734,11 @@ class ARIABridge:
         # Beeinflusst das Timeout fuer stt_request — bei "loading" warten wir laenger,
         # weil das Modell beim ersten Request noch ~1-2 Min runtergeladen werden kann.
         self._remote_stt_ready: bool = False
+        # VNC-Tunnel (Workspace-Desktop): session → {"writer": StreamWriter,
+        # "task": asyncio.Task}. Wir bruecken rohes RFB-TCP (QEMU-VNC auf dem
+        # Host) <-> RVS (vnc_data/vnc_input, Base64-in-JSON).
+        self._vnc_sessions: dict[str, dict] = {}
+        self._vnc_host: str = os.environ.get("ARIA_VNC_HOST", "host.docker.internal")
         # FLUX-Render-Requests die aktuell auf Antwort der flux-bridge (Gamebox) warten.
         # requestId → Future mit dem flux_response-Payload (oder None bei Fehler).
         self._pending_flux: dict[str, asyncio.Future] = {}
@@ -3364,6 +3369,43 @@ class ARIABridge:
                             pid or "main", rel, len(full_text), "ok" if ok else "abgelehnt")
             return
 
+        elif msg_type == "check_desktop":
+            session = str(payload.get("session") or "main")
+            try:
+                port = int(payload.get("port") or 5901)
+            except (TypeError, ValueError):
+                port = 5901
+            asyncio.create_task(self._check_desktop(session, port))
+            return
+
+        elif msg_type == "vnc_open":
+            session = str(payload.get("session") or "main")
+            try:
+                port = int(payload.get("port") or 5901)
+            except (TypeError, ValueError):
+                port = 5901
+            asyncio.create_task(self._vnc_open(session, port))
+            return
+
+        elif msg_type == "vnc_close":
+            session = str(payload.get("session") or "main")
+            asyncio.create_task(self._vnc_close(session))
+            return
+
+        elif msg_type == "vnc_input":
+            session = str(payload.get("session") or "main")
+            b64 = payload.get("b64") or ""
+            sess = self._vnc_sessions.get(session)
+            if sess and b64:
+                try:
+                    data = base64.b64decode(b64)
+                    sess["writer"].write(data)
+                    # drain nicht awaiten (wir sind synchron im Dispatcher) —
+                    # write puffert, der Kernel-Socket schluckt RFB-Input locker.
+                except Exception as exc:
+                    logger.warning("[vnc] input schreiben (%s) fehlgeschlagen: %s", session, exc)
+            return
+
         elif msg_type == "config_request":
             # Eine andere Bridge (whisper/f5tts) bittet um die aktuelle Voice-
             # Config — passiert wenn sie sich connected, weil sie sonst die
@@ -4363,6 +4405,94 @@ class ARIABridge:
             logger.warning("[code_file] Schreiben fehlgeschlagen %s/%s: %s", project_id, rel_path, exc)
             return False
 
+    async def _check_desktop(self, session: str, port: int) -> None:
+        """Probt ob auf dem Host eine QEMU-VNC laeuft und meldet desktop_status."""
+        available = False
+        try:
+            fut = asyncio.open_connection(self._vnc_host, port)
+            reader, writer = await asyncio.wait_for(fut, timeout=2.0)
+            available = True
+            writer.close()
+            try:
+                await writer.wait_closed()
+            except Exception:
+                pass
+        except Exception:
+            available = False
+        await self._send_to_rvs({
+            "type": "desktop_status",
+            "payload": {
+                "available": available,
+                "session": session,
+                "message": (f"Desktop laeuft (VNC {self._vnc_host}:{port})" if available
+                            else "Kein Desktop verbunden"),
+            },
+            "timestamp": int(time.time() * 1000),
+        })
+
+    async def _vnc_open(self, session: str, port: int) -> None:
+        """Oeffnet die TCP-Verbindung zur QEMU-VNC und streamt RFB-Bytes als
+        vnc_data ueber RVS. Bestehende Session wird vorher geschlossen."""
+        await self._vnc_close(session)
+        try:
+            reader, writer = await asyncio.open_connection(self._vnc_host, port)
+        except Exception as exc:
+            logger.warning("[vnc] open %s:%d fehlgeschlagen: %s", self._vnc_host, port, exc)
+            await self._send_to_rvs({
+                "type": "desktop_status",
+                "payload": {"available": False, "session": session,
+                            "message": f"VNC-Verbindung fehlgeschlagen: {exc}"},
+                "timestamp": int(time.time() * 1000),
+            })
+            return
+        task = asyncio.create_task(self._vnc_reader_loop(session, reader))
+        self._vnc_sessions[session] = {"writer": writer, "task": task}
+        logger.info("[vnc] Tunnel offen: session=%s → %s:%d", session, self._vnc_host, port)
+
+    async def _vnc_reader_loop(self, session: str, reader: asyncio.StreamReader) -> None:
+        try:
+            while True:
+                data = await reader.read(16384)
+                if not data:
+                    break
+                await self._send_to_rvs({
+                    "type": "vnc_data",
+                    "payload": {"session": session, "b64": base64.b64encode(data).decode("ascii")},
+                    "timestamp": int(time.time() * 1000),
+                })
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.info("[vnc] reader-loop (%s) beendet: %s", session, exc)
+        finally:
+            # TCP-Ende → App informieren (Desktop weg).
+            if session in self._vnc_sessions:
+                await self._send_to_rvs({
+                    "type": "desktop_status",
+                    "payload": {"available": False, "session": session, "message": "VNC-Verbindung beendet"},
+                    "timestamp": int(time.time() * 1000),
+                })
+
+    async def _vnc_close(self, session: str) -> None:
+        sess = self._vnc_sessions.pop(session, None)
+        if not sess:
+            return
+        task = sess.get("task")
+        if task:
+            task.cancel()
+            try:
+                await task
+            except (asyncio.CancelledError, Exception):
+                pass
+        writer = sess.get("writer")
+        if writer:
+            try:
+                writer.close()
+                await writer.wait_closed()
+            except Exception:
+                pass
+        logger.info("[vnc] Tunnel geschlossen: session=%s", session)
+
     async def _delete_chat_message(self, ts: int) -> dict:
         """Entfernt eine Bubble: aus chat_backup.jsonl + Brain conversation,
         broadcastet chat_message_deleted via RVS.