fix(openai): stream compact responses to Codex clients

Author: specialpointcentral

backend/internal/handler/openai_gateway_compact_body_signal_test.go

@@ -60,6 +60,9 @@ func TestNormalizeOpenAIResponsesCompactRequest_BodySignalPromoted(t *testing.T)
 	seed, exists := c.Get(service.OpenAICompactSessionSeedKeyForTest())
 	require.True(t, exists)
 	require.Equal(t, "pck-signal-1", seed)
+	clientStream, exists := c.Get(service.OpenAICompactClientStreamKeyForTest())
+	require.True(t, exists)
+	require.True(t, clientStream.(bool))
 }
 
 func TestNormalizeOpenAIResponsesCompactRequest_BodySignalTrailingSlash(t *testing.T) {

backend/internal/handler/openai_gateway_handler.go

@@ -590,6 +590,9 @@ func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Con
 	if !isCompactRequest {
 		return body, true
 	}
+	if streamResult := gjson.GetBytes(body, "stream"); streamResult.Type == gjson.True {
+		service.SetOpenAICompactClientStreamRequested(c, true)
+	}
 	if compactSeed := strings.TrimSpace(gjson.GetBytes(body, "prompt_cache_key").String()); compactSeed != "" {
 		c.Set(service.OpenAICompactSessionSeedKeyForTest(), compactSeed)
 	}

backend/internal/service/openai_compact_model_mapping_test.go

@@ -133,3 +133,64 @@ func TestOpenAIGatewayService_OAuthPassthrough_CompactOnlyModelMappingOverridesU
 	require.Equal(t, "gpt-5.4-openai-compact", gjson.GetBytes(upstream.lastBody, "model").String())
 	require.Equal(t, "gpt-5.4", gjson.GetBytes(rec.Body.Bytes(), "model").String())
 }
+
+func TestOpenAIGatewayService_OAuthPassthrough_CompactClientStreamWrapsJSONAsSSE(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", bytes.NewReader(nil))
+	c.Request.Header.Set("User-Agent", "codex_cli_rs/0.1.0")
+	c.Request.Header.Set("Content-Type", "application/json")
+	SetOpenAICompactClientStreamRequested(c, true)
+
+	originalBody := []byte(`{"model":"gpt-5.4","stream":true,"store":true,"instructions":"compact-pass","input":[{"type":"text","text":"compact me"}]}`)
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/plain"}, "x-request-id": []string{"rid-compact-pass-stream"}},
+		Body: io.NopCloser(strings.NewReader(`{
+  "id": "cmp_125",
+  "model": "gpt-5.4",
+  "status": "completed",
+  "output": [
+    {"type": "compaction_summary", "content": "ok"}
+  ],
+  "usage": {"input_tokens": 2, "output_tokens": 3}
+}`)),
+	}}
+
+	svc := &OpenAIGatewayService{httpUpstream: upstream}
+	account := &Account{
+		ID:          4,
+		Name:        "openai-oauth-pass",
+		Platform:    PlatformOpenAI,
+		Type:        AccountTypeOAuth,
+		Concurrency: 1,
+		Credentials: map[string]any{
+			"access_token":       "oauth-token",
+			"chatgpt_account_id": "chatgpt-acc",
+		},
+		Extra:       map[string]any{"openai_passthrough": true},
+		Status:      StatusActive,
+		Schedulable: true,
+	}
+
+	result, err := svc.Forward(context.Background(), c, account, originalBody)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, "application/json", upstream.lastReq.Header.Get("Accept"))
+	require.False(t, gjson.GetBytes(upstream.lastBody, "stream").Exists())
+	require.Contains(t, rec.Header().Get("Content-Type"), "text/event-stream")
+	require.Empty(t, rec.Header().Get("Connection"))
+	for _, line := range strings.Split(rec.Body.String(), "\n") {
+		if line == "" {
+			continue
+		}
+		require.Truef(t, strings.HasPrefix(line, "event:") || strings.HasPrefix(line, "data:"), "malformed SSE line: %q", line)
+	}
+	require.Contains(t, rec.Body.String(), "event: response.completed")
+	require.Contains(t, rec.Body.String(), `"type":"response.completed"`)
+	require.Contains(t, rec.Body.String(), `"sequence_number":1`)
+	require.Contains(t, rec.Body.String(), `"id":"cmp_125"`)
+	require.NotContains(t, rec.Body.String(), "data: [DONE]")
+}

backend/internal/service/openai_gateway_passthrough.go

@@ -1088,6 +1088,16 @@ func (s *OpenAIGatewayService) handleNonStreamingResponsePassthrough(
 	if originalModel != "" && mappedModel != "" && originalModel != mappedModel {
 		body = s.replaceModelInResponseBody(body, mappedModel, originalModel)
 	}
+	if shouldWriteOpenAICompactClientStream(c, body) {
+		writeOpenAICompactCompletedSSE(c, resp.StatusCode, body)
+		return &openaiNonStreamingResultPassthrough{
+			OpenAIUsage:      usage,
+			usage:            usage,
+			responseID:       extractOpenAIResponseIDFromJSONBytes(body),
+			imageCount:       countOpenAIResponseImageOutputsFromJSONBytes(body),
+			imageOutputSizes: collectOpenAIResponseImageOutputSizesFromJSONBytes(body),
+		}, nil
+	}
 	c.Data(resp.StatusCode, contentType, body)
 	return &openaiNonStreamingResultPassthrough{
 		OpenAIUsage:      usage,

backend/internal/service/openai_gateway_request_body.go

@@ -155,6 +155,29 @@ func OpenAICompactSessionSeedKeyForTest() string {
 	return openAICompactSessionSeedKey
 }
 
+func OpenAICompactClientStreamKeyForTest() string {
+	return openAICompactClientStreamKey
+}
+
+func SetOpenAICompactClientStreamRequested(c *gin.Context, requested bool) {
+	if c == nil {
+		return
+	}
+	c.Set(openAICompactClientStreamKey, requested)
+}
+
+func getOpenAICompactClientStreamRequested(c *gin.Context) bool {
+	if c == nil {
+		return false
+	}
+	value, ok := c.Get(openAICompactClientStreamKey)
+	if !ok {
+		return false
+	}
+	requested, _ := value.(bool)
+	return requested
+}
+
 func NormalizeOpenAICompactRequestBodyForTest(body []byte) ([]byte, bool, error) {
 	return normalizeOpenAICompactRequestBody(body)
 }

backend/internal/service/openai_gateway_response_handling.go

@@ -817,6 +817,16 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r
 	}
 
 	responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
+	if shouldWriteOpenAICompactClientStream(c, body) {
+		writeOpenAICompactCompletedSSE(c, resp.StatusCode, body)
+		return &openaiNonStreamingResult{
+			OpenAIUsage:      usage,
+			usage:            usage,
+			responseID:       extractOpenAIResponseIDFromJSONBytes(body),
+			imageCount:       countOpenAIResponseImageOutputsFromJSONBytes(body),
+			imageOutputSizes: collectOpenAIResponseImageOutputSizesFromJSONBytes(body),
+		}, nil
+	}
 
 	contentType := "application/json"
 	if s.cfg != nil && !s.cfg.Security.ResponseHeaders.Enabled {
@@ -836,6 +846,40 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r
 	}, nil
 }
 
+func shouldWriteOpenAICompactClientStream(c *gin.Context, body []byte) bool {
+	return isOpenAIResponsesCompactPath(c) && getOpenAICompactClientStreamRequested(c) && gjson.ValidBytes(body)
+}
+
+func writeOpenAICompactCompletedSSE(c *gin.Context, statusCode int, responseBody []byte) {
+	if statusCode == 0 {
+		statusCode = http.StatusOK
+	}
+	responseBody = compactOpenAIJSONForSSE(responseBody)
+	c.Writer.Header().Set("Content-Type", "text/event-stream")
+	c.Writer.Header().Set("Cache-Control", "no-cache")
+	c.Writer.Header().Set("X-Accel-Buffering", "no")
+	c.Status(statusCode)
+	_, _ = c.Writer.Write([]byte("event: response.completed\n"))
+	_, _ = c.Writer.Write([]byte(`data: {"type":"response.completed","sequence_number":1,"response":`))
+	_, _ = c.Writer.Write(responseBody)
+	_, _ = c.Writer.Write([]byte("}\n\n"))
+	if flusher, ok := c.Writer.(http.Flusher); ok {
+		flusher.Flush()
+	}
+}
+
+func compactOpenAIJSONForSSE(body []byte) []byte {
+	body = bytes.TrimSpace(body)
+	if len(body) == 0 {
+		return body
+	}
+	var compacted bytes.Buffer
+	if err := json.Compact(&compacted, body); err != nil {
+		return body
+	}
+	return compacted.Bytes()
+}
+
 func isEventStreamResponse(header http.Header) bool {
 	contentType := strings.ToLower(header.Get("Content-Type"))
 	return strings.Contains(contentType, "text/event-stream")

backend/internal/service/openai_gateway_service.go

@@ -49,6 +49,7 @@ const (
 	openAIWSRetryBackoffMaxDefault     = 2 * time.Second
 	openAIWSRetryJitterRatioDefault    = 0.2
 	openAICompactSessionSeedKey        = "openai_compact_session_seed"
+	openAICompactClientStreamKey       = "openai_compact_client_stream_requested"
 	codexCLIVersion                    = "0.125.0"
 	// Codex 限额快照仅用于后台展示/诊断，不需要每个成功请求都立即落库。
 	openAICodexSnapshotPersistMinInterval = 30 * time.Second