From 27fc3169bb4eb0509e3aba7dde4ab80286b0ae65 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sun, 26 Jul 2026 00:16:48 +0800 Subject: [PATCH] feat(runtime): improve executor binding logic and add config-based executor handling - Added `UsesConfig` to `XAIAutoExecutor` for determining config-based binding. - Refactored `Service` to support thread-safe executor registration, using a new mutex (`executorRegistrationMu`). - Improved executor rebind logic to prevent unnecessary replacements unless forced or required by config updates. - Updated handling of config updates to correctly replace stale XAI executors. - Enhanced the test suite for edge cases in executor binding and replacement for multiple providers. Closes: #4567 --- .../executor/xai_websockets_executor.go | 5 + sdk/cliproxy/service.go | 65 ++++--- .../service_codex_executor_binding_test.go | 164 +++++++++++++++++- sdk/cliproxy/service_plugin_executor_test.go | 2 +- 4 files changed, 206 insertions(+), 30 deletions(-) diff --git a/internal/runtime/executor/xai_websockets_executor.go b/internal/runtime/executor/xai_websockets_executor.go index 8a884fe72..a28cb9c62 100644 --- a/internal/runtime/executor/xai_websockets_executor.go +++ b/internal/runtime/executor/xai_websockets_executor.go @@ -1594,6 +1594,11 @@ func NewXAIAutoExecutor(cfg *config.Config) *XAIAutoExecutor { func (e *XAIAutoExecutor) Identifier() string { return "xai" } +// UsesConfig reports whether the executor was created for cfg. +func (e *XAIAutoExecutor) UsesConfig(cfg *config.Config) bool { + return e != nil && e.httpExec != nil && e.httpExec.cfg == cfg +} + func (e *XAIAutoExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if e == nil || e.httpExec == nil { return nil diff --git a/sdk/cliproxy/service.go b/sdk/cliproxy/service.go index c1259d884..231ffb68e 100644 --- a/sdk/cliproxy/service.go +++ b/sdk/cliproxy/service.go @@ -53,9 +53,10 @@ type Service struct { configUpdateMu sync.Mutex // configRuntimeMu orders side-effecting runtime application after config commits. - configRuntimeMu sync.Mutex - configSequence uint64 - appliedRoutingState *routingRuntimeState + configRuntimeMu sync.Mutex + executorRegistrationMu sync.Mutex + configSequence uint64 + appliedRoutingState *routingRuntimeState // configPath is the path to the configuration file. configPath string @@ -348,10 +349,13 @@ func (s *Service) syncPluginModelRuntime(ctx context.Context) { if ctx.Err() != nil { return } + s.cfgMu.RLock() + homeEnabled := s.cfg != nil && s.cfg.Home.Enabled + s.cfgMu.RUnlock() s.registerAvailableExecutors(ctx, executorRegistrationOptions{ - includeBaseline: s.cfg != nil && s.cfg.Home.Enabled, + includeBaseline: homeEnabled, includePlugins: true, - forceReplaceAuths: true, + forceReplaceAuths: false, auths: s.coreManager.List(), }) s.refreshPluginModelRegistrations(ctx) @@ -1036,7 +1040,7 @@ func (c *openAICompatibilityRegistrationCache) lookup(compatName string) (*openA return entry, ok } -func (s *Service) hasNativeOpenAICompatExecutorConfig(a *coreauth.Auth, providerKey string) bool { +func (s *Service) hasNativeOpenAICompatExecutorConfig(a *coreauth.Auth, providerKey string, cfg *config.Config) bool { if a == nil { return false } @@ -1052,7 +1056,7 @@ func (s *Service) hasNativeOpenAICompatExecutorConfig(a *coreauth.Auth, provider if strings.EqualFold(strings.TrimSpace(a.Provider), "openai-compatibility") { return true } - if s == nil || s.cfg == nil { + if s == nil || cfg == nil { return false } @@ -1069,8 +1073,8 @@ func (s *Service) hasNativeOpenAICompatExecutorConfig(a *coreauth.Auth, provider candidates = append(candidates, strings.ToLower(provider)) } - for i := range s.cfg.OpenAICompatibility { - compat := &s.cfg.OpenAICompatibility[i] + for i := range cfg.OpenAICompatibility { + compat := &cfg.OpenAICompatibility[i] if compat.Disabled { continue } @@ -1130,10 +1134,15 @@ func (s *Service) registerAvailableExecutors(ctx context.Context, opts executorR if ctx == nil { ctx = context.Background() } + s.executorRegistrationMu.Lock() + defer s.executorRegistrationMu.Unlock() + if ctx.Err() != nil { + return + } // Keep all Service-owned executor registration paths here so native, Home, // auth-derived, and plugin executors stay in the same binding order. if opts.includeBaseline { - s.registerExecutorsForAuths(baselineExecutorAuths(), true) + s.registerExecutorsForAuths(baselineExecutorAuths(), opts.forceReplaceAuths) } if len(opts.auths) > 0 { s.registerExecutorsForAuths(opts.auths, opts.forceReplaceAuths) @@ -1187,6 +1196,9 @@ func (s *Service) registerExecutorForAuth(a *coreauth.Auth, forceReplace bool) { if s == nil || s.coreManager == nil || a == nil { return } + s.cfgMu.RLock() + cfg := s.cfg + s.cfgMu.RUnlock() if strings.EqualFold(strings.TrimSpace(a.Provider), "codex") { if !forceReplace { existingExecutor, hasExecutor := s.coreManager.Executor("codex") @@ -1197,7 +1209,7 @@ func (s *Service) registerExecutorForAuth(a *coreauth.Auth, forceReplace bool) { } } } - s.coreManager.RegisterExecutor(executor.NewCodexAutoExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewCodexAutoExecutor(cfg)) return } // Skip disabled auth entries when (re)binding executors. @@ -1220,29 +1232,38 @@ func (s *Service) registerExecutorForAuth(a *coreauth.Auth, forceReplace bool) { } } } - s.coreManager.RegisterExecutor(executor.NewOpenAICompatExecutor(compatProviderKey, s.cfg)) + s.coreManager.RegisterExecutor(executor.NewOpenAICompatExecutor(compatProviderKey, cfg)) return } switch strings.ToLower(a.Provider) { case constant.Gemini: - s.coreManager.RegisterExecutor(executor.NewGeminiExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewGeminiExecutor(cfg)) case constant.GeminiInteractions: - s.coreManager.RegisterExecutor(executor.NewGeminiInteractionsExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewGeminiInteractionsExecutor(cfg)) case "vertex": - s.coreManager.RegisterExecutor(executor.NewGeminiVertexExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewGeminiVertexExecutor(cfg)) case "aistudio": if s.wsGateway != nil { - s.coreManager.RegisterExecutor(executor.NewAIStudioExecutor(s.cfg, a.ID, s.wsGateway)) + s.coreManager.RegisterExecutor(executor.NewAIStudioExecutor(cfg, a.ID, s.wsGateway)) } return case "antigravity": - s.coreManager.RegisterExecutor(executor.NewAntigravityExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewAntigravityExecutor(cfg)) case "claude": - s.coreManager.RegisterExecutor(executor.NewClaudeExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewClaudeExecutor(cfg)) case "kimi": - s.coreManager.RegisterExecutor(executor.NewKimiExecutor(s.cfg)) + s.coreManager.RegisterExecutor(executor.NewKimiExecutor(cfg)) case "xai": - s.coreManager.RegisterExecutor(executor.NewXAIAutoExecutor(s.cfg)) + if !forceReplace { + existingExecutor, hasExecutor := s.coreManager.Executor("xai") + if hasExecutor { + existingXAIAutoExecutor, isXAIAutoExecutor := existingExecutor.(*executor.XAIAutoExecutor) + if isXAIAutoExecutor && existingXAIAutoExecutor.UsesConfig(cfg) { + return + } + } + } + s.coreManager.RegisterExecutor(executor.NewXAIAutoExecutor(cfg)) default: providerKey := strings.ToLower(strings.TrimSpace(a.Provider)) if providerKey == "" { @@ -1250,7 +1271,7 @@ func (s *Service) registerExecutorForAuth(a *coreauth.Auth, forceReplace bool) { } if s.pluginHost != nil && s.pluginHost.HasExecutorCandidateProvider(providerKey) && - !s.hasNativeOpenAICompatExecutorConfig(a, providerKey) { + !s.hasNativeOpenAICompatExecutorConfig(a, providerKey, cfg) { s.unregisterOpenAICompatExecutor(providerKey) return } @@ -1261,7 +1282,7 @@ func (s *Service) registerExecutorForAuth(a *coreauth.Auth, forceReplace bool) { } } } - s.coreManager.RegisterExecutor(executor.NewOpenAICompatExecutor(providerKey, s.cfg)) + s.coreManager.RegisterExecutor(executor.NewOpenAICompatExecutor(providerKey, cfg)) } } diff --git a/sdk/cliproxy/service_codex_executor_binding_test.go b/sdk/cliproxy/service_codex_executor_binding_test.go index 0cd399ef2..7de704ffa 100644 --- a/sdk/cliproxy/service_codex_executor_binding_test.go +++ b/sdk/cliproxy/service_codex_executor_binding_test.go @@ -1,11 +1,16 @@ package cliproxy import ( + "context" "testing" + "github.com/router-for-me/CLIProxyAPI/v7/internal/pluginhost" "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor" + "github.com/router-for-me/CLIProxyAPI/v7/internal/watcher" + sdkAuth "github.com/router-for-me/CLIProxyAPI/v7/sdk/auth" coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" "github.com/router-for-me/CLIProxyAPI/v7/sdk/config" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" ) func TestEnsureExecutorsForAuth_CodexDoesNotReplaceInNormalMode(t *testing.T) { @@ -64,7 +69,77 @@ func TestEnsureExecutorsForAuthWithMode_CodexForceReplace(t *testing.T) { } } -func TestEnsureExecutorsForAuth_XAIBindsAutoExecutor(t *testing.T) { +func TestSyncPluginModelRuntime_UnrelatedAuthDoesNotReplaceWebsocketExecutor(t *testing.T) { + testCases := []struct { + name string + provider string + homeEnabled bool + }{ + {name: "codex standard mode", provider: "codex"}, + {name: "codex home mode", provider: "codex", homeEnabled: true}, + {name: "xai standard mode", provider: "xai"}, + {name: "xai home mode", provider: "xai", homeEnabled: true}, + } + + for _, tt := range testCases { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + cfg := &config.Config{} + cfg.Home.Enabled = tt.homeEnabled + service := &Service{ + cfg: cfg, + coreManager: coreauth.NewManager(nil, nil, nil), + pluginHost: pluginhost.New(), + } + providerAuth := &coreauth.Auth{ + ID: tt.provider + "-auth", + Provider: tt.provider, + Status: coreauth.StatusActive, + } + unrelatedAuth := &coreauth.Auth{ + ID: "unrelated-auth", + Provider: "claude", + Status: coreauth.StatusActive, + } + t.Cleanup(func() { + GlobalModelRegistry().UnregisterClient(providerAuth.ID) + GlobalModelRegistry().UnregisterClient(unrelatedAuth.ID) + sdkAuth.RegisterPluginAuthParser(nil) + sdktranslator.SetPluginHooks(nil) + }) + + if _, errRegister := service.coreManager.Register(ctx, providerAuth); errRegister != nil { + t.Fatalf("register %s auth: %v", tt.provider, errRegister) + } + if _, errRegister := service.coreManager.Register(ctx, unrelatedAuth); errRegister != nil { + t.Fatalf("register unrelated auth: %v", errRegister) + } + service.ensureExecutorsForAuth(providerAuth) + firstExecutor, okFirst := service.coreManager.Executor(tt.provider) + if !okFirst || firstExecutor == nil { + t.Fatalf("expected %s executor before plugin model sync", tt.provider) + } + + updatedAuth := unrelatedAuth.Clone() + updatedAuth.Label = "updated unrelated auth" + service.handleAuthUpdate(ctx, watcher.AuthUpdate{ + Action: watcher.AuthUpdateActionModify, + ID: updatedAuth.ID, + Auth: updatedAuth, + }) + + secondExecutor, okSecond := service.coreManager.Executor(tt.provider) + if !okSecond || secondExecutor == nil { + t.Fatalf("expected %s executor after plugin model sync", tt.provider) + } + if firstExecutor != secondExecutor { + t.Fatalf("expected unrelated auth sync to preserve the %s executor", tt.provider) + } + }) + } +} + +func TestEnsureExecutorsForAuth_XAIDoesNotReplaceInNormalMode(t *testing.T) { service := &Service{ cfg: &config.Config{}, coreManager: coreauth.NewManager(nil, nil, nil), @@ -76,12 +151,87 @@ func TestEnsureExecutorsForAuth_XAIBindsAutoExecutor(t *testing.T) { } service.ensureExecutorsForAuth(auth) - - gotExecutor, ok := service.coreManager.Executor("xai") - if !ok || gotExecutor == nil { - t.Fatal("expected xai executor after bind") + firstExecutor, okFirst := service.coreManager.Executor("xai") + if !okFirst || firstExecutor == nil { + t.Fatal("expected xai executor after first bind") } - if _, ok := gotExecutor.(*executor.XAIAutoExecutor); !ok { - t.Fatalf("xai executor type = %T, want *executor.XAIAutoExecutor", gotExecutor) + if _, isXAIAutoExecutor := firstExecutor.(*executor.XAIAutoExecutor); !isXAIAutoExecutor { + t.Fatalf("xai executor type = %T, want *executor.XAIAutoExecutor", firstExecutor) + } + + service.ensureExecutorsForAuth(auth) + secondExecutor, okSecond := service.coreManager.Executor("xai") + if !okSecond || secondExecutor == nil { + t.Fatal("expected xai executor after second bind") + } + if firstExecutor != secondExecutor { + t.Fatal("expected xai executor to stay unchanged in normal mode") + } +} + +func TestEnsureExecutorsForAuthWithMode_XAIForceReplace(t *testing.T) { + service := &Service{ + cfg: &config.Config{}, + coreManager: coreauth.NewManager(nil, nil, nil), + } + auth := &coreauth.Auth{ + ID: "xai-auth-2", + Provider: "xai", + Status: coreauth.StatusActive, + } + + service.ensureExecutorsForAuth(auth) + firstExecutor, okFirst := service.coreManager.Executor("xai") + if !okFirst || firstExecutor == nil { + t.Fatal("expected xai executor after first bind") + } + + service.ensureExecutorsForAuthWithMode(auth, true) + secondExecutor, okSecond := service.coreManager.Executor("xai") + if !okSecond || secondExecutor == nil { + t.Fatal("expected xai executor after forced rebind") + } + if firstExecutor == secondExecutor { + t.Fatal("expected xai executor replacement in force mode") + } + if _, isXAIAutoExecutor := secondExecutor.(*executor.XAIAutoExecutor); !isXAIAutoExecutor { + t.Fatalf("xai executor type = %T, want *executor.XAIAutoExecutor", secondExecutor) + } +} + +func TestEnsureExecutorsForAuth_XAIReplacesExecutorAfterConfigUpdate(t *testing.T) { + service := &Service{ + cfg: &config.Config{}, + coreManager: coreauth.NewManager(nil, nil, nil), + pluginHost: pluginhost.New(), + } + t.Cleanup(func() { + sdkAuth.RegisterPluginAuthParser(nil) + sdktranslator.SetPluginHooks(nil) + }) + auth := &coreauth.Auth{ + ID: "xai-auth-config-update", + Provider: "xai", + Status: coreauth.StatusActive, + } + + service.ensureExecutorsForAuth(auth) + firstExecutor, okFirst := service.coreManager.Executor("xai") + if !okFirst || firstExecutor == nil { + t.Fatal("expected xai executor before config update") + } + + service.applyWatcherConfigUpdate(&config.Config{}) + service.ensureExecutorsForAuth(auth) + + secondExecutor, okSecond := service.coreManager.Executor("xai") + if !okSecond || secondExecutor == nil { + t.Fatal("expected xai executor after config update") + } + if firstExecutor == secondExecutor { + t.Fatal("expected stale xai executor replacement after config update") + } + if _, isXAIAutoExecutor := secondExecutor.(*executor.XAIAutoExecutor); !isXAIAutoExecutor { + t.Fatalf("xai executor type = %T, want *executor.XAIAutoExecutor", secondExecutor) } } diff --git a/sdk/cliproxy/service_plugin_executor_test.go b/sdk/cliproxy/service_plugin_executor_test.go index c751cbe25..a6ed15ec4 100644 --- a/sdk/cliproxy/service_plugin_executor_test.go +++ b/sdk/cliproxy/service_plugin_executor_test.go @@ -50,7 +50,7 @@ func TestHasNativeOpenAICompatExecutorConfig(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := service.hasNativeOpenAICompatExecutorConfig(tt.auth, tt.providerKey) + got := service.hasNativeOpenAICompatExecutorConfig(tt.auth, tt.providerKey, service.cfg) if got != tt.want { t.Fatalf("hasNativeOpenAICompatExecutorConfig() = %v, want %v", got, tt.want) }