From ee7bb09b4dc734bbae2f7b09a898885726198599 Mon Sep 17 00:00:00 2001
From: Arik Sfaradi <sfa.arik@gmail.com>
Date: Sun, 14 Dec 2025 17:22:36 +0200
Subject: [PATCH 1/3] Bugfix/livekit (#237)

* pass debug configuration from demo app

* Fix freeze issue -
livekit get tracks that need to be attached to video elem - since we pass the MediaStream - create hidden element and attach it there - keeps the track active

* fixing bug that sometimes the streaming manager is undefined

* in livekit there is no need to use ws - all data is from the dataChannel

* use livekit chat for streams v2

* sendTextMessage is in livekit usage

* handle activity state changes

* use preppareConnection to speed up connection time

* use sendTextMessage to publish messages
topic: 'lk.chat' - indicates that the streamingOrchestrator use built in AgentSession

* fix init issues

* monitor faulure events

* use sendTextMessage

* indicate that streaming started

* refactoring

* resolve conflicts

* cursor review

* review fix - redundant

* review - unused

* review - add doc that indicates its only v2

* review
---
 demo/app.tsx                                  |  11 +-
 demo/environment.ts                           |   1 +
 demo/hooks/useAgentManager.ts                 |   3 +
 .../agent-manager/connect-to-manager.ts       |  14 +-
 src/services/agent-manager/index.ts           |  79 +++---
 src/services/streaming-manager/common.ts      |   7 +
 .../streaming-manager/livekit-manager.ts      | 259 +++++++++++++-----
 7 files changed, 268 insertions(+), 106 deletions(-)

diff --git a/demo/app.tsx b/demo/app.tsx
index 5df62f90..6aa04403 100644
--- a/demo/app.tsx
+++ b/demo/app.tsx
@@ -2,7 +2,7 @@ import { ChatMode, ConnectionState } from '@sdk/types';
 import { useEffect, useRef, useState } from 'preact/hooks';
 
 import './app.css';
-import { agentId, clientKey, didApiUrl, didSocketApiUrl } from './environment';
+import { agentId, clientKey, debug, didApiUrl, didSocketApiUrl } from './environment';
 import { useAgentManager } from './hooks/useAgentManager';
 
 export function App() {
@@ -19,6 +19,7 @@ export function App() {
 
     const { srcObject, connectionState, messages, isSpeaking, connect, disconnect, speak, chat, interrupt } =
         useAgentManager({
+            debug,
             agentId,
             baseURL: didApiUrl,
             wsURL: didSocketApiUrl,
@@ -64,10 +65,10 @@ export function App() {
                             {connectionState === ConnectionState.Connected
                                 ? 'Send'
                                 : connectionState === ConnectionState.Connecting
-                                  ? 'Connecting...'
-                                  : connectionState === ConnectionState.Fail
-                                    ? 'Failed, Try Again'
-                                    : 'Connect'}
+                                    ? 'Connecting...'
+                                    : connectionState === ConnectionState.Fail
+                                        ? 'Failed, Try Again'
+                                        : 'Connect'}
                         </button>
 
                         <button
diff --git a/demo/environment.ts b/demo/environment.ts
index 413966e9..866808a5 100644
--- a/demo/environment.ts
+++ b/demo/environment.ts
@@ -1,4 +1,5 @@
 export const nodeEnv = import.meta.env.VITE_NODE_ENV as string;
+export const debug = (import.meta.env.VITE_DEBUG as boolean) || false;
 export const didApiUrl = import.meta.env.VITE_DID_API_URL as string;
 export const didSocketApiUrl = import.meta.env.VITE_WS_ENDPOINT as string;
 export const mixpanelKey = import.meta.env.VITE_MIXPANEL_KEY as string;
diff --git a/demo/hooks/useAgentManager.ts b/demo/hooks/useAgentManager.ts
index 600359b2..3696d393 100644
--- a/demo/hooks/useAgentManager.ts
+++ b/demo/hooks/useAgentManager.ts
@@ -27,6 +27,7 @@ interface UseAgentManagerOptions {
     externalId?: string;
     mixpanelKey?: string;
     mixpanelAdditionalProperties?: Record<string, any>;
+    debug?: boolean;
 }
 
 export function useAgentManager(props: UseAgentManagerOptions) {
@@ -41,6 +42,7 @@ export function useAgentManager(props: UseAgentManagerOptions) {
         streamOptions,
         mixpanelKey,
         mixpanelAdditionalProperties,
+        debug,
     } = props;
 
     const [isSpeaking, setIsSpeaking] = useState(false);
@@ -104,6 +106,7 @@ export function useAgentManager(props: UseAgentManagerOptions) {
                 mixpanelKey,
                 mixpanelAdditionalProperties,
                 streamOptions,
+                debug,
             });
 
             await newManager.connect();
diff --git a/src/services/agent-manager/connect-to-manager.ts b/src/services/agent-manager/connect-to-manager.ts
index 8151904f..c5df9635 100644
--- a/src/services/agent-manager/connect-to-manager.ts
+++ b/src/services/agent-manager/connect-to-manager.ts
@@ -172,6 +172,8 @@ function connectToManager(
     return new Promise(async (resolve, reject) => {
         try {
             let streamingManager: StreamingManager<CreateStreamOptions | CreateStreamV2Options>;
+            let shouldResolveOnComplete = false;
+
             streamingManager = await createStreamingManager(agent, getAgentStreamOptions(agent, options), {
                 ...options,
                 analytics,
@@ -181,7 +183,13 @@ function connectToManager(
                         options.callbacks.onConnectionStateChange?.(state);
 
                         if (state === ConnectionState.Connected) {
-                            resolve(streamingManager);
+                            // If manager is ready, resolve immediately
+                            // Otherwise, mark to resolve after manager is created
+                            if (streamingManager) {
+                                resolve(streamingManager);
+                            } else {
+                                shouldResolveOnComplete = true;
+                            }
                         }
                     },
                     onVideoStateChange: (state: StreamingState, statsReport?: any) => {
@@ -213,6 +221,10 @@ function connectToManager(
                     },
                 },
             });
+
+            if (shouldResolveOnComplete) {
+                resolve(streamingManager);
+            }
         } catch (error) {
             reject(error);
         }
diff --git a/src/services/agent-manager/index.ts b/src/services/agent-manager/index.ts
index 6c0d81f5..dfdce681 100644
--- a/src/services/agent-manager/index.ts
+++ b/src/services/agent-manager/index.ts
@@ -3,6 +3,7 @@ import {
     AgentManagerOptions,
     Chat,
     ChatMode,
+    ChatResponse,
     ConnectionState,
     CreateStreamOptions,
     Interrupt,
@@ -15,6 +16,7 @@ import { CONNECTION_RETRY_TIMEOUT_MS } from '@sdk/config/consts';
 import { didApiUrl, didSocketApiUrl, mixpanelKey } from '@sdk/config/environment';
 import { ChatCreationFailed, ValidationError } from '@sdk/errors';
 import { getRandom } from '@sdk/utils';
+import { isStreamsV2Agent } from '@sdk/utils/agent';
 import { isChatModeWithoutChat, isTextualChat } from '@sdk/utils/chat';
 import { createAgentsApi } from '../../api/agents';
 import { getAgentInfo, getAnalyticsInfo } from '../../utils/analytics';
@@ -72,6 +74,7 @@ export async function createAgentManager(agent: string, options: AgentManagerOpt
     const agentsApi = createAgentsApi(options.auth, baseURL, options.callbacks.onError, options.externalId);
 
     const agentEntity = await agentsApi.getById(agent);
+    const isStreamsV2 = isStreamsV2Agent(agentEntity.presenter.type);
     analytics.enrich(getAgentInfo(agentEntity));
 
     const { onMessage, clearQueue } = createMessageEventQueue(analytics, items, options, agentEntity, () =>
@@ -100,7 +103,7 @@ export async function createAgentManager(agent: string, options: AgentManagerOpt
         }
 
         const websocketPromise =
-            options.mode === ChatMode.DirectPlayback
+            options.mode === ChatMode.DirectPlayback || isStreamsV2
                 ? Promise.resolve(undefined)
                 : createSocketManager(
                       options.auth,
@@ -253,41 +256,47 @@ export async function createAgentManager(agent: string, options: AgentManagerOpt
             };
 
             const sendChatRequest = async (messages: Message[], chatId: string) => {
-                return retryOperation(
-                    () => {
-                        return agentsApi.chat(
-                            agentEntity.id,
-                            chatId,
-                            {
-                                chatMode: items.chatMode,
-                                streamId: items.streamingManager?.streamId,
-                                sessionId: items.streamingManager?.sessionId,
-                                messages: messages.map(({ matches, ...message }) => message),
-                            },
-                            {
-                                ...getRequestHeaders(items.chatMode),
-                                skipErrorHandler: true,
-                            }
-                        );
+                const chatRequestFn = isStreamsV2
+                    ? async () => {
+                          await items.streamingManager?.sendTextMessage?.(userMessage);
+                          return Promise.resolve({
+                              result: 'This is a mock message until the data channel is implemented.',
+                          } as ChatResponse);
+                      }
+                    : async () => {
+                          return agentsApi.chat(
+                              agentEntity.id,
+                              chatId,
+                              {
+                                  chatMode: items.chatMode,
+                                  streamId: items.streamingManager?.streamId,
+                                  sessionId: items.streamingManager?.sessionId,
+                                  messages: messages.map(({ matches, ...message }) => message),
+                              },
+                              {
+                                  ...getRequestHeaders(items.chatMode),
+                                  skipErrorHandler: true,
+                              }
+                          );
+                      };
+
+                return retryOperation(chatRequestFn, {
+                    limit: 2,
+                    shouldRetryFn: error => {
+                        const isInvalidSessionId = error?.message?.includes('missing or invalid session_id');
+                        const isStreamError = error?.message?.includes('Stream Error');
+
+                        if (!isStreamError && !isInvalidSessionId) {
+                            options.callbacks.onError?.(error);
+                            return false;
+                        }
+                        return true;
                     },
-                    {
-                        limit: 2,
-                        shouldRetryFn: error => {
-                            const isInvalidSessionId = error?.message?.includes('missing or invalid session_id');
-                            const isStreamError = error?.message?.includes('Stream Error');
-
-                            if (!isStreamError && !isInvalidSessionId) {
-                                options.callbacks.onError?.(error);
-                                return false;
-                            }
-                            return true;
-                        },
-                        onRetry: async () => {
-                            await disconnect();
-                            await connect(false);
-                        },
-                    }
-                );
+                    onRetry: async () => {
+                        await disconnect();
+                        await connect(false);
+                    },
+                });
             };
 
             try {
diff --git a/src/services/streaming-manager/common.ts b/src/services/streaming-manager/common.ts
index e34d4f2a..60752df7 100644
--- a/src/services/streaming-manager/common.ts
+++ b/src/services/streaming-manager/common.ts
@@ -25,6 +25,13 @@ export type StreamingManager<T extends CreateStreamOptions | CreateStreamV2Optio
      */
     sendDataChannelMessage(payload: string): void;
 
+    /**
+     * Method to send text messages to the server
+     * @param payload The message payload to send
+     * supported only for livekit manager
+     */
+    sendTextMessage?(payload: string): Promise<void>;
+
     /**
      * Session identifier information, should be returned in the body of all streaming requests
      */
diff --git a/src/services/streaming-manager/livekit-manager.ts b/src/services/streaming-manager/livekit-manager.ts
index addf97a0..68b88fa0 100644
--- a/src/services/streaming-manager/livekit-manager.ts
+++ b/src/services/streaming-manager/livekit-manager.ts
@@ -16,11 +16,14 @@ import { didApiUrl } from '../../config/environment';
 import { createStreamingLogger, StreamingManager } from './common';
 
 import type {
+    ConnectionQuality,
     ConnectionState as LiveKitConnectionState,
+    Participant,
     RemoteParticipant,
     RemoteTrack,
     Room,
     RoomEvent,
+    SubscriptionError,
 } from 'livekit-client';
 
 async function importLiveKit(): Promise<{
@@ -39,6 +42,44 @@ async function importLiveKit(): Promise<{
     }
 }
 
+function attachHiddenElement(track: RemoteTrack, attachedElements: HTMLMediaElement[]): HTMLMediaElement {
+    const hiddenElement = track.attach();
+    attachedElements.push(hiddenElement);
+    hiddenElement.style.position = 'absolute';
+    hiddenElement.style.width = '1px';
+    hiddenElement.style.height = '1px';
+    hiddenElement.style.opacity = '0.01';
+    hiddenElement.style.pointerEvents = 'none';
+    hiddenElement.muted = true;
+    return hiddenElement;
+}
+
+const connectivityQualityToState = {
+    excellent: ConnectivityState.Strong,
+    good: ConnectivityState.Strong,
+    poor: ConnectivityState.Weak,
+    lost: ConnectivityState.Unknown,
+    unknown: ConnectivityState.Unknown,
+};
+
+const internalErrorMassage = JSON.stringify({
+    kind: 'InternalServerError',
+    description: 'Stream Error',
+});
+
+export function handleInitError(
+    error: unknown,
+    log: (message?: any, ...optionalParams: any[]) => void,
+    callbacks: StreamingManagerOptions['callbacks'],
+    markInitialConnectionDone: () => void
+): void {
+    log('Failed to connect to LiveKit room:', error);
+    markInitialConnectionDone();
+    callbacks.onConnectionStateChange?.(ConnectionState.Fail);
+    callbacks.onError?.(error as Error, { streamId: '' });
+    throw error;
+}
+
 export async function createLiveKitStreamingManager<T extends CreateStreamV2Options>(
     agentId: string,
     agent: T,
@@ -58,15 +99,75 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
     let room: Room | null = null;
     let isConnected = false;
     let videoId: string | null = null;
-    let mediaStream: MediaStream | null = null;
     const streamType = StreamType.Fluent;
+    let isInitialConnection = true;
+    const attachedElements: HTMLMediaElement[] = [];
 
     room = new Room({
         adaptiveStream: true,
         dynacast: true,
     });
 
-    room.on(RoomEvent.ConnectionStateChanged, state => {
+    const streamApi = createStreamApiV2(auth, baseURL || didApiUrl, agentId, callbacks.onError);
+    let streamId: string | undefined;
+    let sessionId: string | undefined;
+
+    let token: string | undefined;
+    let url: string | undefined;
+
+    try {
+        const streamResponse = await streamApi.createStream({
+            transport_provider: TransportProvider.Livekit,
+            chat_id: agent.chat_id,
+        });
+
+        const { session_id, session_token, session_url } = streamResponse;
+        callbacks.onStreamCreated?.({ stream_id: session_id, session_id, agent_id: agentId });
+        streamId = session_id;
+        sessionId = session_id;
+        token = session_token;
+        url = session_url;
+
+        await room.prepareConnection(url, token);
+    } catch (error) {
+        handleInitError(error, log, callbacks, () => {
+            isInitialConnection = false;
+        });
+    }
+
+    if (!url || !token || !streamId || !sessionId) {
+        return Promise.reject(new Error('Failed to initialize LiveKit stream'));
+    }
+
+    room.on(RoomEvent.ConnectionStateChanged, handleConnectionStateChanged)
+        .on(RoomEvent.ConnectionQualityChanged, handleConnectionQualityChanged)
+        .on(RoomEvent.ActiveSpeakersChanged, handleActiveSpeakersChanged)
+        .on(RoomEvent.ParticipantConnected, handleParticipantConnected)
+        .on(RoomEvent.TrackSubscribed, handleTrackSubscribed)
+        .on(RoomEvent.TrackUnsubscribed, handleTrackUnsubscribed)
+        .on(RoomEvent.DataReceived, handleDataReceived)
+        .on(RoomEvent.MediaDevicesError, handleMediaDevicesError)
+        .on(RoomEvent.EncryptionError, handleEncryptionError)
+        .on(RoomEvent.TrackSubscriptionFailed, handleTrackSubscriptionFailed);
+
+    callbacks.onConnectionStateChange?.(ConnectionState.New);
+
+    try {
+        await room.connect(url, token);
+        log('LiveKit room joined successfully');
+
+        isInitialConnection = false;
+    } catch (error) {
+        handleInitError(error, log, callbacks, () => {
+            isInitialConnection = false;
+        });
+    }
+
+    analytics.enrich({
+        'stream-type': streamType,
+    });
+
+    function handleConnectionStateChanged(state: LiveKitConnectionState): void {
         log('Connection state changed:', state);
         switch (state) {
             case LiveKitConnectionState.Connecting:
@@ -75,7 +176,14 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
             case LiveKitConnectionState.Connected:
                 log('LiveKit room connected successfully');
                 isConnected = true;
-                callbacks.onConnectionStateChange?.(ConnectionState.Connected);
+                // During initial connection, defer the callback to ensure manager is returned first
+                if (isInitialConnection) {
+                    queueMicrotask(() => {
+                        callbacks.onConnectionStateChange?.(ConnectionState.Connected);
+                    });
+                } else {
+                    callbacks.onConnectionStateChange?.(ConnectionState.Connected);
+                }
                 break;
             case LiveKitConnectionState.Disconnected:
                 log('LiveKit room disconnected');
@@ -91,49 +199,71 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
                 callbacks.onConnectionStateChange?.(ConnectionState.Connecting);
                 break;
         }
-    });
+    }
 
-    room.on(RoomEvent.ConnectionQualityChanged, (quality, participant) => {
-        if (participant?.isLocal && quality === 'poor') {
-            log('Connection quality is poor');
-            callbacks.onConnectivityStateChange?.(ConnectivityState.Weak);
+    function handleConnectionQualityChanged(quality: ConnectionQuality, participant?: Participant): void {
+        log('Connection quality:', quality);
+        if (participant?.isLocal) {
+            callbacks.onConnectivityStateChange?.(connectivityQualityToState[quality]);
         }
-    });
+    }
 
-    // Handle room closure - this would be handled by the server or other means
-    // For now, we'll handle Closed state in disconnect method
+    function handleActiveSpeakersChanged(activeSpeakers: Participant[]): void {
+        log('Active speakers changed:', activeSpeakers);
+        const activeSpeaker = activeSpeakers[0];
+        if (activeSpeaker) {
+            callbacks.onAgentActivityStateChange?.(AgentActivityState.Talking);
+        } else {
+            callbacks.onAgentActivityStateChange?.(AgentActivityState.Idle);
+        }
+    }
 
-    room.on(RoomEvent.ParticipantConnected, (participant: RemoteParticipant) => {
+    function handleParticipantConnected(participant: RemoteParticipant): void {
         log('Participant connected:', participant.identity);
-    });
+    }
 
-    room.on(RoomEvent.TrackSubscribed, (track: RemoteTrack, publication, participant: RemoteParticipant) => {
+    function handleTrackSubscribed(track: RemoteTrack, publication: any, participant: RemoteParticipant): void {
         log(`Track subscribed: ${track.kind} from ${participant.identity}`);
 
-        if (!mediaStream) {
-            mediaStream = new MediaStream([track.mediaStreamTrack]);
-        } else {
-            mediaStream.addTrack(track.mediaStreamTrack);
+        if (track.kind === 'video') {
+            const hiddenElement = attachHiddenElement(track, attachedElements);
+            document.body.appendChild(hiddenElement);
+
+            // Play the hidden element to keep the track "alive"
+            hiddenElement
+                .play()
+                .then(() => {
+                    log('Hidden video element playing');
+                })
+                .catch(e => log('Error playing hidden element:', e));
+
+            log(`Video element created, srcObject: ${hiddenElement.srcObject}`);
+
+            if (hiddenElement.srcObject) {
+                callbacks.onSrcObjectReady?.(hiddenElement.srcObject as MediaStream);
+                callbacks.onVideoStateChange?.(StreamingState.Start);
+            }
         }
 
-        callbacks.onSrcObjectReady?.(mediaStream);
-
-        // Handle video track subscription as stats signal for fluent streams
-        if (track.kind === 'video') {
-            callbacks.onVideoStateChange?.(StreamingState.Start);
+        if (track.kind === 'audio') {
+            // For audio, create element and play it directly
+            const audioElement = track.attach();
+            attachedElements.push(audioElement);
+            audioElement.style.display = 'none';
+            document.body.appendChild(audioElement);
+            audioElement.play().catch(e => log('Error playing audio element:', e));
         }
-    });
+    }
 
-    room.on(RoomEvent.TrackUnsubscribed, (track: RemoteTrack, publication, participant: RemoteParticipant) => {
+    function handleTrackUnsubscribed(track: RemoteTrack, publication: any, participant: RemoteParticipant): void {
         log(`Track unsubscribed: ${track.kind} from ${participant.identity}`);
 
-        // Handle video track unsubscription as stats signal for fluent streams
         if (track.kind === 'video') {
             callbacks.onVideoStateChange?.(StreamingState.Stop);
         }
-    });
+    }
 
-    room.on(RoomEvent.DataReceived, (payload: Uint8Array, participant?: RemoteParticipant) => {
+    function handleDataReceived(payload: Uint8Array, participant?: RemoteParticipant): void {
         const message = new TextDecoder().decode(payload);
         log('Data received:', message);
 
@@ -151,51 +281,52 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
         } catch (e) {
             log('Failed to parse data channel message:', e);
         }
-    });
-
-    callbacks.onConnectionStateChange?.(ConnectionState.New);
-
-    const streamApi = createStreamApiV2(auth, baseURL || didApiUrl, agentId, callbacks.onError);
-    let streamId: string;
-    let sessionId: string;
+    }
 
-    try {
-        const streamResponse = await streamApi.createStream({
-            transport_provider: TransportProvider.Livekit,
-            chat_id: agent.chat_id,
-        });
+    function handleMediaDevicesError(error: Error): void {
+        log('Media devices error:', error);
+        callbacks.onError?.(new Error(internalErrorMassage), { streamId });
+    }
 
-        const { agent_id, session_id, session_token: token, session_url: url } = streamResponse;
-        streamId = agent_id;
-        sessionId = session_id;
+    function handleEncryptionError(error: Error): void {
+        log('Encryption error:', error);
+        callbacks.onError?.(new Error(internalErrorMassage), { streamId });
+    }
 
-        await room.connect(url, token);
-        log('LiveKit room joined successfully');
-    } catch (error) {
-        log('Failed to connect to LiveKit room:', error);
-        callbacks.onConnectionStateChange?.(ConnectionState.Fail);
-        callbacks.onError?.(error as Error, { streamId: '' });
-        throw error;
+    function handleTrackSubscriptionFailed(
+        trackSid: string,
+        participant: RemoteParticipant,
+        reason?: SubscriptionError
+    ): void {
+        log('Track subscription failed:', { trackSid, participant, reason });
     }
 
-    analytics.enrich({
-        'stream-type': streamType,
-    });
+    function cleanDomElements(): void {
+        attachedElements.forEach(el => {
+            if (el.parentNode) {
+                el.parentNode.removeChild(el);
+            }
+        });
+        attachedElements.length = 0;
+    }
 
-    async function sendDataChannelMessage(message: string) {
+    async function sendTextMessage(message: string) {
         if (!isConnected || !room) {
             log('Room is not connected for sending messages');
-            callbacks.onError?.(new Error('Room is not connected for sending messages'), {
+            callbacks.onError?.(new Error(internalErrorMassage), {
                 streamId,
             });
             return;
         }
 
         try {
-            await room.localParticipant.publishData(new TextEncoder().encode(message), { reliable: true });
-        } catch (e: any) {
-            log('Error sending data channel message', e);
-            callbacks.onError?.(e, { streamId });
+            await room.localParticipant.sendText(message, {
+                topic: 'lk.chat',
+            });
+            log('Message sent successfully:', message);
+        } catch (error) {
+            log('Failed to send message:', error);
+            callbacks.onError?.(new Error(internalErrorMassage), { streamId });
         }
     }
 
@@ -206,7 +337,7 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
                 payload,
             });
 
-            return sendDataChannelMessage(message);
+            return sendTextMessage(message);
         },
 
         async disconnect() {
@@ -214,16 +345,14 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
                 await room.disconnect();
                 room = null;
             }
-            if (mediaStream) {
-                mediaStream.getTracks().forEach(track => track.stop());
-                mediaStream = null;
-            }
+            cleanDomElements();
             isConnected = false;
             callbacks.onConnectionStateChange?.(ConnectionState.Completed);
             callbacks.onAgentActivityStateChange?.(AgentActivityState.Idle);
         },
 
-        sendDataChannelMessage,
+        sendDataChannelMessage: sendTextMessage,
+        sendTextMessage,
 
         sessionId,
         streamId,

From 2aac8170141c751832d4b453d06538bef1535257 Mon Sep 17 00:00:00 2001
From: dor-eitan <164745144+dor-eitan@users.noreply.github.com>
Date: Wed, 17 Dec 2025 10:50:25 +0200
Subject: [PATCH 2/3] Merge pull request #238 from
 de-id/bugfix/analytics-rtt-candidate

bugfix: analytics - choose correct rtt
---
 .../streaming-manager/advanced.test.ts        |  6 +--
 .../streaming-manager/business-flows.test.ts  | 46 -------------------
 src/services/streaming-manager/stats/poll.ts  |  5 +-
 .../streaming-manager/stats/report.ts         | 19 +++++++-
 .../streaming-manager/webrtc-core.test.ts     | 19 +-------
 .../streaming-manager/webrtc-manager.ts       |  3 +-
 6 files changed, 24 insertions(+), 74 deletions(-)

diff --git a/src/services/streaming-manager/advanced.test.ts b/src/services/streaming-manager/advanced.test.ts
index 916fd804..dd783c28 100644
--- a/src/services/streaming-manager/advanced.test.ts
+++ b/src/services/streaming-manager/advanced.test.ts
@@ -396,8 +396,7 @@ describe('Streaming Manager Advanced', () => {
                 expect.anything(),
                 expect.anything(),
                 expect.anything(),
-                expect.any(Function),
-                false
+                expect.any(Function)
             );
 
             const connectivityCallback = (pollStats as jest.Mock).mock.calls[0][4];
@@ -422,8 +421,7 @@ describe('Streaming Manager Advanced', () => {
                 expect.anything(),
                 expect.anything(),
                 expect.anything(),
-                expect.anything(),
-                false
+                expect.anything()
             );
 
             expect(manager.streamId).toBe('streamId');
diff --git a/src/services/streaming-manager/business-flows.test.ts b/src/services/streaming-manager/business-flows.test.ts
index 1358e772..2d283da8 100644
--- a/src/services/streaming-manager/business-flows.test.ts
+++ b/src/services/streaming-manager/business-flows.test.ts
@@ -5,7 +5,6 @@
 
 import { StreamApiFactory, StreamingAgentFactory, StreamingManagerOptionsFactory } from '../../test-utils/factories';
 import { AgentActivityState, CreateStreamOptions, StreamType, StreamingManagerOptions } from '../../types/index';
-import { pollStats } from './stats/poll';
 import { createWebRTCStreamingManager as createStreamingManager } from './webrtc-manager';
 
 // Mock createStreamApi
@@ -146,51 +145,6 @@ describe('Streaming Manager Business Flows', () => {
         });
     });
 
-    describe('Warmup Mode Flows', () => {
-        it('should handle warmup=true with legacy stream', async () => {
-            const warmupAgent = { ...agent, stream_warmup: true };
-            const manager = await createStreamingManager(agentId, warmupAgent, options);
-
-            // Warmup should be enabled for legacy streams
-            expect(pollStats).toHaveBeenCalledWith(
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                true // warmup = true
-            );
-
-            expect(manager.streamId).toBe('streamId');
-        });
-
-        it('should handle warmup=true with fluent stream (should disable warmup)', async () => {
-            mockApi.createStream.mockResolvedValueOnce({
-                id: 'streamId',
-                offer: { type: 'offer', sdp: 'sdp' },
-                ice_servers: [],
-                session_id: 'sessionId',
-                fluent: true,
-                interrupt_enabled: false,
-            });
-
-            const warmupAgent = { ...agent, stream_warmup: true };
-            const manager = await createStreamingManager(agentId, warmupAgent, options);
-
-            // Warmup should be disabled for fluent streams
-            expect(pollStats).toHaveBeenCalledWith(
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                false // warmup = false for fluent
-            );
-
-            expect(manager.streamType).toBe(StreamType.Fluent);
-        });
-    });
-
     describe('Error Handling Flows', () => {
         it('should handle connection failure', async () => {
             mockApi.createStream.mockRejectedValueOnce(new Error('Connection failed'));
diff --git a/src/services/streaming-manager/stats/poll.ts b/src/services/streaming-manager/stats/poll.ts
index d0f1d7b4..4ebcce95 100644
--- a/src/services/streaming-manager/stats/poll.ts
+++ b/src/services/streaming-manager/stats/poll.ts
@@ -43,11 +43,8 @@ export function pollStats(
     getIsConnected: () => boolean,
     onConnected: () => void,
     onVideoStateChange?: (state: StreamingState, statsReport?: VideoRTCStatsReport) => void,
-    onConnectivityStateChange?: (state: ConnectivityState) => void,
-    warmup: boolean = false
+    onConnectivityStateChange?: (state: ConnectivityState) => void
 ) {
-    const streamsBeforeReady = warmup ? 1 : 0;
-
     let allStats: SlimRTCStatsReport[] = [];
     let previousStats: SlimRTCStatsReport;
     let notReceivingNumIntervals = 0;
diff --git a/src/services/streaming-manager/stats/report.ts b/src/services/streaming-manager/stats/report.ts
index 2d5363df..2c615891 100644
--- a/src/services/streaming-manager/stats/report.ts
+++ b/src/services/streaming-manager/stats/report.ts
@@ -77,12 +77,26 @@ function extractAnomalies(stats: AnalyticsRTCStatsReport[]): AnalyticsRTCStatsRe
 export function formatStats(stats: RTCStatsReport): SlimRTCStatsReport {
     let codec = '';
     let currRtt: number = 0;
+
     for (const report of stats.values()) {
         if (report && report.type === 'codec' && report.mimeType.startsWith('video')) {
             codec = report.mimeType.split('/')[1];
         }
         if (report && report.type === 'candidate-pair') {
-            currRtt = report.currentRoundTripTime;
+            const rtt = report.currentRoundTripTime;
+            const candidatePair = report as any;
+            const isNominated = candidatePair.nominated === true;
+
+            // Prioritize RTT from the nominated candidate-pair (the active connection path).
+            // This ensures we capture the actual network latency being used, not just any candidate.
+            // Only update if we have a valid positive RTT value to avoid overwriting with invalid data.
+            if (rtt > 0) {
+                if (isNominated) {
+                    currRtt = rtt;
+                } else if (currRtt === 0) {
+                    currRtt = rtt;
+                }
+            }
         }
         if (report && report.type === 'inbound-rtp' && report.kind === 'video') {
             return {
@@ -180,11 +194,14 @@ export function createVideoStatsReport(
             freezeDuration: report.freezeDuration - stats[index - 1].freezeDuration,
         };
     });
+
     const anomalies = extractAnomalies(differentialReport);
     const lowFpsCount = anomalies.reduce((acc, report) => acc + (report.causes!.includes('low fps') ? 1 : 0), 0);
+
     const avgJittersSamples = differentialReport
         .filter(stat => !!stat.avgJitterDelayInInterval)
         .map(stat => stat.avgJitterDelayInInterval);
+
     const avgRttSamples = differentialReport.filter(stat => !!stat.rtt).map(stat => stat.rtt);
 
     return {
diff --git a/src/services/streaming-manager/webrtc-core.test.ts b/src/services/streaming-manager/webrtc-core.test.ts
index 40e46864..0704d5c5 100644
--- a/src/services/streaming-manager/webrtc-core.test.ts
+++ b/src/services/streaming-manager/webrtc-core.test.ts
@@ -58,19 +58,6 @@ describe('Streaming Manager Core', () => {
             expect(options.analytics.enrich).toHaveBeenCalledWith({ 'stream-type': StreamType.Fluent });
         });
 
-        it('should handle warmup mode', async () => {
-            agentStreamOptions = { stream_warmup: true } as any;
-            await createStreamingManager(agentId, agentStreamOptions, options);
-            expect(pollStats).toHaveBeenCalledWith(
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                expect.anything(),
-                true
-            );
-        });
-
         it('should throw error when session_id is missing', async () => {
             mockApi.createStream.mockResolvedValueOnce({
                 id: 'streamId',
@@ -255,8 +242,7 @@ describe('Streaming Manager Core', () => {
                 expect.anything(),
                 expect.anything(),
                 expect.anything(),
-                expect.anything(),
-                false
+                expect.anything()
             );
         });
     });
@@ -370,8 +356,7 @@ describe('Streaming Manager Core', () => {
                 expect.anything(),
                 expect.anything(),
                 expect.anything(),
-                expect.anything(),
-                false
+                expect.anything()
             );
 
             expect(manager.streamId).toBe('streamId');
diff --git a/src/services/streaming-manager/webrtc-manager.ts b/src/services/streaming-manager/webrtc-manager.ts
index d5416cf5..e0afde1b 100644
--- a/src/services/streaming-manager/webrtc-manager.ts
+++ b/src/services/streaming-manager/webrtc-manager.ts
@@ -198,8 +198,7 @@ export async function createWebRTCStreamingManager<T extends CreateStreamOptions
                 report,
                 streamType,
             }),
-        state => callbacks.onConnectivityStateChange?.(state),
-        warmup
+        state => callbacks.onConnectivityStateChange?.(state)
     );
 
     peerConnection.onicecandidate = (event: RTCPeerConnectionIceEvent) => {

From 198fb2e9a5659e5178466df18167b270bea4c08b Mon Sep 17 00:00:00 2001
From: Arik Sfaradi <sfa.arik@gmail.com>
Date: Wed, 17 Dec 2025 13:31:47 +0200
Subject: [PATCH 3/3] Feature/data channel (#240)

* Fix freezes or video  not tracked in agents-ui

* pass onMessage of the eventQueue to livekit manager that gets the text messages through the data channel

* trigger the event queue when getting answer

* unuse

* type

* types

* fix ts

* evert because its exported from the main
---
 .../agent-manager/connect-to-manager.ts       |  3 +
 src/services/agent-manager/index.ts           |  9 +-
 .../streaming-manager/livekit-manager.ts      | 86 +++++++------------
 src/types/entities/video.ts                   |  2 +
 src/types/stream/stream.ts                    |  3 +-
 5 files changed, 43 insertions(+), 60 deletions(-)

diff --git a/src/services/agent-manager/connect-to-manager.ts b/src/services/agent-manager/connect-to-manager.ts
index c5df9635..107980f0 100644
--- a/src/services/agent-manager/connect-to-manager.ts
+++ b/src/services/agent-manager/connect-to-manager.ts
@@ -12,6 +12,7 @@ import {
     AgentsAPI,
     Chat,
     ChatMode,
+    ChatProgressCallback,
     ConnectionState,
     CreateStreamOptions,
     CreateStreamV2Options,
@@ -158,6 +159,8 @@ function trackLegacyVideoAnalytics(
 type ConnectToManagerOptions = AgentManagerOptions & {
     callbacks: AgentManagerOptions['callbacks'] & {
         onVideoIdChange?: (videoId: string | null) => void;
+        /** Internal callback for livekit-manager data channel events */
+        onMessage?: ChatProgressCallback;
     };
     chatId?: string;
 };
diff --git a/src/services/agent-manager/index.ts b/src/services/agent-manager/index.ts
index dfdce681..cf3068b1 100644
--- a/src/services/agent-manager/index.ts
+++ b/src/services/agent-manager/index.ts
@@ -116,7 +116,10 @@ export async function createAgentManager(agent: string, options: AgentManagerOpt
             () => {
                 return initializeStreamAndChat(
                     agentEntity,
-                    { ...options, callbacks: { ...options.callbacks, onVideoIdChange: updateVideoId } },
+                    {
+                        ...options,
+                        callbacks: { ...options.callbacks, onVideoIdChange: updateVideoId, onMessage },
+                    },
                     agentsApi,
                     analytics,
                     items.chat
@@ -259,9 +262,7 @@ export async function createAgentManager(agent: string, options: AgentManagerOpt
                 const chatRequestFn = isStreamsV2
                     ? async () => {
                           await items.streamingManager?.sendTextMessage?.(userMessage);
-                          return Promise.resolve({
-                              result: 'This is a mock message until the data channel is implemented.',
-                          } as ChatResponse);
+                          return Promise.resolve({} as ChatResponse);
                       }
                     : async () => {
                           return agentsApi.chat(
diff --git a/src/services/streaming-manager/livekit-manager.ts b/src/services/streaming-manager/livekit-manager.ts
index 68b88fa0..cb0f166a 100644
--- a/src/services/streaming-manager/livekit-manager.ts
+++ b/src/services/streaming-manager/livekit-manager.ts
@@ -11,6 +11,7 @@ import {
     StreamType,
     TransportProvider,
 } from '@sdk/types';
+import { ChatProgress } from '@sdk/types/entities/agents/manager';
 import { createStreamApiV2 } from '../../api/streams/streamsApiV2';
 import { didApiUrl } from '../../config/environment';
 import { createStreamingLogger, StreamingManager } from './common';
@@ -42,18 +43,6 @@ async function importLiveKit(): Promise<{
     }
 }
 
-function attachHiddenElement(track: RemoteTrack, attachedElements: HTMLMediaElement[]): HTMLMediaElement {
-    const hiddenElement = track.attach();
-    attachedElements.push(hiddenElement);
-    hiddenElement.style.position = 'absolute';
-    hiddenElement.style.width = '1px';
-    hiddenElement.style.height = '1px';
-    hiddenElement.style.opacity = '0.01';
-    hiddenElement.style.pointerEvents = 'none';
-    hiddenElement.muted = true;
-    return hiddenElement;
-}
-
 const connectivityQualityToState = {
     excellent: ConnectivityState.Strong,
     good: ConnectivityState.Strong,
@@ -101,10 +90,10 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
     let videoId: string | null = null;
     const streamType = StreamType.Fluent;
     let isInitialConnection = true;
-    const attachedElements: HTMLMediaElement[] = [];
+    let sharedMediaStream: MediaStream | null = null;
 
     room = new Room({
-        adaptiveStream: true,
+        adaptiveStream: false, // Must be false to use mediaStreamTrack directly
         dynacast: true,
     });
 
@@ -225,33 +214,24 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
     function handleTrackSubscribed(track: RemoteTrack, publication: any, participant: RemoteParticipant): void {
         log(`Track subscribed: ${track.kind} from ${participant.identity}`);
 
-        if (track.kind === 'video') {
-            const hiddenElement = attachHiddenElement(track, attachedElements);
-            document.body.appendChild(hiddenElement);
-
-            // Play the hidden element to keep the track "alive"
-            hiddenElement
-                .play()
-                .then(() => {
-                    log('Hidden video element playing');
-                })
-                .catch(e => log('Error playing hidden element:', e));
-
-            log(`Video element created, srcObject: ${hiddenElement.srcObject}`);
-
-            if (hiddenElement.srcObject) {
-                callbacks.onSrcObjectReady?.(hiddenElement.srcObject as MediaStream);
-                callbacks.onVideoStateChange?.(StreamingState.Start);
-            }
+        const mediaStreamTrack = track.mediaStreamTrack;
+        if (!mediaStreamTrack) {
+            log(`No mediaStreamTrack available for ${track.kind}`);
+            return;
+        }
+
+        // Create shared stream if it doesn't exist, or add track to existing stream
+        if (!sharedMediaStream) {
+            sharedMediaStream = new MediaStream([mediaStreamTrack]);
+            log(`Created shared MediaStream with ${track.kind} track`);
+        } else {
+            sharedMediaStream.addTrack(mediaStreamTrack);
+            log(`Added ${track.kind} track to shared MediaStream`);
         }
 
-        if (track.kind === 'audio') {
-            // For audio, create element and play it directly
-            const audioElement = track.attach();
-            attachedElements.push(audioElement);
-            audioElement.style.display = 'none';
-            document.body.appendChild(audioElement);
-            audioElement.play().catch(e => log('Error playing audio element:', e));
+        if (track.kind === 'video') {
+            callbacks.onSrcObjectReady?.(sharedMediaStream);
+            callbacks.onVideoStateChange?.(StreamingState.Start);
         }
     }
 
@@ -269,14 +249,12 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
 
         try {
             const data = JSON.parse(message);
-            if (data.subject === StreamEvents.StreamStarted && data.metadata?.videoId) {
-                videoId = data.metadata.videoId;
-                callbacks.onVideoIdChange?.(videoId);
-                callbacks.onAgentActivityStateChange?.(AgentActivityState.Talking);
-            } else if (data.subject === StreamEvents.StreamDone) {
-                videoId = null;
-                callbacks.onVideoIdChange?.(videoId);
-                callbacks.onAgentActivityStateChange?.(AgentActivityState.Idle);
+            if (data.subject === StreamEvents.ChatAnswer) {
+                const eventName = ChatProgress.Answer;
+                callbacks.onMessage?.(eventName, {
+                    event: eventName,
+                    ...data,
+                });
             }
         } catch (e) {
             log('Failed to parse data channel message:', e);
@@ -301,13 +279,11 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
         log('Track subscription failed:', { trackSid, participant, reason });
     }
 
-    function cleanDomElements(): void {
-        attachedElements.forEach(el => {
-            if (el.parentNode) {
-                el.parentNode.removeChild(el);
-            }
-        });
-        attachedElements.length = 0;
+    function cleanMediaStream(): void {
+        if (sharedMediaStream) {
+            sharedMediaStream.getTracks().forEach(track => track.stop());
+            sharedMediaStream = null;
+        }
     }
 
     async function sendTextMessage(message: string) {
@@ -345,7 +321,7 @@ export async function createLiveKitStreamingManager<T extends CreateStreamV2Opti
                 await room.disconnect();
                 room = null;
             }
-            cleanDomElements();
+            cleanMediaStream();
             isConnected = false;
             callbacks.onConnectionStateChange?.(ConnectionState.Completed);
             callbacks.onAgentActivityStateChange?.(AgentActivityState.Idle);
diff --git a/src/types/entities/video.ts b/src/types/entities/video.ts
index 108c6084..6792ccac 100644
--- a/src/types/entities/video.ts
+++ b/src/types/entities/video.ts
@@ -10,6 +10,8 @@ export const mapVideoType = (type: string): VideoType => {
             return VideoType.Clip;
         case 'talk':
             return VideoType.Talk;
+        case 'expressive':
+            return VideoType.Expressive;
         default:
             throw new Error(`Unknown video type: ${type}`);
     }
diff --git a/src/types/stream/stream.ts b/src/types/stream/stream.ts
index a915e842..f8cf99a0 100644
--- a/src/types/stream/stream.ts
+++ b/src/types/stream/stream.ts
@@ -1,6 +1,7 @@
 import { Analytics } from '@sdk/services/analytics/mixpanel';
 import { VideoRTCStatsReport } from '@sdk/services/streaming-manager/stats/report';
 import { Auth } from '../auth';
+import { ChatProgressCallback } from '../entities/agents/manager';
 import { CreateClipStreamRequest, CreateTalkStreamRequest, SendClipStreamPayload, SendTalkStreamPayload } from './api';
 import { ICreateStreamRequestResponse, IceCandidate, SendStreamPayloadResponse, Status } from './rtc';
 
@@ -53,7 +54,7 @@ export enum StreamType {
 }
 
 export interface ManagerCallbacks {
-    onMessage?: (event: string, data: string) => void;
+    onMessage?: ChatProgressCallback;
     onConnectionStateChange?: (state: ConnectionState) => void;
     onVideoStateChange?: (state: StreamingState, report?: VideoRTCStatsReport) => void;
     onSrcObjectReady?: (value: MediaStream) => void;