import 'dart:async'; import 'dart:collection'; import 'dart:convert'; import 'dart:io'; import 'dart:math' as math; import 'package:flutter/foundation.dart'; import 'package:permission_handler/permission_handler.dart'; import 'package:web_socket_channel/io.dart'; import 'package:didvan/services/ai/voice_audio_engine.dart'; import 'package:didvan/services/network/request.dart'; import 'package:didvan/services/network/request_helper.dart'; enum VoiceSessionPhase { connecting, reconnecting, listening, hearing, searching, thinking, speaking, muted, error, closed, } class RealtimeVoiceSession extends ChangeNotifier { static const String _webSocketUrl = 'wss://api.x.ai/v1/realtime?model=grok-voice-latest'; static const String _voice = 'leo'; static const int sampleRate = 24000; static const int _maximumPreSessionAudioBytes = sampleRate * 2 * 3; final VoiceAudioEngine _audio = VoiceAudioEngine(); final ListQueue _preSessionAudio = ListQueue(); final List _assistantMessages = []; IOWebSocketChannel? _socket; StreamSubscription? _socketSubscription; StreamSubscription? _audioSubscription; Timer? _reconnectTimer; int _reconnectAttempt = 0; int _connectionGeneration = 0; int _preSessionAudioBytes = 0; int _playbackGeneration = 0; bool _started = false; bool _closing = false; bool _sessionReady = false; bool _playbackActive = false; bool _responseDone = false; bool _userSpeaking = false; bool _micMuted = false; bool _hardwareAecEnabled = false; String? _conversationId; String? _activeResponseId; String _liveAssistantTranscript = ''; Future _playbackChain = Future.value(); DateTime _lastLevelNotification = DateTime.fromMillisecondsSinceEpoch(0); VoiceSessionPhase _phase = VoiceSessionPhase.connecting; String? _errorMessage; double _microphoneLevel = 0; double _assistantLevel = 0; VoiceSessionPhase get phase => _phase; String? get errorMessage => _errorMessage; bool get isMicMuted => _micMuted; bool get isUserSpeaking => _userSpeaking; bool get isAssistantActive => _activeResponseId != null || _playbackActive; bool get hardwareAecEnabled => _hardwareAecEnabled; bool get supportsAutomaticInterruption => _hardwareAecEnabled; bool get isSessionReady => _sessionReady; double get microphoneLevel => _microphoneLevel; double get assistantLevel => _assistantLevel; String get liveAssistantTranscript => _liveAssistantTranscript; List get assistantMessages => List.unmodifiable(_assistantMessages); String get statusText { switch (_phase) { case VoiceSessionPhase.connecting: return 'در حال اتصال'; case VoiceSessionPhase.reconnecting: return 'اتصال مجدد'; case VoiceSessionPhase.listening: return 'گوش می‌دهم'; case VoiceSessionPhase.hearing: return 'صدایت را می‌شنوم'; case VoiceSessionPhase.searching: return 'در حال جست‌وجوی وب'; case VoiceSessionPhase.thinking: return 'در حال فکر کردن'; case VoiceSessionPhase.speaking: return 'هوشان در حال صحبت است'; case VoiceSessionPhase.muted: return 'میکروفن خاموش است'; case VoiceSessionPhase.error: return _errorMessage ?? 'خطایی رخ داد'; case VoiceSessionPhase.closed: return 'مکالمه پایان یافت'; } } String get _instructions { final now = DateTime.now(); final date = '${now.year.toString().padLeft(4, '0')}-' '${now.month.toString().padLeft(2, '0')}-' '${now.day.toString().padLeft(2, '0')}'; return 'تو هوشان، دستیار هوشمند دیدوان هستی. فقط فارسی، کوتاه، روشن و ' 'طبیعی صحبت کن. تاریخ امروز $date و منطقه زمانی Asia/Tehran است. ' 'برای خبر، قیمت، آب‌وهوا، سیاست، ورزش، افراد، شرکت‌ها و هر اطلاعات ' 'زمان‌حساس قبل از پاسخ از web_search استفاده کن. اگر صدای کاربر ' 'نامفهوم یا ناقص بود حدس نزن و از او بخواه دوباره بگوید.'; } Future start() async { if (_started || _closing) return; _started = true; _setPhase(VoiceSessionPhase.connecting); final permission = await Permission.microphone.request(); if (!permission.isGranted) { _setError('دسترسی میکروفن لازم است'); return; } try { // Subscribe before start so the first captured syllable cannot be lost. _audioSubscription = _audio.events.listen( _handleAudioEvent, onError: (Object error, StackTrace stackTrace) { _setError('خطا در مسیر صوتی: $error'); }, ); final capabilities = await _audio.start(sampleRate: sampleRate); _hardwareAecEnabled = capabilities.aecEnabled; debugPrint( '🎧 native voice audio ready: ' 'aec=${capabilities.aecEnabled} ' 'ns=${capabilities.noiseSuppressionEnabled} ' 'rate=${capabilities.sampleRate}', ); await _connect(); } catch (error) { _setError('راه‌اندازی صدا ناموفق بود: $error'); } } void _handleAudioEvent(VoiceAudioEvent event) { if (_closing) return; switch (event.type) { case VoiceAudioEventType.microphone: final pcm = event.pcm; if (pcm != null) _handleMicrophoneFrame(pcm, event.rms); break; case VoiceAudioEventType.playbackStarted: _playbackActive = true; _setPhase(VoiceSessionPhase.speaking); break; case VoiceAudioEventType.playbackIdle: _playbackActive = false; _assistantLevel = 0; if (_responseDone) { _activeResponseId = null; } if (!_userSpeaking && _activeResponseId == null) { _setPhase(_micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening); } else { _notify(); } break; case VoiceAudioEventType.error: _setError(event.message ?? 'خطا در موتور صوتی'); break; } } void _handleMicrophoneFrame(Uint8List pcm, double rms) { _microphoneLevel = math.max(_microphoneLevel * 0.58, (rms * 8).clamp(0.0, 1.0)); final now = DateTime.now(); if (now.difference(_lastLevelNotification).inMilliseconds >= 50) { _lastLevelNotification = now; _notify(); } if (_micMuted) return; // A device without platform AEC is kept strictly half-duplex. This // guarantees that assistant playback can never be submitted as user audio. // The UI exposes immediate tap-to-interrupt on those rare devices. if (!_hardwareAecEnabled && isAssistantActive) return; if (_sessionReady) { _sendAudio(pcm); return; } _preSessionAudio.addLast(pcm); _preSessionAudioBytes += pcm.length; while (_preSessionAudioBytes > _maximumPreSessionAudioBytes && _preSessionAudio.isNotEmpty) { _preSessionAudioBytes -= _preSessionAudio.removeFirst().length; } } Future _fetchEphemeralToken() async { final service = RequestService( '${RequestHelper.newApiBaseUrl}/ai/voice/token', ); await service.post(); if (!service.isSuccess) { debugPrint('Voice token failed: ${service.errorMessage}'); return null; } final token = service.result['token']?.toString(); return token == null || token.isEmpty ? null : token; } Future _connect() async { if (_closing) return; final generation = ++_connectionGeneration; _sessionReady = false; _setPhase( _reconnectAttempt == 0 ? VoiceSessionPhase.connecting : VoiceSessionPhase.reconnecting, ); try { final token = await _fetchEphemeralToken(); if (_closing || generation != _connectionGeneration) return; if (token == null) throw StateError('توکن صوتی دریافت نشد'); final resumeId = _conversationId; final url = resumeId == null ? _webSocketUrl : '$_webSocketUrl&conversation_id=' '${Uri.encodeQueryComponent(resumeId)}'; final webSocket = await WebSocket.connect( url, protocols: ['xai-client-secret.$token'], ); webSocket.pingInterval = const Duration(seconds: 25); if (_closing || generation != _connectionGeneration) { await webSocket.close(); return; } final channel = IOWebSocketChannel(webSocket); _socket = channel; await _socketSubscription?.cancel(); _socketSubscription = channel.stream.listen( _handleServerEvent, onError: (Object error, StackTrace stackTrace) { _handleDisconnect(generation, error); }, onDone: () => _handleDisconnect(generation, null), ); _sendSessionConfiguration(); } catch (error) { if (_closing || generation != _connectionGeneration) return; debugPrint('Voice WebSocket connection failed: $error'); _scheduleReconnect(); } } void _sendSessionConfiguration() { _send({ 'type': 'session.update', 'session': { 'voice': _voice, 'instructions': _instructions, 'turn_detection': { 'type': 'server_vad', 'threshold': 0.5, 'prefix_padding_ms': 700, 'silence_duration_ms': 650, }, 'resumption': {'enabled': true}, 'tools': [ { 'type': 'web_search', 'location': { 'country': 'IR', 'city': 'Tehran', 'timezone': 'Asia/Tehran', }, }, {'type': 'x_search'}, ], 'audio': { 'input': { 'format': { 'type': 'audio/pcm', 'rate': sampleRate, }, 'transcription': { 'model': 'grok-transcribe', 'language_hint': 'fa-IR', }, }, 'output': { 'format': { 'type': 'audio/pcm', 'rate': sampleRate, }, }, }, }, }); } void _handleServerEvent(dynamic raw) { if (_closing) return; try { final decoded = jsonDecode( raw is String ? raw : utf8.decode(raw as List), ); final event = Map.from(decoded as Map); final type = event['type']?.toString() ?? ''; switch (type) { case 'conversation.created': final id = event['conversation']?['id']?.toString(); if (id != null && id.isNotEmpty) _conversationId = id; break; case 'session.updated': _sessionReady = true; _reconnectAttempt = 0; if (!_micMuted) { while (_preSessionAudio.isNotEmpty) { _sendAudio(_preSessionAudio.removeFirst()); } } else { _preSessionAudio.clear(); } _preSessionAudioBytes = 0; _setPhase( _micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening, ); break; case 'input_audio_buffer.speech_started': if (_micMuted) { _send({'type': 'input_audio_buffer.clear'}); break; } _userSpeaking = true; if (isAssistantActive) { // With server VAD xAI cancels generation automatically. Only stop // local playback; sending response.cancel here creates the // "no active response found" race seen in the old implementation. _stopLocalPlayback(); _activeResponseId = null; _responseDone = true; } _setPhase(VoiceSessionPhase.hearing); break; case 'input_audio_buffer.speech_stopped': _userSpeaking = false; _setPhase(VoiceSessionPhase.thinking); break; case 'response.created': _startResponse(event); break; case 'response.output_audio.delta': _handleAssistantAudio(event); break; case 'response.output_audio_transcript.delta': final delta = event['delta']?.toString() ?? ''; if (delta.isNotEmpty) { _liveAssistantTranscript += delta; _notify(); } break; case 'response.output_audio_transcript.done': _commitAssistantTranscript(event['transcript']?.toString()); break; case 'response.output_item.added': final item = jsonEncode(event['item'] ?? '').toLowerCase(); if (item.contains('web_search') || item.contains('x_search')) { _setPhase(VoiceSessionPhase.searching); } break; case 'response.done': if (!_matchesActiveResponse(event['response']?['id']?.toString())) { break; } _responseDone = true; _commitAssistantTranscript(); if (!_playbackActive) { _activeResponseId = null; _setPhase(_micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening); } break; case 'error': _handleServerError(event); break; default: if (kDebugMode && type.isNotEmpty && type != 'ping' && type != 'response.content_part.added' && type != 'response.content_part.done') { debugPrint('Voice event: $type'); } } } catch (error) { debugPrint('Voice event parse failed: $error'); } } void _startResponse(Map event) { if (_liveAssistantTranscript.trim().isNotEmpty) { _commitAssistantTranscript(); } _activeResponseId = event['response']?['id']?.toString() ?? event['response_id']?.toString(); _responseDone = false; _assistantLevel = 0; _playbackGeneration++; unawaited(_audio.stopPlayback(generation: _playbackGeneration)); _setPhase(VoiceSessionPhase.thinking); } void _handleAssistantAudio(Map event) { final responseId = event['response_id']?.toString(); if (_activeResponseId == null || (responseId != null && responseId != _activeResponseId)) { return; } final delta = event['delta']?.toString(); if (delta == null || delta.isEmpty) return; final pcm = base64Decode(delta); _assistantLevel = math.max( _assistantLevel * 0.55, (_pcmRms(pcm) * 4.8).clamp(0.0, 1.0), ); _setPhase(VoiceSessionPhase.speaking); final generation = _playbackGeneration; _playbackChain = _playbackChain .then((_) => _audio.enqueuePlayback( pcm, generation: generation, )) .catchError((Object error, StackTrace stackTrace) { debugPrint('Playback enqueue failed: $error'); }); } bool _matchesActiveResponse(String? responseId) { if (_activeResponseId == null) return false; return responseId == null || responseId == _activeResponseId; } void _handleServerError(Map event) { final nested = event['error']; final error = nested is Map ? Map.from(nested) : const {}; final message = error['message']?.toString() ?? event['message']?.toString() ?? 'خطای سرویس صوتی'; final normalized = message.toLowerCase(); if (normalized.contains('no active response') || normalized.contains('not active')) { return; } _setError(message); } void _commitAssistantTranscript([String? finalText]) { final text = (finalText?.trim().isNotEmpty ?? false) ? finalText!.trim() : _liveAssistantTranscript.trim(); if (text.isNotEmpty && (_assistantMessages.isEmpty || _assistantMessages.last != text)) { _assistantMessages.add(text); } _liveAssistantTranscript = ''; _notify(); } void _sendAudio(Uint8List pcm) { _send({ 'type': 'input_audio_buffer.append', 'audio': base64Encode(pcm), }); } void _send(Map message) { try { _socket?.sink.add(jsonEncode(message)); } catch (error) { debugPrint('Voice WebSocket send failed: $error'); } } void _handleDisconnect(int generation, Object? error) { if (_closing || generation != _connectionGeneration) return; _sessionReady = false; _socket = null; _socketSubscription = null; _stopLocalPlayback(); _activeResponseId = null; _scheduleReconnect(); } void _scheduleReconnect() { if (_closing || _reconnectTimer?.isActive == true) return; _setPhase(VoiceSessionPhase.reconnecting); final seconds = math.min(1 << math.min(_reconnectAttempt, 4), 15); _reconnectAttempt++; _reconnectTimer = Timer(Duration(seconds: seconds), () { _reconnectTimer = null; unawaited(_connect()); }); } void toggleMute() { _micMuted = !_micMuted; _preSessionAudio.clear(); _preSessionAudioBytes = 0; if (_micMuted) { _send({'type': 'input_audio_buffer.clear'}); _setPhase(VoiceSessionPhase.muted); } else { _setPhase( _sessionReady ? VoiceSessionPhase.listening : VoiceSessionPhase.connecting, ); } } void interruptAssistant() { if (!isAssistantActive) return; // Manual stop is the one place where response.cancel is correct. _send({'type': 'response.cancel'}); _stopLocalPlayback(); _activeResponseId = null; _responseDone = true; _setPhase( _micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening); } void _stopLocalPlayback() { _playbackGeneration++; _playbackActive = false; _assistantLevel = 0; unawaited(_audio.stopPlayback(generation: _playbackGeneration)); } double _pcmRms(Uint8List pcm) { if (pcm.length < 2) return 0; final data = ByteData.sublistView(pcm); var sum = 0.0; var count = 0; for (var index = 0; index + 1 < pcm.length; index += 2) { final sample = data.getInt16(index, Endian.little) / 32768.0; sum += sample * sample; count++; } return count == 0 ? 0 : math.sqrt(sum / count); } void _setPhase(VoiceSessionPhase value) { if (_phase == value) { _notify(); return; } _phase = value; _notify(); } void _setError(String message) { _errorMessage = message; _phase = VoiceSessionPhase.error; _notify(); } void _notify() { if (!_closing) notifyListeners(); } Future close() async { if (_closing) return; _closing = true; _phase = VoiceSessionPhase.closed; _connectionGeneration++; _reconnectTimer?.cancel(); _reconnectTimer = null; _sessionReady = false; _preSessionAudio.clear(); _preSessionAudioBytes = 0; final socketSubscription = _socketSubscription; _socketSubscription = null; await socketSubscription?.cancel(); try { await _socket?.sink.close(); } catch (_) {} _socket = null; final audioSubscription = _audioSubscription; _audioSubscription = null; await audioSubscription?.cancel(); try { await _audio.stop(); } catch (_) {} } @override void dispose() { unawaited(close()); super.dispose(); } }