Skip to content

fix: aggregate SSE upstream into chat.completion JSON for non-stream clients #4557

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lingozhi wants to merge 1 commit into
base: main
Choose a base branch
from
+249 −2
Open

fix: aggregate SSE upstream into chat.completion JSON for non-stream clients #4557

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 231 additions & 0 deletions relay/channel/openai/chat_via_responses.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -548,3 +548,234 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
}
return usage, nil
}

// OaiResponsesSSEToChatJSON handles the case where the client requested a
// non-streaming chat completion but the upstream /v1/responses endpoint
// returned an SSE stream (common for reasoning models). It parses the SSE
// events, accumulates output text / tool calls / usage, builds a single
// dto.OpenAITextResponse, and writes it to the client as one JSON body.
//
// This avoids the bug where, in the original code path, when upstream returns
// SSE for a non-stream client request, raw SSE chunks (with empty choices) get
// forwarded directly to the client.
func OaiResponsesSSEToChatJSON(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) {
if resp == nil || resp.Body == nil {
return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError)
}
defer service.CloseResponseBodyGracefully(resp)

responseId := helper.GetResponseID(c)
createAt := time.Now().Unix()
model := info.UpstreamModelName

var (
usage = &dto.Usage{}
outputText strings.Builder
usageText strings.Builder
streamErr *types.NewAPIError
)

toolCallIndexByID := make(map[string]int)
toolCallNameByID := make(map[string]string)
toolCallArgsByID := make(map[string]string)
toolCallOrder := make([]string, 0)
toolCallCanonicalIDByItemID := make(map[string]string)

registerToolCall := func(callID string) {
if _, ok := toolCallIndexByID[callID]; ok {
return
}
toolCallIndexByID[callID] = len(toolCallOrder)
toolCallOrder = append(toolCallOrder, callID)
}

helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
if streamErr != nil {
sr.Stop(streamErr)
return
}

var streamResp dto.ResponsesStreamResponse
if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
return
}
Comment on lines +598 to +602

@coderabbitai coderabbitai Bot Apr 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Stop aggregation on malformed SSE events.

Right now a bad event is only logged and skipped, so this path can still return 200 with a partial chat.completion assembled from incomplete upstream data. The stream handler above aborts on the same condition; this path should do the same.

Proposed fix 
 		var streamResp dto.ResponsesStreamResponse
 		if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
 			logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
-			return
+			streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError)
+			sr.Stop(streamErr)
+			return
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
var streamResp dto.ResponsesStreamResponse
if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
return
}
var streamResp dto.ResponsesStreamResponse
if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError)
sr.Stop(streamErr)
return
}
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@relay/channel/openai/chat_via_responses.go` around lines 598 - 602, The
malformed SSE event handling currently logs the error then continues; change it
so a failed unmarshal of data into streamResp (via common.UnmarshalJsonStr)
aborts the whole aggregation rather than skipping: after logger.LogError(c,
"..."+err.Error()) propagate/return an error to the caller or abort the request
flow exactly like the stream handler does (e.g., call the same abort/cleanup
routine or return a non-200 error), ensuring the incomplete chat completion is
not emitted.


switch streamResp.Type {
case "response.created":
if streamResp.Response != nil {
if streamResp.Response.Model != "" {
model = streamResp.Response.Model
}
if streamResp.Response.CreatedAt != 0 {
createAt = int64(streamResp.Response.CreatedAt)
}
}

case "response.output_text.delta":
if streamResp.Delta != "" {
outputText.WriteString(streamResp.Delta)
usageText.WriteString(streamResp.Delta)
}

case "response.output_item.added", "response.output_item.done":
if streamResp.Item == nil || streamResp.Item.Type != "function_call" {
break
}
itemID := strings.TrimSpace(streamResp.Item.ID)
callID := strings.TrimSpace(streamResp.Item.CallId)
if callID == "" {
callID = itemID
}
if itemID != "" && callID != "" {
toolCallCanonicalIDByItemID[itemID] = callID
}
if callID == "" {
break
}
registerToolCall(callID)
if name := strings.TrimSpace(streamResp.Item.Name); name != "" {
toolCallNameByID[callID] = name
usageText.WriteString(name)
}
if args := streamResp.Item.ArgumentsString(); args != "" {
toolCallArgsByID[callID] = args
}
Comment on lines +621 to +643

@coderabbitai coderabbitai Bot Apr 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Fallback usage accounting for tool calls is inconsistent here.

response.output_item.added and response.output_item.done can both carry the same function name, so usageText may count it twice. At the same time, arguments that only appear on the item payload are never appended at all. When upstream omits usage, the estimated Usage can end up inflated or undercounted.

Proposed fix 
 	toolCallIndexByID := make(map[string]int)
 	toolCallNameByID := make(map[string]string)
 	toolCallArgsByID := make(map[string]string)
+	toolCallNameCounted := make(map[string]bool)
 	toolCallOrder := make([]string, 0)
 	toolCallCanonicalIDByItemID := make(map[string]string)
@@
 			registerToolCall(callID)
 			if name := strings.TrimSpace(streamResp.Item.Name); name != "" {
 				toolCallNameByID[callID] = name
-				usageText.WriteString(name)
+				if !toolCallNameCounted[callID] {
+					usageText.WriteString(name)
+					toolCallNameCounted[callID] = true
+				}
 			}
 			if args := streamResp.Item.ArgumentsString(); args != "" {
+				usageText.WriteString(stringDeltaFromPrefix(toolCallArgsByID[callID], args))
 				toolCallArgsByID[callID] = args
 			}
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@relay/channel/openai/chat_via_responses.go` around lines 621 - 643, When
handling "response.output_item.added"/"response.output_item.done" in
chat_via_responses.go, avoid double-counting the function name and ensure args
from the item payload are used as a fallback when upstream omits usage: after
deriving itemID and callID and calling registerToolCall(callID), only set
toolCallNameByID[callID] and append to usageText if the name isn't already
recorded for that callID (prevent duplicate append on both "added" and "done"),
and if toolCallArgsByID[callID] is empty but streamResp.Item.ArgumentsString()
returns non-empty, populate toolCallArgsByID[callID] and append the args to
usageText as the fallback; use the same callID lookup (and
toolCallCanonicalIDByItemID mapping) to ensure consistent canonicalization.


case "response.function_call_arguments.delta":
itemID := strings.TrimSpace(streamResp.ItemID)
callID := toolCallCanonicalIDByItemID[itemID]
if callID == "" {
callID = itemID
}
if callID == "" {
break
}
registerToolCall(callID)
toolCallArgsByID[callID] += streamResp.Delta
usageText.WriteString(streamResp.Delta)

case "response.completed":
if streamResp.Response != nil {
if streamResp.Response.Model != "" {
model = streamResp.Response.Model
}
if streamResp.Response.CreatedAt != 0 {
createAt = int64(streamResp.Response.CreatedAt)
}
if streamResp.Response.Usage != nil {
if streamResp.Response.Usage.InputTokens != 0 {
usage.PromptTokens = streamResp.Response.Usage.InputTokens
usage.InputTokens = streamResp.Response.Usage.InputTokens
}
if streamResp.Response.Usage.OutputTokens != 0 {
usage.CompletionTokens = streamResp.Response.Usage.OutputTokens
usage.OutputTokens = streamResp.Response.Usage.OutputTokens
}
if streamResp.Response.Usage.TotalTokens != 0 {
usage.TotalTokens = streamResp.Response.Usage.TotalTokens
} else {
usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens
}
if streamResp.Response.Usage.InputTokensDetails != nil {
usage.PromptTokensDetails.CachedTokens = streamResp.Response.Usage.InputTokensDetails.CachedTokens
usage.PromptTokensDetails.ImageTokens = streamResp.Response.Usage.InputTokensDetails.ImageTokens
usage.PromptTokensDetails.AudioTokens = streamResp.Response.Usage.InputTokensDetails.AudioTokens
}
if streamResp.Response.Usage.CompletionTokenDetails.ReasoningTokens != 0 {
usage.CompletionTokenDetails.ReasoningTokens = streamResp.Response.Usage.CompletionTokenDetails.ReasoningTokens
}
}
}

case "response.error", "response.failed":
if streamResp.Response != nil {
if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" {
streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError)
sr.Stop(streamErr)
return
}
}
streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError)
sr.Stop(streamErr)
return

default:
}
})

if streamErr != nil {
return nil, streamErr
}

if usage.TotalTokens == 0 {
usage = service.ResponseText2Usage(c, usageText.String(), info.UpstreamModelName, info.GetEstimatePromptTokens())
}

sawToolCall := len(toolCallOrder) > 0
finishReason := "stop"
if sawToolCall && outputText.Len() == 0 {
finishReason = "tool_calls"
}

msg := dto.Message{
Role: "assistant",
Content: outputText.String(),
}

if sawToolCall {
toolCalls := make([]dto.ToolCallResponse, 0, len(toolCallOrder))
for _, callID := range toolCallOrder {
tc := dto.ToolCallResponse{
ID: callID,
Type: "function",
Function: dto.FunctionResponse{
Name: toolCallNameByID[callID],
Arguments: toolCallArgsByID[callID],
},
}
toolCalls = append(toolCalls, tc)
}
toolCallsBytes, err := common.Marshal(toolCalls)
if err != nil {
return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError)
}
msg.ToolCalls = toolCallsBytes
}

chatResp := &dto.OpenAITextResponse{
Id: responseId,
Model: model,
Object: "chat.completion",
Created: createAt,
Choices: []dto.OpenAITextResponseChoice{
{
Index: 0,
Message: msg,
FinishReason: finishReason,
},
},
Usage: *usage,
}

var (
responseBody []byte
err error
)
switch info.RelayFormat {
case types.RelayFormatClaude:
claudeResp := service.ResponseOpenAI2Claude(chatResp, info)
responseBody, err = common.Marshal(claudeResp)
case types.RelayFormatGemini:
geminiResp := service.ResponseOpenAI2Gemini(chatResp, info)
responseBody, err = common.Marshal(geminiResp)
default:
responseBody, err = common.Marshal(chatResp)
}
if err != nil {
return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError)
}

service.IOCopyBytesGracefully(c, resp, responseBody)
return usage, nil
}
20 changes: 18 additions & 2 deletions relay/chat_completions_via_responses.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,14 +139,17 @@ func chatCompletionsViaResponses(c *gin.Context, info *relaycommon.RelayInfo, ad
statusCodeMappingStr := c.GetString("status_code_mapping")

httpResp = resp.(*http.Response)
info.IsStream = info.IsStream || strings.HasPrefix(httpResp.Header.Get("Content-Type"), "text/event-stream")
clientWantsStream := info.IsStream
upstreamIsSSE := strings.HasPrefix(httpResp.Header.Get("Content-Type"), "text/event-stream")
info.IsStream = clientWantsStream || upstreamIsSSE
if httpResp.StatusCode != http.StatusOK {
newApiErr := service.RelayErrorHandler(c.Request.Context(), httpResp, false)
service.ResetStatusCode(newApiErr, statusCodeMappingStr)
return nil, newApiErr
}

if info.IsStream {
if clientWantsStream {
// Client requested stream — forward as SSE regardless of upstream format.
usage, newApiErr := openaichannel.OaiResponsesToChatStreamHandler(c, info, httpResp)
if newApiErr != nil {
service.ResetStatusCode(newApiErr, statusCodeMappingStr)
Expand All @@ -155,6 +158,19 @@ func chatCompletionsViaResponses(c *gin.Context, info *relaycommon.RelayInfo, ad
return usage, nil
}

if upstreamIsSSE {
// Client wants non-stream JSON, but upstream returned SSE (common for
// reasoning models). Buffer and aggregate the SSE stream into a full
// chat.completion JSON before writing to the client.
info.IsStream = false
usage, newApiErr := openaichannel.OaiResponsesSSEToChatJSON(c, info, httpResp)
if newApiErr != nil {
service.ResetStatusCode(newApiErr, statusCodeMappingStr)
return nil, newApiErr
}
return usage, nil
}

usage, newApiErr := openaichannel.OaiResponsesToChatHandler(c, info, httpResp)
if newApiErr != nil {
service.ResetStatusCode(newApiErr, statusCodeMappingStr)
Expand Down
Loading