Buffer ICE without changing WebRTC component identity

streaming_app.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import copy
 import json
 import multiprocessing
@@ -8,6 +9,7 @@ import tarfile
 import threading
 import time
 import uuid
+from collections import defaultdict
 from pathlib import Path
 
 import gradio as gr
@@ -224,6 +226,47 @@ async def get_turn_configuration():
     return configuration
 
 
+_fastrtc_handle_offer = WebRTC.handle_offer
+
+
+async def reliable_handle_offer(self, body, set_outputs):
+    """Buffer trickled ICE candidates until FastRTC registers their SDP offer."""
+    lock = getattr(self, "_roleforge_signaling_lock", None)
+    if lock is None:
+        lock = asyncio.Lock()
+        self._roleforge_signaling_lock = lock
+        self._roleforge_pending_ice = defaultdict(list)
+
+    async with lock:
+        webrtc_id = body.get("webrtc_id")
+        is_candidate = body.get("type") == "ice-candidate" and "candidate" in body
+
+        if is_candidate and webrtc_id not in self.pcs:
+            pending = self._roleforge_pending_ice[webrtc_id]
+            if len(pending) < 32:
+                pending.append(body)
+            print(f"Buffered early ICE candidate for pending connection: {webrtc_id}")
+            return {"status": "success"}
+
+        response = await _fastrtc_handle_offer(self, body, set_outputs)
+
+        if not is_candidate and webrtc_id in self.pcs:
+            pending = self._roleforge_pending_ice.pop(webrtc_id, [])
+            for candidate in pending:
+                await _fastrtc_handle_offer(self, candidate, set_outputs)
+            print(
+                f"Registered WebRTC offer and replayed {len(pending)} early ICE candidates: "
+                f"{webrtc_id}"
+            )
+
+        return response
+
+
+# Preserve FastRTC's exact WebRTC component class/frontend bundle. Subclassing a
+# custom Gradio component changes its frontend component name and breaks SSR.
+WebRTC.handle_offer = reliable_handle_offer
+
+
 def new_bridge(oauth_token: gr.OAuthToken | None):
     remember_turn_token(oauth_token)
     input_queue = _manager.Queue(INPUT_QUEUE_FRAMES)
@@ -284,6 +327,8 @@ class PersonaPlexQueueHandler(StreamHandler):
             input_sample_rate=OUTPUT_SAMPLE_RATE,
             fps=WEBRTC_FRAME_RATE,
         )
+        self._logged_input = False
+        self._logged_output = False
 
     def _bridge(self):
         # FastRTC prepends the WebRTC component value to additional inputs.
@@ -301,6 +346,9 @@ class PersonaPlexQueueHandler(StreamHandler):
             return
         sample_rate, array = frame
         audio = np.asarray(array, dtype=np.int16).reshape(-1)
+        if not self._logged_input:
+            print("WebRTC audio bridge received its first input frame.")
+            self._logged_input = True
         if bounded_put(input_queue, (int(sample_rate), audio)):
             counters["input_drops"] = int(counters.get("input_drops", 0)) + 1
 
@@ -315,6 +363,9 @@ class PersonaPlexQueueHandler(StreamHandler):
             audio = output_queue.get_nowait()
         except queue.Empty:
             return None
+        if not self._logged_output:
+            print("WebRTC audio bridge emitted its first output frame.")
+            self._logged_output = True
         return OUTPUT_SAMPLE_RATE, np.asarray(audio, dtype=np.int16).reshape(1, -1)
 
     def copy(self):
@@ -407,6 +458,7 @@ def run_streaming_session(
         drain_proxy_queue(input_queue)
         ready_seconds = round(time.perf_counter() - session_started, 3)
         ready_event.set()
+        print("PersonaPlex GPU session is ready and waiting for WebRTC audio.")
         live_started = time.perf_counter()
         status = (
             f"**Engine ready in {ready_seconds:.1f}s.** Start the WebRTC microphone and speak naturally. "