mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-10-19 22:19:12 +08:00
fix(server): prevent data races during concurrent config access and reloads
- Access configuration safely via getters across server routes, middleware, and service routines. - Clone configuration snapshots for runtime and synchronize access in management handlers. - Guard configuration YAML snapshots with mutex locks during reloads. Closes: #6592
This commit is contained in:
44
internal/api/runtime_reload_regression_test.go
Normal file
44
internal/api/runtime_reload_regression_test.go
Normal file
@@ -0,0 +1,44 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/router-for-me/CLIProxyAPI/v8/internal/config"
|
||||
)
|
||||
|
||||
func TestHandlerRuntimeReloadConcurrentWithModelLists(t *testing.T) {
|
||||
server := newTestServer(t)
|
||||
host := &mockModelListInterceptorHost{}
|
||||
start := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for i := 0; i < 500; i++ {
|
||||
server.handlers.SetPluginHost(host)
|
||||
server.handlers.UpdateClients(&config.SDKConfig{RequestLog: i%2 == 0})
|
||||
server.handlers.SetPluginHost(nil)
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for i := 0; i < 500; i++ {
|
||||
req := httptest.NewRequest(http.MethodGet, "/v1/models", nil)
|
||||
req.Header.Set("Authorization", "Bearer test-key")
|
||||
req.Header.Set("User-Agent", "codex_cli_rs/0.100.0")
|
||||
rr := httptest.NewRecorder()
|
||||
server.engine.ServeHTTP(rr, req)
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Errorf("model list status = %d, want 200", rr.Code)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
close(start)
|
||||
wg.Wait()
|
||||
}
|
||||
@@ -236,8 +236,10 @@ func (s *Server) codexAlphaSearchModelRouterHost() handlers.PluginModelRouterHos
|
||||
if s.pluginHost != nil {
|
||||
return s.pluginHost
|
||||
}
|
||||
if s.handlers != nil && s.handlers.ModelRouterHost != nil {
|
||||
return s.handlers.ModelRouterHost
|
||||
if s.handlers != nil {
|
||||
if host := s.handlers.ModelRouterHostSnapshot(); host != nil {
|
||||
return host
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -157,7 +157,8 @@ func rewriteClaudeDDModelInBody(rawJSON []byte) []byte {
|
||||
// Parameters:
|
||||
// - c: The Gin context for the request.
|
||||
func (h *ClaudeCodeAPIHandler) ClaudeModels(c *gin.Context) {
|
||||
disableCloaking := h.Cfg != nil && h.Cfg.ClaudeCode.DisableCloakingModelList
|
||||
cfg := h.CurrentConfig()
|
||||
disableCloaking := cfg != nil && cfg.ClaudeCode.DisableCloakingModelList
|
||||
h.WriteModelListResponse(c, h.HandlerType(), claudemodels.BuildResponse(h.Models(), disableCloaking))
|
||||
}
|
||||
|
||||
@@ -261,7 +262,7 @@ func (h *ClaudeCodeAPIHandler) handleStreamingResponse(c *gin.Context, rawJSON [
|
||||
flusher.Flush()
|
||||
return true
|
||||
}
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
defer stopKeepAlive()
|
||||
|
||||
dataChan, resolvedHeaders, errChan, okBootstrap := h.ExecuteStreamWithPreFTKeepAlive(
|
||||
@@ -451,7 +452,7 @@ func (h *ClaudeCodeAPIHandler) WriteErrorResponse(c *gin.Context, msg *interface
|
||||
c.Writer.Header().Add("Retry-After", value)
|
||||
}
|
||||
}
|
||||
if msg != nil && msg.Addon != nil && handlers.PassthroughHeadersEnabled(h.Cfg) {
|
||||
if msg != nil && msg.Addon != nil && handlers.PassthroughHeadersEnabled(h.CurrentConfig()) {
|
||||
for key, values := range msg.Addon {
|
||||
if len(values) == 0 || handlers.IsCPAReservedResponseHeader(key) {
|
||||
continue
|
||||
|
||||
@@ -225,7 +225,7 @@ func (h *GeminiAPIHandler) handleStreamGenerateContent(c *gin.Context, modelName
|
||||
var keepAliveC <-chan time.Time
|
||||
stopKeepAlive := func() {}
|
||||
if alt == "" {
|
||||
keepAliveC, stopKeepAlive = handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive = handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
}
|
||||
defer stopKeepAlive()
|
||||
|
||||
|
||||
@@ -151,7 +151,7 @@ func (h *GeminiAPIHandler) handleInteractionsStream(c *gin.Context, cliCtx conte
|
||||
handlers.WriteUpstreamHeaders(c.Writer.Header(), upstreamHeaders)
|
||||
committed = true
|
||||
}
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
defer stopKeepAlive()
|
||||
|
||||
var stream handlers.ModelExecutionStream
|
||||
|
||||
@@ -408,6 +408,8 @@ func setGenerateMetadata(meta map[string]any, rawJSON []byte) {
|
||||
// It holds a pool of clients to interact with the backend service and manages
|
||||
// load balancing, client selection, and configuration.
|
||||
type BaseAPIHandler struct {
|
||||
runtimeMu sync.RWMutex
|
||||
|
||||
// AuthManager manages auth lifecycle and execution in the new architecture.
|
||||
AuthManager *coreauth.Manager
|
||||
|
||||
@@ -444,7 +446,25 @@ func NewBaseAPIHandlers(cfg *config.SDKConfig, authManager *coreauth.Manager) *B
|
||||
// Parameters:
|
||||
// - clients: The new slice of AI service clients
|
||||
// - cfg: The new application configuration
|
||||
func (h *BaseAPIHandler) UpdateClients(cfg *config.SDKConfig) { h.Cfg = cfg }
|
||||
func (h *BaseAPIHandler) UpdateClients(cfg *config.SDKConfig) {
|
||||
if h == nil {
|
||||
return
|
||||
}
|
||||
h.runtimeMu.Lock()
|
||||
h.Cfg = cfg
|
||||
h.runtimeMu.Unlock()
|
||||
}
|
||||
|
||||
// CurrentConfig returns the current application configuration snapshot.
|
||||
func (h *BaseAPIHandler) CurrentConfig() *config.SDKConfig {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
h.runtimeMu.RLock()
|
||||
cfg := h.Cfg
|
||||
h.runtimeMu.RUnlock()
|
||||
return cfg
|
||||
}
|
||||
|
||||
// SetPluginHost configures the optional plugin interceptor host.
|
||||
func (h *BaseAPIHandler) SetPluginHost(host PluginInterceptorHost) {
|
||||
@@ -452,10 +472,11 @@ func (h *BaseAPIHandler) SetPluginHost(host PluginInterceptorHost) {
|
||||
return
|
||||
}
|
||||
if isNilPluginInterceptorHost(host) {
|
||||
h.PluginHost = nil
|
||||
return
|
||||
host = nil
|
||||
}
|
||||
h.runtimeMu.Lock()
|
||||
h.PluginHost = host
|
||||
h.runtimeMu.Unlock()
|
||||
}
|
||||
|
||||
// SetModelRouterHost configures the optional plugin model router host.
|
||||
@@ -464,10 +485,25 @@ func (h *BaseAPIHandler) SetModelRouterHost(host PluginModelRouterHost) {
|
||||
return
|
||||
}
|
||||
if isNilPluginModelRouterHost(host) {
|
||||
h.ModelRouterHost = nil
|
||||
return
|
||||
host = nil
|
||||
}
|
||||
h.runtimeMu.Lock()
|
||||
h.ModelRouterHost = host
|
||||
h.runtimeMu.Unlock()
|
||||
}
|
||||
|
||||
// ModelRouterHostSnapshot returns the configured plugin model router host snapshot.
|
||||
func (h *BaseAPIHandler) ModelRouterHostSnapshot() PluginModelRouterHost {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
h.runtimeMu.RLock()
|
||||
host := h.ModelRouterHost
|
||||
h.runtimeMu.RUnlock()
|
||||
if isNilPluginModelRouterHost(host) {
|
||||
return nil
|
||||
}
|
||||
return host
|
||||
}
|
||||
|
||||
func isNilPluginInterceptorHost(host PluginInterceptorHost) bool {
|
||||
@@ -596,7 +632,7 @@ func (h *BaseAPIHandler) GetContextWithCancel(handler interfaces.APIHandler, c *
|
||||
if c != nil {
|
||||
logging.SetResponseStatus(cancelCtx, c.Writer.Status())
|
||||
}
|
||||
if h.Cfg.RequestLog && len(params) == 1 {
|
||||
if cfg := h.CurrentConfig(); cfg != nil && cfg.RequestLog && len(params) == 1 {
|
||||
if captured, exists := c.Get(logging.APIResponseCapturedContextKey); exists {
|
||||
if capturedBool, ok := captured.(bool); ok && capturedBool {
|
||||
cancel()
|
||||
@@ -648,7 +684,7 @@ func (h *BaseAPIHandler) StartNonStreamingKeepAlive(c *gin.Context, ctx context.
|
||||
if h == nil || c == nil {
|
||||
return func() {}
|
||||
}
|
||||
interval := NonStreamingKeepAliveInterval(h.Cfg)
|
||||
interval := NonStreamingKeepAliveInterval(h.CurrentConfig())
|
||||
if interval <= 0 {
|
||||
return func() {}
|
||||
}
|
||||
|
||||
@@ -128,7 +128,7 @@ func (h *BaseAPIHandler) WriteErrorResponse(c *gin.Context, msg *interfaces.Erro
|
||||
c.Writer.Header().Add("Retry-After", value)
|
||||
}
|
||||
}
|
||||
if msg != nil && msg.Addon != nil && PassthroughHeadersEnabled(h.Cfg) {
|
||||
if msg != nil && msg.Addon != nil && PassthroughHeadersEnabled(h.CurrentConfig()) {
|
||||
for key, values := range msg.Addon {
|
||||
if len(values) == 0 || IsCPAReservedResponseHeader(key) {
|
||||
continue
|
||||
@@ -196,7 +196,7 @@ func writeDirectErrorResponse(c *gin.Context, status int, msg *interfaces.ErrorM
|
||||
}
|
||||
|
||||
func (h *BaseAPIHandler) LoggingAPIResponseError(ctx context.Context, err *interfaces.ErrorMessage) {
|
||||
if h.Cfg.RequestLog {
|
||||
if cfg := h.CurrentConfig(); cfg != nil && cfg.RequestLog {
|
||||
if ginContext, ok := ctx.Value("gin").(*gin.Context); ok {
|
||||
if apiResponseErrors, isExist := ginContext.Get("API_RESPONSE_ERROR"); isExist {
|
||||
if slicesAPIResponseError, isOk := apiResponseErrors.([]*interfaces.ErrorMessage); isOk {
|
||||
|
||||
@@ -112,7 +112,7 @@ func (h *BaseAPIHandler) executeWithAuthManagerFormats(ctx context.Context, entr
|
||||
executedReq, executedOpts := afterAuthCapture.apply(req, opts)
|
||||
ctx = enrichContextWithSessionHierarchy(ctx, executedOpts.Headers, executedReq.Payload, executedOpts.Metadata)
|
||||
rawResponseHeaders := cloneHeader(resp.Headers)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
responseHeaders := downstreamHeadersFromExecutor(rawResponseHeaders, passthroughHeaders)
|
||||
body, responseHeaders := h.applyResponseInterceptors(ctx, lifecycle.requestID(), responseProtocol, normalizedModel, originalRequestedModel, executedOpts, rawResponseHeaders, responseHeaders, executedOpts.OriginalRequest, executedReq.Payload, resp.Payload, http.StatusOK, execOptions.SkipInterceptorPluginID, passthroughHeaders)
|
||||
lifecycle.complete(pluginapi.RequestCompletionSucceeded, http.StatusOK, nil)
|
||||
@@ -182,7 +182,7 @@ func (h *BaseAPIHandler) executeCountWithAuthManager(ctx context.Context, handle
|
||||
executedReq, executedOpts := afterAuthCapture.apply(req, opts)
|
||||
ctx = enrichContextWithSessionHierarchy(ctx, executedOpts.Headers, executedReq.Payload, executedOpts.Metadata)
|
||||
rawResponseHeaders := cloneHeader(resp.Headers)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
responseHeaders := downstreamHeadersFromExecutor(rawResponseHeaders, passthroughHeaders)
|
||||
body, responseHeaders := h.applyResponseInterceptors(ctx, lifecycle.requestID(), handlerType, normalizedModel, originalRequestedModel, executedOpts, rawResponseHeaders, responseHeaders, executedOpts.OriginalRequest, executedReq.Payload, resp.Payload, http.StatusOK, execOptions.SkipInterceptorPluginID, passthroughHeaders)
|
||||
lifecycle.complete(pluginapi.RequestCompletionSucceeded, http.StatusOK, nil)
|
||||
@@ -232,7 +232,7 @@ func (h *BaseAPIHandler) executeWithPluginExecutor(ctx context.Context, entryPro
|
||||
reporter.EnsurePublished(execCtx)
|
||||
}
|
||||
rawResponseHeaders := cloneHeader(resp.Headers)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
responseHeaders := downstreamHeadersFromExecutor(rawResponseHeaders, passthroughHeaders)
|
||||
body, responseHeaders := h.applyResponseInterceptors(execCtx, lifecycle.requestID(), responseProtocol, modelName, originalRequestedModel, opts, rawResponseHeaders, responseHeaders, opts.OriginalRequest, req.Payload, resp.Payload, http.StatusOK, execOptions.SkipInterceptorPluginID, passthroughHeaders)
|
||||
lifecycle.complete(pluginapi.RequestCompletionSucceeded, http.StatusOK, nil)
|
||||
@@ -268,7 +268,7 @@ func (h *BaseAPIHandler) countWithPluginExecutor(ctx context.Context, handlerTyp
|
||||
return nil, nil, errMsg
|
||||
}
|
||||
rawResponseHeaders := cloneHeader(resp.Headers)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeaders := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
responseHeaders := downstreamHeadersFromExecutor(rawResponseHeaders, passthroughHeaders)
|
||||
body, responseHeaders := h.applyResponseInterceptors(ctx, lifecycle.requestID(), handlerType, modelName, originalRequestedModel, opts, rawResponseHeaders, responseHeaders, opts.OriginalRequest, req.Payload, resp.Payload, http.StatusOK, execOptions.SkipInterceptorPluginID, passthroughHeaders)
|
||||
lifecycle.complete(pluginapi.RequestCompletionSucceeded, http.StatusOK, nil)
|
||||
@@ -400,11 +400,19 @@ func (h *BaseAPIHandler) pluginExecutorHost() PluginExecutorHost {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
if executorHost, ok := h.ModelRouterHost.(PluginExecutorHost); ok && executorHost != nil {
|
||||
return executorHost
|
||||
h.runtimeMu.RLock()
|
||||
routerHost := h.ModelRouterHost
|
||||
pluginHost := h.PluginHost
|
||||
h.runtimeMu.RUnlock()
|
||||
if !isNilPluginModelRouterHost(routerHost) {
|
||||
if executorHost, ok := routerHost.(PluginExecutorHost); ok && executorHost != nil {
|
||||
return executorHost
|
||||
}
|
||||
}
|
||||
if executorHost, ok := h.PluginHost.(PluginExecutorHost); ok && executorHost != nil {
|
||||
return executorHost
|
||||
if !isNilPluginInterceptorHost(pluginHost) {
|
||||
if executorHost, ok := pluginHost.(PluginExecutorHost); ok && executorHost != nil {
|
||||
return executorHost
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -340,7 +340,13 @@ func (h *BaseAPIHandler) interceptorHost() PluginInterceptorHost {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
return h.PluginHost
|
||||
h.runtimeMu.RLock()
|
||||
host := h.PluginHost
|
||||
h.runtimeMu.RUnlock()
|
||||
if isNilPluginInterceptorHost(host) {
|
||||
return nil
|
||||
}
|
||||
return host
|
||||
}
|
||||
|
||||
func streamInterceptorsEnabled(host PluginInterceptorHost) bool {
|
||||
|
||||
@@ -301,14 +301,17 @@ func (h *BaseAPIHandler) modelRouterHost() PluginModelRouterHost {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
if !isNilPluginModelRouterHost(h.ModelRouterHost) {
|
||||
return h.ModelRouterHost
|
||||
h.runtimeMu.RLock()
|
||||
routerHost := h.ModelRouterHost
|
||||
pluginHost := h.PluginHost
|
||||
h.runtimeMu.RUnlock()
|
||||
if !isNilPluginModelRouterHost(routerHost) {
|
||||
return routerHost
|
||||
}
|
||||
host := h.interceptorHost()
|
||||
if host == nil {
|
||||
if isNilPluginInterceptorHost(pluginHost) {
|
||||
return nil
|
||||
}
|
||||
router, ok := host.(PluginModelRouterHost)
|
||||
router, ok := pluginHost.(PluginModelRouterHost)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -92,7 +92,7 @@ func (h *BaseAPIHandler) streamWithPluginExecutor(ctx context.Context, entryProt
|
||||
return nil, nil, errChan
|
||||
}
|
||||
|
||||
passthroughHeadersEnabled := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeadersEnabled := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
interceptorHost := h.interceptorHost()
|
||||
streamInterceptorsActive := streamInterceptorsEnabled(interceptorHost)
|
||||
rawStreamHeaders := cloneHeader(streamResult.Headers)
|
||||
@@ -375,7 +375,7 @@ func (h *BaseAPIHandler) executeStreamWithAuthManagerFormats(ctx context.Context
|
||||
if executedReq, executedOpts := executedRequest(); len(executedOpts.Headers) > 0 || len(executedReq.Payload) > 0 || len(executedOpts.Metadata) > 0 {
|
||||
ctx = enrichContextWithSessionHierarchy(ctx, executedOpts.Headers, executedReq.Payload, executedOpts.Metadata)
|
||||
}
|
||||
passthroughHeadersEnabled := executionPassthroughHeaders(h.Cfg, execOptions.InternalSource)
|
||||
passthroughHeadersEnabled := executionPassthroughHeaders(h.CurrentConfig(), execOptions.InternalSource)
|
||||
interceptorHost := h.interceptorHost()
|
||||
streamInterceptorsActive := streamInterceptorsEnabled(interceptorHost)
|
||||
// Resolve bootstrap retries and header initialization before returning so the
|
||||
@@ -539,7 +539,7 @@ func (h *BaseAPIHandler) executeStreamWithAuthManagerFormats(ctx context.Context
|
||||
}
|
||||
}
|
||||
|
||||
maxBootstrapRetries := StreamingBootstrapRetries(h.Cfg)
|
||||
maxBootstrapRetries := StreamingBootstrapRetries(h.CurrentConfig())
|
||||
if h.AuthManager.HomeEnabled() {
|
||||
maxBootstrapRetries = 0
|
||||
}
|
||||
|
||||
@@ -11,10 +11,15 @@ func (h *OpenAIAPIHandler) codexClientModelsResponse(clientVersion ...string) ma
|
||||
if len(clientVersion) > 0 {
|
||||
version = clientVersion[0]
|
||||
}
|
||||
optimizeMultiAgentV2 := h != nil && h.Cfg != nil && h.Cfg.Client.Codex.OptimizeMultiAgentV2
|
||||
optimizeMultiAgentV2 := false
|
||||
var applyPatchCapabilityForModel codexmodels.ApplyPatchCapabilityForModelFunc
|
||||
if h != nil && h.Cfg != nil && h.Cfg.Client.Codex.EnableApplyPatch {
|
||||
applyPatchCapabilityForModel = h.SupportsApplyPatchModel
|
||||
if h != nil {
|
||||
if cfg := h.CurrentConfig(); cfg != nil {
|
||||
optimizeMultiAgentV2 = cfg.Client.Codex.OptimizeMultiAgentV2
|
||||
if cfg.Client.Codex.EnableApplyPatch {
|
||||
applyPatchCapabilityForModel = h.SupportsApplyPatchModel
|
||||
}
|
||||
}
|
||||
}
|
||||
var toolSearchCapabilityForModel codexmodels.ToolSearchCapabilityForModelFunc
|
||||
if h != nil && responsesclaude.ClientToolSearchEnabled() {
|
||||
|
||||
@@ -507,7 +507,7 @@ func (h *OpenAIAPIHandler) handleStreamingResponse(c *gin.Context, rawJSON []byt
|
||||
flusher.Flush()
|
||||
return true
|
||||
}
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
defer stopKeepAlive()
|
||||
|
||||
dataChan, resolvedHeaders, errChan, okBootstrap := h.ExecuteStreamWithPreFTKeepAlive(
|
||||
@@ -666,7 +666,7 @@ func (h *OpenAIAPIHandler) handleCompletionsStreamingResponse(c *gin.Context, ra
|
||||
flusher.Flush()
|
||||
return true
|
||||
}
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
defer stopKeepAlive()
|
||||
|
||||
dataChan, resolvedHeaders, errChan, okBootstrap := h.ExecuteStreamWithPreFTKeepAlive(
|
||||
|
||||
@@ -79,7 +79,7 @@ func (h *OpenAIAPIHandler) newImagesStreamKeepAliveTicker() (func(), <-chan time
|
||||
if h == nil || h.BaseAPIHandler == nil {
|
||||
return func() {}, nil
|
||||
}
|
||||
keepAliveC, stop := handlers.NewStreamingKeepAliveTicker(handlers.StreamingKeepAliveInterval(h.Cfg))
|
||||
keepAliveC, stop := handlers.NewStreamingKeepAliveTicker(handlers.StreamingKeepAliveInterval(h.CurrentConfig()))
|
||||
return stop, keepAliveC
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ func (h *OpenAIAPIHandler) newImagesPreFirstTokenKeepAliveTicker() (func(), <-ch
|
||||
if h == nil || h.BaseAPIHandler == nil {
|
||||
return func() {}, nil
|
||||
}
|
||||
keepAliveC, stop := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stop := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
return stop, keepAliveC
|
||||
}
|
||||
|
||||
@@ -624,9 +624,11 @@ func parseBoolField(raw string, fallback bool) bool {
|
||||
}
|
||||
|
||||
func (h *OpenAIAPIHandler) ImagesGenerations(c *gin.Context) {
|
||||
if h != nil && h.BaseAPIHandler != nil && h.BaseAPIHandler.Cfg != nil && h.BaseAPIHandler.Cfg.DisableImageGeneration == internalconfig.DisableImageGenerationAll {
|
||||
c.AbortWithStatus(http.StatusNotFound)
|
||||
return
|
||||
if h != nil {
|
||||
if cfg := h.CurrentConfig(); cfg != nil && cfg.DisableImageGeneration == internalconfig.DisableImageGenerationAll {
|
||||
c.AbortWithStatus(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
rawJSON, err := handlers.ReadRequestBody(c)
|
||||
@@ -728,9 +730,11 @@ func (h *OpenAIAPIHandler) ImagesGenerations(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *OpenAIAPIHandler) ImagesEdits(c *gin.Context) {
|
||||
if h != nil && h.BaseAPIHandler != nil && h.BaseAPIHandler.Cfg != nil && h.BaseAPIHandler.Cfg.DisableImageGeneration == internalconfig.DisableImageGenerationAll {
|
||||
c.AbortWithStatus(http.StatusNotFound)
|
||||
return
|
||||
if h != nil {
|
||||
if cfg := h.CurrentConfig(); cfg != nil && cfg.DisableImageGeneration == internalconfig.DisableImageGenerationAll {
|
||||
c.AbortWithStatus(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
contentType := strings.ToLower(strings.TrimSpace(c.GetHeader("Content-Type")))
|
||||
|
||||
@@ -551,7 +551,11 @@ func (h *OpenAIResponsesAPIHandler) OpenAIResponsesModels(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *OpenAIResponsesAPIHandler) prepareCodexMultiAgentV2Tools(c *gin.Context, payload []byte) []byte {
|
||||
if h == nil || h.Cfg == nil {
|
||||
if h == nil {
|
||||
return payload
|
||||
}
|
||||
cfg := h.CurrentConfig()
|
||||
if cfg == nil {
|
||||
return payload
|
||||
}
|
||||
|
||||
@@ -570,7 +574,7 @@ func (h *OpenAIResponsesAPIHandler) prepareCodexMultiAgentV2Tools(c *gin.Context
|
||||
requestCtx,
|
||||
requestHeaders,
|
||||
payload,
|
||||
h.Cfg.Client.Codex.OptimizeMultiAgentV2,
|
||||
cfg.Client.Codex.OptimizeMultiAgentV2,
|
||||
homeEnabled,
|
||||
)
|
||||
if prepared && c != nil {
|
||||
@@ -580,7 +584,11 @@ func (h *OpenAIResponsesAPIHandler) prepareCodexMultiAgentV2Tools(c *gin.Context
|
||||
}
|
||||
|
||||
func (h *OpenAIResponsesAPIHandler) prepareCodexOrphanDelegation(c *gin.Context, payload []byte) []byte {
|
||||
if h == nil || h.Cfg == nil || !h.Cfg.CodexOrphanDelegationCompatibility {
|
||||
if h == nil {
|
||||
return payload
|
||||
}
|
||||
cfg := h.CurrentConfig()
|
||||
if cfg == nil || !cfg.CodexOrphanDelegationCompatibility {
|
||||
return payload
|
||||
}
|
||||
requestCtx := context.Background()
|
||||
@@ -754,7 +762,7 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ
|
||||
}
|
||||
return true
|
||||
}
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.Cfg)
|
||||
keepAliveC, stopKeepAlive := handlers.NewPreFirstTokenKeepAliveTicker(h.CurrentConfig())
|
||||
defer stopKeepAlive()
|
||||
|
||||
dataChan, resolvedHeaders, errChan, okBootstrap := h.ExecuteStreamWithPreFTKeepAlive(
|
||||
|
||||
@@ -278,14 +278,21 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) {
|
||||
c.Request = c.Request.WithContext(socketCtx)
|
||||
// The reader owns the client socket so an in-flight interrupt is not stuck
|
||||
// behind the response currently being forwarded.
|
||||
requestLogEnabled := h != nil && h.Cfg != nil && h.Cfg.RequestLog
|
||||
requestLogEnabled := false
|
||||
codexResponseSteering := false
|
||||
if h != nil {
|
||||
if cfg := h.CurrentConfig(); cfg != nil {
|
||||
requestLogEnabled = cfg.RequestLog
|
||||
codexResponseSteering = cfg.CodexResponseSteering
|
||||
}
|
||||
}
|
||||
wsTimelineLog := newWebsocketTimelineLog(requestLogEnabled, websocketTimelineSourceFromContext(c))
|
||||
localInterrupt := newResponsesLocalInterrupt()
|
||||
input, inputDone := readResponsesWebsocketInput(socketCtx, cancelSocket, conn, func(payload []byte) error {
|
||||
return h.forwardResponsesWebsocketInterrupt(socketCtx, passthroughSessionID, payload)
|
||||
}, localInterrupt, writer, wsTimelineLog)
|
||||
var duplexInput <-chan cliproxyexecutor.WebsocketInput
|
||||
if h != nil && h.Cfg != nil && h.Cfg.CodexResponseSteering {
|
||||
if codexResponseSteering {
|
||||
duplexInput = input
|
||||
}
|
||||
downstreamSessionKey := websocketDownstreamSessionKey(c.Request)
|
||||
|
||||
@@ -61,7 +61,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket(
|
||||
var keepAliveC <-chan time.Time
|
||||
keepAliveInterval := time.Duration(0)
|
||||
if h != nil {
|
||||
keepAliveInterval = handlers.StreamingKeepAliveInterval(h.Cfg)
|
||||
keepAliveInterval = handlers.StreamingKeepAliveInterval(h.CurrentConfig())
|
||||
}
|
||||
if opts.keepAliveInterval != nil {
|
||||
keepAliveInterval = *opts.keepAliveInterval
|
||||
|
||||
@@ -296,11 +296,13 @@ func firstPostForm(c *gin.Context, keys ...string) string {
|
||||
}
|
||||
|
||||
func (h *OpenAIAPIHandler) videoAuthBindingTTL() time.Duration {
|
||||
if h != nil && h.BaseAPIHandler != nil && h.Cfg != nil {
|
||||
raw := strings.TrimSpace(h.Cfg.VideoResultAuthCacheTTL)
|
||||
if raw != "" {
|
||||
if ttl, err := time.ParseDuration(raw); err == nil && ttl > 0 {
|
||||
return ttl
|
||||
if h != nil {
|
||||
if cfg := h.CurrentConfig(); cfg != nil {
|
||||
raw := strings.TrimSpace(cfg.VideoResultAuthCacheTTL)
|
||||
if raw != "" {
|
||||
if ttl, err := time.ParseDuration(raw); err == nil && ttl > 0 {
|
||||
return ttl
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -947,8 +949,10 @@ func (h *OpenAIAPIHandler) videoContentHTTPClient(c *gin.Context) *http.Client {
|
||||
ctx = c.Request.Context()
|
||||
}
|
||||
var cfg *config.Config
|
||||
if h != nil && h.BaseAPIHandler != nil && h.Cfg != nil {
|
||||
cfg = &config.Config{SDKConfig: *h.Cfg}
|
||||
if h != nil {
|
||||
if sdkCfg := h.CurrentConfig(); sdkCfg != nil {
|
||||
cfg = &config.Config{SDKConfig: *sdkCfg}
|
||||
}
|
||||
}
|
||||
return helps.NewProxyAwareHTTPClient(ctx, cfg, h.videoContentDownloadAuth(c), 0)
|
||||
}
|
||||
|
||||
@@ -148,7 +148,7 @@ func (h *BaseAPIHandler) ForwardStream(c *gin.Context, flusher http.Flusher, can
|
||||
}
|
||||
}
|
||||
|
||||
keepAliveInterval := StreamingKeepAliveInterval(h.Cfg)
|
||||
keepAliveInterval := StreamingKeepAliveInterval(h.CurrentConfig())
|
||||
if opts.KeepAliveInterval != nil {
|
||||
keepAliveInterval = *opts.KeepAliveInterval
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user