fix(responses): preserve streamed output items

Keep each message and reasoning item at its announced output index.
Include the answer in completed responses with reasoning or fallback
function calls, and retain reasoning supplied through backend deltas.

Add regression coverage for stream indices, final output, plain text,
and automatic tool parsing.

Assisted-by: Codex:GPT-6

core/http/endpoints/openresponses/responses.go

@@ -2424,6 +2424,8 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 	}
 
 	// Non-tool-call streaming path
+	messageOutputIndex := outputIndex
+	var reasoningOutputIndex int
 	// Emit output_item.added for message
 	currentMessageID = fmt.Sprintf("msg_%s", uuid.New().String())
 	messageItem := &schema.ORItemField{
@@ -2436,7 +2438,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 	sendSSEEvent(c, &schema.ORStreamEvent{
 		Type:           "response.output_item.added",
 		SequenceNumber: sequenceNumber,
-		OutputIndex:    &outputIndex,
+		OutputIndex:    &messageOutputIndex,
 		Item:           messageItem,
 	})
 	sequenceNumber++
@@ -2448,7 +2450,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 		Type:           "response.content_part.added",
 		SequenceNumber: sequenceNumber,
 		ItemID:         currentMessageID,
-		OutputIndex:    &outputIndex,
+		OutputIndex:    &messageOutputIndex,
 		ContentIndex:   &currentContentIndex,
 		Part:           &emptyTextPart,
 	})
@@ -2471,10 +2473,11 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 		}
 
 		// Handle reasoning item
-		if extractor.Reasoning() != "" {
+		if extractor.Reasoning() != "" || reasoningDelta != "" {
 			// Check if we need to create reasoning item
 			if currentReasoningID == "" {
 				outputIndex++
+				reasoningOutputIndex = outputIndex
 				currentReasoningID = fmt.Sprintf("reasoning_%s", uuid.New().String())
 				reasoningItem := &schema.ORItemField{
 					Type:   "reasoning",
@@ -2484,7 +2487,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 				sendSSEEvent(c, &schema.ORStreamEvent{
 					Type:           "response.output_item.added",
 					SequenceNumber: sequenceNumber,
-					OutputIndex:    &outputIndex,
+					OutputIndex:    &reasoningOutputIndex,
 					Item:           reasoningItem,
 				})
 				sequenceNumber++
@@ -2496,7 +2499,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 					Type:           "response.content_part.added",
 					SequenceNumber: sequenceNumber,
 					ItemID:         currentReasoningID,
-					OutputIndex:    &outputIndex,
+					OutputIndex:    &reasoningOutputIndex,
 					ContentIndex:   &currentReasoningContentIndex,
 					Part:           &emptyPart,
 				})
@@ -2509,7 +2512,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 					Type:           "response.output_text.delta",
 					SequenceNumber: sequenceNumber,
 					ItemID:         currentReasoningID,
-					OutputIndex:    &outputIndex,
+					OutputIndex:    &reasoningOutputIndex,
 					ContentIndex:   &currentReasoningContentIndex,
 					Delta:          strPtr(reasoningDelta),
 					Logprobs:       emptyLogprobs(),
@@ -2526,7 +2529,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 				Type:           "response.output_text.delta",
 				SequenceNumber: sequenceNumber,
 				ItemID:         currentMessageID,
-				OutputIndex:    &outputIndex,
+				OutputIndex:    &messageOutputIndex,
 				ContentIndex:   &currentContentIndex,
 				Delta:          strPtr(contentDelta),
 				Logprobs:       emptyLogprobs(),
@@ -2595,7 +2598,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 			Type:           "response.output_text.done",
 			SequenceNumber: sequenceNumber,
 			ItemID:         currentReasoningID,
-			OutputIndex:    &outputIndex,
+			OutputIndex:    &reasoningOutputIndex,
 			ContentIndex:   &currentReasoningContentIndex,
 			Text:           strPtr(finalReasoning),
 			Logprobs:       emptyLogprobs(),
@@ -2608,7 +2611,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 			Type:           "response.content_part.done",
 			SequenceNumber: sequenceNumber,
 			ItemID:         currentReasoningID,
-			OutputIndex:    &outputIndex,
+			OutputIndex:    &reasoningOutputIndex,
 			ContentIndex:   &currentReasoningContentIndex,
 			Part:           &reasoningPart,
 		})
@@ -2624,7 +2627,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 		sendSSEEvent(c, &schema.ORStreamEvent{
 			Type:           "response.output_item.done",
 			SequenceNumber: sequenceNumber,
-			OutputIndex:    &outputIndex,
+			OutputIndex:    &reasoningOutputIndex,
 			Item:           reasoningItem,
 		})
 		sequenceNumber++
@@ -2658,7 +2661,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 		Type:           "response.output_text.done",
 		SequenceNumber: sequenceNumber,
 		ItemID:         currentMessageID,
-		OutputIndex:    &outputIndex,
+		OutputIndex:    &messageOutputIndex,
 		ContentIndex:   &currentContentIndex,
 		Text:           strPtr(result),
 		Logprobs:       logprobsPtr(mcpStreamLogprobs),
@@ -2671,7 +2674,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 		Type:           "response.content_part.done",
 		SequenceNumber: sequenceNumber,
 		ItemID:         currentMessageID,
-		OutputIndex:    &outputIndex,
+		OutputIndex:    &messageOutputIndex,
 		ContentIndex:   &currentContentIndex,
 		Part:           &resultPart,
 	})
@@ -2683,7 +2686,7 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 	sendSSEEvent(c, &schema.ORStreamEvent{
 		Type:           "response.output_item.done",
 		SequenceNumber: sequenceNumber,
-		OutputIndex:    &outputIndex,
+		OutputIndex:    &messageOutputIndex,
 		Item:           messageItem,
 	})
 	sequenceNumber++
@@ -2723,34 +2726,9 @@ func handleOpenResponsesStream(c echo.Context, responseID string, createdAt int6
 	// Emit response.completed
 	now := time.Now().Unix()
 
-	// Collect final output items (reasoning first, then messages, then tool calls)
-	var finalOutputItems []schema.ORItemField
-	// Add reasoning item if it exists
-	if currentReasoningID != "" && finalReasoning != "" {
-		finalOutputItems = append(finalOutputItems, schema.ORItemField{
-			Type:    "reasoning",
-			ID:      currentReasoningID,
-			Status:  "completed",
-			Content: []schema.ORContentPart{makeOutputTextPart(finalReasoning)},
-		})
-	}
-	// Add message item
-	if len(collectedOutputItems) > 0 {
-		// Use collected items (may include reasoning already)
-		for _, item := range collectedOutputItems {
-			if item.Type == "message" {
-				finalOutputItems = append(finalOutputItems, item)
-			}
-		}
-	} else {
-		finalOutputItems = append(finalOutputItems, *messageItem)
-	}
-	// Add function_call items from fallback
-	for _, item := range collectedOutputItems {
-		if item.Type == "function_call" {
-			finalOutputItems = append(finalOutputItems, item)
-		}
-	}
+	// The final output array must use the indices announced in the stream.
+	// The message is opened first, followed by reasoning and fallback calls.
+	finalOutputItems := append([]schema.ORItemField{*messageItem}, collectedOutputItems...)
 	responseCompleted := buildORResponse(responseID, createdAt, &now, "completed", input, finalOutputItems, &schema.ORUsage{
 		InputTokens:  noToolTokenUsage.Prompt,
 		OutputTokens: noToolTokenUsage.Completion,

core/http/endpoints/openresponses/responses_stream_test.go

@@ -0,0 +1,148 @@
+// SPDX-License-Identifier: MIT
+package openresponses
+
+import (
+	"context"
+	"encoding/json"
+	"net/http/httptest"
+	"strings"
+
+	"github.com/labstack/echo/v4"
+	"github.com/mudler/LocalAI/core/backend"
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/core/schema"
+	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
+	"github.com/mudler/LocalAI/pkg/model"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Responses stream item consistency", func() {
+	DescribeTable("preserves every item and its announced output index", func(tokens []string, chatDeltas []*pb.ChatDelta, wantReasoning, wantAnswer string, fallback bool) {
+		originalInference := backend.ModelInferenceFunc
+		DeferCleanup(func() { backend.ModelInferenceFunc = originalInference })
+		backend.ModelInferenceFunc = func(
+			ctx context.Context, prompt string, messages schema.Messages,
+			images, videos, audios []string, loader *model.ModelLoader,
+			cfg *config.ModelConfig, cl *config.ModelConfigLoader, app *config.ApplicationConfig,
+			tokenCallback func(string, backend.TokenUsage) bool, tools, toolChoice string,
+			logprobs, topLogprobs *int, logitBias map[string]float64, metadata map[string]string,
+		) (func() (backend.LLMResponse, error), error) {
+			return func() (backend.LLMResponse, error) {
+				for i, token := range tokens {
+					usage := backend.TokenUsage{}
+					if len(chatDeltas) > 0 {
+						usage.ChatDeltas = []*pb.ChatDelta{chatDeltas[i]}
+					}
+					if !tokenCallback(token, usage) {
+						break
+					}
+				}
+				return backend.LLMResponse{Response: strings.Join(tokens, ""), ChatDeltas: chatDeltas, Usage: backend.TokenUsage{Prompt: 3, Completion: 8}}, nil
+			}, nil
+		}
+		cfg := &config.ModelConfig{}
+		cfg.FunctionsConfig.AutomaticToolParsingFallback = fallback
+		cfg.FunctionsConfig.JSONRegexMatch = []string{`(?s)<tool_call>(.*?)</tool_call>`}
+		recorder := httptest.NewRecorder()
+		request := httptest.NewRequest("POST", "/v1/responses", nil)
+		c := echo.New().NewContext(request, recorder)
+		input := &schema.OpenResponsesRequest{Model: "test-model", Input: "hello", Stream: true}
+		err := handleOpenResponsesStream(c, "resp_test", 1, input, cfg, nil, nil, config.NewApplicationConfig(), "hello", &schema.OpenAIRequest{Context: request.Context()}, nil, false, false, nil, nil)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(recorder.Body.String()).To(HaveSuffix("data: [DONE]\n\n"))
+
+		var events []schema.ORStreamEvent
+		var completed *schema.ORResponseResource
+		for _, line := range strings.Split(recorder.Body.String(), "\n") {
+			if !strings.HasPrefix(line, "data: ") || line == "data: [DONE]" {
+				continue
+			}
+			var event schema.ORStreamEvent
+			Expect(json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &event)).To(Succeed())
+			Expect(event.Type).NotTo(Equal("error"))
+			events = append(events, event)
+			if event.Type == "response.completed" {
+				completed = event.Response
+			}
+		}
+		Expect(completed).NotTo(BeNil())
+		wantCount := 1
+		if wantReasoning != "" {
+			wantCount++
+		}
+		if fallback {
+			wantCount++
+		}
+		Expect(completed.Output).To(HaveLen(wantCount), "final output must retain the answer alongside reasoning and fallback calls")
+
+		indices := map[string]int{}
+		done := map[string]int{}
+		deltas := map[string]string{}
+		for i, event := range events {
+			Expect(event.SequenceNumber).To(Equal(i))
+			if event.Type == "response.output_item.added" {
+				Expect(event.Item).NotTo(BeNil())
+				Expect(event.OutputIndex).NotTo(BeNil())
+				Expect(indices).NotTo(HaveKey(event.Item.ID))
+				Expect(*event.OutputIndex).To(Equal(len(indices)))
+				indices[event.Item.ID] = *event.OutputIndex
+			}
+			id := event.ItemID
+			if event.Item != nil {
+				id = event.Item.ID
+			}
+			if id == "" {
+				continue
+			}
+			Expect(indices).To(HaveKey(id))
+			Expect(event.OutputIndex).NotTo(BeNil())
+			Expect(*event.OutputIndex).To(Equal(indices[id]), "event %s changes the index for %s", event.Type, id)
+			Expect(completed.Output[indices[id]].ID).To(Equal(id))
+			if event.Type == "response.output_item.done" {
+				done[id]++
+				Expect(event.Item.Status).To(Equal("completed"))
+				Expect(event.Item.Type).To(Equal(completed.Output[indices[id]].Type))
+				if event.Item.Type == "function_call" {
+					Expect(event.Item.Name).To(Equal(completed.Output[indices[id]].Name))
+					Expect(event.Item.Arguments).To(Equal(completed.Output[indices[id]].Arguments))
+				} else {
+					Expect(event.Item.Content).To(Equal(completed.Output[indices[id]].Content))
+				}
+			}
+			if event.Type == "response.output_text.delta" {
+				deltas[id] += *event.Delta
+			}
+		}
+		Expect(indices).To(HaveLen(wantCount))
+		for _, item := range completed.Output {
+			Expect(done[item.ID]).To(Equal(1))
+			switch item.Type {
+			case "message", "reasoning":
+				want := wantAnswer
+				if item.Type == "reasoning" {
+					want = wantReasoning
+				}
+				parts, ok := item.Content.([]any)
+				Expect(ok).To(BeTrue())
+				Expect(parts).To(HaveLen(1))
+				Expect(parts[0].(map[string]any)["text"]).To(Equal(want))
+				if !fallback {
+					Expect(deltas[item.ID]).To(Equal(want))
+				}
+			case "function_call":
+				Expect(item.Name).To(Equal("get_weather"))
+				Expect(item.Arguments).To(MatchJSON(`{"city":"Rome"}`))
+				Expect(item.CallID).NotTo(BeEmpty())
+			default:
+				Fail("unexpected output item type: " + item.Type)
+			}
+		}
+	},
+		Entry("tagged reasoning and answer", []string{"<think>", "Let me think.", "</think>", "The answer is 42."}, nil, "Let me think.", "The answer is 42.", false),
+		Entry("backend reasoning and answer deltas", []string{"", ""}, []*pb.ChatDelta{{ReasoningContent: "Let me think."}, {Content: "The answer is 42."}}, "Let me think.", "The answer is 42.", false),
+		Entry("plain text", []string{"Hello", " world."}, nil, "", "Hello world.", false),
+		Entry("automatic fallback tool call", []string{`<tool_call>{"name":"get_weather","arguments":{"city":"Rome"}}</tool_call>`}, nil, "", "", true),
+		Entry("reasoning and automatic fallback tool call", []string{"<think>", "Let me think.", "</think>", `<tool_call>{"name":"get_weather","arguments":{"city":"Rome"}}</tool_call>`}, nil, "Let me think.", "", true),
+	)
+})

docs/content/features/text-generation.md

@@ -340,6 +340,15 @@ curl http://localhost:8080/v1/responses \
   }'
 ```
 
+#### Streaming responses
+
+Set `"stream": true` to receive Server-Sent Events. Each `response.output_item.added` event assigns an `output_index` to an item.
+Use that index and the item ID to associate later deltas and completion events with the same item.
+
+If a request without explicit tools produces reasoning, the stream uses separate items for reasoning and answer text.
+Each item keeps its original index throughout the stream.
+The `response.completed` event includes both items in the same index order, followed by any automatically parsed tool calls.
+
 #### Background Processing
 
 Run requests in the background for long-running tasks: