streamGoogle function

AssistantMessageEventStream streamGoogle(Model model, Context context, [ GoogleOptions? options, Client? client, ])

Streams an assistant message from a Google Generative AI endpoint.

Ported from pi's stream in google-generative-ai.ts. The endpoint is {model.baseUrl}/models/{modelId}:streamGenerateContent?alt=sse (default base URL https://generativelanguage.googleapis.com/v1beta on the model descriptor).

Errors-as-events invariant (non-negotiable): this function never throws. Network failures, non-200 responses, provider error chunks, malformed SSE, and aborts all terminate the returned stream with an ErrorEvent carrying StopReason.error or StopReason.aborted.

client overrides the HTTP client (used by tests with http.testing.MockClient); when omitted, an owned client is created and closed when the stream finishes.

Implementation

AssistantMessageEventStream streamGoogle(
  Model model,
  Context context, [
  GoogleOptions? options,
  http.Client? client,
]) {
  final eventStream = AssistantMessageEventStream();
  final cancelToken = options?.cancelToken;
  // No injected client: the shared keep-alive client (never closed per call).
  final httpClient = client ?? sharedProviderHttpClient();

  // Blocks accumulate in the shared state holder; each event carries a
  // fresh immutable snapshot of them (pi mutates one `output` object).
  final state = ProviderStreamState(model);

  unawaited(
    runProviderStream(
      eventStream,
      state,
      cancelToken,
      httpClient,
      ownsClient: false, // shared or injected — never closed per call
      body: () async {
        final apiKey = _getClientApiKey(
          model.provider,
          options?.apiKey,
          options?.headers,
        );

        http.Request buildRequest(Map<String, dynamic> params) =>
            http.Request(
                'POST',
                Uri.parse(
                  '${model.baseUrl}/models/${model.id}'
                  ':streamGenerateContent?alt=sse',
                ),
              )
              ..headers.addAll(_buildHeaders(model, options, apiKey))
              ..body = jsonEncode(params);

        final params = await applyPayloadHook(
          _buildParams(model, context, options),
          model,
          options?.onPayload,
        );

        cancelToken?.throwIfCancelled();

        http.StreamedResponse response;
        try {
          response = await startProviderResponse(
            eventStream,
            state,
            httpClient,
            buildRequest(params),
            cancelToken,
            options?.onResponse,
          );
        } on ProviderHttpError catch (error) {
          // Two Gemini 400s are recoverable by rewriting the request and
          // retrying ONCE — the turn survives and the model can explain the
          // situation to the user instead of dying with a raw API error:
          //
          // 1. 'Unable to process input image': an inline image the backend
          //    cannot decode → every image downgraded to placeholder text.
          // 2. 'missing a thought_signature': a replayed functionCall whose
          //    signature was never captured / dropped / belongs to another
          //    model → the tool call and its result become plain text notes.
          final List<Message>? retryMessages;
          if (_isUndecodableImageError(error) &&
              messagesContainImages(context.messages)) {
            retryMessages = downgradeUndecodableImages(context.messages);
          } else if (_isMissingThoughtSignatureError(error) &&
              _hasUnreplayableToolCall(context.messages, model)) {
            retryMessages = _textifyUnreplayableToolCalls(
              context.messages,
              model,
            );
          } else {
            retryMessages = null;
          }
          if (retryMessages == null) rethrow;

          final retryContext = Context(
            systemPrompt: context.systemPrompt,
            messages: retryMessages,
            tools: context.tools,
          );
          final retryParams = await applyPayloadHook(
            _buildParams(model, retryContext, options),
            model,
            options?.onPayload,
          );
          cancelToken?.throwIfCancelled();
          response = await startProviderResponse(
            eventStream,
            state,
            httpClient,
            buildRequest(retryParams),
            cancelToken,
            options?.onResponse,
          );
        }

        // pi tracks `currentBlock` (the open text/thinking block) inline;
        // here that cursor lives in the chunk handler.
        final handler = _GoogleChunkHandler(eventStream, state, model);

        await _consumeSseStream(response, cancelToken, handler);

        handler.endCurrentBlock();

        if (cancelToken?.isCancelled ?? false) {
          throw const AbortedError();
        }
        if (state.stopReason == StopReason.aborted ||
            state.stopReason == StopReason.error) {
          throw StateError(state.errorMessage ?? 'An unknown error occurred');
        }

        eventStream.push(
          DoneEvent(reason: state.stopReason, message: state.snapshot()),
        );
      },
    ),
  );

  return eventStream;
}