diff --git a/pkg/runtime/streaming.go b/pkg/runtime/streaming.go index 1b69e27e50..30ca571cd4 100644 --- a/pkg/runtime/streaming.go +++ b/pkg/runtime/streaming.go @@ -356,6 +356,26 @@ mainLoop: } } + // Text in a terminal chunk belongs to the answer too: some + // OpenAI-compatible providers send the last delta together with + // finish_reason, so accumulate it before the early return below. + if choice.Delta.ReasoningContent != "" { + responseStarted = true + events.Emit(AgentChoiceReasoning(a.Name(), sess.ID, choice.Delta.ReasoningContent, messageID)) + fullReasoningContent.WriteString(choice.Delta.ReasoningContent) + } + + // Capture thinking signature for Anthropic extended thinking + if choice.Delta.ThinkingSignature != "" { + responseStarted = true + thinkingSignature = choice.Delta.ThinkingSignature + } + + if choice.Delta.Content != "" { + responseStarted = true + appendContent(markerFilter.Push(choice.Delta.Content)) + } + if choice.FinishReason == chat.FinishReasonStop || choice.FinishReason == chat.FinishReasonLength || choice.FinishReason == chat.FinishReasonRefusal { finishAssistantText() recordUsage() @@ -403,23 +423,6 @@ mainLoop: providerFinishReason = choice.FinishReason } - if choice.Delta.ReasoningContent != "" { - responseStarted = true - events.Emit(AgentChoiceReasoning(a.Name(), sess.ID, choice.Delta.ReasoningContent, messageID)) - fullReasoningContent.WriteString(choice.Delta.ReasoningContent) - } - - // Capture thinking signature for Anthropic extended thinking - if choice.Delta.ThinkingSignature != "" { - responseStarted = true - thinkingSignature = choice.Delta.ThinkingSignature - } - - if choice.Delta.Content != "" { - responseStarted = true - appendContent(markerFilter.Push(choice.Delta.Content)) - } - case <-ctx.Done(): // Context cancelled (SIGTERM, Ctrl+C, or idle-timeout cancel from // this function). Return promptly so graceful shutdown can proceed. diff --git a/pkg/runtime/streaming_test.go b/pkg/runtime/streaming_test.go index 4422c8684a..a344143b1e 100644 --- a/pkg/runtime/streaming_test.go +++ b/pkg/runtime/streaming_test.go @@ -216,6 +216,43 @@ func TestHandleStream_ToolCallThenSeparateStop(t *testing.T) { assert.False(t, res.Stopped) } +// TestHandleStream_ContentInTerminalChunk covers providers that send the last +// text delta in the same chunk as the terminal finish_reason: that text must +// end up in the result and in the live events instead of being dropped. +func TestHandleStream_ContentInTerminalChunk(t *testing.T) { + t.Parallel() + + for _, reason := range []chat.FinishReason{chat.FinishReasonStop, chat.FinishReasonLength} { + t.Run(string(reason), func(t *testing.T) { + t.Parallel() + + b := newStreamBuilder(). + AddReasoning("thinking "). + AddContent("M"). + AddContent("1-deepseek-v4") + b.responses = append(b.responses, chat.MessageStreamResponse{ + Choices: []chat.MessageStreamChoice{{ + Index: 0, + FinishReason: reason, + Delta: chat.MessageDelta{ + Content: ".1-flash", + ReasoningContent: "done", + }, + }}, + Usage: &chat.Usage{InputTokens: 1, OutputTokens: 1}, + }) + + res, live := runMarkerStream(t, b.Build()) + + assert.Equal(t, "M1-deepseek-v4.1-flash", res.Content) + assert.Equal(t, res.Content, live) + assert.Equal(t, "thinking done", res.ReasoningContent) + assert.Equal(t, reason, res.FinishReason) + assert.True(t, res.Stopped) + }) + } +} + // TestHandleStream_MediaAccumulatesAlongsideText verifies that a // generated-media delta streamed alongside text is accumulated into // streamResult.Media without disturbing the existing text/finish-reason