From 1f9d298e4981967947f2002f48ccd77f5bfc7845 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Sun, 9 Aug 2026 01:40:00 +0800 Subject: [PATCH 1/4] feat: support DeepSeek Responses and configurable timeout --- core/relay/adaptor/ai360/adaptor.go | 5 +- core/relay/adaptor/ali/adaptor.go | 20 +++++- core/relay/adaptor/azure/main.go | 5 +- core/relay/adaptor/azure/main_test.go | 8 +++ core/relay/adaptor/azure2/main.go | 5 +- core/relay/adaptor/baichuan/adaptor.go | 5 +- core/relay/adaptor/cloudflare/adaptor.go | 5 +- core/relay/adaptor/deepseek/adaptor.go | 22 +++++- core/relay/adaptor/deepseek/adaptor_test.go | 30 +++++++++ core/relay/adaptor/deepseek/constants.go | 16 +++++ core/relay/adaptor/doubao/main.go | 11 ++- core/relay/adaptor/geminiopenai/adaptor.go | 5 +- core/relay/adaptor/groq/adaptor.go | 5 +- core/relay/adaptor/jina/adaptor.go | 5 +- core/relay/adaptor/lingyiwanwu/adaptor.go | 5 +- core/relay/adaptor/mistral/adaptor.go | 5 +- core/relay/adaptor/novita/adaptor.go | 5 +- core/relay/adaptor/openai/adaptor.go | 48 +++++++++++-- core/relay/adaptor/openai/config.go | 60 ++++++++++++++++- core/relay/adaptor/openai/config_test.go | 43 ++++++++++++ core/relay/adaptor/openai/response.go | 20 +++++- core/relay/adaptor/openai/response_test.go | 74 +++++++++++++++++---- core/relay/adaptor/openrouter/adaptor.go | 7 +- core/relay/adaptor/qianfan/config.go | 12 ++++ core/relay/adaptor/sangforaicp/adaptor.go | 3 +- core/relay/adaptor/stepfun/adaptor.go | 5 +- core/relay/adaptor/tencent/adaptor.go | 5 +- core/relay/adaptor/xai/adaptor.go | 5 +- core/relay/adaptor/xunfei/adaptor.go | 7 +- 29 files changed, 385 insertions(+), 66 deletions(-) create mode 100644 core/relay/adaptor/openai/config_test.go diff --git a/core/relay/adaptor/ai360/adaptor.go b/core/relay/adaptor/ai360/adaptor.go index 19a67d807..2ee85bfd5 100644 --- a/core/relay/adaptor/ai360/adaptor.go +++ b/core/relay/adaptor/ai360/adaptor.go @@ -23,7 +23,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "360 AI open platform\nOpenAI-compatible chat endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "360 AI open platform\nOpenAI-compatible chat endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/ali/adaptor.go b/core/relay/adaptor/ali/adaptor.go index 5f8d20c06..8f3c5fac9 100644 --- a/core/relay/adaptor/ali/adaptor.go +++ b/core/relay/adaptor/ali/adaptor.go @@ -395,7 +395,20 @@ func (a *Adaptor) DoResponse( mode.ResponsesDelete, mode.ResponsesCancel, mode.ResponsesInputItems: - return openai.DoResponse(meta, store, c, resp) + if meta.Mode != mode.Responses || !utils.IsStreamResponse(resp) { + return openai.DoResponse(meta, store, c, resp) + } + + options, err := openai.LoadDoResponseOptions(meta) + if err != nil { + return adaptor.DoResponseResult{}, relaymodel.WrapperOpenAIError( + err, + "load_channel_config_failed", + http.StatusInternalServerError, + ) + } + + return openai.DoResponse(meta, store, c, resp, options) default: return adaptor.DoResponseResult{}, relaymodel.WrapperOpenAIErrorWithMessage( fmt.Sprintf("unsupported mode: %s", meta.Mode), @@ -407,7 +420,8 @@ func (a *Adaptor) DoResponse( func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "OpenAI compatibility\nNative Responses API support\nNetwork search metering support\nImage generation/edit support: https://help.aliyun.com/zh/model-studio/qwen-image-api and https://help.aliyun.com/zh/model-studio/qwen-image-edit-api\nVideo generation support: DashScope /api/v1/services/aigc/video-generation/video-synthesis\nRerank support: https://help.aliyun.com/zh/model-studio/text-rerank-api\nSTT support: https://help.aliyun.com/zh/model-studio/sambert-speech-synthesis/\nAnthropic support: /api/v2/apps/claude-code-proxy\nGemini support", - Models: ModelList, + Readme: "OpenAI compatibility\nNative Responses API support\nNetwork search metering support\nImage generation/edit support: https://help.aliyun.com/zh/model-studio/qwen-image-api and https://help.aliyun.com/zh/model-studio/qwen-image-edit-api\nVideo generation support: DashScope /api/v1/services/aigc/video-generation/video-synthesis\nRerank support: https://help.aliyun.com/zh/model-studio/text-rerank-api\nSTT support: https://help.aliyun.com/zh/model-studio/sambert-speech-synthesis/\nAnthropic support: /api/v2/apps/claude-code-proxy\nGemini support", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/azure/main.go b/core/relay/adaptor/azure/main.go index e407578b8..8b9fd4784 100644 --- a/core/relay/adaptor/azure/main.go +++ b/core/relay/adaptor/azure/main.go @@ -439,7 +439,8 @@ func (a *Adaptor) Metadata() adaptor.Metadata { "Azure OpenAI endpoint\nModel names do not contain '.' character, dots will be removed\nFor example: gpt-3.5-turbo becomes gpt-35-turbo\nAPI version is optional, default is '%s'\nSupports Gemini-compatible request conversion", DefaultAPIVersion, ), - KeyHelp: "key or key|api-version", - Models: openai.ModelList, + KeyHelp: "key or key|api-version", + ConfigSchema: openai.ConfigSchema(), + Models: openai.ModelList, } } diff --git a/core/relay/adaptor/azure/main_test.go b/core/relay/adaptor/azure/main_test.go index eb189f321..e377f42b9 100644 --- a/core/relay/adaptor/azure/main_test.go +++ b/core/relay/adaptor/azure/main_test.go @@ -102,6 +102,14 @@ func TestGetRequestURL(t *testing.T) { } } +func TestMetadataIncludesOpenAIConfigSchema(t *testing.T) { + metadata := (&azure.Adaptor{}).Metadata() + properties, ok := metadata.ConfigSchema["properties"].(map[string]any) + require.True(t, ok) + _, ok = properties["responses_first_event_timeout"] + assert.True(t, ok) +} + func TestGetRequestURL_ResponsesOnlyModels(t *testing.T) { adaptor := &azure.Adaptor{} diff --git a/core/relay/adaptor/azure2/main.go b/core/relay/adaptor/azure2/main.go index a88f9b6ff..f314fa21e 100644 --- a/core/relay/adaptor/azure2/main.go +++ b/core/relay/adaptor/azure2/main.go @@ -43,7 +43,8 @@ func (a *Adaptor) Metadata() adaptor.Metadata { "Azure AI Foundry / Azure OpenAI compatible endpoint\nModel names can contain '.' character\nAPI version is optional, default is '%s'\nSupports Gemini-compatible request conversion", azure.DefaultAPIVersion, ), - KeyHelp: "key or key|api-version", - Models: openai.ModelList, + KeyHelp: "key or key|api-version", + ConfigSchema: openai.ConfigSchema(), + Models: openai.ModelList, } } diff --git a/core/relay/adaptor/baichuan/adaptor.go b/core/relay/adaptor/baichuan/adaptor.go index 8a6c14a6e..2c905f552 100644 --- a/core/relay/adaptor/baichuan/adaptor.go +++ b/core/relay/adaptor/baichuan/adaptor.go @@ -23,7 +23,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Baichuan open platform\nOpenAI-compatible chat endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Baichuan open platform\nOpenAI-compatible chat endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/cloudflare/adaptor.go b/core/relay/adaptor/cloudflare/adaptor.go index 74a68a109..ecf56e5e3 100644 --- a/core/relay/adaptor/cloudflare/adaptor.go +++ b/core/relay/adaptor/cloudflare/adaptor.go @@ -100,7 +100,8 @@ func (a *Adaptor) GetRequestURL( func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Cloudflare Workers AI\nDefault base URL uses the account-scoped REST API\nAlso supports AI Gateway Workers AI endpoints ending with `/workers-ai`\nChat and embeddings use OpenAI-compatible paths; other modes use `/run/{model}`", - Models: ModelList, + Readme: "Cloudflare Workers AI\nDefault base URL uses the account-scoped REST API\nAlso supports AI Gateway Workers AI endpoints ending with `/workers-ai`\nChat and embeddings use OpenAI-compatible paths; other modes use `/run/{model}`", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/deepseek/adaptor.go b/core/relay/adaptor/deepseek/adaptor.go index 3b0932ffb..d56f61237 100644 --- a/core/relay/adaptor/deepseek/adaptor.go +++ b/core/relay/adaptor/deepseek/adaptor.go @@ -40,12 +40,29 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) SupportMode(mt *meta.Meta) bool { m := adaptor.ModeFromMeta(mt) + if m == mode.Responses { + return supportsResponsesModel(mt) + } + return m == mode.ChatCompletions || m == mode.Completions || m == mode.Anthropic || m == mode.Gemini } +func supportsResponsesModel(mt *meta.Meta) bool { + if mt == nil { + return false + } + + modelName := strings.ToLower(mt.ActualModel) + if modelName == "" { + modelName = strings.ToLower(mt.OriginModel) + } + + return modelName == "deepseek-v4-flash" +} + func (a *Adaptor) SetupRequestHeader( meta *meta.Meta, store adaptor.Store, @@ -145,8 +162,9 @@ func (a *Adaptor) DoResponse( func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "DeepSeek API\nOpenAI-compatible chat and completions endpoints\nSupports native Anthropic-compatible endpoint and Gemini-compatible request conversion", - Models: ModelList, + Readme: "DeepSeek API\nOpenAI-compatible chat and completions endpoints\nSupports native Responses API for deepseek-v4-flash\nSupports native Anthropic-compatible endpoint and Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/deepseek/adaptor_test.go b/core/relay/adaptor/deepseek/adaptor_test.go index 79d8e4d69..6ba9aed26 100644 --- a/core/relay/adaptor/deepseek/adaptor_test.go +++ b/core/relay/adaptor/deepseek/adaptor_test.go @@ -123,6 +123,12 @@ func TestDeepseekGetRequestURLOpenAIModes(t *testing.T) { baseURL: baseURL, wantURL: "https://api.deepseek.com/v1/chat/completions", }, + { + name: "responses official base", + mode: mode.Responses, + baseURL: baseURL, + wantURL: "https://api.deepseek.com/v1/responses", + }, } for _, tc := range testCases { @@ -141,6 +147,30 @@ func TestDeepseekGetRequestURLOpenAIModes(t *testing.T) { } } +func TestDeepseekSupportModeResponses(t *testing.T) { + a := &Adaptor{} + + assert.True(t, a.SupportMode(&meta.Meta{ + Mode: mode.Responses, + ActualModel: "deepseek-v4-flash", + })) + assert.False(t, a.SupportMode(&meta.Meta{ + Mode: mode.Responses, + ActualModel: "deepseek-chat", + })) + assert.False(t, a.SupportMode(&meta.Meta{ + Mode: mode.ResponsesGet, + })) +} + +func TestDeepseekMetadataIncludesOpenAIConfigSchema(t *testing.T) { + metadata := (&Adaptor{}).Metadata() + properties, ok := metadata.ConfigSchema["properties"].(map[string]any) + require.True(t, ok) + _, ok = properties["responses_first_event_timeout"] + assert.True(t, ok) +} + func TestDeepseekSetupRequestHeaderAnthropic(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/core/relay/adaptor/deepseek/constants.go b/core/relay/adaptor/deepseek/constants.go index b37543796..9ee545948 100644 --- a/core/relay/adaptor/deepseek/constants.go +++ b/core/relay/adaptor/deepseek/constants.go @@ -6,6 +6,22 @@ import ( ) var ModelList = []model.ModelConfig{ + { + Model: "deepseek-v4-flash", + Type: mode.ChatCompletions, + Owner: model.ModelOwnerDeepSeek, + Price: model.Price{ + InputPrice: 0.00014, + CachedPrice: 0.0000028, + OutputPrice: 0.00028, + }, + Config: model.NewModelConfig( + model.WithModelConfigMaxContextTokens(1000000), + model.WithModelConfigMaxOutputTokens(384000), + model.WithModelConfigToolChoice(true), + ), + }, + { Model: "deepseek-chat", Type: mode.ChatCompletions, diff --git a/core/relay/adaptor/doubao/main.go b/core/relay/adaptor/doubao/main.go index f9df727de..0ada2db1e 100644 --- a/core/relay/adaptor/doubao/main.go +++ b/core/relay/adaptor/doubao/main.go @@ -247,8 +247,9 @@ func (a *Adaptor) SupportMode(mt *meta.Meta) bool { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Doubao / Volcano Engine endpoint\nSupports bot-style models, native Responses API, Gemini-compatible request conversion, and network search metering fields", - Models: ModelList, + Readme: "Doubao / Volcano Engine endpoint\nSupports bot-style models, native Responses API, Gemini-compatible request conversion, and network search metering fields", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } @@ -376,6 +377,12 @@ func (a *Adaptor) DoResponse( return openai.GeminiStreamHandler(meta, c, resp) } return openai.GeminiHandler(meta, c, resp) + case mode.Responses, + mode.ResponsesGet, + mode.ResponsesDelete, + mode.ResponsesCancel, + mode.ResponsesInputItems: + return a.Adaptor.DoResponse(meta, store, c, resp) default: return openai.DoResponse(meta, store, c, resp) } diff --git a/core/relay/adaptor/geminiopenai/adaptor.go b/core/relay/adaptor/geminiopenai/adaptor.go index e9f9d7a56..b4b89f406 100644 --- a/core/relay/adaptor/geminiopenai/adaptor.go +++ b/core/relay/adaptor/geminiopenai/adaptor.go @@ -24,7 +24,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "https://ai.google.dev/gemini-api/docs/openai\nGoogle Gemini OpenAI-compatible endpoint", - Models: gemini.ModelList, + Readme: "https://ai.google.dev/gemini-api/docs/openai\nGoogle Gemini OpenAI-compatible endpoint", + ConfigSchema: openai.ConfigSchema(), + Models: gemini.ModelList, } } diff --git a/core/relay/adaptor/groq/adaptor.go b/core/relay/adaptor/groq/adaptor.go index 886f29e95..85b713f81 100644 --- a/core/relay/adaptor/groq/adaptor.go +++ b/core/relay/adaptor/groq/adaptor.go @@ -23,7 +23,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Groq API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Groq API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/jina/adaptor.go b/core/relay/adaptor/jina/adaptor.go index 483098ef8..f19c3d6ae 100644 --- a/core/relay/adaptor/jina/adaptor.go +++ b/core/relay/adaptor/jina/adaptor.go @@ -55,7 +55,8 @@ func (a *Adaptor) DoResponse( func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "https://jina.ai\nSupports embeddings and rerank\nAlso supports Gemini-compatible request conversion through the OpenAI-compatible layer", - Models: ModelList, + Readme: "https://jina.ai\nSupports embeddings and rerank\nAlso supports Gemini-compatible request conversion through the OpenAI-compatible layer", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/lingyiwanwu/adaptor.go b/core/relay/adaptor/lingyiwanwu/adaptor.go index a9a90bb03..fdf08063a 100644 --- a/core/relay/adaptor/lingyiwanwu/adaptor.go +++ b/core/relay/adaptor/lingyiwanwu/adaptor.go @@ -27,7 +27,8 @@ func (a *Adaptor) GetBalance(_ *model.Channel) (float64, error) { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Lingyi Wanwu API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Lingyi Wanwu API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/mistral/adaptor.go b/core/relay/adaptor/mistral/adaptor.go index 467e46522..66860fe47 100644 --- a/core/relay/adaptor/mistral/adaptor.go +++ b/core/relay/adaptor/mistral/adaptor.go @@ -23,7 +23,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Mistral API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Mistral API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/novita/adaptor.go b/core/relay/adaptor/novita/adaptor.go index 9f5cada95..0e1bc401f 100644 --- a/core/relay/adaptor/novita/adaptor.go +++ b/core/relay/adaptor/novita/adaptor.go @@ -23,7 +23,8 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Novita AI API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Novita AI API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/openai/adaptor.go b/core/relay/adaptor/openai/adaptor.go index 82cf0c53f..e96401ed8 100644 --- a/core/relay/adaptor/openai/adaptor.go +++ b/core/relay/adaptor/openai/adaptor.go @@ -434,17 +434,39 @@ func ConvertRequest( } } -//nolint:gocyclo func DoResponse( meta *meta.Meta, store adaptor.Store, c *gin.Context, resp *http.Response, + options ...DoResponseOptions, +) (result adaptor.DoResponseResult, err adaptor.Error) { + responseOptions := defaultConfig().doResponseOptions() + if len(options) > 0 { + responseOptions = options[0] + } + + return doResponse(meta, store, c, resp, responseOptions) +} + +//nolint:gocyclo +func doResponse( + meta *meta.Meta, + store adaptor.Store, + c *gin.Context, + resp *http.Response, + options DoResponseOptions, ) (result adaptor.DoResponseResult, err adaptor.Error) { switch meta.Mode { case mode.Responses: if utils.IsStreamResponse(resp) { - result, err = ResponseStreamHandler(meta, store, c, resp) + result, err = responseStreamHandler( + meta, + store, + c, + resp, + options.ResponsesFirstEventTimeout, + ) } else { result, err = ResponseHandler(meta, store, c, resp) } @@ -588,13 +610,29 @@ func (a *Adaptor) DoResponse( c *gin.Context, resp *http.Response, ) (result adaptor.DoResponseResult, err adaptor.Error) { - return DoResponse(meta, store, c, resp) + options := defaultConfig().doResponseOptions() + if meta.Mode == mode.Responses && utils.IsStreamResponse(resp) { + var configErr error + + cfg, configErr := a.loadConfig(meta) + if configErr != nil { + return adaptor.DoResponseResult{}, relaymodel.WrapperOpenAIError( + configErr, + "load_channel_config_failed", + http.StatusInternalServerError, + ) + } + + options = cfg.doResponseOptions() + } + + return DoResponse(meta, store, c, resp, options) } func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "OpenAI native API\nSupports chat, completions, embeddings, moderations, image, audio, rerank, PDF parsing, video generation, and Responses API\nAlso supports Anthropic-compatible and Gemini-compatible request conversion on top of the OpenAI endpoint\nChannel config `map_reasoning_to_reasoning_content` rewrites upstream `reasoning` fields to `reasoning_content` in chat completion responses", - ConfigSchema: configSchema(), + Readme: "OpenAI native API\nSupports chat, completions, embeddings, moderations, image, audio, rerank, PDF parsing, video generation, and Responses API\nAlso supports Anthropic-compatible and Gemini-compatible request conversion on top of the OpenAI endpoint\nChannel config `responses_first_event_timeout` sets the maximum seconds to wait for the first effective Responses stream event\nChannel config `map_reasoning_to_reasoning_content` rewrites upstream `reasoning` fields to `reasoning_content` in chat completion responses", + ConfigSchema: ConfigSchema(), Models: ModelList, } } diff --git a/core/relay/adaptor/openai/config.go b/core/relay/adaptor/openai/config.go index 169f3bfbe..6c371a897 100644 --- a/core/relay/adaptor/openai/config.go +++ b/core/relay/adaptor/openai/config.go @@ -1,13 +1,60 @@ package openai -import "github.com/labring/aiproxy/core/relay/meta" +import ( + "time" + + "github.com/labring/aiproxy/core/relay/meta" +) + +const defaultResponsesFirstEventTimeoutSeconds uint32 = 2 type Config struct { - MapReasoningToReasoningContent bool `json:"map_reasoning_to_reasoning_content"` + MapReasoningToReasoningContent bool `json:"map_reasoning_to_reasoning_content"` + ResponsesFirstEventTimeout uint32 `json:"responses_first_event_timeout"` +} + +// DoResponseOptions contains the response handling options needed by the OpenAI Responses path. +type DoResponseOptions struct { + ResponsesFirstEventTimeout time.Duration +} + +func defaultConfig() Config { + return Config{ + ResponsesFirstEventTimeout: defaultResponsesFirstEventTimeoutSeconds, + } +} + +// ConfigSchema returns the channel configuration schema shared by OpenAI-compatible adaptors. +func ConfigSchema() map[string]any { + return configSchema() +} + +func (c Config) responsesFirstEventTimeout() time.Duration { + return time.Duration(c.ResponsesFirstEventTimeout) * time.Second +} + +func (c Config) doResponseOptions() DoResponseOptions { + return DoResponseOptions{ + ResponsesFirstEventTimeout: c.responsesFirstEventTimeout(), + } +} + +// LoadDoResponseOptions loads OpenAI response handling options from channel config. +func LoadDoResponseOptions(meta *meta.Meta) (DoResponseOptions, error) { + cfg := defaultConfig() + if meta == nil { + return cfg.doResponseOptions(), nil + } + + if err := meta.ChannelConfigs.LoadConfig(&cfg); err != nil { + return DoResponseOptions{}, err + } + + return cfg.doResponseOptions(), nil } func (a *Adaptor) loadConfig(meta *meta.Meta) (Config, error) { - cfg := Config{} + cfg := defaultConfig() return a.configCache.Load(meta, cfg) } @@ -15,6 +62,13 @@ func configSchema() map[string]any { return map[string]any{ "type": "object", "properties": map[string]any{ + "responses_first_event_timeout": map[string]any{ + "type": "integer", + "title": "Responses first event timeout", + "description": "Maximum seconds to buffer initial Responses API lifecycle events while waiting for the first output or error event. Increase this value to allow late upstream errors to trigger channel retries.", + "default": defaultResponsesFirstEventTimeoutSeconds, + "minimum": 0, + }, "map_reasoning_to_reasoning_content": map[string]any{ "type": "boolean", "title": "Map reasoning To reasoning_content", diff --git a/core/relay/adaptor/openai/config_test.go b/core/relay/adaptor/openai/config_test.go new file mode 100644 index 000000000..773e5d739 --- /dev/null +++ b/core/relay/adaptor/openai/config_test.go @@ -0,0 +1,43 @@ +//nolint:testpackage +package openai + +import ( + "testing" + "time" + + "github.com/labring/aiproxy/core/model" + "github.com/labring/aiproxy/core/relay/meta" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestResponsesFirstEventTimeoutConfig(t *testing.T) { + t.Parallel() + + a := &Adaptor{} + + defaultCfg, err := a.loadConfig(&meta.Meta{}) + require.NoError(t, err) + assert.Equal(t, 2*time.Second, defaultCfg.responsesFirstEventTimeout()) + + configured, err := a.loadConfig(&meta.Meta{ + ChannelConfigs: model.ChannelConfigs{ + "responses_first_event_timeout": 30, + }, + }) + require.NoError(t, err) + assert.Equal(t, 30*time.Second, configured.responsesFirstEventTimeout()) +} + +func TestConfigSchemaIncludesResponsesFirstEventTimeout(t *testing.T) { + t.Parallel() + + properties, ok := configSchema()["properties"].(map[string]any) + require.True(t, ok) + + field, ok := properties["responses_first_event_timeout"].(map[string]any) + require.True(t, ok) + assert.Equal(t, "integer", field["type"]) + assert.Equal(t, defaultResponsesFirstEventTimeoutSeconds, field["default"]) + assert.Equal(t, 0, field["minimum"]) +} diff --git a/core/relay/adaptor/openai/response.go b/core/relay/adaptor/openai/response.go index 07a2b94c8..4449fe6fe 100644 --- a/core/relay/adaptor/openai/response.go +++ b/core/relay/adaptor/openai/response.go @@ -21,8 +21,6 @@ import ( "github.com/sirupsen/logrus" ) -var responseStreamInitialBufferTimeout = 2 * time.Second - // ConvertResponseRequest converts a response creation request func ConvertResponseRequest( meta *meta.Meta, @@ -192,6 +190,22 @@ func ResponseStreamHandler( store adaptor.Store, c *gin.Context, resp *http.Response, +) (adaptor.DoResponseResult, adaptor.Error) { + return responseStreamHandler( + meta, + store, + c, + resp, + defaultConfig().responsesFirstEventTimeout(), + ) +} + +func responseStreamHandler( + meta *meta.Meta, + store adaptor.Store, + c *gin.Context, + resp *http.Response, + firstEventTimeout time.Duration, ) (adaptor.DoResponseResult, adaptor.Error) { if !adaptor.IsSuccessfulResponseStatus(mode.Responses, resp.StatusCode) { return adaptor.DoResponseResult{}, ErrorHanlder(resp) @@ -305,7 +319,7 @@ readLoop: pendingEvents = append(pendingEvents, append([]byte(nil), data...)) if bufferTimer == nil { - bufferTimer = time.NewTimer(responseStreamInitialBufferTimeout) + bufferTimer = time.NewTimer(firstEventTimeout) } continue diff --git a/core/relay/adaptor/openai/response_test.go b/core/relay/adaptor/openai/response_test.go index b8638e45b..88daf3dfe 100644 --- a/core/relay/adaptor/openai/response_test.go +++ b/core/relay/adaptor/openai/response_test.go @@ -8,7 +8,6 @@ import ( "net/http" "net/http/httptest" "strings" - "sync" "testing" "time" @@ -16,6 +15,7 @@ import ( "github.com/labring/aiproxy/core/model" "github.com/labring/aiproxy/core/relay/adaptor" "github.com/labring/aiproxy/core/relay/meta" + "github.com/labring/aiproxy/core/relay/mode" relaymodel "github.com/labring/aiproxy/core/relay/model" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,8 +26,6 @@ type responseTestStore struct { savedIfNotExist []adaptor.StoreCache } -var responseStreamInitialBufferTimeoutTestMu sync.Mutex - func (s *responseTestStore) GetStore(string, int, string) (adaptor.StoreCache, error) { return adaptor.StoreCache{}, nil } @@ -353,17 +351,8 @@ func TestResponseStreamHandlerAcceptsObjectFunctionCallArguments(t *testing.T) { } func TestResponseStreamHandlerStartsBufferTimeoutFromFirstDelayedEvent(t *testing.T) { - responseStreamInitialBufferTimeoutTestMu.Lock() - defer responseStreamInitialBufferTimeoutTestMu.Unlock() - gin.SetMode(gin.TestMode) - oldTimeout := responseStreamInitialBufferTimeout - responseStreamInitialBufferTimeout = time.Millisecond - t.Cleanup(func() { - responseStreamInitialBufferTimeout = oldTimeout - }) - reader, writer := io.Pipe() defer writer.Close() @@ -399,13 +388,72 @@ func TestResponseStreamHandlerStartsBufferTimeoutFromFirstDelayedEvent(t *testin Header: make(http.Header), } - result, err := ResponseStreamHandler(&meta.Meta{}, &responseTestStore{}, c, resp) + result, err := responseStreamHandler( + &meta.Meta{}, + &responseTestStore{}, + c, + resp, + time.Millisecond, + ) require.Nil(t, err) assert.Equal(t, "resp_timeout", result.UpstreamID) assert.Contains(t, recorder.Body.String(), "response.in_progress") assert.Contains(t, recorder.Body.String(), "response.completed") } +func TestAdaptorDoResponseUsesResponsesFirstEventTimeout(t *testing.T) { + gin.SetMode(gin.TestMode) + + reader, writer := io.Pipe() + defer writer.Close() + + go func() { + _, _ = writer.Write([]byte(strings.Join([]string{ + "event: response.created", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_configured_timeout\",\"object\":\"response\",\"created_at\":1,\"status\":\"in_progress\",\"model\":\"gpt-5.4\",\"output\":[],\"parallel_tool_calls\":true,\"store\":false}}", + "", + }, "\n"))) + + time.Sleep(20 * time.Millisecond) + + _, _ = writer.Write([]byte(strings.Join([]string{ + "event: error", + "data: {\"type\":\"error\",\"error\":{\"type\":\"server_error\",\"code\":\"server_error\",\"message\":\"stream failed\"}}", + "", + }, "\n"))) + _ = writer.Close() + }() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequestWithContext( + t.Context(), + http.MethodPost, + "/v1/responses", + nil, + ) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: reader, + Header: http.Header{ + "Content-Type": {"text/event-stream"}, + }, + } + m := &meta.Meta{ + Mode: mode.Responses, + ChannelConfigs: model.ChannelConfigs{ + "responses_first_event_timeout": 0, + }, + } + + result, err := (&Adaptor{}).DoResponse(m, &responseTestStore{}, c, resp) + require.Nil(t, err) + assert.Equal(t, "resp_configured_timeout", result.UpstreamID) + assert.Contains(t, recorder.Body.String(), "response.created") + assert.Contains(t, recorder.Body.String(), "stream failed") +} + func TestResponseHandlerWebSearchCountFromToolUsage(t *testing.T) { t.Parallel() gin.SetMode(gin.TestMode) diff --git a/core/relay/adaptor/openrouter/adaptor.go b/core/relay/adaptor/openrouter/adaptor.go index bd1f5a879..dc3bef6e3 100644 --- a/core/relay/adaptor/openrouter/adaptor.go +++ b/core/relay/adaptor/openrouter/adaptor.go @@ -46,13 +46,14 @@ func (a *Adaptor) DoResponse( return openai.Handler(meta, c, resp, openai.ReasoningToReasoningContentPreHandler) default: - return openai.DoResponse(meta, store, c, resp) + return a.Adaptor.DoResponse(meta, store, c, resp) } } func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "OpenRouter OpenAI-compatible endpoint\nThe upstream `reasoning` field is normalized to `reasoning_content`\nAlso supports Gemini-compatible request conversion", - Models: openai.ModelList, + Readme: "OpenRouter OpenAI-compatible endpoint\nThe upstream `reasoning` field is normalized to `reasoning_content`\nAlso supports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: openai.ModelList, } } diff --git a/core/relay/adaptor/qianfan/config.go b/core/relay/adaptor/qianfan/config.go index 2e294d17e..296d62cd1 100644 --- a/core/relay/adaptor/qianfan/config.go +++ b/core/relay/adaptor/qianfan/config.go @@ -16,6 +16,18 @@ func configSchema() map[string]any { return map[string]any{ "type": "object", "properties": map[string]any{ + "responses_first_event_timeout": map[string]any{ + "type": "integer", + "title": "Responses first event timeout", + "description": "Maximum seconds to buffer initial Responses API lifecycle events while waiting for the first output or error event. Increase this value to allow late upstream errors to trigger channel retries.", + "default": 2, + "minimum": 0, + }, + "map_reasoning_to_reasoning_content": map[string]any{ + "type": "boolean", + "title": "Map reasoning To reasoning_content", + "description": "Rewrite upstream chat completion `reasoning` fields to `reasoning_content` in both streaming and non-streaming responses.", + }, "appid": map[string]any{ "type": "string", "title": "AppID Header", diff --git a/core/relay/adaptor/sangforaicp/adaptor.go b/core/relay/adaptor/sangforaicp/adaptor.go index 04cea7f06..c903fc48e 100644 --- a/core/relay/adaptor/sangforaicp/adaptor.go +++ b/core/relay/adaptor/sangforaicp/adaptor.go @@ -21,6 +21,7 @@ func (a *Adaptor) DefaultBaseURL() string { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Sangfor AICP OpenAI-compatible endpoint", + Readme: "Sangfor AICP OpenAI-compatible endpoint", + ConfigSchema: openai.ConfigSchema(), } } diff --git a/core/relay/adaptor/stepfun/adaptor.go b/core/relay/adaptor/stepfun/adaptor.go index 1ddca59a4..86151a38c 100644 --- a/core/relay/adaptor/stepfun/adaptor.go +++ b/core/relay/adaptor/stepfun/adaptor.go @@ -44,7 +44,8 @@ func (a *Adaptor) GetBalance(_ *model.Channel) (float64, error) { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "StepFun API\nOpenAI-compatible endpoint\nTTS requests use a default voice when not provided\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "StepFun API\nOpenAI-compatible endpoint\nTTS requests use a default voice when not provided\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/tencent/adaptor.go b/core/relay/adaptor/tencent/adaptor.go index e46b5e93d..7828246d0 100644 --- a/core/relay/adaptor/tencent/adaptor.go +++ b/core/relay/adaptor/tencent/adaptor.go @@ -29,7 +29,8 @@ func (a *Adaptor) GetBalance(_ *model.Channel) (float64, error) { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "Tencent Hunyuan API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", - Models: ModelList, + Readme: "Tencent Hunyuan API\nOpenAI-compatible endpoint\nSupports Gemini-compatible request conversion", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/xai/adaptor.go b/core/relay/adaptor/xai/adaptor.go index 501f15353..1e4702386 100644 --- a/core/relay/adaptor/xai/adaptor.go +++ b/core/relay/adaptor/xai/adaptor.go @@ -40,7 +40,8 @@ func (a *Adaptor) DoResponse( func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "xAI API\nOpenAI-compatible endpoint", - Models: ModelList, + Readme: "xAI API\nOpenAI-compatible endpoint", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } diff --git a/core/relay/adaptor/xunfei/adaptor.go b/core/relay/adaptor/xunfei/adaptor.go index 35aaef7ed..5c3d0073a 100644 --- a/core/relay/adaptor/xunfei/adaptor.go +++ b/core/relay/adaptor/xunfei/adaptor.go @@ -46,8 +46,9 @@ func (a *Adaptor) GetBalance(_ *model.Channel) (float64, error) { func (a *Adaptor) Metadata() adaptor.Metadata { return adaptor.Metadata{ - Readme: "iFlytek Spark API\nOpenAI-compatible endpoint\nKey format uses `app_id|app_token`\nSupports Gemini-compatible request conversion", - KeyHelp: "app_id|app_token", - Models: ModelList, + Readme: "iFlytek Spark API\nOpenAI-compatible endpoint\nKey format uses `app_id|app_token`\nSupports Gemini-compatible request conversion", + KeyHelp: "app_id|app_token", + ConfigSchema: openai.ConfigSchema(), + Models: ModelList, } } From 7b010e08692cc1fe378828a6f1146d4d10ff4af0 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Sun, 9 Aug 2026 01:48:03 +0800 Subject: [PATCH 2/4] fix: use pinned pnpm in release builds --- .github/workflows/release.yml | 4 ++-- Dockerfile | 4 +--- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 20baab87f..80c2d1b8a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -41,8 +41,8 @@ jobs: - name: Build working-directory: web run: | - npm install -g pnpm - pnpm install && pnpm run build + corepack pnpm install --frozen-lockfile + corepack pnpm run build - name: Upload Artifact uses: actions/upload-artifact@v7 diff --git a/Dockerfile b/Dockerfile index 0e1f41aa5..8797d5145 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,9 +4,7 @@ WORKDIR /aiproxy/web COPY ./web/ ./ -RUN npm install -g pnpm - -RUN pnpm install && pnpm run build +RUN corepack pnpm install --frozen-lockfile && corepack pnpm run build FROM golang:1.26-alpine AS builder From 80b20dbb9da81e9db3d27237329d200402d4ef47 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Sun, 9 Aug 2026 02:37:42 +0800 Subject: [PATCH 3/4] test: cover channel retry rounds and filtering --- core/controller/relay-channel.go | 73 +++--- core/controller/relay-channel_test.go | 277 ++++++++++++++++++----- core/controller/relay-controller.go | 68 ++---- core/controller/relay-controller_test.go | 260 +++++++++++++++++++++ 4 files changed, 518 insertions(+), 160 deletions(-) diff --git a/core/controller/relay-channel.go b/core/controller/relay-channel.go index f57077252..140cab37c 100644 --- a/core/controller/relay-channel.go +++ b/core/controller/relay-channel.go @@ -255,27 +255,6 @@ func getChannelErrorRate(errorRates map[int64]float64, channelID int64) float64 return errorRates[channelID] } -func pickMinErrorRateHasPermissionChannel( - current *model.Channel, - currentErrorRate float64, - candidate *model.Channel, - candidateErrorRate float64, -) *model.Channel { - if candidate == nil { - return current - } - - if current == nil { - return candidate - } - - if candidateErrorRate < currentErrorRate { - return candidate - } - - return current -} - func pickChannel( channels []*model.Channel, errorRates map[int64]float64, @@ -656,28 +635,21 @@ func getRetryChannel( } } - if state.exhausted { - if state.lastMinErrorRateHasPermissionChannel == nil { + if state.designatedChannel != nil { + // Explicitly selected channels stay pinned for the request. + channelID := int64(state.designatedChannel.ID) + if _, ignored := state.ignoreChannelIDs[channelID]; ignored { return nil, ErrChannelsExhausted } - // Check if the lowest-error has-permission channel has high error rate. - // If so, return exhausted to prevent retrying with a bad channel - channelID := int64(state.lastMinErrorRateHasPermissionChannel.ID) if errorRate := getChannelErrorRate(errorRates, channelID); errorRate > maxRetryErrorRate { return nil, ErrChannelsExhausted } - return state.lastMinErrorRateHasPermissionChannel, nil + return state.designatedChannel, nil } - filteredChannels := filterChannels( - state.migratedChannels, - errorRates, - maxRetryErrorRate, - state.ignoreChannelIDs, - state.failedChannelIDs, - ) + filteredChannels := getRetryCandidates(state, errorRates) if len(state.preferChannelIDs) > 0 { newChannel := pickPreferredChannel( @@ -694,27 +666,36 @@ func getRetryChannel( errorRates, ) if err != nil { - if !errors.Is(err, ErrChannelsExhausted) || - state.lastMinErrorRateHasPermissionChannel == nil { + if !errors.Is(err, ErrChannelsExhausted) || len(state.failedChannelIDs) == 0 { return nil, err } - // Check if the lowest-error has-permission channel has high error rate. - // If so, return exhausted to prevent retrying with a bad channel - channelID := int64(state.lastMinErrorRateHasPermissionChannel.ID) - if errorRate := getChannelErrorRate(errorRates, channelID); errorRate > maxRetryErrorRate { - return nil, ErrChannelsExhausted - } - - // Check if the lowest-error has-permission channel is still healthy before using it. - state.exhausted = true + // Start a new round so every currently eligible channel gets another attempt. + state.failedChannelIDs = make(map[int64]struct{}) + state.preferChannelIDs = nil - return state.lastMinErrorRateHasPermissionChannel, nil + return pickChannel( + getRetryCandidates(state, errorRates), + errorRates, + ) } return newChannel, nil } +func getRetryCandidates( + state *retryState, + errorRates map[int64]float64, +) []*model.Channel { + return filterChannels( + state.migratedChannels, + errorRates, + maxRetryErrorRate, + state.ignoreChannelIDs, + state.failedChannelIDs, + ) +} + func filterChannels( channels []*model.Channel, errorRates map[int64]float64, diff --git a/core/controller/relay-channel_test.go b/core/controller/relay-channel_test.go index 32e786257..0be9600a3 100644 --- a/core/controller/relay-channel_test.go +++ b/core/controller/relay-channel_test.go @@ -6,6 +6,7 @@ import ( "net/http/httptest" "path/filepath" "testing" + "time" "github.com/gin-gonic/gin" "github.com/labring/aiproxy/core/middleware" @@ -168,7 +169,7 @@ func TestGetRetryChannelPrefersPreferredChannels(t *testing.T) { }) t.Run( - "returns exhausted when failed channels consume all retry candidates", + "starts a new round when failed channels consume all retry candidates", func(t *testing.T) { t.Parallel() @@ -185,59 +186,14 @@ func TestGetRetryChannelPrefersPreferredChannels(t *testing.T) { ) channel, err := getRetryChannel(context.Background(), state) - require.ErrorIs(t, err, ErrChannelsExhausted) - assert.Nil(t, channel) + require.NoError(t, err) + assert.NotNil(t, channel) + assert.Empty(t, state.failedChannelIDs) }, ) } -func TestPickMinErrorRateHasPermissionChannel(t *testing.T) { - t.Parallel() - - current := &model.Channel{ID: 1} - candidate := &model.Channel{ID: 2} - - t.Run("returns candidate when current is nil", func(t *testing.T) { - t.Parallel() - - picked := pickMinErrorRateHasPermissionChannel( - nil, - 0, - candidate, - 0.2, - ) - require.NotNil(t, picked) - assert.Equal(t, 2, picked.ID) - }) - - t.Run("keeps current when candidate error rate is higher", func(t *testing.T) { - t.Parallel() - - picked := pickMinErrorRateHasPermissionChannel( - current, - 0.1, - candidate, - 0.3, - ) - require.NotNil(t, picked) - assert.Equal(t, 1, picked.ID) - }) - - t.Run("switches to candidate when candidate error rate is lower", func(t *testing.T) { - t.Parallel() - - picked := pickMinErrorRateHasPermissionChannel( - current, - 0.4, - candidate, - 0.2, - ) - require.NotNil(t, picked) - assert.Equal(t, 2, picked.ID) - }) -} - -func TestGetRetryChannelFallsBackToLowestErrorRateHasPermissionChannel(t *testing.T) { +func TestGetRetryChannelStartsNewRoundAfterCandidatesAreExhausted(t *testing.T) { t.Parallel() ch1 := &model.Channel{ @@ -254,23 +210,220 @@ func TestGetRetryChannelFallsBackToLowestErrorRateHasPermissionChannel(t *testin } state := &retryState{ - meta: meta.NewMeta( - ch1, - mode.Responses, - "gpt-5", - model.ModelConfig{}, - ), - migratedChannels: []*model.Channel{ch1, ch2}, - failedChannelIDs: map[int64]struct{}{}, - ignoreChannelIDs: map[int64]struct{}{1: {}, 2: {}}, - lastMinErrorRateHasPermissionChannel: ch2, + meta: meta.NewMeta(ch1, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{ch1, ch2}, + failedChannelIDs: map[int64]struct{}{1: {}, 2: {}}, + ignoreChannelIDs: nil, + preferChannelIDs: []int{1}, + } + + channel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + require.NotNil(t, channel) + assert.Contains(t, []int{1, 2}, channel.ID) + assert.Empty(t, state.failedChannelIDs) + assert.Empty(t, state.preferChannelIDs) + + state.failedChannelIDs[int64(channel.ID)] = struct{}{} + nextChannel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + require.NotNil(t, nextChannel) + assert.NotEqual(t, channel.ID, nextChannel.ID) +} + +func TestGetRetryChannelKeepsPermissionFailuresIgnoredAcrossRounds(t *testing.T) { + t.Parallel() + + ch1 := &model.Channel{ID: 1, Type: model.ChannelTypeOpenAI, Status: model.ChannelStatusEnabled} + ch2 := &model.Channel{ID: 2, Type: model.ChannelTypeOpenAI, Status: model.ChannelStatusEnabled} + state := &retryState{ + meta: meta.NewMeta(ch1, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{ch1, ch2}, + failedChannelIDs: map[int64]struct{}{1: {}}, + ignoreChannelIDs: map[int64]struct{}{2: {}}, + } + + channel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + require.NotNil(t, channel) + assert.Equal(t, 1, channel.ID) + assert.Empty(t, state.failedChannelIDs) +} + +func TestGetRetryChannelKeepsDesignatedChannelPinned(t *testing.T) { + t.Parallel() + + ch1 := &model.Channel{ID: 1, Type: model.ChannelTypeOpenAI, Status: model.ChannelStatusEnabled} + ch2 := &model.Channel{ID: 2, Type: model.ChannelTypeOpenAI, Status: model.ChannelStatusEnabled} + state := &retryState{ + designatedChannel: ch1, + meta: meta.NewMeta(ch1, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{ch2}, + failedChannelIDs: map[int64]struct{}{1: {}, 2: {}}, + } + + channel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + assert.Equal(t, ch1.ID, channel.ID) + + state.ignoreChannelIDs = map[int64]struct{}{1: {}} + channel, err = getRetryChannel(context.Background(), state) + require.ErrorIs(t, err, ErrChannelsExhausted) + assert.Nil(t, channel) +} + +func TestFilterChannelsAppliesRetryEligibilityRules(t *testing.T) { + t.Parallel() + + enabled := &model.Channel{ + ID: 1, + Status: model.ChannelStatusEnabled, + } + disabled := &model.Channel{ + ID: 2, + Status: model.ChannelStatusDisabled, + } + highError := &model.Channel{ + ID: 3, + Status: model.ChannelStatusEnabled, + } + exactThreshold := &model.Channel{ + ID: 6, + Status: model.ChannelStatusEnabled, + } + ignored := &model.Channel{ + ID: 4, + Status: model.ChannelStatusEnabled, + } + multiIgnored := &model.Channel{ + ID: 5, + Status: model.ChannelStatusEnabled, + } + + filtered := filterChannels( + []*model.Channel{nil, enabled, disabled, highError, ignored, multiIgnored, exactThreshold}, + map[int64]float64{3: maxRetryErrorRate + 0.01, 6: maxRetryErrorRate}, + maxRetryErrorRate, + map[int64]struct{}{4: {}}, + map[int64]struct{}{5: {}}, + ) + + gotIDs := make([]int, len(filtered)) + for i, channel := range filtered { + gotIDs[i] = channel.ID + } + + assert.Equal(t, []int{enabled.ID, exactThreshold.ID}, gotIDs) +} + +func TestGetRetryChannelVisitsEveryEligibleChannelBeforeNextRound(t *testing.T) { + t.Parallel() + + channels := []*model.Channel{ + {ID: 1, Status: model.ChannelStatusEnabled}, + {ID: 2, Status: model.ChannelStatusEnabled}, + {ID: 3, Status: model.ChannelStatusEnabled}, + } + state := &retryState{ + meta: meta.NewMeta(channels[0], mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: channels, + failedChannelIDs: map[int64]struct{}{1: {}, 2: {}, 3: {}}, + } + seen := make(map[int]struct{}, len(channels)) + + for len(seen) < len(channels) { + channel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + require.NotNil(t, channel) + _, alreadySeen := seen[channel.ID] + assert.False(t, alreadySeen, "channel %d was selected twice in one round", channel.ID) + seen[channel.ID] = struct{}{} + state.failedChannelIDs[int64(channel.ID)] = struct{}{} + } + + channel, err := getRetryChannel(context.Background(), state) + require.NoError(t, err) + require.NotNil(t, channel) + assert.Empty(t, state.failedChannelIDs) +} + +func TestFilterChannelsDisablesErrorRateFilterAtZero(t *testing.T) { + t.Parallel() + + channels := []*model.Channel{ + {ID: 1, Status: model.ChannelStatusEnabled}, + {ID: 2, Status: model.ChannelStatusDisabled}, + } + + filtered := filterChannels( + channels, + map[int64]float64{1: 1}, + 0, + ) + + require.Len(t, filtered, 1) + assert.Equal(t, 1, filtered[0].ID) +} + +func TestGetRetryChannelReturnsExhaustedWhenNoRoundCanBeReset(t *testing.T) { + t.Parallel() + + channel := &model.Channel{ + ID: 1, + Status: model.ChannelStatusDisabled, + } + state := &retryState{ + meta: meta.NewMeta(channel, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{channel}, + failedChannelIDs: map[int64]struct{}{}, + } + + got, err := getRetryChannel(context.Background(), state) + require.ErrorIs(t, err, ErrChannelsExhausted) + assert.Nil(t, got) + assert.Empty(t, state.failedChannelIDs) +} + +func TestGetRetryChannelReturnsExhaustedWhenRoundResetHasNoEligibleChannel(t *testing.T) { + t.Parallel() + + channel := &model.Channel{ + ID: 1, + Status: model.ChannelStatusEnabled, + } + state := &retryState{ + meta: meta.NewMeta(channel, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{channel}, + failedChannelIDs: map[int64]struct{}{1: {}}, + ignoreChannelIDs: map[int64]struct{}{1: {}}, + } + + got, err := getRetryChannel(context.Background(), state) + require.ErrorIs(t, err, ErrChannelsExhausted) + assert.Nil(t, got) + assert.Empty(t, state.failedChannelIDs) +} + +func TestGetRetryChannelRoundResetPreservesBackoffState(t *testing.T) { + t.Parallel() + + ch1 := &model.Channel{ID: 1, Status: model.ChannelStatusEnabled} + ch2 := &model.Channel{ID: 2, Status: model.ChannelStatusEnabled} + base := time.Unix(100, 0) + state := &retryState{ + meta: meta.NewMeta(ch1, mode.Responses, "gpt-5", model.ModelConfig{}), + migratedChannels: []*model.Channel{ch1, ch2}, + failedChannelIDs: map[int64]struct{}{1: {}, 2: {}}, + channelRetryInfo: map[int]channelRetryInfo{ + 1: {failures: 2, lastEndAt: base}, + }, } channel, err := getRetryChannel(context.Background(), state) require.NoError(t, err) require.NotNil(t, channel) - assert.Equal(t, 2, channel.ID) - assert.True(t, state.exhausted) + assert.Empty(t, state.failedChannelIDs) + assert.Equal(t, channelRetryInfo{failures: 2, lastEndAt: base}, state.channelRetryInfo[1]) } func TestGetPriorityWeight(t *testing.T) { diff --git a/core/controller/relay-controller.go b/core/controller/relay-controller.go index dad01cd02..7e000a548 100644 --- a/core/controller/relay-controller.go +++ b/core/controller/relay-controller.go @@ -20,7 +20,6 @@ import ( "github.com/labring/aiproxy/core/common/conv" "github.com/labring/aiproxy/core/middleware" "github.com/labring/aiproxy/core/model" - "github.com/labring/aiproxy/core/monitor" "github.com/labring/aiproxy/core/relay/adaptor" "github.com/labring/aiproxy/core/relay/adaptors" "github.com/labring/aiproxy/core/relay/controller" @@ -546,12 +545,11 @@ func buildBodyDetailOption(meta *meta.Meta) controller.BodyDetailOption { } type retryState struct { - retryTimes int - lastMinErrorRateHasPermissionChannel *model.Channel - preferChannelIDs []int - ignoreChannelIDs map[int64]struct{} - exhausted bool - failedChannelIDs map[int64]struct{} // Track all failed channels in this request + retryTimes int + designatedChannel *model.Channel + preferChannelIDs []int + ignoreChannelIDs map[int64]struct{} + failedChannelIDs map[int64]struct{} // Track failed channels in the current retry round meta *meta.Meta price model.Price @@ -622,7 +620,7 @@ func initRetryState( } if channel.designatedChannel { - state.exhausted = true + state.designatedChannel = channel.channel } if !monitorplugin.ChannelHasPermission(result.Error) { @@ -631,8 +629,6 @@ func initRetryState( } state.ignoreChannelIDs[int64(channel.channel.ID)] = struct{}{} - } else { - state.lastMinErrorRateHasPermissionChannel = channel.channel } return state @@ -693,7 +689,7 @@ func (s *retryState) remainingRelayDelay( func retryLoop(c *gin.Context, mode mode.Mode, state *retryState, relayController RelayHandler) { log := common.GetLogger(c) - // do not use for i := range state.retryTimes, because the retryTimes is constant + // retryTimes can grow when permission failures add more eligible-channel attempts i := 0 for { @@ -819,49 +815,17 @@ func handleRetryResult( hasPermission := monitorplugin.ChannelHasPermission(state.result.Error) - if state.exhausted { - if !hasPermission { - return true - } - } else { - if !hasPermission { - if state.ignoreChannelIDs == nil { - state.ignoreChannelIDs = make(map[int64]struct{}) - } - - state.ignoreChannelIDs[int64(newChannel.ID)] = struct{}{} - state.retryTimes++ - } else { - if state.lastMinErrorRateHasPermissionChannel == nil { - state.lastMinErrorRateHasPermissionChannel = newChannel - return false - } - - currentErrorRate, err := monitor.GetChannelModelErrorRate( - ctx.Request.Context(), - state.meta.OriginModel, - int64(state.lastMinErrorRateHasPermissionChannel.ID), - ) - if err != nil { - return false - } - - newErrorRate, err := monitor.GetChannelModelErrorRate( - ctx.Request.Context(), - state.meta.OriginModel, - int64(newChannel.ID), - ) - if err != nil { - return false - } + if state.designatedChannel != nil { + return !hasPermission + } - state.lastMinErrorRateHasPermissionChannel = pickMinErrorRateHasPermissionChannel( - state.lastMinErrorRateHasPermissionChannel, - currentErrorRate, - newChannel, - newErrorRate, - ) + if !hasPermission { + if state.ignoreChannelIDs == nil { + state.ignoreChannelIDs = make(map[int64]struct{}) } + + state.ignoreChannelIDs[int64(newChannel.ID)] = struct{}{} + state.retryTimes++ } return false diff --git a/core/controller/relay-controller_test.go b/core/controller/relay-controller_test.go index 140cb50dd..a911f6433 100644 --- a/core/controller/relay-controller_test.go +++ b/core/controller/relay-controller_test.go @@ -2,16 +2,21 @@ package controller import ( + "context" "net/http" + "net/http/httptest" "reflect" "testing" "time" + "github.com/gin-gonic/gin" "github.com/glebarez/sqlite" "github.com/labring/aiproxy/core/model" + "github.com/labring/aiproxy/core/relay/adaptor" relaycontroller "github.com/labring/aiproxy/core/relay/controller" "github.com/labring/aiproxy/core/relay/meta" "github.com/labring/aiproxy/core/relay/mode" + relaymodel "github.com/labring/aiproxy/core/relay/model" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/gorm" @@ -107,12 +112,267 @@ func TestCalculateRelayBackoffDelay(t *testing.T) { t.Parallel() assert.Zero(t, calculateRelayBackoffDelay(0, 500*time.Millisecond)) + assert.Equal(t, time.Second, calculateRelayBackoffDelay(1, -time.Second)) assert.Equal(t, 1500*time.Millisecond, calculateRelayBackoffDelay(1, 500*time.Millisecond)) + assert.Equal(t, 2*time.Second, calculateRelayBackoffDelay(1, 2*time.Second)) assert.Equal(t, 2500*time.Millisecond, calculateRelayBackoffDelay(2, 500*time.Millisecond)) assert.Equal(t, 5*time.Second, calculateRelayBackoffDelay(20, time.Second)) assert.Equal(t, 2*time.Second, calculateRelayBackoffDelay(1, time.Second)) } +func TestHandleRelayResultDecidesRetryLifecycle(t *testing.T) { + t.Parallel() + + newContext := func(ctx context.Context) *gin.Context { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequestWithContext(ctx, http.MethodPost, "/", nil) + + return c + } + err := relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{ + Message: "upstream unavailable", + }) + + tests := []struct { + name string + bizErr adaptor.Error + retry bool + retryTimes int + wantDone bool + }{ + { + name: "successful request is done", + wantDone: true, + }, + { + name: "retryable error with budget continues", + bizErr: err, + retry: true, + retryTimes: 2, + wantDone: false, + }, + { + name: "retry disabled finishes", + bizErr: err, + retryTimes: 2, + wantDone: true, + }, + { + name: "zero retry budget finishes", + bizErr: err, + retry: true, + wantDone: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + assert.Equal(t, tt.wantDone, handleRelayResult( + newContext(context.Background()), + tt.bizErr, + tt.retry, + tt.retryTimes, + )) + }) + } + + canceled, cancel := context.WithCancel(context.Background()) + cancel() + assert.True(t, handleRelayResult(newContext(canceled), err, true, 2)) +} + +func TestInitRetryStateRecordsInitialFailure(t *testing.T) { + t.Parallel() + + ch1 := &model.Channel{ID: 1, Status: model.ChannelStatusEnabled} + ch2 := &model.Channel{ID: 2, Status: model.ChannelStatusEnabled} + endAt := time.Unix(500, 0) + initial := &initialChannel{ + channel: ch1, + migratedChannels: []*model.Channel{ch1, ch2}, + preferChannelIDs: []int{2}, + ignoreChannelIDs: map[int64]struct{}{9: {}}, + } + requestMeta := meta.NewMeta(ch1, mode.Responses, "gpt-5", model.ModelConfig{}) + result := &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusTooManyRequests, relaymodel.OpenAIError{ + Message: "rate limited", + }), + } + + state := initRetryState(3, initial, requestMeta, result, model.Price{}, endAt) + + assert.Equal(t, 3, state.retryTimes) + assert.Equal(t, []int{2}, state.preferChannelIDs) + assert.Equal(t, initial.ignoreChannelIDs, state.ignoreChannelIDs) + assert.Equal(t, []*model.Channel{ch1, ch2}, state.migratedChannels) + assert.Contains(t, state.failedChannelIDs, int64(ch1.ID)) + assert.Equal(t, 1, state.channelRetryInfo[ch1.ID].failures) + assert.Equal(t, endAt, state.channelRetryInfo[ch1.ID].lastEndAt) + assert.Nil(t, state.designatedChannel) +} + +func TestInitRetryStateMarksInitialPermissionFailureAsIgnored(t *testing.T) { + t.Parallel() + + channel := &model.Channel{ID: 7, Status: model.ChannelStatusEnabled} + requestMeta := meta.NewMeta(channel, mode.Responses, "gpt-5", model.ModelConfig{}) + result := &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusUnauthorized, relaymodel.OpenAIError{ + Message: "invalid key", + }), + } + + state := initRetryState( + 2, + &initialChannel{channel: channel}, + requestMeta, + result, + model.Price{}, + time.Unix(600, 0), + ) + + assert.Contains(t, state.failedChannelIDs, int64(channel.ID)) + assert.Contains(t, state.ignoreChannelIDs, int64(channel.ID)) + assert.Empty(t, state.channelRetryInfo) +} + +func TestInitRetryStateTracksDesignatedChannel(t *testing.T) { + t.Parallel() + + channel := &model.Channel{ID: 8, Status: model.ChannelStatusEnabled} + requestMeta := meta.NewMeta(channel, mode.Responses, "gpt-5", model.ModelConfig{}) + result := &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{ + Message: "upstream error", + }), + } + + state := initRetryState( + 1, + &initialChannel{channel: channel, designatedChannel: true}, + requestMeta, + result, + model.Price{}, + time.Unix(700, 0), + ) + + assert.Same(t, channel, state.designatedChannel) +} + +func TestHandleRetryResultUpdatesAutomaticRetryState(t *testing.T) { + t.Parallel() + + newContext := func() *gin.Context { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequestWithContext(context.Background(), http.MethodPost, "/", nil) + + return c + } + newState := func(err adaptor.Error) *retryState { + return &retryState{ + retryTimes: 2, + result: &relaycontroller.HandleResult{Error: err}, + } + } + permissionError := relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{ + Message: "upstream unavailable", + }) + noPermissionError := relaymodel.NewOpenAIError(http.StatusUnauthorized, relaymodel.OpenAIError{ + Message: "invalid key", + }) + channel := &model.Channel{ID: 11, Status: model.ChannelStatusEnabled} + + t.Run("permissioned failure keeps retrying without hard filtering", func(t *testing.T) { + t.Parallel() + + state := newState(permissionError) + + done := handleRetryResult(newContext(), true, channel, state) + + assert.False(t, done) + assert.Equal(t, 2, state.retryTimes) + assert.Empty(t, state.ignoreChannelIDs) + }) + + t.Run("permission failure is hard filtered and extends retry budget", func(t *testing.T) { + t.Parallel() + + state := newState(noPermissionError) + + done := handleRetryResult(newContext(), true, channel, state) + + assert.False(t, done) + assert.Equal(t, 3, state.retryTimes) + assert.Contains(t, state.ignoreChannelIDs, int64(channel.ID)) + }) + + t.Run("retry disabled finishes immediately", func(t *testing.T) { + t.Parallel() + + state := newState(permissionError) + + done := handleRetryResult(newContext(), false, channel, state) + + assert.True(t, done) + }) + + t.Run("nil result error finishes immediately", func(t *testing.T) { + t.Parallel() + + state := newState(nil) + + done := handleRetryResult(newContext(), true, channel, state) + + assert.True(t, done) + }) +} + +func TestHandleRetryResultKeepsDesignatedChannelSemantics(t *testing.T) { + t.Parallel() + + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequestWithContext(context.Background(), http.MethodPost, "/", nil) + channel := &model.Channel{ID: 12, Status: model.ChannelStatusEnabled} + + permissionedState := &retryState{ + designatedChannel: channel, + result: &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{}), + }, + } + assert.False(t, handleRetryResult(c, true, channel, permissionedState)) + + noPermissionState := &retryState{ + designatedChannel: channel, + result: &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusForbidden, relaymodel.OpenAIError{}), + }, + } + assert.True(t, handleRetryResult(c, true, channel, noPermissionState)) + assert.Empty(t, noPermissionState.ignoreChannelIDs) +} + +func TestHandleRetryResultStopsOnCanceledContext(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequestWithContext(ctx, http.MethodPost, "/", nil) + + cancel() + + state := &retryState{ + result: &relaycontroller.HandleResult{ + Error: relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{}), + }, + } + + assert.True(t, handleRetryResult(c, true, &model.Channel{ID: 13}, state)) +} + func TestRelayControllerVideoModesValidateRequests(t *testing.T) { t.Parallel() From 7fe0d475945a80cc77e645466d5819806faeb8d8 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Sun, 9 Aug 2026 02:48:08 +0800 Subject: [PATCH 4/4] feat: log response processing cost --- core/relay/plugin/monitor/monitor.go | 4 + core/relay/plugin/monitor/monitor_test.go | 112 ++++++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/core/relay/plugin/monitor/monitor.go b/core/relay/plugin/monitor/monitor.go index 981391ca9..ff3ea7b58 100644 --- a/core/relay/plugin/monitor/monitor.go +++ b/core/relay/plugin/monitor/monitor.go @@ -231,8 +231,12 @@ func (m *ChannelMonitor) DoResponse( resp *http.Response, do adaptor.DoResponse, ) (adaptor.DoResponseResult, adaptor.Error) { + responseAt := time.Now() result, relayErr := do.DoResponse(meta, store, c, resp) + responseCost := common.TruncateDuration(time.Since(responseAt)) + common.GetLogger(c).Data["resp_cost"] = responseCost.String() + if result.Usage.TotalTokens > 0 { count, overLimitCount, secondCount := reqlimit.PushChannelModelTokensRequest( context.Background(), diff --git a/core/relay/plugin/monitor/monitor_test.go b/core/relay/plugin/monitor/monitor_test.go index 64bcf0fc8..aa29e5f4d 100644 --- a/core/relay/plugin/monitor/monitor_test.go +++ b/core/relay/plugin/monitor/monitor_test.go @@ -2,15 +2,39 @@ package monitor import ( + "context" "net/http" + "net/http/httptest" "testing" + "time" + "github.com/gin-gonic/gin" + "github.com/labring/aiproxy/core/common" "github.com/labring/aiproxy/core/common/config" + "github.com/labring/aiproxy/core/model" + "github.com/labring/aiproxy/core/relay/adaptor" relaymeta "github.com/labring/aiproxy/core/relay/meta" + "github.com/labring/aiproxy/core/relay/mode" relaymodel "github.com/labring/aiproxy/core/relay/model" "github.com/stretchr/testify/require" ) +type monitorDoResponseFunc func( + *relaymeta.Meta, + adaptor.Store, + *gin.Context, + *http.Response, +) (adaptor.DoResponseResult, adaptor.Error) + +func (fn monitorDoResponseFunc) DoResponse( + meta *relaymeta.Meta, + store adaptor.Store, + c *gin.Context, + resp *http.Response, +) (adaptor.DoResponseResult, adaptor.Error) { + return fn(meta, store, c, resp) +} + func TestGetChannelWarnErrorRateUsesChannelValueEvenWhenAutoBalanceDisabled(t *testing.T) { meta := &relaymeta.Meta{} meta.Channel.WarnErrorRate = 0.42 @@ -109,3 +133,91 @@ func TestChannelHasPermissionForForbiddenErrorCode(t *testing.T) { }) } } + +func TestChannelMonitorDoResponseRecordsResponseCost(t *testing.T) { + gin.SetMode(gin.TestMode) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequestWithContext(context.Background(), http.MethodPost, "/", nil) + entry := common.NewLogger() + common.SetLogger(c.Request, entry) + + requestMeta := relaymeta.NewMeta( + &model.Channel{ID: 901, Type: model.ChannelTypeOpenAI}, + mode.ChatCompletions, + "resp-cost-test", + model.ModelConfig{}, + ) + requestMeta.Channel.MaxErrorRate = 0 + + result, relayErr := (&ChannelMonitor{}).DoResponse( + requestMeta, + nil, + c, + &http.Response{StatusCode: http.StatusOK}, + monitorDoResponseFunc(func( + *relaymeta.Meta, + adaptor.Store, + *gin.Context, + *http.Response, + ) (adaptor.DoResponseResult, adaptor.Error) { + time.Sleep(2 * time.Millisecond) + return adaptor.DoResponseResult{}, nil + }), + ) + + require.NoError(t, relayErr) + require.Empty(t, result.UpstreamID) + require.Contains(t, entry.Data, "resp_cost") + cost, ok := entry.Data["resp_cost"].(string) + require.True(t, ok) + require.NotEmpty(t, cost) + parsedCost, err := time.ParseDuration(cost) + require.NoError(t, err) + require.Greater(t, parsedCost, time.Duration(0)) +} + +func TestChannelMonitorDoResponseRecordsResponseCostOnError(t *testing.T) { + gin.SetMode(gin.TestMode) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequestWithContext(context.Background(), http.MethodPost, "/", nil) + entry := common.NewLogger() + common.SetLogger(c.Request, entry) + + requestMeta := relaymeta.NewMeta( + &model.Channel{ID: 902, Type: model.ChannelTypeOpenAI}, + mode.ChatCompletions, + "resp-cost-error-test", + model.ModelConfig{}, + ) + relayErrExpected := relaymodel.NewOpenAIError(http.StatusBadGateway, relaymodel.OpenAIError{ + Message: "upstream error", + }) + + _, relayErr := (&ChannelMonitor{}).DoResponse( + requestMeta, + nil, + c, + &http.Response{StatusCode: http.StatusBadGateway}, + monitorDoResponseFunc(func( + *relaymeta.Meta, + adaptor.Store, + *gin.Context, + *http.Response, + ) (adaptor.DoResponseResult, adaptor.Error) { + return adaptor.DoResponseResult{}, relayErrExpected + }), + ) + + require.ErrorIs(t, relayErr, relayErrExpected) + require.Contains(t, entry.Data, "resp_cost") + cost, ok := entry.Data["resp_cost"].(string) + require.True(t, ok) + require.NotEmpty(t, cost) + parsedCost, err := time.ParseDuration(cost) + require.NoError(t, err) + require.Greater(t, parsedCost, time.Duration(0)) +}