trainhub-flutter/lib/data/services/llm_client.dart

import 'dart:convert';

import 'package:dio/dio.dart';
import 'package:trainhub_flutter/core/constants/ai_constants.dart';
import 'package:trainhub_flutter/data/services/ai_settings_service.dart';

/// Streams chat completions from the configured AI provider.
///
/// - Local llama.cpp, OpenAI and Gemini speak the OpenAI-compatible
///   `/chat/completions` SSE format.
/// - Anthropic uses its native `/v1/messages` SSE format.
///
/// Owns all HTTP/SSE plumbing so that controllers only deal with a plain
/// `Stream<String>` of content deltas.
class LlmClient {
  LlmClient(this._settingsService, {Dio? dio})
    : _dio =
          dio ??
          Dio(
            BaseOptions(
              connectTimeout: AiConstants.serverConnectTimeout,
              receiveTimeout: AiConstants.serverReceiveTimeout,
            ),
          );

  final AiSettingsService _settingsService;
  final Dio _dio;

  AiProvider get activeProvider => _settingsService.settings.provider;

  /// Streams content deltas for a chat completion.
  ///
  /// [messages] must already include the system prompt and history.
  /// Pass a [cancelToken] to allow aborting generation mid-stream.
  /// Throws [DioException] if the server is unreachable; errors that occur
  /// mid-stream are surfaced through the returned stream.
  Stream<String> streamChat(
    List<Map<String, String>> messages, {
    CancelToken? cancelToken,
  }) {
    final settings = _settingsService.settings;
    return switch (settings.provider) {
      AiProvider.anthropic => _streamAnthropic(
        messages,
        settings,
        cancelToken: cancelToken,
      ),
      _ => _streamOpenAiCompatible(
        messages,
        settings,
        cancelToken: cancelToken,
      ),
    };
  }

  // ---------------------------------------------------------------------------
  // OpenAI-compatible providers: local llama.cpp, OpenAI, Gemini
  // ---------------------------------------------------------------------------

  Stream<String> _streamOpenAiCompatible(
    List<Map<String, String>> messages,
    AiSettings settings, {
    CancelToken? cancelToken,
  }) async* {
    final provider = settings.provider;
    final isBuiltIn = provider == AiProvider.local;

    final url = switch (provider) {
      AiProvider.local => AiConstants.chatApiUrl,
      AiProvider.ollama || AiProvider.lmstudio =>
        '${settings.baseUrlFor(provider)}/v1/chat/completions',
      AiProvider.openai => 'https://api.openai.com/v1/chat/completions',
      AiProvider.gemini =>
        'https://generativelanguage.googleapis.com/v1beta/openai/chat/completions',
      AiProvider.anthropic => throw StateError('handled separately'),
    };

    final apiKey = settings.apiKeyFor(provider);
    if (provider.needsApiKey && apiKey == null) {
      throw Exception('No API key configured for ${provider.label}.');
    }

    final response = await _dio.post<ResponseBody>(
      url,
      cancelToken: cancelToken,
      options: Options(
        responseType: ResponseType.stream,
        headers: apiKey == null ? null : {'Authorization': 'Bearer $apiKey'},
      ),
      data: {
        if (!isBuiltIn) 'model': settings.modelFor(provider),
        'messages': messages,
        if (isBuiltIn) 'temperature': AiConstants.chatTemperature,
        'max_tokens': AiConstants.chatMaxTokens,
        // llama.cpp-specific: reuses the KV-cache of the shared prefix
        // between turns — makes multi-turn conversations respond much faster.
        if (isBuiltIn) 'cache_prompt': true,
        'stream': true,
      },
    );

    await for (final line in _sseLines(response)) {
      if (!line.startsWith('data: ')) continue;
      final dataStr = line.substring(6).trim();
      if (dataStr == '[DONE]') return;
      if (dataStr.isEmpty) continue;
      try {
        final data = jsonDecode(dataStr);
        final delta =
            (data['choices']?[0]?['delta']?['content'] ?? '') as String;
        if (delta.isNotEmpty) yield delta;
      } catch (_) {
        // Malformed line — dropped. Thanks to the line buffering this only
        // happens on genuinely corrupt data, not on chunk boundaries.
      }
    }
  }

  // ---------------------------------------------------------------------------
  // Anthropic native Messages API
  // ---------------------------------------------------------------------------

  Stream<String> _streamAnthropic(
    List<Map<String, String>> messages,
    AiSettings settings, {
    CancelToken? cancelToken,
  }) async* {
    final apiKey = settings.apiKeyFor(AiProvider.anthropic);
    if (apiKey == null) {
      throw Exception('No API key configured for Anthropic.');
    }

    // Anthropic takes the system prompt as a top-level field, not a message.
    final system = messages
        .where((m) => m['role'] == 'system')
        .map((m) => m['content'])
        .join('\n\n');
    final chat = messages.where((m) => m['role'] != 'system').toList();

    final response = await _dio.post<ResponseBody>(
      'https://api.anthropic.com/v1/messages',
      cancelToken: cancelToken,
      options: Options(
        responseType: ResponseType.stream,
        headers: {'x-api-key': apiKey, 'anthropic-version': '2023-06-01'},
      ),
      data: {
        'model': settings.modelFor(AiProvider.anthropic),
        'max_tokens': AiConstants.chatMaxTokens,
        if (system.isNotEmpty) 'system': system,
        'messages': chat,
        'stream': true,
      },
    );

    await for (final line in _sseLines(response)) {
      if (!line.startsWith('data: ')) continue;
      final dataStr = line.substring(6).trim();
      if (dataStr.isEmpty) continue;
      try {
        final data = jsonDecode(dataStr);
        final type = data['type'] as String?;
        if (type == 'content_block_delta') {
          final delta = data['delta'];
          if (delta?['type'] == 'text_delta') {
            final text = (delta['text'] ?? '') as String;
            if (text.isNotEmpty) yield text;
          }
        } else if (type == 'message_stop') {
          return;
        } else if (type == 'error') {
          throw Exception(
            'Anthropic API error: ${data['error']?['message'] ?? 'unknown'}',
          );
        }
      } on FormatException {
        // Malformed line — dropped.
      }
    }
  }

  // ---------------------------------------------------------------------------
  // SSE line splitting with chunk-boundary buffering
  // ---------------------------------------------------------------------------

  /// SSE lines can be split across TCP chunks, so carry the unfinished tail
  /// of each chunk over to the next one instead of dropping it.
  Stream<String> _sseLines(Response<ResponseBody> response) async* {
    var pending = '';
    await for (final chunk in response.data!.stream) {
      pending += utf8.decode(chunk, allowMalformed: true);
      final lines = pending.split('\n');
      pending = lines.removeLast();
      for (final line in lines) {
        yield line;
      }
    }
    if (pending.isNotEmpty) yield pending;
  }
}