Update flare-ui/src/app/services/conversation-manager.service.ts

flare-ui/src/app/services/conversation-manager.service.ts

@@ -1,5 +1,9 @@
+// conversation-manager.service.ts
+// Path: /flare-ui/src/app/services/conversation-manager.service.ts
+
 import { Injectable, OnDestroy } from '@angular/core';
-import { Subject, Subscription, BehaviorSubject } from 'rxjs';
+import { Subject, Subscription, BehaviorSubject, throwError } from 'rxjs';
+import { catchError, retry } from 'rxjs/operators';
 import { WebSocketService } from './websocket.service';
 import { AudioStreamService } from './audio-stream.service';
 
@@ -9,13 +13,30 @@ export type ConversationState =
   | 'processing_stt' 
   | 'processing_llm' 
   | 'processing_tts' 
-  | 'playing_audio';
+  | 'playing_audio'
+  | 'error';
 
 export interface ConversationMessage {
-  role: 'user' | 'assistant';
+  role: 'user' | 'assistant' | 'system';
   text: string;
   timestamp: Date;
   audioUrl?: string;
+  error?: boolean;
+}
+
+export interface ConversationConfig {
+  language?: string;
+  stt_engine?: string;
+  tts_engine?: string;
+  enable_barge_in?: boolean;
+  max_silence_duration?: number;
+}
+
+export interface ConversationError {
+  type: 'websocket' | 'audio' | 'permission' | 'network' | 'unknown';
+  message: string;
+  details?: any;
+  timestamp: Date;
 }
 
 @Injectable({
@@ -25,6 +46,12 @@ export class ConversationManagerService implements OnDestroy {
   private subscriptions = new Subscription();
   private audioQueue: string[] = [];
   private isInterrupting = false;
+  private sessionId: string | null = null;
+  private conversationConfig: ConversationConfig = {
+    language: 'tr-TR',
+    stt_engine: 'google',
+    enable_barge_in: true
+  };
   
   // State management
   private currentStateSubject = new BehaviorSubject<ConversationState>('idle');
@@ -38,8 +65,13 @@ export class ConversationManagerService implements OnDestroy {
   private transcriptionSubject = new BehaviorSubject<string>('');
   public transcription$ = this.transcriptionSubject.asObservable();
   
+  // Error handling
+  private errorSubject = new Subject<ConversationError>();
+  public error$ = this.errorSubject.asObservable();
+  
   // Audio player reference
   private audioPlayer: HTMLAudioElement | null = null;
+  private audioPlayerPromise: Promise<void> | null = null;
 
   constructor(
     private wsService: WebSocketService,
@@ -50,50 +82,114 @@ export class ConversationManagerService implements OnDestroy {
     this.cleanup();
   }
 
-  async startConversation(sessionId: string): Promise<void> {
+  async startConversation(sessionId: string, config?: ConversationConfig): Promise<void> {
     try {
-      // Connect WebSocket
-      await this.wsService.connect(sessionId);
+      if (!sessionId) {
+        throw new Error('Session ID is required');
+      }
+
+      // Update configuration
+      if (config) {
+        this.conversationConfig = { ...this.conversationConfig, ...config };
+      }
+
+      this.sessionId = sessionId;
+      
+      // Reset state
+      this.clearMessages();
+      this.currentStateSubject.next('idle');
+      
+      // Connect WebSocket first
+      await this.wsService.connect(sessionId).catch(error => {
+        throw new Error(`WebSocket connection failed: ${error.message}`);
+      });
       
       // Start audio recording
-      await this.audioService.startRecording();
+      await this.audioService.startRecording().catch(error => {
+        // Disconnect WebSocket if audio fails
+        this.wsService.disconnect();
+        throw new Error(`Audio recording failed: ${error.message}`);
+      });
       
       // Set up subscriptions
       this.setupSubscriptions();
       
-      // Send start signal
-      this.wsService.sendControl('start_session', {
-        language: 'tr-TR',
-        stt_engine: 'google',
-        enable_barge_in: true
-      });
+      // Send start signal with configuration
+      this.wsService.sendControl('start_session', this.conversationConfig);
       
-      console.log('✅ Conversation started');
+      console.log('✅ Conversation started with config:', this.conversationConfig);
       
-    } catch (error) {
+    } catch (error: any) {
       console.error('Failed to start conversation:', error);
+      
+      const conversationError: ConversationError = {
+        type: this.determineErrorType(error),
+        message: error.message || 'Failed to start conversation',
+        details: error,
+        timestamp: new Date()
+      };
+      
+      this.errorSubject.next(conversationError);
+      this.currentStateSubject.next('error');
+      this.cleanup();
+      
       throw error;
     }
   }
 
   stopConversation(): void {
-    this.cleanup();
+    try {
+      // Send stop signal if connected
+      if (this.wsService.isConnected()) {
+        this.wsService.sendControl('stop_session');
+      }
+      
+      this.cleanup();
+      this.addSystemMessage('Conversation ended');
+      
+    } catch (error) {
+      console.error('Error stopping conversation:', error);
+      this.cleanup();
+    }
   }
 
   private setupSubscriptions(): void {
     // Audio chunks from microphone
     this.subscriptions.add(
-      this.audioService.audioChunk$.subscribe(chunk => {
-        if (!this.isInterrupting) {
-          this.wsService.sendAudioChunk(chunk.data);
+      this.audioService.audioChunk$.subscribe({
+        next: (chunk) => {
+          if (!this.isInterrupting && this.wsService.isConnected()) {
+            try {
+              this.wsService.sendAudioChunk(chunk.data);
+            } catch (error) {
+              console.error('Failed to send audio chunk:', error);
+            }
+          }
+        },
+        error: (error) => {
+          console.error('Audio stream error:', error);
+          this.handleAudioError(error);
         }
       })
     );
     
+    // Audio stream errors
+    this.subscriptions.add(
+      this.audioService.error$.subscribe(error => {
+        this.handleAudioError(error);
+      })
+    );
+    
     // WebSocket messages
     this.subscriptions.add(
-      this.wsService.message$.subscribe(message => {
-        this.handleMessage(message);
+      this.wsService.message$.subscribe({
+        next: (message) => {
+          this.handleMessage(message);
+        },
+        error: (error) => {
+          console.error('WebSocket message error:', error);
+          this.handleWebSocketError(error);
+        }
       })
     );
     
@@ -114,37 +210,67 @@ export class ConversationManagerService implements OnDestroy {
       })
     );
     
-    // Errors
+    // WebSocket errors
     this.subscriptions.add(
       this.wsService.error$.subscribe(error => {
         console.error('WebSocket error:', error);
-        this.addSystemMessage(`Hata: ${error}`);
+        this.handleWebSocketError({ message: error });
+      })
+    );
+    
+    // WebSocket connection state
+    this.subscriptions.add(
+      this.wsService.connection$.subscribe(connected => {
+        if (!connected && this.currentStateSubject.value !== 'idle') {
+          this.addSystemMessage('Connection lost. Attempting to reconnect...');
+        }
       })
     );
   }
 
   private handleMessage(message: any): void {
-    switch (message.type) {
-      case 'transcription':
-        if (message['is_final']) {
-          this.addMessage('user', message['text']);
-          this.transcriptionSubject.next('');
-        }
-        break;
-        
-      case 'assistant_response':
-        this.addMessage('assistant', message['text']);
-        break;
-        
-      case 'tts_audio':
-        this.handleTTSAudio(message);
-        break;
-        
-      case 'control':
-        if (message['action'] === 'stop_playback') {
-          this.stopAudioPlayback();
-        }
-        break;
+    try {
+      switch (message.type) {
+        case 'transcription':
+          if (message['is_final']) {
+            this.addMessage('user', message['text']);
+            this.transcriptionSubject.next('');
+          }
+          break;
+          
+        case 'assistant_response':
+          this.addMessage('assistant', message['text']);
+          break;
+          
+        case 'tts_audio':
+          this.handleTTSAudio(message);
+          break;
+          
+        case 'control':
+          if (message['action'] === 'stop_playback') {
+            this.stopAudioPlayback();
+          }
+          break;
+          
+        case 'error':
+          this.handleServerError(message);
+          break;
+          
+        case 'session_config':
+          // Update configuration from server
+          if (message['config']) {
+            this.conversationConfig = { ...this.conversationConfig, ...message['config'] };
+          }
+          break;
+      }
+    } catch (error) {
+      console.error('Error handling message:', error);
+      this.errorSubject.next({
+        type: 'unknown',
+        message: 'Failed to process message',
+        details: error,
+        timestamp: new Date()
+      });
     }
   }
 
@@ -158,28 +284,42 @@ export class ConversationManagerService implements OnDestroy {
     } else if (to === 'playing_audio') {
       // Start playing accumulated audio
       this.playQueuedAudio();
+    } else if (to === 'error') {
+      // Handle error state
+      this.addSystemMessage('An error occurred in the conversation flow');
     }
   }
 
   private handleTTSAudio(message: any): void {
-    // Accumulate audio chunks
-    this.audioQueue.push(message['data']);
-    
-    if (message['is_last']) {
-      // All chunks received, create audio blob
-      const audioData = this.audioQueue.join('');
-      const audioBlob = this.base64ToBlob(audioData, 'audio/mpeg');
-      const audioUrl = URL.createObjectURL(audioBlob);
-      
-      // Update last message with audio URL
-      const messages = this.messagesSubject.value;
-      if (messages.length > 0 && messages[messages.length - 1].role === 'assistant') {
-        messages[messages.length - 1].audioUrl = audioUrl;
-        this.messagesSubject.next([...messages]);
+    try {
+      // Validate audio data
+      if (!message['data']) {
+        console.warn('TTS audio message missing data');
+        return;
       }
       
-      // Clear queue
-      this.audioQueue = [];
+      // Accumulate audio chunks
+      this.audioQueue.push(message['data']);
+      
+      if (message['is_last']) {
+        // All chunks received, create audio blob
+        const audioData = this.audioQueue.join('');
+        const audioBlob = this.base64ToBlob(audioData, message['mime_type'] || 'audio/mpeg');
+        const audioUrl = URL.createObjectURL(audioBlob);
+        
+        // Update last message with audio URL
+        const messages = this.messagesSubject.value;
+        if (messages.length > 0 && messages[messages.length - 1].role === 'assistant') {
+          messages[messages.length - 1].audioUrl = audioUrl;
+          this.messagesSubject.next([...messages]);
+        }
+        
+        // Clear queue
+        this.audioQueue = [];
+      }
+    } catch (error) {
+      console.error('Error handling TTS audio:', error);
+      this.audioQueue = []; // Clear queue on error
     }
   }
 
@@ -192,27 +332,75 @@ export class ConversationManagerService implements OnDestroy {
     }
   }
 
-  private playAudio(audioUrl: string): void {
-    if (!this.audioPlayer) {
-      this.audioPlayer = new Audio();
+  private async playAudio(audioUrl: string): Promise<void> {
+    try {
+      if (!this.audioPlayer) {
+        this.audioPlayer = new Audio();
+        this.setupAudioPlayerHandlers();
+      }
+      
+      this.audioPlayer.src = audioUrl;
+      
+      // Store the play promise to handle interruptions properly
+      this.audioPlayerPromise = this.audioPlayer.play();
+      
+      await this.audioPlayerPromise;
+      
+    } catch (error: any) {
+      // Check if error is due to interruption
+      if (error.name === 'AbortError') {
+        console.log('Audio playback interrupted');
+      } else {
+        console.error('Audio playback error:', error);
+        this.errorSubject.next({
+          type: 'audio',
+          message: 'Failed to play audio response',
+          details: error,
+          timestamp: new Date()
+        });
+      }
+    } finally {
+      this.audioPlayerPromise = null;
     }
-    
-    this.audioPlayer.src = audioUrl;
-    this.audioPlayer.play().catch(error => {
-      console.error('Audio playback error:', error);
-    });
+  }
+
+  private setupAudioPlayerHandlers(): void {
+    if (!this.audioPlayer) return;
     
     this.audioPlayer.onended = () => {
       // Notify that audio playback ended
-      this.wsService.sendControl('audio_ended');
+      if (this.wsService.isConnected()) {
+        try {
+          this.wsService.sendControl('audio_ended');
+        } catch (error) {
+          console.error('Failed to send audio_ended signal:', error);
+        }
+      }
+      this.currentStateSubject.next('idle');
+    };
+    
+    this.audioPlayer.onerror = (error) => {
+      console.error('Audio player error:', error);
       this.currentStateSubject.next('idle');
     };
   }
 
   private stopAudioPlayback(): void {
-    if (this.audioPlayer) {
-      this.audioPlayer.pause();
-      this.audioPlayer.currentTime = 0;
+    try {
+      if (this.audioPlayer) {
+        this.audioPlayer.pause();
+        this.audioPlayer.currentTime = 0;
+        
+        // Cancel any pending play promise
+        if (this.audioPlayerPromise) {
+          this.audioPlayerPromise.catch(() => {
+            // Ignore abort errors
+          });
+          this.audioPlayerPromise = null;
+        }
+      }
+    } catch (error) {
+      console.error('Error stopping audio playback:', error);
     }
   }
 
@@ -220,17 +408,27 @@ export class ConversationManagerService implements OnDestroy {
   performBargeIn(): void {
     const currentState = this.currentStateSubject.value;
     
-    if (currentState !== 'idle' && currentState !== 'listening') {
+    if (currentState !== 'idle' && currentState !== 'listening' && currentState !== 'error') {
       console.log('🛑 Performing barge-in');
       this.isInterrupting = true;
       
       // Stop audio playback
       this.stopAudioPlayback();
       
+      // Clear audio queue
+      this.audioQueue = [];
+      
       // Notify server
-      this.wsService.sendControl('interrupt', {
-        at_state: currentState
-      });
+      if (this.wsService.isConnected()) {
+        try {
+          this.wsService.sendControl('interrupt', {
+            at_state: currentState,
+            timestamp: Date.now()
+          });
+        } catch (error) {
+          console.error('Failed to send interrupt signal:', error);
+        }
+      }
       
       // Reset interruption flag after a delay
       setTimeout(() => {
@@ -239,40 +437,161 @@ export class ConversationManagerService implements OnDestroy {
     }
   }
 
-  private addMessage(role: 'user' | 'assistant', text: string): void {
+  private addMessage(role: 'user' | 'assistant', text: string, error: boolean = false): void {
+    if (!text || text.trim().length === 0) {
+      return;
+    }
+    
     const messages = this.messagesSubject.value;
     messages.push({
       role,
       text,
-      timestamp: new Date()
+      timestamp: new Date(),
+      error
     });
     this.messagesSubject.next([...messages]);
   }
 
   private addSystemMessage(text: string): void {
     console.log(`📢 System: ${text}`);
+    const messages = this.messagesSubject.value;
+    messages.push({
+      role: 'system',
+      text,
+      timestamp: new Date()
+    });
+    this.messagesSubject.next([...messages]);
   }
 
   private base64ToBlob(base64: string, mimeType: string): Blob {
-    const byteCharacters = atob(base64);
-    const byteNumbers = new Array(byteCharacters.length);
+    try {
+      const byteCharacters = atob(base64);
+      const byteNumbers = new Array(byteCharacters.length);
+      
+      for (let i = 0; i < byteCharacters.length; i++) {
+        byteNumbers[i] = byteCharacters.charCodeAt(i);
+      }
+      
+      const byteArray = new Uint8Array(byteNumbers);
+      return new Blob([byteArray], { type: mimeType });
+    } catch (error) {
+      console.error('Error converting base64 to blob:', error);
+      throw new Error('Failed to convert audio data');
+    }
+  }
+
+  private handleAudioError(error: any): void {
+    const conversationError: ConversationError = {
+      type: error.type || 'audio',
+      message: error.message || 'Audio error occurred',
+      details: error,
+      timestamp: new Date()
+    };
+    
+    this.errorSubject.next(conversationError);
+    
+    // Add user-friendly message
+    if (error.type === 'permission') {
+      this.addSystemMessage('Microphone permission denied. Please allow microphone access.');
+    } else if (error.type === 'device') {
+      this.addSystemMessage('Microphone not found or not accessible.');
+    } else {
+      this.addSystemMessage('Audio error occurred. Please check your microphone.');
+    }
+    
+    // Update state
+    this.currentStateSubject.next('error');
+  }
+
+  private handleWebSocketError(error: any): void {
+    const conversationError: ConversationError = {
+      type: 'websocket',
+      message: error.message || 'WebSocket error occurred',
+      details: error,
+      timestamp: new Date()
+    };
+    
+    this.errorSubject.next(conversationError);
+    this.addSystemMessage('Connection error. Please check your internet connection.');
+    
+    // Don't set error state for temporary connection issues
+    if (this.wsService.getReconnectionInfo().isReconnecting) {
+      this.addSystemMessage('Attempting to reconnect...');
+    } else {
+      this.currentStateSubject.next('error');
+    }
+  }
+
+  private handleServerError(message: any): void {
+    const errorType = message['error_type'] || 'unknown';
+    const errorMessage = message['message'] || 'Server error occurred';
+    
+    const conversationError: ConversationError = {
+      type: errorType === 'race_condition' ? 'network' : 'unknown',
+      message: errorMessage,
+      details: message,
+      timestamp: new Date()
+    };
+    
+    this.errorSubject.next(conversationError);
+    
+    // Add user-friendly message based on error type
+    if (errorType === 'race_condition') {
+      this.addSystemMessage('Session conflict detected. Please restart the conversation.');
+    } else if (errorType === 'stt_error') {
+      this.addSystemMessage('Speech recognition error. Please try speaking again.');
+    } else if (errorType === 'tts_error') {
+      this.addSystemMessage('Text-to-speech error. Response will be shown as text only.');
+    } else {
+      this.addSystemMessage(`Error: ${errorMessage}`);
+    }
+  }
+
+  private determineErrorType(error: any): ConversationError['type'] {
+    if (error.type) {
+      return error.type;
+    }
+    
+    if (error.message?.includes('WebSocket') || error.message?.includes('connection')) {
+      return 'websocket';
+    }
+    
+    if (error.message?.includes('microphone') || error.message?.includes('audio')) {
+      return 'audio';
+    }
+    
+    if (error.message?.includes('permission')) {
+      return 'permission';
+    }
     
-    for (let i = 0; i < byteCharacters.length; i++) {
-      byteNumbers[i] = byteCharacters.charCodeAt(i);
+    if (error.message?.includes('network') || error.status === 0) {
+      return 'network';
     }
     
-    const byteArray = new Uint8Array(byteNumbers);
-    return new Blob([byteArray], { type: mimeType });
+    return 'unknown';
   }
 
   private cleanup(): void {
-    this.subscriptions.unsubscribe();
-    this.audioService.stopRecording();
-    this.wsService.disconnect();
-    this.stopAudioPlayback();
-    this.audioQueue = [];
-    this.currentStateSubject.next('idle');
-    console.log('🧹 Conversation cleaned up');
+    try {
+      this.subscriptions.unsubscribe();
+      this.subscriptions = new Subscription();
+      
+      this.audioService.stopRecording();
+      this.wsService.disconnect();
+      this.stopAudioPlayback();
+      
+      if (this.audioPlayer) {
+        this.audioPlayer = null;
+      }
+      
+      this.audioQueue = [];
+      this.isInterrupting = false;
+      this.currentStateSubject.next('idle');
+      
+      console.log('🧹 Conversation cleaned up');
+    } catch (error) {
+      console.error('Error during cleanup:', error);
+    }
   }
 
   // Public methods for UI
@@ -286,5 +605,37 @@ export class ConversationManagerService implements OnDestroy {
 
   clearMessages(): void {
     this.messagesSubject.next([]);
+    this.transcriptionSubject.next('');
+  }
+
+  updateConfig(config: Partial<ConversationConfig>): void {
+    this.conversationConfig = { ...this.conversationConfig, ...config };
+    
+    // Send config update if connected
+    if (this.wsService.isConnected()) {
+      try {
+        this.wsService.sendControl('update_config', config);
+      } catch (error) {
+        console.error('Failed to update config:', error);
+      }
+    }
+  }
+
+  getConfig(): ConversationConfig {
+    return { ...this.conversationConfig };
+  }
+
+  isConnected(): boolean {
+    return this.wsService.isConnected();
+  }
+
+  // Retry connection
+  async retryConnection(): Promise<void> {
+    if (!this.sessionId) {
+      throw new Error('No session ID available for retry');
+    }
+    
+    this.currentStateSubject.next('idle');
+    await this.startConversation(this.sessionId, this.conversationConfig);
   }
 }