diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 238d359b9e..8283b54a19 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -914,6 +914,11 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( trimmedData = strings.TrimSpace(string(normalizedData)) line = "data: " + string(normalizedData) } + if normalizedData, normalized := normalizeCompletedImageGenerationStatus(dataBytes); normalized { + dataBytes = normalizedData + trimmedData = strings.TrimSpace(string(normalizedData)) + line = "data: " + string(normalizedData) + } eventType := strings.TrimSpace(gjson.Get(trimmedData, "type").String()) if eventType == "response.failed" { failedMessage = extractOpenAISSEErrorMessage(dataBytes) diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 5841ce6556..845428b641 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -281,6 +281,11 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp forceFlushFailedEvent = true sawFailedEvent = true } + if normalizedData, normalized := normalizeCompletedImageGenerationStatus(dataBytes); normalized { + dataBytes = normalizedData + data = string(normalizedData) + line = "data: " + data + } imageCounter.AddSSEData(dataBytes) // Correct Codex tool calls if needed (apply_patch -> edit, etc.) @@ -1071,6 +1076,9 @@ func extractCodexFinalResponse(body string) ([]byte, bool) { if finalResponse != nil { return } + if normalized, changed := normalizeCompletedImageGenerationStatus(data); changed { + data = normalized + } eventType := gjson.GetBytes(data, "type").String() if eventType == "response.done" || eventType == "response.completed" { if response := gjson.GetBytes(data, "response"); response.Exists() && response.Type == gjson.JSON && response.Raw != "" { @@ -1084,6 +1092,59 @@ func extractCodexFinalResponse(body string) ([]byte, bool) { return nil, false } +func normalizeCompletedImageGenerationStatus(data []byte) ([]byte, bool) { + if len(data) == 0 || !gjson.ValidBytes(data) { + return data, false + } + + shouldNormalize := func(item gjson.Result) bool { + if !item.Exists() || !item.IsObject() || + strings.TrimSpace(item.Get("type").String()) != "image_generation_call" { + return false + } + switch strings.TrimSpace(item.Get("status").String()) { + case "generating", "in_progress": + return strings.TrimSpace(item.Get("result").String()) != "" + default: + return false + } + } + + eventType := strings.TrimSpace(gjson.GetBytes(data, "type").String()) + switch eventType { + case "response.output_item.done": + if !shouldNormalize(gjson.GetBytes(data, "item")) { + return data, false + } + updated, err := sjson.SetBytes(data, "item.status", "completed") + if err != nil { + return data, false + } + return updated, true + case "response.completed", "response.done": + output := gjson.GetBytes(data, "response.output") + if !output.Exists() || !output.IsArray() { + return data, false + } + updated := data + changed := false + for i, item := range output.Array() { + if !shouldNormalize(item) { + continue + } + next, err := sjson.SetBytes(updated, "response.output."+strconv.Itoa(i)+".status", "completed") + if err != nil { + return data, false + } + updated = next + changed = true + } + return updated, changed + default: + return data, false + } +} + func normalizeResponsesStreamingTerminalOutput(data []byte, acc *apicompat.BufferedResponseAccumulator, imageOutputs []json.RawMessage) ([]byte, bool) { eventType := strings.TrimSpace(gjson.GetBytes(data, "type").String()) switch eventType { @@ -1124,7 +1185,8 @@ func responsesStreamEventMayContributeToOutput(eventType string) bool { } // collectRawResponsesOutputItemsFromSSE 按到达顺序收集 SSE 流中 -// response.output_item.done 携带的原始 item。item 以 raw JSON 逐字节保留， +// response.output_item.done 携带的原始 item。除已产生结果但仍停留在进行中 +// 的图片状态外，item 以 raw JSON 逐字节保留， // 避免经窄结构体重建时丢弃 encrypted_content/summary/opaque 等 compact // 专属或未来新增字段（#3777 问题 2）。若整条流没有任何 done 事件，退回 // 收集 output_item.added 中的 compaction 类 item——compaction 结果没有 @@ -1151,6 +1213,9 @@ func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) { items = append(items, json.RawMessage(item.Raw)) } forEachOpenAISSEDataPayload(bodyText, func(data []byte) { + if normalized, changed := normalizeCompletedImageGenerationStatus(data); changed { + data = normalized + } if strings.TrimSpace(gjson.GetBytes(data, "type").String()) != "response.output_item.done" { return } diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index bc14350394..507fb9d5c3 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -2961,7 +2961,7 @@ func TestHandleSSEToJSON_ReconstructsImageGenerationOutputItemDone(t *testing.T) Header: http.Header{"Content-Type": []string{"text/event-stream"}}, } body := []byte(strings.Join([]string{ - `data: {"type":"response.output_item.done","item":{"id":"ig_123","type":"image_generation_call","result":"aGVsbG8=","revised_prompt":"draw a cat","output_format":"png"}}`, + `data: {"type":"response.output_item.done","item":{"id":"ig_123","type":"image_generation_call","status":"generating","result":"aGVsbG8=","revised_prompt":"draw a cat","output_format":"png"}}`, `data: {"type":"response.completed","response":{"id":"resp_img","model":"gpt-5.4","output":[],"usage":{"input_tokens":7,"output_tokens":9,"output_tokens_details":{"image_tokens":4}}}}`, `data: [DONE]`, }, "\n")) @@ -2972,6 +2972,7 @@ func TestHandleSSEToJSON_ReconstructsImageGenerationOutputItemDone(t *testing.T) require.Equal(t, 4, usage.ImageOutputTokens) require.NotContains(t, rec.Body.String(), "data:") require.Equal(t, "image_generation_call", gjson.Get(rec.Body.String(), "output.0.type").String()) + require.Equal(t, "completed", gjson.Get(rec.Body.String(), "output.0.status").String()) require.Equal(t, "aGVsbG8=", gjson.Get(rec.Body.String(), "output.0.result").String()) require.Equal(t, "draw a cat", gjson.Get(rec.Body.String(), "output.0.revised_prompt").String()) } diff --git a/backend/internal/service/openai_image_generation_controls_test.go b/backend/internal/service/openai_image_generation_controls_test.go index af0cdf669c..6c9180d3b9 100644 --- a/backend/internal/service/openai_image_generation_controls_test.go +++ b/backend/internal/service/openai_image_generation_controls_test.go @@ -455,13 +455,13 @@ func TestOpenAIGatewayServiceHandleResponsesImageOutputs_Streaming(t *testing.T) gin.SetMode(gin.TestMode) svc := newOpenAIImageGenerationControlTestService(&httpUpstreamRecorder{}) - c, _ := newOpenAIImageGenerationControlTestContext(true, "unit-test-agent/1.0") + c, recorder := newOpenAIImageGenerationControlTestContext(true, "unit-test-agent/1.0") resp := &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader( - "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"result\":\"final-image\"}}\n\n" + - "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_image_stream\",\"model\":\"gpt-5.5\",\"output\":[{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"result\":\"final-image\"}],\"usage\":{\"input_tokens\":11,\"output_tokens\":5,\"output_tokens_details\":{\"image_tokens\":4}}}}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"status\":\"generating\",\"result\":\"final-image\"}}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_image_stream\",\"model\":\"gpt-5.5\",\"output\":[{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"status\":\"generating\",\"result\":\"final-image\"}],\"usage\":{\"input_tokens\":11,\"output_tokens\":5,\"output_tokens_details\":{\"image_tokens\":4}}}}\n\n", )), } @@ -474,6 +474,73 @@ func TestOpenAIGatewayServiceHandleResponsesImageOutputs_Streaming(t *testing.T) require.Equal(t, 11, result.usage.InputTokens) require.Equal(t, 5, result.usage.OutputTokens) require.Equal(t, 4, result.usage.ImageOutputTokens) + require.NotContains(t, recorder.Body.String(), `"status":"generating"`) + require.Equal(t, 2, strings.Count(recorder.Body.String(), `"status":"completed"`)) +} + +func TestOpenAIGatewayServiceHandleResponsesImageOutputs_StreamingPassthrough(t *testing.T) { + gin.SetMode(gin.TestMode) + + svc := newOpenAIImageGenerationControlTestService(&httpUpstreamRecorder{}) + c, recorder := newOpenAIImageGenerationControlTestContext(true, "unit-test-agent/1.0") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader( + "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"status\":\"in_progress\",\"result\":\"final-image\"}}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_image_stream\",\"model\":\"gpt-5.5\",\"output\":[{\"id\":\"ig_stream_1\",\"type\":\"image_generation_call\",\"status\":\"in_progress\",\"result\":\"final-image\"}],\"usage\":{\"input_tokens\":11,\"output_tokens\":5,\"output_tokens_details\":{\"image_tokens\":4}}}}\n\n", + )), + } + + result, err := svc.handleStreamingResponsePassthrough(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "gpt-5.5", "gpt-5.5") + + require.NoError(t, err) + require.NotNil(t, result) + require.NotContains(t, recorder.Body.String(), `"status":"in_progress"`) + require.Equal(t, 2, strings.Count(recorder.Body.String(), `"status":"completed"`)) +} + +func TestNormalizeCompletedImageGenerationStatus(t *testing.T) { + tests := []struct { + name string + input string + want string + wantChanged bool + }{ + { + name: "output item done with result", + input: `{"type":"response.output_item.done","item":{"type":"image_generation_call","status":"generating","result":"image-data"}}`, + want: `{"type":"response.output_item.done","item":{"type":"image_generation_call","status":"completed","result":"image-data"}}`, + wantChanged: true, + }, + { + name: "terminal response only changes completed image result", + input: `{"type":"response.completed","response":{"output":[{"type":"image_generation_call","status":"in_progress","result":"image-data"},{"type":"image_generation_call","status":"failed","result":"partial-data"}]}}`, + want: `{"type":"response.completed","response":{"output":[{"type":"image_generation_call","status":"completed","result":"image-data"},{"type":"image_generation_call","status":"failed","result":"partial-data"}]}}`, + wantChanged: true, + }, + { + name: "done item without result", + input: `{"type":"response.output_item.done","item":{"type":"image_generation_call","status":"generating"}}`, + want: `{"type":"response.output_item.done","item":{"type":"image_generation_call","status":"generating"}}`, + wantChanged: false, + }, + { + name: "non-final image event", + input: `{"type":"response.output_item.added","item":{"type":"image_generation_call","status":"generating","result":"image-data"}}`, + want: `{"type":"response.output_item.added","item":{"type":"image_generation_call","status":"generating","result":"image-data"}}`, + wantChanged: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, changed := normalizeCompletedImageGenerationStatus([]byte(tt.input)) + + require.Equal(t, tt.wantChanged, changed) + require.JSONEq(t, tt.want, string(got)) + }) + } } // TestHandleStreamingResponse_CyberPolicyCapturesRealUpstreamTokens 锁定流式