fix(openai): normalize chat stream termination

relay/channel/openai/relay-openai.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sort"
 	"strings"
 
 	"github.com/QuantumNous/new-api/common"
@@ -121,6 +122,7 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
 	var secondLastStreamData string // 存储倒数第二个stream data，用于音频模型
 	seenStreamToolCalls := make(map[string]struct{})
 	var streamFunctionCallNames []string
+	streamChoiceStates := make(map[int]openAIStreamChoiceState)
 
 	// 检查是否为音频模型
 	isAudioModel := strings.Contains(strings.ToLower(model), "audio")
@@ -133,6 +135,19 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
 			}
 		}
 		if len(data) > 0 {
+			var streamResponse dto.ChatCompletionsStreamResponse
+			if err := common.UnmarshalJsonStr(data, &streamResponse); err == nil {
+				for _, choice := range streamResponse.Choices {
+					state := streamChoiceStates[choice.Index]
+					if choice.FinishReason != nil && *choice.FinishReason != "" {
+						state.finished = true
+					}
+					if len(choice.Delta.ToolCalls) > 0 {
+						state.toolCalls = true
+					}
+					streamChoiceStates[choice.Index] = state
+				}
+			}
 			// 对音频模型，保存倒数第二个stream data
 			if isAudioModel && lastStreamData != "" {
 				secondLastStreamData = lastStreamData
@@ -165,17 +180,69 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
 		}
 	}
 
+	// A finish_reason is an explicit terminal signal even when an upstream
+	// omits [DONE]. EOF/timeout without either signal is only a partial stream
+	// and must not be normalized into a successful downstream completion.
+	streamCompleted := openAIStreamCompleted(info, streamChoiceStates)
+	var synthesizedFinishResponse *dto.ChatCompletionsStreamResponse
+	if streamCompleted && info.RelayFormat == types.RelayFormatOpenAI &&
+		info.StreamStatus != nil && info.StreamStatus.EndReason == relaycommon.StreamEndReasonDone {
+		missingChoiceIndexes := make([]int, 0)
+		for choiceIndex, state := range streamChoiceStates {
+			if !state.finished {
+				missingChoiceIndexes = append(missingChoiceIndexes, choiceIndex)
+			}
+		}
+		if len(missingChoiceIndexes) > 0 {
+			sort.Ints(missingChoiceIndexes)
+			synthesizedFinishResponse = &dto.ChatCompletionsStreamResponse{
+				Id:      responseId,
+				Object:  "chat.completion.chunk",
+				Created: createAt,
+				Model:   model,
+			}
+			synthesizedFinishResponse.SetSystemFingerprint(systemFingerprint)
+			for _, choiceIndex := range missingChoiceIndexes {
+				finishReason := types.FinishReasonStop
+				if streamChoiceStates[choiceIndex].toolCalls {
+					finishReason = types.FinishReasonToolCalls
+				}
+				synthesizedFinishResponse.Choices = append(synthesizedFinishResponse.Choices,
+					dto.ChatCompletionsStreamResponseChoice{
+						FinishReason: common.GetPointer(finishReason),
+						Index:        choiceIndex,
+					})
+			}
+		}
+	}
+
 	// 处理最后的响应
 	shouldSendLastResp := true
-	if err := handleLastResponse(lastStreamData, &responseId, &createAt, &systemFingerprint, &model, &usage,
-		&containStreamUsage, info, &shouldSendLastResp); err != nil {
-		logger.LogError(c, fmt.Sprintf("error handling last response: %s, lastStreamData: [%s]", err.Error(), lastStreamData))
+	if streamCompleted {
+		if err := handleLastResponse(lastStreamData, &responseId, &createAt, &systemFingerprint, &model, &usage,
+			&containStreamUsage, info, &shouldSendLastResp); err != nil {
+			logger.LogError(c, fmt.Sprintf("error handling last response: %s, lastStreamData: [%s]", err.Error(), lastStreamData))
+		}
 	}
 
-	if info.RelayFormat == types.RelayFormatOpenAI {
+	if info.RelayFormat == types.RelayFormatOpenAI && streamCompleted {
+		// [DONE] is an explicit upstream completion signal. Some compatible
+		// providers omit the terminal JSON chunk, so emit one before a trailing
+		// usage-only chunk for clients that require finish_reason.
+		lastStreamHasChoices := true
+		var lastStreamResponse dto.ChatCompletionsStreamResponse
+		if err := common.UnmarshalJsonStr(lastStreamData, &lastStreamResponse); err == nil {
+			lastStreamHasChoices = len(lastStreamResponse.Choices) > 0
+		}
+		if synthesizedFinishResponse != nil && !lastStreamHasChoices {
+			_ = helper.ObjectData(c, synthesizedFinishResponse)
+		}
 		if shouldSendLastResp {
 			_ = sendStreamData(c, info, lastStreamData, info.ChannelSetting.ForceFormat, info.ChannelSetting.ThinkingToContent)
 		}
+		if synthesizedFinishResponse != nil && lastStreamHasChoices {
+			_ = helper.ObjectData(c, synthesizedFinishResponse)
+		}
 	}
 
 	if !containStreamUsage {
@@ -189,11 +256,46 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
 		info.CountBillableToolCall(dto.BuildInCallFunctionCall, name)
 	}
 
+	if !streamCompleted {
+		if info.StreamStatus != nil {
+			info.StreamStatus.RecordError("openai stream ended without finish_reason or [DONE]")
+			logger.LogError(c, fmt.Sprintf("openai stream ended before terminal chunk: %s", info.StreamStatus.Summary()))
+		}
+		return usage, nil
+	}
+
 	HandleFinalResponse(c, info, lastStreamData, responseId, createAt, model, systemFingerprint, usage, containStreamUsage)
 
 	return usage, nil
 }
 
+type openAIStreamChoiceState struct {
+	finished  bool
+	toolCalls bool
+}
+
+func openAIStreamCompleted(info *relaycommon.RelayInfo, choiceStates map[int]openAIStreamChoiceState) bool {
+	if info == nil || info.StreamStatus == nil {
+		return true
+	}
+	switch info.StreamStatus.EndReason {
+	case relaycommon.StreamEndReasonDone:
+		return true
+	case relaycommon.StreamEndReasonEOF:
+		if len(choiceStates) == 0 {
+			return false
+		}
+		for _, state := range choiceStates {
+			if !state.finished {
+				return false
+			}
+		}
+		return true
+	default:
+		return false
+	}
+}
+
 func collectStreamFunctionCallNames(data string, seen map[string]struct{}, names *[]string) {
 	var streamResponse dto.ChatCompletionsStreamResponse
 	if err := common.UnmarshalJsonStr(data, &streamResponse); err != nil {