feat(#3996): stream model-generated media deltas

pkg/chat/chat.go

@@ -153,6 +153,13 @@ type MessageDelta struct {
 	ThoughtSignature  []byte              `json:"thought_signature,omitempty"`
 	FunctionCall      *tools.FunctionCall `json:"function_call,omitempty"`
 	ToolCalls         []tools.ToolCall    `json:"tool_calls,omitempty"`
+	// Media carries provider-generated binary content (e.g. inline image
+	// blobs) streamed alongside or instead of text. A single chunk can carry
+	// more than one blob (e.g. Gemini returning multiple inline parts across
+	// candidates in one chunk), so this is a slice rather than a single
+	// pointer — a scalar field silently dropped every blob but the last. See
+	// [MediaDelta].
+	Media []MediaDelta `json:"media,omitempty"`
 }
 
 // MessageStreamChoice represents a choice in a streaming response

pkg/chat/media.go

@@ -0,0 +1,29 @@
+package chat
+
+// MediaDelta carries binary media generated by a provider while a response
+// is streaming (e.g. an inline image blob from Gemini). It is the shared,
+// provider-agnostic representation for this content: adapters for any
+// provider populate it the same way, and the runtime accumulator is the only
+// consumer, so no provider gets a side channel of its own.
+//
+// Data holds the live bytes only for the duration of the stream. Once the
+// turn completes, the runtime materializes them into a session artifact and
+// the persisted assistant message keeps only a reference — see
+// [DocumentSource.ArtifactPath]. Data is never written to session storage.
+type MediaDelta struct {
+	// Data is the raw generated bytes. Present only on the streaming delta;
+	// never persisted.
+	Data []byte `json:"data,omitempty"`
+
+	// MimeType is the media's MIME type as reported by the provider (e.g.
+	// "image/png").
+	MimeType string `json:"mime_type,omitempty"`
+
+	// Name is the provider-supplied display name, if any. May be empty; the
+	// runtime accumulator synthesizes one when needed.
+	Name string `json:"name,omitempty"`
+
+	// Size is the byte length of Data, cached because Data itself is
+	// dropped once the artifact is materialized.
+	Size int64 `json:"size,omitempty"`
+}

pkg/model/provider/gemini/adapter.go

@@ -112,18 +112,22 @@ func (g *StreamAdapter) run() {
 		}
 
 		if resp != nil {
-			// Check for text content without using Text() to avoid warnings
+			// Check for text content and generated inline media without using
+			// Text() to avoid warnings
 			hasText := false
+			hasMedia := false
 			for _, candidate := range resp.Candidates {
 				if candidate.Content != nil {
 					for _, part := range candidate.Content.Parts {
 						if part.Text != "" {
 							hasText = true
-							break
+						}
+						if part.InlineData != nil && len(part.InlineData.Data) > 0 {
+							hasMedia = true
 						}
 					}
 				}
-				if hasText {
+				if hasText && hasMedia {
 					break
 				}
 			}
@@ -134,9 +138,9 @@ func (g *StreamAdapter) run() {
 			// calls. Forward such chunks so downstream can capture token usage.
 			hasUsage := resp.UsageMetadata != nil
 
-			// Send response if it has content, function calls, or usage metadata
-			if hasText || hasFuncs || hasUsage {
-				hasContent = hasContent || hasText
+			// Send response if it has content, generated media, function calls, or usage metadata
+			if hasText || hasMedia || hasFuncs || hasUsage {
+				hasContent = hasContent || hasText || hasMedia
 				hasToolCalls = hasToolCalls || hasFuncs
 				lastResponse = resp // Store for final message
 				if !g.send(result{resp: resp}) {
@@ -232,6 +236,7 @@ func (g *StreamAdapter) Recv() (chat.MessageStreamResponse, error) {
 		var reasoningTextSb strings.Builder
 		var textContentSb strings.Builder
 		var thoughtSignature []byte
+		var media []chat.MediaDelta
 		for _, candidate := range res.resp.Candidates {
 			if candidate.Content != nil {
 				for _, part := range candidate.Content.Parts {
@@ -246,6 +251,21 @@ func (g *StreamAdapter) Recv() (chat.MessageStreamResponse, error) {
 							textContentSb.WriteString(part.Text)
 						}
 					}
+
+					// Inline generated media (e.g. an image from an
+					// image-output model). Gemini can return more than one
+					// inline blob per chunk — multiple parts in one candidate,
+					// or multiple candidates — so every blob is appended
+					// rather than overwriting a single field, which used to
+					// silently drop all but the last one.
+					if part.InlineData != nil && len(part.InlineData.Data) > 0 {
+						media = append(media, chat.MediaDelta{
+							Data:     part.InlineData.Data,
+							MimeType: part.InlineData.MIMEType,
+							Name:     part.InlineData.DisplayName,
+							Size:     int64(len(part.InlineData.Data)),
+						})
+					}
 				}
 			}
 		}
@@ -260,6 +280,9 @@ func (g *StreamAdapter) Recv() (chat.MessageStreamResponse, error) {
 		if len(thoughtSignature) > 0 {
 			resp.Choices[0].Delta.ThoughtSignature = thoughtSignature
 		}
+		if len(media) > 0 {
+			resp.Choices[0].Delta.Media = media
+		}
 
 		// Handle function calls
 		if funcs := res.resp.FunctionCalls(); len(funcs) > 0 {

pkg/model/provider/gemini/adapter_test.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/genai"
 
@@ -183,3 +184,147 @@ func TestStreamAdapter_FunctionCalls(t *testing.T) {
 		require.Empty(t, finalResp.Choices[0].Delta.ToolCalls)
 	})
 }
+
+func TestStreamAdapter_GeneratedImage(t *testing.T) {
+	t.Parallel()
+
+	t.Run("inline image-only chunk is forwarded as Delta.Media", func(t *testing.T) {
+		imgBytes := []byte{0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a}
+		mockResp := &genai.GenerateContentResponse{
+			Candidates: []*genai.Candidate{
+				{
+					Content: &genai.Content{
+						Parts: []*genai.Part{
+							{
+								InlineData: &genai.Blob{
+									Data:        imgBytes,
+									MIMEType:    "image/png",
+									DisplayName: "cat.png",
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+
+		iter := func(fn func(*genai.GenerateContentResponse, error) bool) {
+			fn(mockResp, nil)
+		}
+		adapter := NewStreamAdapter(iter, "test-model", true)
+
+		// A chunk carrying only inline media (no text, no function calls, no
+		// usage) must still be forwarded rather than dropped by the run()
+		// content gate.
+		resp, err := adapter.Recv()
+		require.NoError(t, err)
+		require.Len(t, resp.Choices[0].Delta.Media, 1, "image-only chunk must be forwarded")
+		assert.Equal(t, imgBytes, resp.Choices[0].Delta.Media[0].Data)
+		assert.Equal(t, "image/png", resp.Choices[0].Delta.Media[0].MimeType)
+		assert.Equal(t, "cat.png", resp.Choices[0].Delta.Media[0].Name)
+		assert.Equal(t, int64(len(imgBytes)), resp.Choices[0].Delta.Media[0].Size)
+		assert.Empty(t, resp.Choices[0].Delta.Content, "no text was in this chunk")
+
+		// The synthesized done event must still report a normal stop —
+		// generated media does not change finish-reason handling.
+		done, err := adapter.Recv()
+		require.NoError(t, err)
+		assert.Equal(t, chat.FinishReasonStop, done.Choices[0].FinishReason)
+	})
+
+	t.Run("text and inline image in the same chunk both surface", func(t *testing.T) {
+		imgBytes := []byte{0x01, 0x02, 0x03}
+		mockResp := &genai.GenerateContentResponse{
+			Candidates: []*genai.Candidate{
+				{
+					Content: &genai.Content{
+						Parts: []*genai.Part{
+							{Text: "here you go"},
+							{InlineData: &genai.Blob{Data: imgBytes, MIMEType: "image/jpeg"}},
+						},
+					},
+				},
+			},
+		}
+
+		iter := func(fn func(*genai.GenerateContentResponse, error) bool) {
+			fn(mockResp, nil)
+		}
+		adapter := NewStreamAdapter(iter, "test-model", true)
+
+		resp, err := adapter.Recv()
+		require.NoError(t, err)
+		assert.Equal(t, "here you go", resp.Choices[0].Delta.Content, "text handling must remain intact alongside media")
+		require.Len(t, resp.Choices[0].Delta.Media, 1)
+		assert.Equal(t, imgBytes, resp.Choices[0].Delta.Media[0].Data)
+		assert.Equal(t, "image/jpeg", resp.Choices[0].Delta.Media[0].MimeType)
+		// Provider omitted a display name; the adapter must not invent one —
+		// that is the runtime accumulator's job.
+		assert.Empty(t, resp.Choices[0].Delta.Media[0].Name)
+	})
+
+	t.Run("empty inline data is not surfaced as media", func(t *testing.T) {
+		// A Blob with zero bytes must not produce a spurious empty Media
+		// delta (and, since it carries no text either, must not be
+		// forwarded as a chunk at all).
+		mockResp := &genai.GenerateContentResponse{
+			Candidates: []*genai.Candidate{
+				{
+					Content: &genai.Content{
+						Parts: []*genai.Part{
+							{InlineData: &genai.Blob{Data: nil, MIMEType: "image/png"}},
+						},
+					},
+				},
+			},
+		}
+
+		iter := func(fn func(*genai.GenerateContentResponse, error) bool) {
+			fn(mockResp, nil)
+		}
+		adapter := NewStreamAdapter(iter, "test-model", true)
+
+		// Nothing at all was produced (no text, no media, no tool calls, no
+		// usage), so the stream ends directly with EOF — same as any other
+		// content-free chunk, no synthesized done event.
+		_, err := adapter.Recv()
+		require.ErrorIs(t, err, io.EOF)
+	})
+
+	t.Run("multiple inline blobs across parts and candidates in one chunk are all retained", func(t *testing.T) {
+		// Gemini can pack more than one generated blob into a single chunk:
+		// multiple parts within a candidate, and/or multiple candidates.
+		// Every blob must survive, not just the last one seen.
+		mockResp := &genai.GenerateContentResponse{
+			Candidates: []*genai.Candidate{
+				{
+					Content: &genai.Content{
+						Parts: []*genai.Part{
+							{InlineData: &genai.Blob{Data: []byte{0x01}, MIMEType: "image/png", DisplayName: "first.png"}},
+							{InlineData: &genai.Blob{Data: []byte{0x02}, MIMEType: "image/jpeg", DisplayName: "second.jpg"}},
+						},
+					},
+				},
+				{
+					Content: &genai.Content{
+						Parts: []*genai.Part{
+							{InlineData: &genai.Blob{Data: []byte{0x03}, MIMEType: "image/webp", DisplayName: "third.webp"}},
+						},
+					},
+				},
+			},
+		}
+
+		iter := func(fn func(*genai.GenerateContentResponse, error) bool) {
+			fn(mockResp, nil)
+		}
+		adapter := NewStreamAdapter(iter, "test-model", true)
+
+		resp, err := adapter.Recv()
+		require.NoError(t, err)
+		require.Len(t, resp.Choices[0].Delta.Media, 3, "every blob in the chunk must be retained")
+		assert.Equal(t, "first.png", resp.Choices[0].Delta.Media[0].Name)
+		assert.Equal(t, "second.jpg", resp.Choices[0].Delta.Media[1].Name)
+		assert.Equal(t, "third.webp", resp.Choices[0].Delta.Media[2].Name)
+	})
+}

pkg/runtime/streaming.go

@@ -47,9 +47,13 @@ type streamResult struct {
 	ReasoningContent  string
 	ThinkingSignature string
 	ThoughtSignature  []byte
-	Stopped           bool
-	FinishReason      chat.FinishReason
-	Usage             *chat.Usage
+	// Media accumulates every [chat.MediaDelta] streamed during the turn
+	// (e.g. generated images). Populated regardless of provider — see
+	// chat.MessageDelta.Media.
+	Media        []chat.MediaDelta
+	Stopped      bool
+	FinishReason chat.FinishReason
+	Usage        *chat.Usage
 }
 
 // handleStream reads a chat.MessageStream to completion, emitting streaming
@@ -110,6 +114,7 @@ func handleStream(ctx context.Context, cancelStream context.CancelCauseFunc, str
 	var thinkingSignature string
 	var thoughtSignature []byte
 	var toolCalls []tools.ToolCall
+	var media []chat.MediaDelta
 	var messageUsage *chat.Usage
 	var providerFinishReason chat.FinishReason
 
@@ -205,6 +210,20 @@ mainLoop:
 				thoughtSignature = choice.Delta.ThoughtSignature
 			}
 
+			// Provider-generated binary media (e.g. inline image blobs) is
+			// accumulated up front, before the terminal finish-reason check
+			// below. A provider (e.g. Gemini) can pack generated media and a
+			// terminal finish_reason ("stop"/"length"/"refusal") into the SAME
+			// chunk; accumulating after that check would return before this
+			// chunk's media was ever added, silently dropping it. Not emitted
+			// as a streaming event: no consumer needs a live partial-media
+			// event yet, and the runtime materializes the final bytes into a
+			// session artifact once the turn completes (see
+			// recordAssistantMessage).
+			if len(choice.Delta.Media) > 0 {
+				media = append(media, choice.Delta.Media...)
+			}
+
 			// Accumulate tool call deltas from this chunk *before* evaluating the
 			// finish reason below. Some OpenAI-compatible providers (e.g. LiteLLM
 			// in front of Gemini) pack a complete tool call and a terminal
@@ -296,6 +315,7 @@ mainLoop:
 					ReasoningContent:  fullReasoningContent.String(),
 					ThinkingSignature: thinkingSignature,
 					ThoughtSignature:  thoughtSignature,
+					Media:             media,
 					Stopped:           len(toolCalls) == 0, // stop only when there are no tool calls to execute
 					FinishReason:      finishReason,
 					Usage:             messageUsage,
@@ -362,6 +382,8 @@ mainLoop:
 	// Invariant: a bare-EOF turn (no per-choice finish_reason) is terminal
 	// whenever there are no tool calls — the outer loop has nothing to continue
 	// on. Turns with tool calls keep Stopped=false so the loop executes them.
+	// Media is irrelevant to this decision: with no tool calls pending, the turn
+	// is over either way (media or not) — stopping is what ends it correctly.
 	// NOTE(krissetto): this can likely be removed once compaction works properly with all providers (aka dmr)
 	stoppedNoToolCalls := len(toolCalls) == 0
 
@@ -396,6 +418,7 @@ mainLoop:
 		ReasoningContent:  fullReasoningContent.String(),
 		ThinkingSignature: thinkingSignature,
 		ThoughtSignature:  thoughtSignature,
+		Media:             media,
 		Stopped:           stoppedNoToolCalls,
 		FinishReason:      finishReason,
 		Usage:             messageUsage,