streamEvents method

@override Stream<ProviderEvent> streamEvents(List<ChatMessage> conversation)

Streams structured events for conversation's next assistant turn.

conversation is oldest-first; its last element is the new user turn.

Implementation

@override
Stream<ProviderEvent> streamEvents(List<ChatMessage> conversation) async* {
  final request = http.Request('POST', _endpoint)
    ..headers['Authorization'] = 'Bearer $apiKey'
    ..headers['Content-Type'] = 'application/json'
    ..headers['Accept'] = 'text/event-stream'
    ..body = jsonEncode({
      'model': model,
      'stream': true,
      'messages': _toOpenAiMessages(conversation),
      if (tools != null) 'tools': tools,
      ...?extraParams,
    });

  final response = await _httpClient.send(request);
  if (response.statusCode < 200 || response.statusCode >= 300) {
    final body = await response.stream.bytesToString();
    throw AgentProviderException(response.statusCode, body);
  }

  final pendingCalls = <int, _PendingToolCall>{};

  await for (final payload in sseDataEvents(response.stream)) {
    if (payload == '[DONE]') break;
    if (payload.isEmpty) continue;

    final Map<String, Object?> json;
    try {
      json = jsonDecode(payload) as Map<String, Object?>;
    } on FormatException {
      // OpenRouter occasionally sends non-JSON keep-alive comments; skip
      // rather than aborting the whole stream.
      continue;
    }

    // A payload that parses as valid JSON but has an unexpected shape
    // (e.g. `choices` isn't a List) throws a TypeError from the casts
    // below -- treat it the same as a FormatException: skip the one bad
    // chunk rather than failing the whole turn and losing everything
    // accumulated so far (including `pendingCalls`).
    try {
      // Checked with `is` rather than `as` so a malformed `error` field
      // can't itself throw a TypeError that gets swallowed by the catch
      // below -- a real mid-stream error must always surface.
      final error = json['error'];
      if (error is Map<String, Object?>) {
        final code = error['code'];
        throw AgentProviderException(
          code is int ? code : response.statusCode,
          error['message']?.toString() ?? jsonEncode(error),
        );
      }

      final choices = json['choices'] as List<Object?>?;
      final delta = choices == null || choices.isEmpty
          ? null
          : (choices.first as Map<String, Object?>)['delta']
              as Map<String, Object?>?;
      if (delta == null) continue;

      final content = delta['content'] as String?;
      if (content != null && content.isNotEmpty) {
        yield TextDelta(content);
      }

      final toolCalls = delta['tool_calls'] as List<Object?>?;
      if (toolCalls != null) {
        for (final raw in toolCalls) {
          final entry = raw as Map<String, Object?>;
          final index = entry['index'] as int? ?? 0;
          final function = entry['function'] as Map<String, Object?>?;
          final pending =
              pendingCalls.putIfAbsent(index, _PendingToolCall.new);
          final id = entry['id'] as String?;
          if (id != null) pending.id = id;
          final name = function?['name'] as String?;
          if (name != null) pending.name = name;
          final argsChunk = function?['arguments'] as String?;
          if (argsChunk != null) pending.arguments.write(argsChunk);
        }
      }
    } on TypeError {
      continue;
    }
  }

  for (final pending in pendingCalls.values) {
    final call = pending.toToolCall();
    if (call != null) yield ToolCallEvent(call);
  }
}