diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index c6b790b3..c93ba7be 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -77,6 +77,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { reqLog = reqLog.With(zap.String("model", reqModel), zap.Bool("stream", reqStream)) setOpsRequestContext(c, reqModel, reqStream, body) + service.LogOpenAIGatewayRequestBody(reqLog, c.Request.URL.Path, body, reqStream) if h.errorPassthroughService != nil { service.BindErrorPassthroughService(c, h.errorPassthroughService) diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 3d8609fa..f7c0d1a9 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -174,6 +174,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { } setOpsRequestContext(c, reqModel, reqStream, body) + service.LogOpenAIGatewayRequestBody(reqLog, c.Request.URL.Path, body, reqStream) // 提前校验 function_call_output 是否具备可关联上下文，避免上游 400。 if !h.validateFunctionCallOutputRequest(c, body, reqLog) { @@ -556,6 +557,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { reqLog = reqLog.With(zap.String("model", reqModel), zap.Bool("stream", reqStream)) setOpsRequestContext(c, reqModel, reqStream, body) + service.LogOpenAIGatewayRequestBody(reqLog, c.Request.URL.Path, body, reqStream) // 绑定错误透传服务，允许 service 层在非 failover 错误场景复用规则。 if h.errorPassthroughService != nil { diff --git a/backend/internal/service/openai_codex_transform.go b/backend/internal/service/openai_codex_transform.go index d0534d8c..7dcd7f44 100644 --- a/backend/internal/service/openai_codex_transform.go +++ b/backend/internal/service/openai_codex_transform.go @@ -131,6 +131,12 @@ func applyCodexOAuthTransform(reqBody map[string]any, isCodexCLI bool, isCompact } } + // OpenAI OAuth（ChatGPT internal）当前不支持 flex。 + // 保留 priority/fast 归一化链路，单独移除 flex，避免把上游 400 包装成 502。 + if stripUnsupportedCodexOAuthServiceTier(reqBody) { + result.Modified = true + } + // 兼容遗留的 functions 和 function_call，转换为 tools 和 tool_choice if functionsRaw, ok := reqBody["functions"]; ok { if functions, k := functionsRaw.([]any); k { @@ -210,6 +216,18 @@ func applyCodexOAuthTransform(reqBody map[string]any, isCodexCLI bool, isCompact return result } +func stripUnsupportedCodexOAuthServiceTier(reqBody map[string]any) bool { + if reqBody == nil { + return false + } + serviceTier := extractOpenAIServiceTier(reqBody) + if serviceTier == nil || *serviceTier != "flex" { + return false + } + delete(reqBody, "service_tier") + return true +} + func normalizeCodexModel(model string) string { if model == "" { return "gpt-5.1" diff --git a/backend/internal/service/openai_codex_transform_test.go b/backend/internal/service/openai_codex_transform_test.go index b52f0566..576ca48e 100644 --- a/backend/internal/service/openai_codex_transform_test.go +++ b/backend/internal/service/openai_codex_transform_test.go @@ -344,6 +344,31 @@ func TestApplyCodexOAuthTransform_StringInputWithToolsField(t *testing.T) { require.Len(t, input, 1) } +func TestApplyCodexOAuthTransform_StripsUnsupportedFlexServiceTier(t *testing.T) { + reqBody := map[string]any{ + "model": "gpt-5.4", + "service_tier": "flex", + } + + result := applyCodexOAuthTransform(reqBody, false, false) + + require.True(t, result.Modified) + _, exists := reqBody["service_tier"] + require.False(t, exists) +} + +func TestApplyCodexOAuthTransform_PreservesFastServiceTierForPriorityNormalization(t *testing.T) { + reqBody := map[string]any{ + "model": "gpt-5.4", + "service_tier": "fast", + } + + result := applyCodexOAuthTransform(reqBody, false, false) + + require.True(t, result.Modified) + require.Equal(t, "fast", reqBody["service_tier"]) +} + func TestExtractSystemMessagesFromInput(t *testing.T) { t.Run("no system messages", func(t *testing.T) { reqBody := map[string]any{ diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index c90f5afa..a8669c0a 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -292,11 +292,24 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( } chatResp := apicompat.ResponsesToChatCompletions(finalResponse, originalModel) + encodedBody, err := json.Marshal(chatResp) + if err != nil { + return nil, fmt.Errorf("marshal chat completions response: %w", err) + } if s.responseHeaderFilter != nil { responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) } - c.JSON(http.StatusOK, chatResp) + LogOpenAIGatewayResponseBody( + openAIChatCompletionsLog(ctx), + openAIGatewayLogPath(c), + http.StatusOK, + "application/json; charset=utf-8", + encodedBody, + false, + zap.String("upstream_request_id", strings.TrimSpace(requestID)), + ) + c.Data(http.StatusOK, "application/json; charset=utf-8", encodedBody) return &OpenAIForwardResult{ RequestID: requestID, @@ -330,6 +343,15 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( c.Writer.Header().Set("Connection", "keep-alive") c.Writer.Header().Set("X-Accel-Buffering", "no") c.Writer.WriteHeader(http.StatusOK) + LogOpenAIGatewayResponseBody( + openAIChatCompletionsLog(ctx), + openAIGatewayLogPath(c), + http.StatusOK, + "text/event-stream", + nil, + true, + zap.String("upstream_request_id", strings.TrimSpace(requestID)), + ) state := apicompat.NewResponsesEventToChatState() state.Model = originalModel diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index e7499447..8f9430f8 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -299,11 +299,24 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( } anthropicResp := apicompat.ResponsesToAnthropic(finalResponse, originalModel) + encodedBody, err := json.Marshal(anthropicResp) + if err != nil { + return nil, fmt.Errorf("marshal anthropic response: %w", err) + } if s.responseHeaderFilter != nil { responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) } - c.JSON(http.StatusOK, anthropicResp) + LogOpenAIGatewayResponseBody( + openAIMessagesCompatLog(ctx), + openAIGatewayLogPath(c), + http.StatusOK, + "application/json; charset=utf-8", + encodedBody, + false, + zap.String("upstream_request_id", strings.TrimSpace(requestID)), + ) + c.Data(http.StatusOK, "application/json; charset=utf-8", encodedBody) return &OpenAIForwardResult{ RequestID: requestID, @@ -339,6 +352,15 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( c.Writer.Header().Set("Connection", "keep-alive") c.Writer.Header().Set("X-Accel-Buffering", "no") c.Writer.WriteHeader(http.StatusOK) + LogOpenAIGatewayResponseBody( + openAIMessagesCompatLog(ctx), + openAIGatewayLogPath(c), + http.StatusOK, + "text/event-stream", + nil, + true, + zap.String("upstream_request_id", strings.TrimSpace(requestID)), + ) state := apicompat.NewResponsesEventToAnthropicState() state.Model = originalModel diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index b975d62d..7cb0f73a 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -2870,6 +2870,15 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( if v := resp.Header.Get("x-request-id"); v != "" { c.Header("x-request-id", v) } + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(ctx), "service.openai_gateway"), + openAIGatewayLogPath(c), + http.StatusOK, + "text/event-stream", + nil, + true, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) w := c.Writer flusher, ok := w.(http.Flusher) @@ -2993,6 +3002,15 @@ func (s *OpenAIGatewayService) handleNonStreamingResponsePassthrough( if contentType == "" { contentType = "application/json" } + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(ctx), "service.openai_gateway"), + openAIGatewayLogPath(c), + resp.StatusCode, + contentType, + body, + false, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) c.Data(resp.StatusCode, contentType, body) return usage, nil } @@ -3308,6 +3326,16 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( writeError compatErrorWriter, ) (*OpenAIForwardResult, error) { body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(c.Request.Context()), "service.openai_gateway"), + openAIGatewayLogPath(c), + resp.StatusCode, + resp.Header.Get("Content-Type"), + body, + false, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + zap.String("account_platform", strings.TrimSpace(account.Platform)), + ) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) if upstreamMsg == "" { @@ -3431,6 +3459,15 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp if v := resp.Header.Get("x-request-id"); v != "" { c.Header("x-request-id", v) } + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(ctx), "service.openai_gateway"), + openAIGatewayLogPath(c), + http.StatusOK, + "text/event-stream", + nil, + true, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) w := c.Writer flusher, ok := w.(http.Flusher) @@ -3875,6 +3912,15 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r } } + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(ctx), "service.openai_gateway"), + openAIGatewayLogPath(c), + resp.StatusCode, + contentType, + body, + false, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) c.Data(resp.StatusCode, contentType, body) return usage, nil @@ -3920,6 +3966,15 @@ func (s *OpenAIGatewayService) handleOAuthSSEToJSON(resp *http.Response, c *gin. contentType := "application/json; charset=utf-8" c.Header("Content-Type", contentType) + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(c.Request.Context()), "service.openai_gateway"), + openAIGatewayLogPath(c), + resp.StatusCode, + contentType, + body, + false, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) c.Data(resp.StatusCode, contentType, body) return usage, nil @@ -3961,15 +4016,33 @@ func (s *OpenAIGatewayService) writeOpenAINonStreamingProtocolError(resp *http.R setOpsUpstreamError(c, http.StatusBadGateway, message, "") responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8") - c.JSON(http.StatusBadGateway, gin.H{ + errorBody := gin.H{ "error": gin.H{ "type": "upstream_error", "message": message, }, - }) + } + encoded, _ := json.Marshal(errorBody) + LogOpenAIGatewayResponseBody( + logger.WithComponent(logger.FromContext(c.Request.Context()), "service.openai_gateway"), + openAIGatewayLogPath(c), + http.StatusBadGateway, + "application/json; charset=utf-8", + encoded, + false, + zap.String("upstream_request_id", strings.TrimSpace(resp.Header.Get("x-request-id"))), + ) + c.JSON(http.StatusBadGateway, errorBody) return fmt.Errorf("non-streaming openai protocol error: %s", message) } +func openAIGatewayLogPath(c *gin.Context) string { + if c == nil || c.Request == nil || c.Request.URL == nil { + return "" + } + return c.Request.URL.Path +} + func extractCodexFinalResponse(body string) ([]byte, bool) { lines := strings.Split(body, "\n") for _, line := range lines { diff --git a/backend/internal/service/openai_oauth_passthrough_test.go b/backend/internal/service/openai_oauth_passthrough_test.go index f51a7491..6d028deb 100644 --- a/backend/internal/service/openai_oauth_passthrough_test.go +++ b/backend/internal/service/openai_oauth_passthrough_test.go @@ -722,6 +722,50 @@ func TestOpenAIGatewayService_OAuthPassthrough_StreamingSetsFirstTokenMs(t *test require.Equal(t, "priority", *result.ServiceTier) } +func TestOpenAIGatewayService_OAuthPassthrough_StripsUnsupportedFlexServiceTier(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.1.0") + + originalBody := []byte(`{"model":"gpt-5.2","stream":false,"service_tier":"flex","input":[{"type":"text","text":"hi"}]}`) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid"}}, + Body: io.NopCloser(strings.NewReader(`{"output":[],"usage":{"input_tokens":1,"output_tokens":1,"input_tokens_details":{"cached_tokens":0}}}`)), + } + upstream := &httpUpstreamRecorder{resp: resp} + + svc := &OpenAIGatewayService{ + cfg: &config.Config{Gateway: config.GatewayConfig{ForceCodexCLI: false}}, + httpUpstream: upstream, + } + + account := &Account{ + ID: 123, + Name: "acc", + 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, + RateMultiplier: f64p(1), + } + + result, err := svc.Forward(context.Background(), c, account, originalBody) + require.NoError(t, err) + require.NotNil(t, result) + require.Nil(t, result.ServiceTier) + require.NotNil(t, upstream.lastReq) + require.False(t, gjson.GetBytes(upstream.lastBody, "service_tier").Exists()) + require.Equal(t, false, gjson.GetBytes(upstream.lastBody, "store").Bool()) + require.Equal(t, true, gjson.GetBytes(upstream.lastBody, "stream").Bool()) +} + func TestOpenAIGatewayService_OAuthPassthrough_StreamClientDisconnectStillCollectsUsage(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/ops_log_runtime.go b/backend/internal/service/ops_log_runtime.go index ed8aefa9..fac83bf7 100644 --- a/backend/internal/service/ops_log_runtime.go +++ b/backend/internal/service/ops_log_runtime.go @@ -15,13 +15,15 @@ import ( func defaultOpsRuntimeLogConfig(cfg *config.Config) *OpsRuntimeLogConfig { out := &OpsRuntimeLogConfig{ - Level: "info", - EnableSampling: false, - SamplingInitial: 100, - SamplingNext: 100, - Caller: true, - StacktraceLevel: "error", - RetentionDays: 30, + Level: "info", + EnableSampling: false, + SamplingInitial: 100, + SamplingNext: 100, + Caller: true, + StacktraceLevel: "error", + RetentionDays: 30, + GatewayBodyLogEnabled: false, + GatewayBodyLogMaxBytes: defaultOpenAIGatewayBodyLogMaxBytes, } if cfg == nil { return out @@ -59,6 +61,9 @@ func normalizeOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig, defaults *OpsRuntime if cfg.RetentionDays <= 0 { cfg.RetentionDays = defaults.RetentionDays } + if cfg.GatewayBodyLogMaxBytes <= 0 { + cfg.GatewayBodyLogMaxBytes = defaults.GatewayBodyLogMaxBytes + } } func validateOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error { @@ -84,6 +89,9 @@ func validateOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error { if cfg.RetentionDays < 1 || cfg.RetentionDays > 3650 { return errors.New("retention_days must be between 1 and 3650") } + if cfg.GatewayBodyLogMaxBytes < 1 || cfg.GatewayBodyLogMaxBytes > 1<<20 { + return errors.New("gateway_body_log_max_bytes must be between 1 and 1048576") + } return nil } @@ -227,6 +235,7 @@ func applyOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error { }); err != nil { return err } + applyOpenAIGatewayBodyLogRuntimeConfig(cfg.GatewayBodyLogEnabled, cfg.GatewayBodyLogMaxBytes) return nil } diff --git a/backend/internal/service/ops_log_runtime_test.go b/backend/internal/service/ops_log_runtime_test.go index 658b4812..758b7c94 100644 --- a/backend/internal/service/ops_log_runtime_test.go +++ b/backend/internal/service/ops_log_runtime_test.go @@ -385,6 +385,12 @@ func TestDefaultNormalizeAndValidateRuntimeLogConfig(t *testing.T) { if defaults.Level != "debug" || defaults.StacktraceLevel != "fatal" || defaults.RetentionDays != 7 { t.Fatalf("unexpected defaults: %+v", defaults) } + if defaults.GatewayBodyLogEnabled { + t.Fatalf("gateway body log should be disabled by default") + } + if defaults.GatewayBodyLogMaxBytes != defaultOpenAIGatewayBodyLogMaxBytes { + t.Fatalf("unexpected default gateway body log max bytes: %d", defaults.GatewayBodyLogMaxBytes) + } cfg := &OpsRuntimeLogConfig{ Level: " ", @@ -402,6 +408,9 @@ func TestDefaultNormalizeAndValidateRuntimeLogConfig(t *testing.T) { if cfg.SamplingInitial != 50 || cfg.SamplingNext != 20 || cfg.RetentionDays != 7 { t.Fatalf("normalize numeric defaults failed: %+v", cfg) } + if cfg.GatewayBodyLogMaxBytes != defaultOpenAIGatewayBodyLogMaxBytes { + t.Fatalf("normalize gateway body log max bytes failed: %+v", cfg) + } if err := validateOpsRuntimeLogConfig(cfg); err != nil { t.Fatalf("validate normalized config should pass: %v", err) } @@ -418,6 +427,7 @@ func TestValidateRuntimeLogConfigErrors(t *testing.T) { {name: "bad initial", cfg: &OpsRuntimeLogConfig{Level: "info", StacktraceLevel: "error", SamplingInitial: 0, SamplingNext: 1, RetentionDays: 1}}, {name: "bad next", cfg: &OpsRuntimeLogConfig{Level: "info", StacktraceLevel: "error", SamplingInitial: 1, SamplingNext: 0, RetentionDays: 1}}, {name: "bad retention", cfg: &OpsRuntimeLogConfig{Level: "info", StacktraceLevel: "error", SamplingInitial: 1, SamplingNext: 1, RetentionDays: 0}}, + {name: "bad gateway body log max bytes", cfg: &OpsRuntimeLogConfig{Level: "info", StacktraceLevel: "error", SamplingInitial: 1, SamplingNext: 1, RetentionDays: 1, GatewayBodyLogMaxBytes: -1}}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -568,3 +578,40 @@ func TestApplyRuntimeLogConfigHelpers(t *testing.T) { var nilSvc *OpsService nilSvc.applyRuntimeLogConfigOnStartup(context.Background()) } + +func TestApplyRuntimeLogConfigHelpers_AppliesGatewayBodyLogState(t *testing.T) { + if err := logger.Init(logger.InitOptions{ + Level: "info", + Format: "json", + ServiceName: "sub2api", + Environment: "test", + Output: logger.OutputOptions{ + ToStdout: true, + ToFile: false, + }, + }); err != nil { + t.Fatalf("init logger: %v", err) + } + + err := applyOpsRuntimeLogConfig(&OpsRuntimeLogConfig{ + Level: "info", + EnableSampling: false, + SamplingInitial: 1, + SamplingNext: 1, + Caller: true, + StacktraceLevel: "error", + RetentionDays: 1, + GatewayBodyLogEnabled: true, + GatewayBodyLogMaxBytes: 321, + }) + if err != nil { + t.Fatalf("applyOpsRuntimeLogConfig() error: %v", err) + } + + if !OpenAIGatewayBodyLogEnabled() { + t.Fatalf("gateway body log should be enabled") + } + if OpenAIGatewayBodyLogMaxBytes() != 321 { + t.Fatalf("gateway body log max bytes = %d, want 321", OpenAIGatewayBodyLogMaxBytes()) + } +} diff --git a/backend/internal/service/ops_settings_models.go b/backend/internal/service/ops_settings_models.go index fa18b05f..12fbfa6a 100644 --- a/backend/internal/service/ops_settings_models.go +++ b/backend/internal/service/ops_settings_models.go @@ -69,17 +69,19 @@ type OpsMetricThresholds struct { } type OpsRuntimeLogConfig struct { - Level string `json:"level"` - EnableSampling bool `json:"enable_sampling"` - SamplingInitial int `json:"sampling_initial"` - SamplingNext int `json:"sampling_thereafter"` - Caller bool `json:"caller"` - StacktraceLevel string `json:"stacktrace_level"` - RetentionDays int `json:"retention_days"` - Source string `json:"source,omitempty"` - UpdatedAt string `json:"updated_at,omitempty"` - UpdatedByUserID int64 `json:"updated_by_user_id,omitempty"` - Extra map[string]any `json:"extra,omitempty"` + Level string `json:"level"` + EnableSampling bool `json:"enable_sampling"` + SamplingInitial int `json:"sampling_initial"` + SamplingNext int `json:"sampling_thereafter"` + Caller bool `json:"caller"` + StacktraceLevel string `json:"stacktrace_level"` + RetentionDays int `json:"retention_days"` + GatewayBodyLogEnabled bool `json:"gateway_body_log_enabled"` + GatewayBodyLogMaxBytes int `json:"gateway_body_log_max_bytes"` + Source string `json:"source,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` + UpdatedByUserID int64 `json:"updated_by_user_id,omitempty"` + Extra map[string]any `json:"extra,omitempty"` } type OpsAlertRuntimeSettings struct {