Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,11 @@ 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)
clientStreamBool, ok := clientStream.(bool)
require.True(t, ok)
require.True(t, clientStreamBool)
}

func TestNormalizeOpenAIResponsesCompactRequest_BodySignalTrailingSlash(t *testing.T) {
Expand Down
3 changes: 3 additions & 0 deletions backend/internal/handler/openai_gateway_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
52 changes: 52 additions & 0 deletions backend/internal/service/openai_compact_model_mapping_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -133,3 +133,55 @@ 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_CompactClientStreamDoesNotWrapJSONAsSSE(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.NotContains(t, rec.Header().Get("Content-Type"), "text/event-stream")
require.NotContains(t, rec.Body.String(), "event: response.completed")
require.Equal(t, "cmp_125", gjson.Get(rec.Body.String(), "id").String())
require.NotContains(t, rec.Body.String(), "data: [DONE]")
}
23 changes: 23 additions & 0 deletions backend/internal/service/openai_gateway_request_body.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
54 changes: 54 additions & 0 deletions backend/internal/service/openai_gateway_response_handling.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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")
Expand Down Expand Up @@ -899,6 +943,16 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
}

responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
if ok && shouldWriteOpenAICompactClientStream(c, body) {
writeOpenAICompactCompletedSSE(c, resp.StatusCode, body)
return &openaiNonStreamingResult{
OpenAIUsage: usage,
usage: usage,
responseID: extractOpenAIResponseIDFromJSONBytes(body),
imageCount: countOpenAIImageOutputsFromSSEBody(bodyText),
imageOutputSizes: collectOpenAIImageOutputSizesFromSSEBody(bodyText),
}, nil
}

contentType := "application/json; charset=utf-8"
if !ok {
Expand Down
1 change: 1 addition & 0 deletions backend/internal/service/openai_gateway_service.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
30 changes: 30 additions & 0 deletions backend/internal/service/openai_gateway_service_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2856,6 +2856,36 @@ func TestHandleSSEToJSON_CompletedEventReturnsJSON(t *testing.T) {
require.NotContains(t, rec.Body.String(), "data:")
}

func TestHandleSSEToJSON_CompactClientStreamCompletedEventReturnsSSE(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", nil)
SetOpenAICompactClientStreamRequested(c, true)

svc := &OpenAIGatewayService{cfg: &config.Config{}}
resp := &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
}
body := []byte(strings.Join([]string{
`data: {"type":"response.in_progress","response":{"id":"resp_compact_sse"}}`,
`data: {"type":"response.completed","response":{"id":"resp_compact_sse","model":"gpt-5.4","usage":{"input_tokens":7,"output_tokens":9,"input_tokens_details":{"cached_tokens":1}}}}`,
`data: [DONE]`,
}, "\n"))

usage, err := svc.handleSSEToJSON(resp, c, body, "gpt-5.4", "gpt-5.4")
require.NoError(t, err)
require.NotNil(t, usage)
require.Equal(t, 7, usage.InputTokens)
require.Equal(t, 9, usage.OutputTokens)
require.Contains(t, rec.Header().Get("Content-Type"), "text/event-stream")
require.Contains(t, rec.Body.String(), "event: response.completed")
require.Contains(t, rec.Body.String(), `"type":"response.completed"`)
require.Contains(t, rec.Body.String(), `"id":"resp_compact_sse"`)
require.NotContains(t, rec.Body.String(), "data: [DONE]")
}

func TestHandleNonStreamingResponse_APIKeyFallsBackToSSEBodyWhenContentTypeIsWrong(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
Expand Down
Loading