feat: refactored webrtc for bi-directional

Author: iamprashant

package.json

@@ -1,6 +1,6 @@
 {
     "name": "@rapidaai/react",
-    "version": "1.1.77",
+    "version": "1.1.78",
     "description": "An easy to use react client for building generative ai application using Rapida platform.",
     "repository": {
         "type": "git",

src/agents/voice-agent.ts

@@ -85,7 +85,10 @@ export class VoiceAgent extends Agent {
           // (the server's ConversationInitialization response) to confirm
           // the conversation is truly established.
           // console.log(`${LOG_PREFIX} transport connected, waiting for conversation initialization`);
-        } else if (state === "disconnected" || state === "closed" || state === "failed") {
+        } else if (state === "disconnected") {
+          // Browser WebRTC can briefly report disconnected during ICE changes.
+          // Keep the selected channel stable while recovery/renegotiation runs.
+        } else if (state === "closed" || state === "failed") {
           // Only mark as disconnected if the gRPC session is also gone.
           // When switching from audio → text the WebRTC peer fires
           // "disconnected"/"closed", but the gRPC stream is still alive.
@@ -679,32 +682,35 @@ export class VoiceAgent extends Agent {
     this.agentConfig.inputOptions.channel = input;
     this.agentConfig.outputOptions.channel = input;
 
-    // Track whether a fresh connection was just created.
-    // If so, ConversationInitialization already carried the stream mode
-    // — no need to send a redundant ConversationConfiguration.
-    const wasConnected = this.webrtcTransport?.isGrpcConnected ?? false;
+    // Fresh Audio connect captures mic during initialization. Runtime switches
+    // recapture before sending configuration so the offer sees a stable track.
+    const wasGrpcConnectedBeforeChannelChange = this.webrtcTransport?.isGrpcConnected ?? false;
 
     // Ensure the session exists
     await this.ensureConnected();
 
     // Add or remove the audio transport
     if (input === Channel.Audio) {
-      try {
-        await this.webrtcTransport?.reconnectAudio();
-      } catch (error) {
-        // gRPC stream was lost between ensureConnected and reconnectAudio.
-        // Trigger a full reconnect on the next user action.
-        this.emit(AgentEvent.ErrorEvent, "client", `Audio reconnect failed: ${error}`);
-        this.webrtcTransport = null;
+      if (wasGrpcConnectedBeforeChannelChange) {
+        try {
+          await this.webrtcTransport?.reconnectAudio();
+        } catch (error) {
+          // gRPC stream was lost between ensureConnected and reconnectAudio.
+          // Trigger a full reconnect on the next user action.
+          this.emit(AgentEvent.ErrorEvent, "client", `Audio reconnect failed: ${error}`);
+          this.webrtcTransport = null;
+        }
       }
     } else {
-      await this.webrtcTransport?.disconnectAudioOnly();
+      if (wasGrpcConnectedBeforeChannelChange) {
+        await this.webrtcTransport?.disconnectAudioOnly();
+      }
     }
 
     // Only send ConversationConfiguration for runtime mode switches.
     // On a fresh connection the ConversationInitialization message
     // already includes the stream mode, so sending it again is redundant.
-    if (wasConnected) {
+    if (wasGrpcConnectedBeforeChannelChange) {
       this.webrtcTransport?.sendConversationConfiguration();
     }
 

src/audio/audio-media-manager.ts

@@ -102,15 +102,17 @@ export class AudioMediaManager {
       }
     }
 
-    const track = this.localStream.getAudioTracks()[0];
-    if (track) {
-      track.enabled = true;
-
-      // Store the actual device being used when none was explicitly selected
-      const settings = track.getSettings();
-      if (settings.deviceId && !this.agentConfig.inputOptions.device) {
-        this.agentConfig.inputOptions.device = settings.deviceId;
-      }
+    const localAudioTrack = this.localStream.getAudioTracks()[0];
+    if (!localAudioTrack) {
+      throw new Error("[AudioMediaManager] Microphone capture returned no audio track");
+    }
+
+    localAudioTrack.enabled = true;
+
+    // Store the actual device being used when none was explicitly selected.
+    const localAudioTrackSettings = localAudioTrack.getSettings();
+    if (localAudioTrackSettings.deviceId && !this.agentConfig.inputOptions.device) {
+      this.agentConfig.inputOptions.device = localAudioTrackSettings.deviceId;
     }
 
     // Audio context for visualization - use vendor-prefixed version if needed
@@ -165,7 +167,7 @@ export class AudioMediaManager {
     const base: MediaTrackConstraints = {
       echoCancellation: true,
       noiseSuppression: true,
-      autoGainControl: false,
+      autoGainControl: true,
     };
 
     if (this.agentConfig.inputOptions.device) {
@@ -183,11 +185,9 @@ export class AudioMediaManager {
       channelCount: { ideal: 1 },
       echoCancellation: true,
       noiseSuppression: true,
-      // Disable AGC for voice AI: when the assistant finishes speaking and the
-      // user starts talking, AGC would abruptly boost the mic gain (it was
-      // suppressed while output was playing), clipping the first syllable and
-      // causing a jarring volume jump. The STT model handles normalization.
-      autoGainControl: false,
+      // Keep AGC enabled to prevent low microphone capture levels on quieter
+      // mics and mobile hardware.
+      autoGainControl: true,
     };
 
     if (this.agentConfig.inputOptions.device) {
@@ -213,8 +213,8 @@ export class AudioMediaManager {
         ...base,
         // @ts-ignore Chrome/Edge-specific
         googEchoCancellation: true,
-        // @ts-ignore AGC disabled — matches autoGainControl: false above
-        googAutoGainControl: false,
+        // @ts-ignore Chrome/Edge-specific AGC.
+        googAutoGainControl: true,
         // @ts-ignore
         googNoiseSuppression: true,
         // @ts-ignore removes low-frequency rumble (fan/AC noise)
@@ -362,9 +362,19 @@ export class AudioMediaManager {
       // the load algorithm (which internally pauses), so play() must wait until
       // the element is ready or AbortError follows.
       if (el.readyState < 3 /* HAVE_FUTURE_DATA */) {
-        await new Promise<void>((resolve) => {
-          el.addEventListener('canplay', resolve as EventListener, { once: true });
-          setTimeout(resolve, 200);
+        await new Promise<void>((resolvePromise) => {
+          let isPromiseResolved = false;
+          // Resolve from whichever arrives first: `canplay` or fallback timeout.
+          const resolveOnce = () => {
+            if (isPromiseResolved) return;
+            isPromiseResolved = true;
+            clearTimeout(fallbackTimeoutId);
+            el.removeEventListener("canplay", handleCanPlay);
+            resolvePromise();
+          };
+          const handleCanPlay: EventListener = () => resolveOnce();
+          const fallbackTimeoutId = setTimeout(resolveOnce, 200);
+          el.addEventListener("canplay", handleCanPlay, { once: true });
         });
       }
 
@@ -417,9 +427,19 @@ export class AudioMediaManager {
       // complete before resuming. 200 ms timeout guards against streams
       // with no active audio tracks where canplay may never fire.
       if (el.readyState < 3 /* HAVE_FUTURE_DATA */) {
-        await new Promise<void>((resolve) => {
-          el.addEventListener('canplay', resolve as EventListener, { once: true });
-          setTimeout(resolve, 200);
+        await new Promise<void>((resolvePromise) => {
+          let isPromiseResolved = false;
+          // Resolve from whichever arrives first: `canplay` or fallback timeout.
+          const resolveOnce = () => {
+            if (isPromiseResolved) return;
+            isPromiseResolved = true;
+            clearTimeout(fallbackTimeoutId);
+            el.removeEventListener("canplay", handleCanPlay);
+            resolvePromise();
+          };
+          const handleCanPlay: EventListener = () => resolveOnce();
+          const fallbackTimeoutId = setTimeout(resolveOnce, 200);
+          el.addEventListener("canplay", handleCanPlay, { once: true });
         });
       }
 
@@ -504,6 +524,12 @@ export class AudioMediaManager {
       }
     }
 
+    const localAudioTrack = this.localStream.getAudioTracks()[0];
+    if (!localAudioTrack) {
+      throw new Error("[AudioMediaManager] Microphone device switch returned no audio track");
+    }
+    localAudioTrack.enabled = !this.isMuted;
+
     // Reconnect analyser (disconnect old source first to avoid leaking audio nodes)
     if (this.audioContext && this._inputAnalyser && this.localStream) {
       this._inputAnalyser.disconnect();
@@ -554,7 +580,7 @@ export class AudioMediaManager {
 
   /** Get the new audio track after setInputDevice (caller wires it to peer) */
   getLocalAudioTrack(): MediaStreamTrack | undefined {
-    return this.localStream?.getAudioTracks()[0];
+    return this.localStream?.getAudioTracks().find(track => track.readyState === "live");
   }
 
   // ---------------------------------------------------------------------------

src/audio/message-protocol-handler.ts

@@ -45,12 +45,13 @@ import type { AudioMediaManager } from "./audio-media-manager";
  * - Handle conversation init/config responses
  * - Handle assistant / user / interruption / tool call messages
  *
- * This is pure dispatch logic — it holds no state of its own.
+ * State: tracks the active WebRTC signaling session to ignore stale offers/ICE.
  */
 export class MessageProtocolHandler {
   // Serialization chain ensures signaling messages (CONFIG → OFFER → ICE)
   // are processed in order even though gRPC callbacks fire without await.
   private processingChain: Promise<void> = Promise.resolve();
+  private activeSignalingSessionId: string | null = null;
 
   constructor(
     private callbacks: AgentCallback,
@@ -159,45 +160,74 @@ export class MessageProtocolHandler {
   // ---------------------------------------------------------------------------
 
   private async handleServerSignaling(signaling: ServerSignaling): Promise<void> {
-    const sessionId = signaling.getSessionid();
-    if (sessionId) this.signaling.setSessionId(sessionId);
-
+    const signalingSessionId = signaling.getSessionid();
     const messageCase = signaling.getMessageCase();
 
+    if (messageCase !== ServerSignaling.MessageCase.CONFIG) {
+      if (this.activeSignalingSessionId && signalingSessionId && signalingSessionId !== this.activeSignalingSessionId) {
+        console.debug("[Protocol] Ignoring stale WebRTC signaling", {
+          activeSignalingSessionId: this.activeSignalingSessionId,
+          signalingSessionId,
+          messageCase,
+        });
+        return;
+      }
+      if (!this.activeSignalingSessionId && signalingSessionId) {
+        this.activeSignalingSessionId = signalingSessionId;
+        this.signaling.setSessionId(signalingSessionId);
+      }
+    }
+
     switch (messageCase) {
       case ServerSignaling.MessageCase.CONFIG: {
-        const config = signaling.getConfig();
-        const iceServers = config?.getIceserversList()?.map(srv => ({
-          urls: srv.getUrlsList(),
-          username: srv.getUsername() || undefined,
-          credential: srv.getCredential() || undefined,
-        })) as RTCIceServer[] | undefined;
-        this.peer.setup(iceServers);
+        try {
+          this.activeSignalingSessionId = signalingSessionId || null;
+          if (this.activeSignalingSessionId) {
+            this.signaling.setSessionId(this.activeSignalingSessionId);
+          }
+
+          const config = signaling.getConfig();
+          const iceServers = config?.getIceserversList()?.map(srv => ({
+            urls: srv.getUrlsList(),
+            username: srv.getUsername() || undefined,
+            credential: srv.getCredential() || undefined,
+          })) as RTCIceServer[] | undefined;
+          this.peer.setup(iceServers);
+        } catch (error) {
+          this.peer.close();
+          console.error("Failed to handle WebRTC config signaling", error);
+          this.callbacks.onConnectionStateChange?.("failed");
+          this.callbacks.onError?.(new Error(`WebRTC config signaling failed: ${error}`));
+        }
         break;
       }
 
       case ServerSignaling.MessageCase.SDP: {
+        const sdpMessage = signaling.getSdp();
+        const sdpType = sdpMessage?.getType();
         try {
-          const sdpMsg = signaling.getSdp();
-          if (!sdpMsg) break;
+          if (!sdpMessage) break;
 
-          const sdpType = sdpMsg.getType();
-          const sdp = sdpMsg.getSdp();
+          const sdp = sdpMessage.getSdp();
 
           if (sdpType === WebRTCSDP.SDPType.OFFER) {
             // console.log("[Protocol] Handling SDP OFFER");
-            await this.peer.handleOffer(sdp, this.audio.mediaStream);
-            const answerSdp = this.peer.getLocalSDP();
+            const answerSdp = await this.peer.handleOffer(sdp, this.audio.mediaStream);
             if (answerSdp) {
               this.signaling.sendWebRTCAnswer(answerSdp);
             } else {
-              console.error("No local SDP after creating answer");
+              console.debug("[Protocol] SDP offer was superseded before answer was sent");
             }
           } else if (sdpType === WebRTCSDP.SDPType.ANSWER) {
             // console.log("[Protocol] Handling SDP ANSWER");
             await this.peer.handleAnswer(sdp);
           }
         } catch (error) {
+          if (sdpType === WebRTCSDP.SDPType.OFFER) {
+            // Cleanup strategy: on failed offer handling, close stale peer state.
+            // Next CONFIG/OFFER cycle will recreate the peer cleanly.
+            this.peer.close();
+          }
           console.error("Failed to handle SDP signaling", error);
           this.callbacks.onConnectionStateChange?.("failed");
           this.callbacks.onError?.(new Error(`SDP signaling failed: ${error}`));