feat(ai-gemini): stream structured outputs natively

.changeset/tidy-gemini-streams.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-gemini': minor
+---
+
+Add native structured-output streaming for the Gemini and experimental Gemini Interactions text adapters.

packages/ai-gemini/src/adapters/text.ts

@@ -222,6 +222,107 @@ export class GeminiTextAdapter<
     }
   }
 
+  /** Stream Gemini's native JSON response and emit the parsed object before completion. */
+  async *structuredOutputStream(
+    options: StructuredOutputOptions<GeminiTextProviderOptions>,
+  ): AsyncIterable<StreamChunk> {
+    const { chatOptions: requestedChatOptions, outputSchema } = options
+    const chatOptions = {
+      ...requestedChatOptions,
+      runId: requestedChatOptions.runId ?? generateId(this.name),
+    }
+    const mappedOptions = this.mapCommonOptionsToGemini(chatOptions)
+
+    try {
+      chatOptions.logger.request(
+        `activity=structuredOutputStream provider=gemini model=${this.model} messages=${chatOptions.messages.length}`,
+        { provider: 'gemini', model: this.model },
+      )
+      const result = await this.client.models.generateContentStream({
+        ...mappedOptions,
+        config: {
+          ...mappedOptions.config,
+          responseMimeType: 'application/json',
+          responseSchema: outputSchema,
+        },
+      })
+
+      let rawText = ''
+      let finished:
+        | Extract<StreamChunk, { type: typeof EventType.RUN_FINISHED }>
+        | undefined
+      let failed = false
+      for await (const chunk of this.processStreamChunks(
+        result,
+        chatOptions,
+        chatOptions.logger,
+      )) {
+        if (chunk.type === EventType.TEXT_MESSAGE_CONTENT) {
+          rawText += chunk.delta
+        }
+        if (chunk.type === EventType.RUN_ERROR) failed = true
+        if (chunk.type === EventType.RUN_FINISHED) {
+          finished = chunk
+        } else {
+          yield chunk
+        }
+      }
+
+      if (failed) return
+      if (!finished) {
+        yield structuredStreamError(
+          chatOptions,
+          'Gemini structured-output stream ended without a terminal event',
+          'truncated-stream',
+        )
+        return
+      }
+      if (!rawText) {
+        yield structuredStreamError(
+          chatOptions,
+          'Gemini structured-output stream contained no content',
+          'empty-response',
+        )
+        return
+      }
+
+      let object: unknown
+      try {
+        object = JSON.parse(rawText)
+      } catch {
+        yield structuredStreamError(
+          chatOptions,
+          'Failed to parse Gemini structured-output stream as JSON',
+          'parse-error',
+        )
+        return
+      }
+
+      yield {
+        type: EventType.CUSTOM,
+        name: 'structured-output.complete',
+        value: { object, raw: rawText },
+        model: chatOptions.model,
+        timestamp: Date.now(),
+      }
+      yield finished
+    } catch (error) {
+      const rawEvent = toRunErrorRawEvent(error)
+      const message =
+        error instanceof Error
+          ? error.message
+          : 'An unknown error occurred during structured output streaming.'
+      chatOptions.logger.errors('gemini.structuredOutputStream fatal', {
+        error,
+        source: 'gemini.structuredOutputStream',
+      })
+      yield {
+        ...structuredStreamError(chatOptions, message, 'provider-error'),
+        ...(rawEvent !== undefined && { rawEvent }),
+      }
+    }
+  }
+
   /**
    * Extract text content from a non-streaming response
    */
@@ -952,6 +1053,22 @@ export class GeminiTextAdapter<
   }
 }
 
+function structuredStreamError(
+  options: TextOptions<GeminiTextProviderOptions>,
+  message: string,
+  code: string,
+): Extract<StreamChunk, { type: typeof EventType.RUN_ERROR }> {
+  return {
+    type: EventType.RUN_ERROR,
+    runId: options.runId,
+    model: options.model,
+    timestamp: Date.now(),
+    message,
+    code,
+    error: { message, code },
+  }
+}
+
 /**
  * Creates a Gemini text adapter with explicit API key.
  * Type resolution happens here at the call site.

packages/ai-gemini/src/experimental/text-interactions/adapter.ts

@@ -438,6 +438,148 @@ export class GeminiTextInteractionsAdapter<
       )
     }
   }
+
+  async *structuredOutputStream(
+    options: StructuredOutputOptions<GeminiTextInteractionsProviderOptions>,
+  ): AsyncIterable<StreamChunk> {
+    const { chatOptions, outputSchema } = options
+    const runId = chatOptions.runId ?? generateId(this.name)
+    const threadId = chatOptions.threadId ?? generateId(this.name)
+    const timestamp = Date.now()
+    const effectivePreviousInteractionId =
+      chatOptions.modelOptions?.previous_interaction_id ??
+      this.interactionIdByThread.get(threadId)
+    const baseRequest = buildInteractionsRequest({
+      ...chatOptions,
+      modelOptions: {
+        ...chatOptions.modelOptions,
+        previous_interaction_id: effectivePreviousInteractionId,
+      },
+    })
+    const request: GeminiInteractionsRequestBody = {
+      ...baseRequest,
+      stream: true,
+      response_format: {
+        type: 'text',
+        mime_type: 'application/json',
+        schema: outputSchema,
+      },
+    }
+
+    try {
+      chatOptions.logger.request(
+        `activity=structuredOutputStream provider=gemini-text-interactions model=${this.model} messages=${chatOptions.messages.length}`,
+        { provider: this.name, model: this.model, request },
+      )
+      const stream = (await this.client.interactions.create(
+        request as GeminiInteractionsRequestBody &
+          Parameters<typeof this.client.interactions.create>[0],
+        { signal: chatOptions.abortController?.signal },
+      )) as AsyncIterable<InteractionSSEEvent>
+
+      let rawText = ''
+      let finished:
+        | Extract<StreamChunk, { type: typeof EventType.RUN_FINISHED }>
+        | undefined
+      let failed = false
+      for await (const chunk of translateInteractionEvents(
+        stream,
+        chatOptions.model,
+        runId,
+        threadId,
+        chatOptions.parentRunId,
+        timestamp,
+        this.name,
+        chatOptions.logger,
+      )) {
+        if (chunk.type === EventType.TEXT_MESSAGE_CONTENT)
+          rawText += chunk.delta
+        if (
+          chunk.type === EventType.CUSTOM &&
+          chunk.name === 'gemini.interactionId'
+        ) {
+          const value =
+            chunk.value as GeminiInteractionsCustomEventValue<'gemini.interactionId'>
+          this.interactionIdByThread.set(threadId, value.interactionId)
+        }
+        if (chunk.type === EventType.RUN_ERROR) failed = true
+        if (chunk.type === EventType.RUN_FINISHED) finished = chunk
+        else yield chunk
+      }
+
+      if (failed) return
+      if (!finished) {
+        yield interactionsStructuredStreamError(
+          chatOptions,
+          runId,
+          'Gemini Interactions structured-output stream ended without a terminal event',
+          'truncated-stream',
+        )
+        return
+      }
+      if (!rawText) {
+        yield interactionsStructuredStreamError(
+          chatOptions,
+          runId,
+          'Gemini Interactions structured-output stream contained no content',
+          'empty-response',
+        )
+        return
+      }
+      let object: unknown
+      try {
+        object = JSON.parse(rawText)
+      } catch {
+        yield interactionsStructuredStreamError(
+          chatOptions,
+          runId,
+          'Failed to parse Gemini Interactions structured-output stream as JSON',
+          'parse-error',
+        )
+        return
+      }
+      yield {
+        type: EventType.CUSTOM,
+        name: 'structured-output.complete',
+        value: { object, raw: rawText },
+        model: chatOptions.model,
+        timestamp,
+      }
+      yield finished
+    } catch (error) {
+      const message =
+        error instanceof Error
+          ? error.message
+          : 'An unknown error occurred during structured output streaming.'
+      chatOptions.logger.errors(
+        'gemini-text-interactions.structuredOutputStream fatal',
+        { error, source: 'gemini-text-interactions.structuredOutputStream' },
+      )
+      yield interactionsStructuredStreamError(
+        chatOptions,
+        runId,
+        message,
+        'provider-error',
+      )
+    }
+  }
+}
+
+function interactionsStructuredStreamError(
+  options: TextOptions<GeminiTextInteractionsProviderOptions>,
+  runId: string,
+  message: string,
+  code: string,
+): Extract<StreamChunk, { type: typeof EventType.RUN_ERROR }> {
+  return {
+    type: EventType.RUN_ERROR,
+    runId,
+    model: options.model,
+    timestamp: Date.now(),
+    message,
+    code,
+    error: { message, code },
+  }
 }
 
 /** @experimental Interactions API is in Beta. */

packages/ai-gemini/tests/gemini-adapter.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, beforeEach, vi } from 'vitest'
 import { z } from 'zod'
-import { chat, summarize } from '@tanstack/ai'
+import { EventType, chat, summarize } from '@tanstack/ai'
 import type { Tool, StreamChunk } from '@tanstack/ai'
 import {
   Type,
@@ -963,6 +963,67 @@ describe('GeminiAdapter through AI', () => {
     expect(adapter.supportsCombinedToolsAndSchema()).toBe(false)
   })
 
+  it('streams structured output natively and completes before RUN_FINISHED', async () => {
+    mocks.generateContentStreamSpy.mockResolvedValue(
+      createStream([
+        {
+          candidates: [{ content: { parts: [{ text: '{"city":' }] } }],
+        },
+        {
+          candidates: [
+            {
+              content: { parts: [{ text: '"Madrid"}' }] },
+              finishReason: 'STOP',
+            },
+          ],
+          usageMetadata: { totalTokenCount: 4 },
+        },
+      ]),
+    )
+
+    const events: Array<StreamChunk> = []
+    for await (const event of createTextAdapter().structuredOutputStream({
+      chatOptions: {
+        model: 'gemini-2.5-pro',
+        messages: [{ role: 'user', content: 'Return a city' }],
+        logger: {
+          request: vi.fn(),
+          provider: vi.fn(),
+          errors: vi.fn(),
+        } as any,
+      },
+      outputSchema: {
+        type: 'object',
+        properties: { city: { type: 'string' } },
+      },
+    })) {
+      events.push(event)
+    }
+
+    const payload = mocks.generateContentStreamSpy.mock.calls[0]![0]
+    expect(payload.config).toMatchObject({
+      responseMimeType: 'application/json',
+      responseSchema: expect.objectContaining({ type: 'object' }),
+    })
+    expect(
+      events.filter((event) => event.type === 'TEXT_MESSAGE_CONTENT'),
+    ).toHaveLength(2)
+    const completeIndex = events.findIndex(
+      (event) =>
+        event.type === EventType.CUSTOM &&
+        event.name === 'structured-output.complete',
+    )
+    const finishedIndex = events.findIndex(
+      (event) => event.type === EventType.RUN_FINISHED,
+    )
+    expect(completeIndex).toBeGreaterThan(-1)
+    expect(completeIndex).toBeLessThan(finishedIndex)
+    expect((events[completeIndex] as any).value).toEqual({
+      object: { city: 'Madrid' },
+      raw: '{"city":"Madrid"}',
+    })
+  })
+
   it('routes summarize() through the gemini chat-stream path', async () => {
     const summaryText = 'Short and sweet.'
     const streamChunks = [