History

fix(auth): drop stale auth updates using monotonic watcher revisions

- Track monotonic watcher revisions across persisted auth updates to filter out out-of-order events. - Validate registration epochs before applying auth updates and deletions to prevent stale state overwrites. - Synchronize auth status patches through post-persist hooks using detached background contexts. - Guard auth status modifications with a dedicated handler mutex.

Luis Pater committed 2 months ago

Commit d1702fd1 parent 5c80a0112 files · +1231 −23Browse files

Unified · Split

12 files
  1. internal/api/handlers/management/auth_files_fields.go +29 −2
  2. internal/api/handlers/management/auth_files_status_sync_test.go +351 −0
  3. internal/api/handlers/management/handler.go +1 −0
  4. internal/api/server.go +8 −0
  5. internal/watcher/dispatcher.go +16 −8
  6. internal/watcher/watcher.go +18 −0
  7. sdk/cliproxy/builder.go +9 −3
  8. sdk/cliproxy/service.go +2 −0
  9. sdk/cliproxy/service_auth.go +51 −1
  10. sdk/cliproxy/service_auth_sync_test.go +710 −0
  11. sdk/cliproxy/types.go +33 −9
  12. sdk/cliproxy/watcher.go +3 −0
internal/api/handlers/management/auth_files_fields.go+29−2
@@ -20,6 +20,7 @@ import ( · 19 unmodified lines
20 "github.com/router-for-me/CLIProxyAPI/v7/internal/watcher/synthesizer"20 "github.com/router-for-me/CLIProxyAPI/v7/internal/watcher/synthesizer"
21 sdkAuth "github.com/router-for-me/CLIProxyAPI/v7/sdk/auth"21 sdkAuth "github.com/router-for-me/CLIProxyAPI/v7/sdk/auth"
22 coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"22 coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
23 log "github.com/sirupsen/logrus"
23)24)
24 25
25// PatchAuthFileStatus toggles the disabled state of an auth file26// PatchAuthFileStatus toggles the disabled state of an auth file
@@ -50,6 +51,9 @@ func (h *Handler) PatchAuthFileStatus(c *gin.Context) { · 24 unmodified lines
50 return51 return
51 }52 }
52 53
54 h.authStatusMu.Lock()
55 defer h.authStatusMu.Unlock()
56
53 ctx := c.Request.Context()57 ctx := c.Request.Context()
54 58
55 targetAuth, _ := h.lookupAuthFile(name, authIndex)59 targetAuth, _ := h.lookupAuthFile(name, authIndex)
@@ -108,10 +112,22 @@ func (h *Handler) PatchAuthFileStatus(c *gin.Context) { · 52 unmodified lines
108 }112 }
109 113
110 applyAuthDisabledState(targetAuth, *req.Disabled)114 applyAuthDisabledState(targetAuth, *req.Disabled)
111 if _, err := h.authManager.Update(ctx, targetAuth); err != nil {115 updatedAuth, err := h.authManager.Update(ctx, targetAuth)
116 if err != nil {
112 c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to update auth: %v", err)})117 c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to update auth: %v", err)})
113 return118 return
114 }119 }
120 if h.postAuthPersistHook != nil {
121 hookAuth := updatedAuth
122 if hookAuth == nil {
123 hookAuth = targetAuth
124 }
125 if errHook := h.postAuthPersistHook(ctx, hookAuth); errHook != nil {
126 log.Errorf("post-auth persist hook failed for status update on %s: %v", targetAuth.ID, errHook)
127 c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to synchronize auth runtime: %v", errHook)})
128 return
129 }
130 }
115 131
116 c.JSON(http.StatusOK, gin.H{"status": "ok", "disabled": *req.Disabled})132 c.JSON(http.StatusOK, gin.H{"status": "ok", "disabled": *req.Disabled})
117}133}
@@ -146,9 +162,20 @@ func (h *Handler) patchPluginVirtualSourceStatus(ctx context.Context, targetAuth · 28 unmodified lines
146 }162 }
147 applyAuthDisabledState(auth, disabled)163 applyAuthDisabledState(auth, disabled)
148 auth.UpdatedAt = now164 auth.UpdatedAt = now
149 if _, errUpdate := h.authManager.Update(ctx, auth); errUpdate != nil {165 updated, errUpdate := h.authManager.Update(ctx, auth)
166 if errUpdate != nil {
150 return fmt.Errorf("failed to update auth %s: %w", auth.ID, errUpdate)167 return fmt.Errorf("failed to update auth %s: %w", auth.ID, errUpdate)
151 }168 }
169 if h.postAuthPersistHook != nil {
170 hookAuth := updated
171 if hookAuth == nil {
172 hookAuth = auth
173 }
174 if errHook := h.postAuthPersistHook(ctx, hookAuth); errHook != nil {
175 log.Errorf("post-auth persist hook failed for plugin virtual auth %s: %v", auth.ID, errHook)
176 return fmt.Errorf("failed to synchronize plugin virtual auth %s: %w", auth.ID, errHook)
177 }
178 }
152 }179 }
153 return nil180 return nil
154}181}
internal/api/handlers/management/auth_files_status_sync_test.go+351−0
@@ -0,0 +1,351 @@
1package management
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "net/http"
8 "net/http/httptest"
9 "os"
10 "path/filepath"
11 "strings"
12 "testing"
13
14 "github.com/gin-gonic/gin"
15 "github.com/router-for-me/CLIProxyAPI/v7/internal/config"
16 "github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
17 coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
18)
19
20func TestPatchAuthFileStatusInvokesPostAuthPersistHook(t *testing.T) {
21 t.Setenv("MANAGEMENT_PASSWORD", "")
22
23 authDir := t.TempDir()
24 fileName := "codex-test.json"
25 filePath := filepath.Join(authDir, fileName)
26 if errWrite := os.WriteFile(filePath, []byte(`{"type":"codex","disabled":false}`), 0o600); errWrite != nil {
27 t.Fatalf("write auth file: %v", errWrite)
28 }
29
30 manager := coreauth.NewManager(nil, nil, nil)
31 auth := &coreauth.Auth{
32 ID: fileName,
33 FileName: fileName,
34 Provider: "codex",
35 Status: coreauth.StatusActive,
36 Attributes: map[string]string{
37 "path": filePath,
38 },
39 }
40 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
41 t.Fatalf("register auth: %v", errRegister)
42 }
43
44 h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
45
46 var hookCalls []*coreauth.Auth
47 h.SetPostAuthPersistHook(func(_ context.Context, updated *coreauth.Auth) error {
48 if updated != nil {
49 hookCalls = append(hookCalls, updated.Clone())
50 }
51 return nil
52 })
53
54 // 1. Disable the credential
55 rec := httptest.NewRecorder()
56 ctx, _ := gin.CreateTestContext(rec)
57 req := httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"codex-test.json","disabled":true}`))
58 req.Header.Set("Content-Type", "application/json")
59 ctx.Request = req
60
61 h.PatchAuthFileStatus(ctx)
62 if rec.Code != http.StatusOK {
63 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusOK, rec.Body.String())
64 }
65
66 if len(hookCalls) != 1 {
67 t.Fatalf("expected 1 hook call after disable, got %d", len(hookCalls))
68 }
69 if !hookCalls[0].Disabled {
70 t.Fatalf("expected hook call auth to be disabled, got %+v", hookCalls[0])
71 }
72
73 // 2. Re-enable the credential
74 rec = httptest.NewRecorder()
75 ctx, _ = gin.CreateTestContext(rec)
76 req = httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"codex-test.json","disabled":false}`))
77 req.Header.Set("Content-Type", "application/json")
78 ctx.Request = req
79
80 h.PatchAuthFileStatus(ctx)
81 if rec.Code != http.StatusOK {
82 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusOK, rec.Body.String())
83 }
84
85 if len(hookCalls) != 2 {
86 t.Fatalf("expected 2 hook calls after re-enable, got %d", len(hookCalls))
87 }
88 if hookCalls[1].Disabled {
89 t.Fatalf("expected second hook call auth to be enabled, got %+v", hookCalls[1])
90 }
91}
92
93func TestPatchAuthFileStatusRestoresModelsViaSyncHook(t *testing.T) {
94 t.Setenv("MANAGEMENT_PASSWORD", "")
95
96 authDir := t.TempDir()
97 fileName := "codex-models.json"
98 filePath := filepath.Join(authDir, fileName)
99 if errWrite := os.WriteFile(filePath, []byte(`{"type":"codex","disabled":false}`), 0o600); errWrite != nil {
100 t.Fatalf("write auth file: %v", errWrite)
101 }
102
103 reg := registry.GetGlobalRegistry()
104 authID := fileName
105 t.Cleanup(func() {
106 reg.UnregisterClient(authID)
107 })
108
109 manager := coreauth.NewManager(nil, nil, nil)
110 auth := &coreauth.Auth{
111 ID: authID,
112 FileName: fileName,
113 Provider: "codex",
114 Status: coreauth.StatusActive,
115 Attributes: map[string]string{
116 "path": filePath,
117 },
118 }
119 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
120 t.Fatalf("register auth: %v", errRegister)
121 }
122
123 // Initial model registration
124 reg.RegisterClient(authID, "codex", []*registry.ModelInfo{
125 {ID: "gpt-6-astra", DisplayName: "GPT-6 Astra"},
126 })
127
128 h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
129 // Wire postAuthPersistHook to synchronize registry models as runtimeAuthSyncHook does
130 h.SetPostAuthPersistHook(func(_ context.Context, a *coreauth.Auth) error {
131 if a == nil || a.ID == "" {
132 return nil
133 }
134 if a.Disabled {
135 reg.UnregisterClient(a.ID)
136 } else {
137 reg.RegisterClient(a.ID, a.Provider, []*registry.ModelInfo{
138 {ID: "gpt-6-astra", DisplayName: "GPT-6 Astra"},
139 })
140 }
141 return nil
142 })
143
144 // Step 1: Confirm models exist before disabling
145 rec := httptest.NewRecorder()
146 ctx, _ := gin.CreateTestContext(rec)
147 ctx.Request = httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
148 h.GetAuthFileModels(ctx)
149 if rec.Code != http.StatusOK {
150 t.Fatalf("GetAuthFileModels initial status = %d, body = %s", rec.Code, rec.Body.String())
151 }
152 var resp struct {
153 Models []map[string]any `json:"models"`
154 }
155 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) != 1 {
156 t.Fatalf("expected 1 initial model, got: %s", rec.Body.String())
157 }
158
159 // Step 2: Disable the credential
160 rec = httptest.NewRecorder()
161 ctx, _ = gin.CreateTestContext(rec)
162 req := httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"codex-models.json","disabled":true}`))
163 req.Header.Set("Content-Type", "application/json")
164 ctx.Request = req
165 h.PatchAuthFileStatus(ctx)
166 if rec.Code != http.StatusOK {
167 t.Fatalf("PatchAuthFileStatus disable failed: %s", rec.Body.String())
168 }
169
170 rec = httptest.NewRecorder()
171 ctx, _ = gin.CreateTestContext(rec)
172 ctx.Request = httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
173 h.GetAuthFileModels(ctx)
174 resp.Models = nil
175 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) != 0 {
176 t.Fatalf("expected 0 models after disable, got: %s", rec.Body.String())
177 }
178
179 // Step 3: Re-enable the credential
180 rec = httptest.NewRecorder()
181 ctx, _ = gin.CreateTestContext(rec)
182 req = httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"codex-models.json","disabled":false}`))
183 req.Header.Set("Content-Type", "application/json")
184 ctx.Request = req
185 h.PatchAuthFileStatus(ctx)
186 if rec.Code != http.StatusOK {
187 t.Fatalf("PatchAuthFileStatus re-enable failed: %s", rec.Body.String())
188 }
189
190 // Step 4: Confirm models are restored without restart
191 rec = httptest.NewRecorder()
192 ctx, _ = gin.CreateTestContext(rec)
193 ctx.Request = httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
194 h.GetAuthFileModels(ctx)
195 resp.Models = nil
196 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) != 1 {
197 t.Fatalf("expected 1 model after re-enable, got: %s", rec.Body.String())
198 }
199}
200
201func TestPatchPluginVirtualSourceStatusInvokesPostAuthPersistHook(t *testing.T) {
202 t.Setenv("MANAGEMENT_PASSWORD", "")
203
204 authDir := t.TempDir()
205 fileName := "source-sync.json"
206 filePath := filepath.Join(authDir, fileName)
207 if errWrite := os.WriteFile(filePath, []byte(`{"type":"gemini-cli","disabled":false}`), 0o600); errWrite != nil {
208 t.Fatalf("write source auth file: %v", errWrite)
209 }
210
211 manager := coreauth.NewManager(nil, nil, nil)
212 for _, id := range []string{"source-sync.json", "virtual-project-a", "virtual-project-b"} {
213 auth := pluginVirtualAuthForTest(authDir, fileName, id)
214 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
215 t.Fatalf("register virtual auth %s: %v", id, errRegister)
216 }
217 }
218
219 h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
220
221 var hookCalls []*coreauth.Auth
222 h.SetPostAuthPersistHook(func(_ context.Context, updated *coreauth.Auth) error {
223 if updated != nil {
224 hookCalls = append(hookCalls, updated.Clone())
225 }
226 return nil
227 })
228
229 // Disable the source file
230 rec := httptest.NewRecorder()
231 ctx, _ := gin.CreateTestContext(rec)
232 req := httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"source-sync.json","disabled":true}`))
233 req.Header.Set("Content-Type", "application/json")
234 ctx.Request = req
235
236 h.PatchAuthFileStatus(ctx)
237 if rec.Code != http.StatusOK {
238 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusOK, rec.Body.String())
239 }
240
241 if len(hookCalls) != 3 {
242 t.Fatalf("expected 3 hook calls for 3 virtual auths on disable, got %d", len(hookCalls))
243 }
244 for _, call := range hookCalls {
245 if !call.Disabled {
246 t.Fatalf("expected virtual auth %s to be disabled in hook call", call.ID)
247 }
248 }
249
250 // Re-enable the source file
251 hookCalls = nil
252 rec = httptest.NewRecorder()
253 ctx, _ = gin.CreateTestContext(rec)
254 req = httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"source-sync.json","disabled":false}`))
255 req.Header.Set("Content-Type", "application/json")
256 ctx.Request = req
257
258 h.PatchAuthFileStatus(ctx)
259 if rec.Code != http.StatusOK {
260 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusOK, rec.Body.String())
261 }
262
263 if len(hookCalls) != 3 {
264 t.Fatalf("expected 3 hook calls for 3 virtual auths on re-enable, got %d", len(hookCalls))
265 }
266 for _, call := range hookCalls {
267 if call.Disabled {
268 t.Fatalf("expected virtual auth %s to be enabled in hook call", call.ID)
269 }
270 }
271}
272
273func TestPatchAuthFileStatusHookErrorReturns500(t *testing.T) {
274 t.Setenv("MANAGEMENT_PASSWORD", "")
275
276 authDir := t.TempDir()
277 fileName := "codex-hook-err.json"
278 filePath := filepath.Join(authDir, fileName)
279 if errWrite := os.WriteFile(filePath, []byte(`{"type":"codex","disabled":false}`), 0o600); errWrite != nil {
280 t.Fatalf("write auth file: %v", errWrite)
281 }
282
283 manager := coreauth.NewManager(nil, nil, nil)
284 auth := &coreauth.Auth{
285 ID: fileName,
286 FileName: fileName,
287 Provider: "codex",
288 Status: coreauth.StatusActive,
289 Attributes: map[string]string{
290 "path": filePath,
291 },
292 }
293 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
294 t.Fatalf("register auth: %v", errRegister)
295 }
296
297 h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
298 h.SetPostAuthPersistHook(func(_ context.Context, _ *coreauth.Auth) error {
299 return errors.New("simulated sync hook failure")
300 })
301
302 rec := httptest.NewRecorder()
303 ctx, _ := gin.CreateTestContext(rec)
304 req := httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"codex-hook-err.json","disabled":true}`))
305 req.Header.Set("Content-Type", "application/json")
306 ctx.Request = req
307
308 h.PatchAuthFileStatus(ctx)
309 if rec.Code != http.StatusInternalServerError {
310 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusInternalServerError, rec.Body.String())
311 }
312 if !strings.Contains(rec.Body.String(), "simulated sync hook failure") {
313 t.Fatalf("body = %s, want simulated sync hook failure error message", rec.Body.String())
314 }
315}
316
317func TestPatchPluginVirtualSourceStatusHookErrorReturnsError(t *testing.T) {
318 t.Setenv("MANAGEMENT_PASSWORD", "")
319
320 authDir := t.TempDir()
321 fileName := "source-hook-err.json"
322 filePath := filepath.Join(authDir, fileName)
323 if errWrite := os.WriteFile(filePath, []byte(`{"type":"gemini-cli","disabled":false}`), 0o600); errWrite != nil {
324 t.Fatalf("write source auth file: %v", errWrite)
325 }
326
327 manager := coreauth.NewManager(nil, nil, nil)
328 auth := pluginVirtualAuthForTest(authDir, fileName, "source-hook-err.json")
329 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
330 t.Fatalf("register virtual auth: %v", errRegister)
331 }
332
333 h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
334 h.SetPostAuthPersistHook(func(_ context.Context, _ *coreauth.Auth) error {
335 return errors.New("plugin sync failed")
336 })
337
338 rec := httptest.NewRecorder()
339 ctx, _ := gin.CreateTestContext(rec)
340 req := httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"source-hook-err.json","disabled":true}`))
341 req.Header.Set("Content-Type", "application/json")
342 ctx.Request = req
343
344 h.PatchAuthFileStatus(ctx)
345 if rec.Code != http.StatusInternalServerError {
346 t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusInternalServerError, rec.Body.String())
347 }
348 if !strings.Contains(rec.Body.String(), "plugin sync failed") {
349 t.Fatalf("body = %s, want plugin sync failed error message", rec.Body.String())
350 }
351}
internal/api/handlers/management/handler.go+1−0
@@ -41,6 +41,7 @@ type Handler struct { · 40 unmodified lines
41 cfg *config.Config41 cfg *config.Config
42 configFilePath string42 configFilePath string
43 mu sync.Mutex43 mu sync.Mutex
44 authStatusMu sync.Mutex
44 reloadMu sync.Mutex45 reloadMu sync.Mutex
45 reloadGeneration uint6446 reloadGeneration uint64
46 appliedReloadGeneration uint6447 appliedReloadGeneration uint64
internal/api/server.go+8−0
@@ -254,6 +254,14 @@ func NewServer(cfg *config.Config, authManager *auth.Manager, accessManager *sdk · 253 unmodified lines
254 return s254 return s
255}255}
256 256
257// Handler returns the HTTP handler used by the server.
258func (s *Server) Handler() http.Handler {
259 if s == nil || s.server == nil {
260 return nil
261 }
262 return s.server.Handler
263}
264
257// Start begins listening for and serving HTTP or HTTPS requests.265// Start begins listening for and serving HTTP or HTTPS requests.
258// It's a blocking call and will only return on an unrecoverable error.266// It's a blocking call and will only return on an unrecoverable error.
259//267//
internal/watcher/dispatcher.go+16−8
@@ -81,11 +81,16 @@ func (w *Watcher) dispatchRuntimeAuthUpdate(update AuthUpdate) bool { · 80 unmodified lines
81}81}
82 82
83func (w *Watcher) dispatchPersistedAuthUpdate(update AuthUpdate) bool {83func (w *Watcher) dispatchPersistedAuthUpdate(update AuthUpdate) bool {
84 if w == nil {84 ok, _ := w.dispatchPersistedAuthUpdateWithRevision(&update)
85 return false85 return ok
86}
87
88func (w *Watcher) dispatchPersistedAuthUpdateWithRevision(update *AuthUpdate) (bool, uint64) {
89 if w == nil || update == nil {
90 return false, 0
86 }91 }
87 if update.Auth == nil || update.Auth.ID == "" {92 if update.Auth == nil || update.Auth.ID == "" {
88 return false93 return false, 0
89 }94 }
90 path := ""95 path := ""
91 if update.Auth.Attributes != nil {96 if update.Auth.Attributes != nil {
@@ -96,7 +101,7 @@ func (w *Watcher) dispatchPersistedAuthUpdate(update AuthUpdate) bool { · 4 unmodified lines
96 }101 }
97 normalized := w.normalizeAuthPath(path)102 normalized := w.normalizeAuthPath(path)
98 if normalized == "" {103 if normalized == "" {
99 return false104 return false, 0
100 }105 }
101 clone := update.Auth.Clone()106 clone := update.Auth.Clone()
102 w.clientsMutex.Lock()107 w.clientsMutex.Lock()
@@ -116,15 +121,18 @@ func (w *Watcher) dispatchPersistedAuthUpdate(update AuthUpdate) bool { · 13 unmodified lines
116 if update.ID == "" {121 if update.ID == "" {
117 update.ID = clone.ID122 update.ID = clone.ID
118 }123 }
119 update.Auth = clone.Clone()124 updateCopy := *update
120 updates := []AuthUpdate{update}125 updateCopy.Auth = clone.Clone()
126 updates := []AuthUpdate{updateCopy}
121 w.stampAuthUpdatesLocked(updates)127 w.stampAuthUpdatesLocked(updates)
128 rev := updates[0].revision
129 update.revision = rev
122 w.clientsMutex.Unlock()130 w.clientsMutex.Unlock()
123 if w.getAuthQueue() == nil {131 if w.getAuthQueue() == nil {
124 return false132 return false, rev
125 }133 }
126 w.dispatchAuthUpdates(updates)134 w.dispatchAuthUpdates(updates)
127 return true135 return true, rev
128}136}
129 137
130func (w *Watcher) refreshAuthState(force bool) {138func (w *Watcher) refreshAuthState(force bool) {
internal/watcher/watcher.go+18−0
@@ -84,6 +84,18 @@ type AuthUpdate struct { · 83 unmodified lines
84 revision uint64 // Watcher-local ordering, independent of runtime auth generations.84 revision uint64 // Watcher-local ordering, independent of runtime auth generations.
85}85}
86 86
87// Revision returns the monotonic watcher revision assigned to this update.
88func (u AuthUpdate) Revision() uint64 {
89 return u.revision
90}
91
92// SetRevision updates the revision counter for this update.
93func (u *AuthUpdate) SetRevision(rev uint64) {
94 if u != nil {
95 u.revision = rev
96 }
97}
98
87const (99const (
88 // replaceCheckDelay is a short delay to allow atomic replace (rename) to settle100 // replaceCheckDelay is a short delay to allow atomic replace (rename) to settle
89 // before deciding whether a Remove event indicates a real deletion.101 // before deciding whether a Remove event indicates a real deletion.
@@ -170,6 +182,12 @@ func (w *Watcher) DispatchPersistedAuthUpdate(update AuthUpdate) bool { · 80 unmodified lines
170 return w.dispatchPersistedAuthUpdate(update)182 return w.dispatchPersistedAuthUpdate(update)
171}183}
172 184
185// DispatchPersistedAuthUpdateWithRevision pushes already-persisted file auth updates through the watcher queue
186// and returns the stamped monotonic revision.
187func (w *Watcher) DispatchPersistedAuthUpdateWithRevision(update *AuthUpdate) (bool, uint64) {
188 return w.dispatchPersistedAuthUpdateWithRevision(update)
189}
190
173// SnapshotCoreAuths converts current clients snapshot into core auth entries.191// SnapshotCoreAuths converts current clients snapshot into core auth entries.
174func (w *Watcher) SnapshotCoreAuths() []*coreauth.Auth {192func (w *Watcher) SnapshotCoreAuths() []*coreauth.Auth {
175 w.clientsMutex.RLock()193 w.clientsMutex.RLock()
sdk/cliproxy/builder.go+9−3
@@ -320,10 +320,16 @@ func (s *Service) runtimeAuthSyncHook() coreauth.PostAuthHook { · 319 unmodified lines
320 ID: auth.ID,320 ID: auth.ID,
321 Auth: auth,321 Auth: auth,
322 }322 }
323 if s.watcher != nil && s.watcher.DispatchPersistedAuthUpdate(update) {323 if s.watcher != nil {
324 return nil324 _, rev := s.watcher.DispatchPersistedAuthUpdateWithRevision(&update)
325 if rev > 0 {
326 update.SetRevision(rev)
327 }
325 }328 }
326 s.handleAuthUpdate(coreauth.WithSkipPersist(ctx), update)329 // Detach from request cancellation so runtime model registration always completes
330 // once the credential has been persisted to disk.
331 syncCtx := coreauth.WithSkipPersist(context.Background())
332 s.handleAuthUpdate(syncCtx, update)
327 return nil333 return nil
328 }334 }
329}335}
sdk/cliproxy/service.go+2−0
@@ -38,6 +38,8 @@ type Service struct { · 37 unmodified lines
38 // configRuntimeMu orders side-effecting runtime application after config commits.38 // configRuntimeMu orders side-effecting runtime application after config commits.
39 configRuntimeMu sync.Mutex39 configRuntimeMu sync.Mutex
40 executorRegistrationMu sync.Mutex40 executorRegistrationMu sync.Mutex
41 authUpdateMu sync.Mutex
42 authRevisions map[string]uint64
41 configSequence uint6443 configSequence uint64
42 appliedRoutingState *routingRuntimeState44 appliedRoutingState *routingRuntimeState
43 45
sdk/cliproxy/service_auth.go+51−1
@@ -94,7 +94,34 @@ func (s *Service) handleAuthUpdates(ctx context.Context, updates []watcher.AuthU · 93 unmodified lines
94 if s == nil {94 if s == nil {
95 return95 return
96 }96 }
97 updates = coalesceAuthUpdates(updates)97 s.authUpdateMu.Lock()
98 defer s.authUpdateMu.Unlock()
99
100 if s.authRevisions == nil {
101 s.authRevisions = make(map[string]uint64)
102 }
103
104 filtered := make([]watcher.AuthUpdate, 0, len(updates))
105 for _, update := range updates {
106 id := authUpdateID(update)
107 if id == "" {
108 filtered = append(filtered, update)
109 continue
110 }
111 rev := update.Revision()
112 if rev > 0 {
113 if prevRev, exists := s.authRevisions[id]; exists && rev <= prevRev {
114 log.Debugf("skipping stale auth update for %s: rev %d <= processed %d", id, rev, prevRev)
115 continue
116 }
117 s.authRevisions[id] = rev
118 }
119 filtered = append(filtered, update)
120 }
121 if len(filtered) == 0 {
122 return
123 }
124 updates = coalesceAuthUpdates(filtered)
98 s.cfgMu.RLock()125 s.cfgMu.RLock()
99 cfg := s.cfg126 cfg := s.cfg
100 s.cfgMu.RUnlock()127 s.cfgMu.RUnlock()
@@ -134,6 +161,12 @@ func (s *Service) handleAuthUpdates(ctx context.Context, updates []watcher.AuthU · 33 unmodified lines
134 if id == "" {161 if id == "" {
135 continue162 continue
136 }163 }
164 if existing, ok := s.coreManager.GetByID(id); ok && existing != nil && update.Auth != nil {
165 if isStaleCoreAuth(existing, update.Auth) {
166 log.Debugf("skipping stale auth delete for %s: incoming gen=%d, existing gen=%d", id, update.Auth.Generation, existing.Generation)
167 continue
168 }
169 }
137 s.applyCoreAuthRemoval(registrationCtx, id)170 s.applyCoreAuthRemoval(registrationCtx, id)
138 needsAliasRebuild = true171 needsAliasRebuild = true
139 default:172 default:
@@ -282,6 +315,10 @@ func (s *Service) prepareCoreAuthForModelRegistration(ctx context.Context, auth · 142 unmodified lines
282 op := "register"315 op := "register"
283 var err error316 var err error
284 if existing, ok := s.coreManager.GetByID(auth.ID); ok {317 if existing, ok := s.coreManager.GetByID(auth.ID); ok {
318 if isStaleCoreAuth(existing, auth) {
319 log.Debugf("skipping stale auth update for %s: incoming gen=%d, existing gen=%d", auth.ID, auth.Generation, existing.Generation)
320 return existing
321 }
285 auth.CreatedAt = existing.CreatedAt322 auth.CreatedAt = existing.CreatedAt
286 if !existing.Disabled && existing.Status != coreauth.StatusDisabled && !auth.Disabled && auth.Status != coreauth.StatusDisabled {323 if !existing.Disabled && existing.Status != coreauth.StatusDisabled && !auth.Disabled && auth.Status != coreauth.StatusDisabled {
287 auth.LastRefreshedAt = existing.LastRefreshedAt324 auth.LastRefreshedAt = existing.LastRefreshedAt
@@ -307,6 +344,19 @@ func (s *Service) prepareCoreAuthForModelRegistration(ctx context.Context, auth · 19 unmodified lines
307 return auth344 return auth
308}345}
309 346
347// isStaleCoreAuth reports whether an incoming auth update is older than the current
348// state in coreManager, based on registration epoch.
349func isStaleCoreAuth(existing, incoming *coreauth.Auth) bool {
350 if existing == nil || incoming == nil {
351 return false
352 }
353 // If incoming has an explicit registration epoch that is older than existing, it's stale.
354 if incoming.RegistrationEpoch > 0 && incoming.RegistrationEpoch < existing.RegistrationEpoch {
355 return true
356 }
357 return false
358}
359
310func (s *Service) completeModelRegistrationForAuth(ctx context.Context, auth *coreauth.Auth) {360func (s *Service) completeModelRegistrationForAuth(ctx context.Context, auth *coreauth.Auth) {
311 s.completeModelRegistrationForAuthWithCache(ctx, auth, nil)361 s.completeModelRegistrationForAuthWithCache(ctx, auth, nil)
312}362}
sdk/cliproxy/service_auth_sync_test.go+710−0
@@ -0,0 +1,710 @@
1package cliproxy
2
3import (
4 "context"
5 "encoding/json"
6 "net/http"
7 "net/http/httptest"
8 "os"
9 "path/filepath"
10 "strings"
11 "testing"
12 "time"
13
14 "golang.org/x/crypto/bcrypt"
15
16 "github.com/router-for-me/CLIProxyAPI/v7/internal/api"
17 internalregistry "github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
18 "github.com/router-for-me/CLIProxyAPI/v7/internal/watcher"
19 coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
20 cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
21 "github.com/router-for-me/CLIProxyAPI/v7/sdk/config"
22)
23
24func TestRuntimeAuthSyncHook_SynchronousModelAndSchedulerRestoration(t *testing.T) {
25 reg := internalregistry.GetGlobalRegistry()
26 authID := "codex-sync-test-auth"
27 reg.UnregisterClient(authID)
28 t.Cleanup(func() {
29 reg.UnregisterClient(authID)
30 })
31
32 manager := coreauth.NewManager(nil, nil, nil)
33 manager.RegisterExecutor(&syncTestExecutor{})
34 auth := &coreauth.Auth{
35 ID: authID,
36 Provider: "codex",
37 Status: coreauth.StatusActive,
38 Attributes: map[string]string{
39 "plan_type": "pro",
40 "path": "/path/to/codex-sync.json",
41 },
42 }
43 if _, errRegister := manager.Register(context.Background(), auth); errRegister != nil {
44 t.Fatalf("register auth: %v", errRegister)
45 }
46
47 // Set up a mock watcher wrapper where dispatchPersistedAuth records updates
48 // but DOES NOT consume or execute anything asynchronously, verifying that
49 // the synchronous restoration does not depend on the watcher background queue.
50 var persistedUpdates []watcher.AuthUpdate
51 watcherWrapper := &WatcherWrapper{
52 dispatchPersistedAuth: func(update watcher.AuthUpdate) bool {
53 persistedUpdates = append(persistedUpdates, update)
54 return true
55 },
56 }
57
58 service := &Service{
59 cfg: &config.Config{},
60 coreManager: manager,
61 watcher: watcherWrapper,
62 }
63
64 hook := service.runtimeAuthSyncHook()
65 if hook == nil {
66 t.Fatal("runtimeAuthSyncHook() returned nil")
67 }
68
69 // 1. Initial sync (enabled -> registers models)
70 if err := hook(context.Background(), auth); err != nil {
71 t.Fatalf("hook failed for initial sync: %v", err)
72 }
73 manager.RegisterExecutor(&syncTestExecutor{})
74 models := reg.GetModelsForClient(authID)
75 if len(models) == 0 {
76 t.Fatal("expected models registered for active auth, got none")
77 }
78
79 // Verify scheduler selection works for enabled auth
80 resp, errExec := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{})
81 if errExec != nil || string(resp.Payload) != authID {
82 t.Fatalf("expected scheduler to select active auth, got resp=%q, err=%v", string(resp.Payload), errExec)
83 }
84
85 // 2. Disable the credential
86 disabledAuth := auth.Clone()
87 disabledAuth.Disabled = true
88 disabledAuth.Status = coreauth.StatusDisabled
89 disabledAuth.StatusMessage = "disabled via management API"
90 disabledAuth.UpdatedAt = time.Now()
91
92 if _, errUpdate := manager.Update(context.Background(), disabledAuth); errUpdate != nil {
93 t.Fatalf("manager update disable failed: %v", errUpdate)
94 }
95 if err := hook(context.Background(), disabledAuth); err != nil {
96 t.Fatalf("hook failed for disable: %v", err)
97 }
98 manager.RegisterExecutor(&syncTestExecutor{})
99
100 // Verify immediately: models must be unregistered from global registry
101 modelsAfterDisable := reg.GetModelsForClient(authID)
102 if len(modelsAfterDisable) != 0 {
103 t.Fatalf("expected 0 models for disabled client, got %d", len(modelsAfterDisable))
104 }
105
106 // Verify scheduler selection fails for disabled auth
107 if _, errExecDisabled := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{}); errExecDisabled == nil {
108 t.Fatal("expected scheduler execution to fail for disabled auth, but got nil error")
109 }
110
111 // 3. Re-enable the credential without quota
112 enabledAuth := disabledAuth.Clone()
113 enabledAuth.Disabled = false
114 enabledAuth.Status = coreauth.StatusActive
115 enabledAuth.StatusMessage = ""
116 enabledAuth.UpdatedAt = time.Now()
117
118 if _, errUpdate := manager.Update(context.Background(), enabledAuth); errUpdate != nil {
119 t.Fatalf("manager update enable failed: %v", errUpdate)
120 }
121
122 // 4. Re-enable hook invocation
123 if err := hook(context.Background(), enabledAuth); err != nil {
124 t.Fatalf("hook failed for re-enable: %v", err)
125 }
126 manager.RegisterExecutor(&syncTestExecutor{})
127
128 // Verify immediately: models must be restored in the global registry without restart
129 modelsAfterEnable := reg.GetModelsForClient(authID)
130 if len(modelsAfterEnable) == 0 {
131 t.Fatal("expected models to be restored synchronously upon re-enable, got none")
132 }
133
134 hasAstra := false
135 for _, m := range modelsAfterEnable {
136 if m != nil && m.ID == "gpt-6-astra" {
137 hasAstra = true
138 break
139 }
140 }
141 if !hasAstra {
142 t.Fatalf("expected gpt-6-astra in restored models: %+v", modelsAfterEnable)
143 }
144
145 // Verify scheduler selection is restored for the re-enabled auth
146 respAfterEnable, errExecEnable := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{})
147 if errExecEnable != nil || string(respAfterEnable.Payload) != authID {
148 t.Fatalf("expected scheduler to select restored auth, got resp=%q, err=%v", string(respAfterEnable.Payload), errExecEnable)
149 }
150
151 // 5. Verify quota cooldown separation: set quota exceeded on the enabled auth
152 quotaAuth := enabledAuth.Clone()
153 quotaAuth.Quota = coreauth.QuotaState{
154 Exceeded: true,
155 Reason: "credential_quota",
156 NextRecoverAt: time.Now().Add(10 * time.Minute),
157 }
158 if _, errUpdate := manager.Update(context.Background(), quotaAuth); errUpdate != nil {
159 t.Fatalf("manager update quota failed: %v", errUpdate)
160 }
161 if err := hook(context.Background(), quotaAuth); err != nil {
162 t.Fatalf("hook failed for quota update: %v", err)
163 }
164 manager.RegisterExecutor(&syncTestExecutor{})
165
166 // Verify quota separation: models remain registered in global registry
167 modelsWithQuota := reg.GetModelsForClient(authID)
168 if len(modelsWithQuota) == 0 {
169 t.Fatal("expected models to remain registered when auth is in quota cooldown")
170 }
171 // But scheduler blocks execution due to quota exhaustion
172 if _, errQuotaExec := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{}); errQuotaExec == nil {
173 t.Fatal("expected scheduler to block execution when quota exceeded")
174 }
175
176 // Verify watcher notification occurred
177 if len(persistedUpdates) == 0 {
178 t.Fatal("expected watcher to be notified of persisted auth updates")
179 }
180}
181
182type syncTestExecutor struct{}
183
184func (e *syncTestExecutor) Identifier() string { return "codex" }
185
186func (e *syncTestExecutor) Execute(ctx context.Context, a *coreauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
187 return cliproxyexecutor.Response{Payload: []byte(a.ID)}, nil
188}
189
190func (e *syncTestExecutor) ExecuteStream(context.Context, *coreauth.Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) {
191 return nil, &coreauth.Error{HTTPStatus: http.StatusNotImplemented, Message: "not implemented"}
192}
193
194func (e *syncTestExecutor) Refresh(_ context.Context, auth *coreauth.Auth) (*coreauth.Auth, error) {
195 return auth, nil
196}
197
198func (e *syncTestExecutor) CountTokens(context.Context, *coreauth.Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
199 return cliproxyexecutor.Response{}, &coreauth.Error{HTTPStatus: http.StatusNotImplemented, Message: "not implemented"}
200}
201
202func (e *syncTestExecutor) HttpRequest(context.Context, *coreauth.Auth, *http.Request) (*http.Response, error) {
203 return nil, &coreauth.Error{HTTPStatus: http.StatusNotImplemented, Message: "not implemented"}
204}
205
206func TestRuntimeAuthSyncHook_NilAndEmptyAuth(t *testing.T) {
207 service := &Service{}
208 hook := service.runtimeAuthSyncHook()
209 if hook == nil {
210 t.Fatal("runtimeAuthSyncHook() returned nil")
211 }
212
213 if err := hook(context.Background(), nil); err != nil {
214 t.Fatalf("expected nil error for nil auth, got: %v", err)
215 }
216
217 if err := hook(context.Background(), &coreauth.Auth{}); err != nil {
218 t.Fatalf("expected nil error for empty auth, got: %v", err)
219 }
220}
221
222func TestEndToEndStatusPatch_RestoresModelsThroughServerPipeline(t *testing.T) {
223 authDir := t.TempDir()
224 fileName := "codex-e2e.json"
225 filePath := filepath.Join(authDir, fileName)
226 if errWrite := os.WriteFile(filePath, []byte(`{"type":"codex","disabled":false}`), 0o600); errWrite != nil {
227 t.Fatalf("write auth file: %v", errWrite)
228 }
229
230 reg := internalregistry.GetGlobalRegistry()
231 authID := fileName
232 reg.UnregisterClient(authID)
233 t.Cleanup(func() {
234 reg.UnregisterClient(authID)
235 })
236
237 secretKey := "test-secret"
238 hashedSecret, errHash := bcrypt.GenerateFromPassword([]byte(secretKey), bcrypt.DefaultCost)
239 if errHash != nil {
240 t.Fatalf("bcrypt hash failed: %v", errHash)
241 }
242
243 cfg := &config.Config{
244 AuthDir: authDir,
245 RemoteManagement: config.RemoteManagement{
246 SecretKey: string(hashedSecret),
247 AllowRemote: true,
248 },
249 }
250
251 service, errBuild := NewBuilder().
252 WithConfig(cfg).
253 WithConfigPath(filepath.Join(authDir, "config.yaml")).
254 Build()
255 if errBuild != nil {
256 t.Fatalf("Build() failed: %v", errBuild)
257 }
258
259 server := api.NewServer(service.cfg, service.coreManager, service.accessManager, service.configPath, service.serverOptions...)
260 if server == nil {
261 t.Fatal("NewServer() returned nil")
262 }
263
264 auth := &coreauth.Auth{
265 ID: authID,
266 FileName: fileName,
267 Provider: "codex",
268 Status: coreauth.StatusActive,
269 Attributes: map[string]string{
270 "plan_type": "pro",
271 "path": filePath,
272 },
273 }
274 if _, errRegister := service.coreManager.Register(context.Background(), auth); errRegister != nil {
275 t.Fatalf("register auth: %v", errRegister)
276 }
277
278 // Initial model sync
279 syncHook := service.runtimeAuthSyncHook()
280 if err := syncHook(context.Background(), auth); err != nil {
281 t.Fatalf("initial sync failed: %v", err)
282 }
283
284 handler := server.Handler()
285
286 // 1. Confirm models exist before disabling
287 rec := httptest.NewRecorder()
288 req := httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
289 req.Header.Set("Authorization", "Bearer test-secret")
290 handler.ServeHTTP(rec, req)
291 if rec.Code != http.StatusOK {
292 t.Fatalf("GetAuthFileModels initial status = %d, body = %s", rec.Code, rec.Body.String())
293 }
294 var resp struct {
295 Models []map[string]any `json:"models"`
296 }
297 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) == 0 {
298 t.Fatalf("expected initial models, got %s", rec.Body.String())
299 }
300
301 // 2. Disable credential via PATCH /v0/management/auth-files/status
302 rec = httptest.NewRecorder()
303 req = httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"`+fileName+`","disabled":true}`))
304 req.Header.Set("Authorization", "Bearer test-secret")
305 req.Header.Set("Content-Type", "application/json")
306 handler.ServeHTTP(rec, req)
307 if rec.Code != http.StatusOK {
308 t.Fatalf("PATCH status disable failed: %s", rec.Body.String())
309 }
310
311 // 3. Confirm GET /v0/management/auth-files/models returns empty list immediately
312 rec = httptest.NewRecorder()
313 req = httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
314 req.Header.Set("Authorization", "Bearer test-secret")
315 handler.ServeHTTP(rec, req)
316 if rec.Code != http.StatusOK {
317 t.Fatalf("GetAuthFileModels after disable status = %d, body = %s", rec.Code, rec.Body.String())
318 }
319 resp.Models = nil
320 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) != 0 {
321 t.Fatalf("expected 0 models after disable, got: %s", rec.Body.String())
322 }
323
324 // 4. Re-enable credential via PATCH /v0/management/auth-files/status
325 rec = httptest.NewRecorder()
326 req = httptest.NewRequest(http.MethodPatch, "/v0/management/auth-files/status", strings.NewReader(`{"name":"`+fileName+`","disabled":false}`))
327 req.Header.Set("Authorization", "Bearer test-secret")
328 req.Header.Set("Content-Type", "application/json")
329 handler.ServeHTTP(rec, req)
330 if rec.Code != http.StatusOK {
331 t.Fatalf("PATCH status re-enable failed: %s", rec.Body.String())
332 }
333
334 // 5. Confirm models are synchronously restored immediately without restart
335 rec = httptest.NewRecorder()
336 req = httptest.NewRequest(http.MethodGet, "/v0/management/auth-files/models?name="+fileName, nil)
337 req.Header.Set("Authorization", "Bearer test-secret")
338 handler.ServeHTTP(rec, req)
339 if rec.Code != http.StatusOK {
340 t.Fatalf("GetAuthFileModels after re-enable status = %d, body = %s", rec.Code, rec.Body.String())
341 }
342 resp.Models = nil
343 if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil || len(resp.Models) == 0 {
344 t.Fatalf("expected models to be restored after re-enable, got: %s", rec.Body.String())
345 }
346}
347
348func TestRuntimeAuthSync_StaleAsyncEventDoesNotOverwriteSynchronousEnable(t *testing.T) {
349 reg := internalregistry.GetGlobalRegistry()
350 authID := "codex-stale-test-auth"
351 reg.UnregisterClient(authID)
352 t.Cleanup(func() {
353 reg.UnregisterClient(authID)
354 })
355
356 manager := coreauth.NewManager(nil, nil, nil)
357 manager.RegisterExecutor(&syncTestExecutor{})
358
359 var currentRevision uint64
360 watcherWrapper := &WatcherWrapper{
361 dispatchPersistedAuthWithRev: func(update *watcher.AuthUpdate) (bool, uint64) {
362 currentRevision++
363 return true, currentRevision
364 },
365 }
366
367 service := &Service{
368 cfg: &config.Config{},
369 coreManager: manager,
370 watcher: watcherWrapper,
371 }
372
373 auth := &coreauth.Auth{
374 ID: authID,
375 Provider: "codex",
376 Status: coreauth.StatusActive,
377 Attributes: map[string]string{
378 "plan_type": "pro",
379 },
380 }
381 registeredAuth, errRegister := manager.Register(context.Background(), auth)
382 if errRegister != nil {
383 t.Fatalf("register auth: %v", errRegister)
384 }
385
386 hook := service.runtimeAuthSyncHook()
387
388 // Initial sync
389 if err := hook(context.Background(), registeredAuth); err != nil {
390 t.Fatalf("initial hook sync failed: %v", err)
391 }
392 if len(reg.GetModelsForClient(authID)) == 0 {
393 t.Fatal("expected models after initial sync")
394 }
395
396 // Step 1: Credential is disabled (Generation advances)
397 disabledAuth := registeredAuth.Clone()
398 disabledAuth.Disabled = true
399 disabledAuth.Status = coreauth.StatusDisabled
400 disabledAuth.UpdatedAt = time.Now()
401
402 updatedDisabled, errDisable := manager.Update(context.Background(), disabledAuth)
403 if errDisable != nil {
404 t.Fatalf("update disable failed: %v", errDisable)
405 }
406
407 // Capture the stale async event that would normally sit in an async queue
408 staleAsyncDisableUpdate := watcher.AuthUpdate{
409 Action: watcher.AuthUpdateActionModify,
410 ID: authID,
411 Auth: updatedDisabled.Clone(),
412 }
413 staleAsyncDisableUpdate.SetRevision(2)
414
415 // Step 2: Synchronous disable hook runs (assigns revision 2)
416 if err := hook(context.Background(), updatedDisabled); err != nil {
417 t.Fatalf("hook disable failed: %v", err)
418 }
419 if len(reg.GetModelsForClient(authID)) != 0 {
420 t.Fatal("expected 0 models after disable")
421 }
422
423 // Step 3: Credential is immediately re-enabled (advances revision to 3)
424 enabledAuth := updatedDisabled.Clone()
425 enabledAuth.Disabled = false
426 enabledAuth.Status = coreauth.StatusActive
427 enabledAuth.UpdatedAt = time.Now()
428
429 updatedEnabled, errEnable := manager.Update(context.Background(), enabledAuth)
430 if errEnable != nil {
431 t.Fatalf("update enable failed: %v", errEnable)
432 }
433
434 // Step 4: Synchronous enable hook runs (assigns revision 3)
435 if err := hook(context.Background(), updatedEnabled); err != nil {
436 t.Fatalf("hook enable failed: %v", err)
437 }
438 manager.RegisterExecutor(&syncTestExecutor{})
439
440 // Models must be restored immediately
441 if len(reg.GetModelsForClient(authID)) == 0 {
442 t.Fatal("expected models restored immediately upon synchronous enable")
443 }
444
445 // Step 5: The delayed stale disable update from Step 1 is delivered now via consumer
446 service.handleAuthUpdate(context.Background(), staleAsyncDisableUpdate)
447 manager.RegisterExecutor(&syncTestExecutor{})
448
449 // Step 6: Verify the stale async event was prevented from overwriting the newer enable state!
450 currentAuth, ok := manager.GetByID(authID)
451 if !ok || currentAuth == nil {
452 t.Fatal("expected auth to exist in manager")
453 }
454 if currentAuth.Disabled {
455 t.Fatal("stale async event overwrote the synchronous enable state in manager!")
456 }
457 modelsAfterStale := reg.GetModelsForClient(authID)
458 if len(modelsAfterStale) == 0 {
459 t.Fatal("stale async event unregistered models from global registry!")
460 }
461
462 // Scheduler execution must still succeed and select the enabled auth
463 resp, errExec := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{})
464 if errExec != nil || string(resp.Payload) != authID {
465 t.Fatalf("expected scheduler to select active auth, got resp=%q err=%v", string(resp.Payload), errExec)
466 }
467}
468
469func TestRuntimeAuthSync_StaleWatcherUnversionedEventDoesNotOverwriteSynchronousEnable(t *testing.T) {
470 reg := internalregistry.GetGlobalRegistry()
471 authID := "codex-unversioned-stale-test-auth"
472 reg.UnregisterClient(authID)
473 t.Cleanup(func() {
474 reg.UnregisterClient(authID)
475 })
476
477 manager := coreauth.NewManager(nil, nil, nil)
478 manager.RegisterExecutor(&syncTestExecutor{})
479
480 var currentRevision uint64
481 watcherWrapper := &WatcherWrapper{
482 dispatchPersistedAuthWithRev: func(update *watcher.AuthUpdate) (bool, uint64) {
483 currentRevision++
484 return true, currentRevision
485 },
486 }
487
488 service := &Service{
489 cfg: &config.Config{},
490 coreManager: manager,
491 watcher: watcherWrapper,
492 }
493
494 auth := &coreauth.Auth{
495 ID: authID,
496 Provider: "codex",
497 Status: coreauth.StatusActive,
498 Attributes: map[string]string{
499 "plan_type": "pro",
500 },
501 }
502 registeredAuth, errRegister := manager.Register(context.Background(), auth)
503 if errRegister != nil {
504 t.Fatalf("register auth: %v", errRegister)
505 }
506
507 hook := service.runtimeAuthSyncHook()
508
509 // Initial sync
510 if err := hook(context.Background(), registeredAuth); err != nil {
511 t.Fatalf("initial hook sync failed: %v", err)
512 }
513 manager.RegisterExecutor(&syncTestExecutor{})
514 if len(reg.GetModelsForClient(authID)) == 0 {
515 t.Fatal("expected models after initial sync")
516 }
517
518 // Step 1: Simulate the watcher generating an unversioned Auth snapshot during a disable
519 // event on disk (as synthesizer.SynthesizeAuthFile produces: RegistrationEpoch=0, Generation=0).
520 // Timestamp is recorded at observation time T1.
521 unversionedDisableTime := time.Now().Add(-50 * time.Millisecond)
522 unversionedDisableAuth := &coreauth.Auth{
523 ID: authID,
524 Provider: "codex",
525 Disabled: true,
526 Status: coreauth.StatusDisabled,
527 CreatedAt: unversionedDisableTime,
528 UpdatedAt: unversionedDisableTime,
529 RegistrationEpoch: 0, // Unversioned from file watcher synthesizer
530 Generation: 0, // Unversioned from file watcher synthesizer
531 Attributes: map[string]string{
532 "plan_type": "pro",
533 },
534 }
535 delayedWatcherUpdate := watcher.AuthUpdate{
536 Action: watcher.AuthUpdateActionModify,
537 ID: authID,
538 Auth: unversionedDisableAuth,
539 }
540 delayedWatcherUpdate.SetRevision(2)
541
542 // Step 2: In the meantime, the credential was synchronously re-enabled via management API
543 // at T2 > T1, updating coreManager with RegistrationEpoch >= 1, Generation >= 2, UpdatedAt = T2.
544 enabledAuth := registeredAuth.Clone()
545 enabledAuth.Disabled = false
546 enabledAuth.Status = coreauth.StatusActive
547 enabledAuth.UpdatedAt = time.Now()
548
549 updatedEnabled, errEnable := manager.Update(context.Background(), enabledAuth)
550 if errEnable != nil {
551 t.Fatalf("update enable failed: %v", errEnable)
552 }
553
554 // Synchronous enable hook runs and restores models in GlobalModelRegistry
555 if err := hook(context.Background(), updatedEnabled); err != nil {
556 t.Fatalf("hook enable failed: %v", err)
557 }
558 manager.RegisterExecutor(&syncTestExecutor{})
559
560 if len(reg.GetModelsForClient(authID)) == 0 {
561 t.Fatal("expected models restored immediately upon synchronous enable")
562 }
563
564 // Step 3: Now the delayed unversioned watcher event from Step 1 arrives via the queue consumer
565 service.handleAuthUpdate(context.Background(), delayedWatcherUpdate)
566 manager.RegisterExecutor(&syncTestExecutor{})
567
568 // Scheduler execution must still succeed and select the enabled auth
569 resp, errExec := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{})
570 if errExec != nil || string(resp.Payload) != authID {
571 t.Fatalf("expected scheduler to select active auth, got resp=%q err=%v", string(resp.Payload), errExec)
572 }
573}
574
575func TestRuntimeAuthSync_FileSnapshotEnqueuedThenRequestExecutionThenConsume(t *testing.T) {
576 reg := internalregistry.GetGlobalRegistry()
577 authID := "codex-interleaved-request-auth"
578 reg.UnregisterClient(authID)
579 t.Cleanup(func() {
580 reg.UnregisterClient(authID)
581 })
582
583 manager := coreauth.NewManager(nil, nil, nil)
584 manager.RegisterExecutor(&syncTestExecutor{})
585
586 service := &Service{
587 cfg: &config.Config{},
588 coreManager: manager,
589 }
590
591 auth := &coreauth.Auth{
592 ID: authID,
593 Provider: "codex",
594 Status: coreauth.StatusActive,
595 Attributes: map[string]string{
596 "plan_type": "pro",
597 },
598 }
599 registeredAuth, errRegister := manager.Register(context.Background(), auth)
600 if errRegister != nil {
601 t.Fatalf("register auth: %v", errRegister)
602 }
603
604 hook := service.runtimeAuthSyncHook()
605
606 // Initial sync (Revision 1)
607 if err := hook(context.Background(), registeredAuth); err != nil {
608 t.Fatalf("initial hook sync failed: %v", err)
609 }
610 manager.RegisterExecutor(&syncTestExecutor{})
611 if len(reg.GetModelsForClient(authID)) == 0 {
612 t.Fatal("expected models after initial sync")
613 }
614
615 // Step 1: A file change occurs on disk (e.g. user disables credential in auth file).
616 // The watcher generates an AuthUpdate with revision 2 and enqueues it.
617 fileSnapshotAuth := registeredAuth.Clone()
618 fileSnapshotAuth.Disabled = true
619 fileSnapshotAuth.Status = coreauth.StatusDisabled
620 fileSnapshotAuth.UpdatedAt = time.Now()
621
622 fileUpdate := watcher.AuthUpdate{
623 Action: watcher.AuthUpdateActionModify,
624 ID: authID,
625 Auth: fileSnapshotAuth,
626 }
627 fileUpdate.SetRevision(2)
628
629 // Step 2: Before the queue consumer can process fileUpdate, an API request completes.
630 // MarkResult updates the auth in coreManager: auth.Generation++ and auth.UpdatedAt = now.
631 manager.MarkResult(context.Background(), coreauth.Result{
632 AuthID: authID,
633 Provider: "codex",
634 Model: "gpt-6-astra",
635 Success: true,
636 })
637
638 // Step 3: Now the queue consumer processes the fileUpdate (revision 2).
639 // Because revision 2 > revision 1, the file update MUST NOT be discarded due to
640 // the intermediate request's UpdatedAt timestamp!
641 service.handleAuthUpdate(context.Background(), fileUpdate)
642 manager.RegisterExecutor(&syncTestExecutor{})
643
644 // Step 4: Verify the file update successfully took effect!
645 currentAuth, ok := manager.GetByID(authID)
646 if !ok || currentAuth == nil {
647 t.Fatal("expected auth to exist in manager")
648 }
649 if !currentAuth.Disabled {
650 t.Fatal("file update was incorrectly dropped after intermediate request execution!")
651 }
652 modelsAfterDisable := reg.GetModelsForClient(authID)
653 if len(modelsAfterDisable) != 0 {
654 t.Fatalf("expected 0 models after file update disable, got %d", len(modelsAfterDisable))
655 }
656}
657
658func TestRuntimeAuthSyncHook_ContextCancellationDoesNotAbortSync(t *testing.T) {
659 reg := internalregistry.GetGlobalRegistry()
660 authID := "codex-canceled-ctx-auth"
661 reg.UnregisterClient(authID)
662 t.Cleanup(func() {
663 reg.UnregisterClient(authID)
664 })
665
666 manager := coreauth.NewManager(nil, nil, nil)
667 manager.RegisterExecutor(&syncTestExecutor{})
668
669 service := &Service{
670 cfg: &config.Config{},
671 coreManager: manager,
672 }
673
674 auth := &coreauth.Auth{
675 ID: authID,
676 Provider: "codex",
677 Status: coreauth.StatusActive,
678 Attributes: map[string]string{
679 "plan_type": "pro",
680 },
681 }
682 registeredAuth, errRegister := manager.Register(context.Background(), auth)
683 if errRegister != nil {
684 t.Fatalf("register auth: %v", errRegister)
685 }
686
687 hook := service.runtimeAuthSyncHook()
688
689 // Create a context that is ALREADY canceled (e.g. client aborted request right after save)
690 canceledCtx, cancel := context.WithCancel(context.Background())
691 cancel()
692
693 // Calling hook with canceled context must NOT abort registration
694 if err := hook(canceledCtx, registeredAuth); err != nil {
695 t.Fatalf("hook failed with canceled context: %v", err)
696 }
697 manager.RegisterExecutor(&syncTestExecutor{})
698
699 // Verify models were registered successfully despite canceled incoming context
700 models := reg.GetModelsForClient(authID)
701 if len(models) == 0 {
702 t.Fatal("expected models to be registered even when incoming context is canceled")
703 }
704
705 // Verify scheduler can execute
706 resp, errExec := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: "gpt-6-astra"}, cliproxyexecutor.Options{})
707 if errExec != nil || string(resp.Payload) != authID {
708 t.Fatalf("expected scheduler execution to succeed, got resp=%q err=%v", string(resp.Payload), errExec)
709 }
710}
sdk/cliproxy/types.go+33−9
@@ -101,13 +101,14 @@ type WatcherWrapper struct { · 100 unmodified lines
101 start func(ctx context.Context) error101 start func(ctx context.Context) error
102 stop func() error102 stop func() error
103 103
104 setConfig func(cfg *config.Config)104 setConfig func(cfg *config.Config)
105 snapshotAuths func() []*coreauth.Auth105 snapshotAuths func() []*coreauth.Auth
106 setUpdateQueue func(queue chan<- watcher.AuthUpdate)106 setUpdateQueue func(queue chan<- watcher.AuthUpdate)
107 dispatchRuntimeUpdate func(update watcher.AuthUpdate) bool107 dispatchRuntimeUpdate func(update watcher.AuthUpdate) bool
108 dispatchPersistedAuth func(update watcher.AuthUpdate) bool108 dispatchPersistedAuth func(update watcher.AuthUpdate) bool
109 setPluginAuthParser func(parser PluginAuthParser)109 dispatchPersistedAuthWithRev func(update *watcher.AuthUpdate) (bool, uint64)
110 reloadConfigIfChanged func()110 setPluginAuthParser func(parser PluginAuthParser)
111 reloadConfigIfChanged func()
111}112}
112 113
113// Start proxies to the underlying watcher Start implementation.114// Start proxies to the underlying watcher Start implementation.
@@ -163,10 +164,33 @@ func (w *WatcherWrapper) DispatchRuntimeAuthUpdate(update watcher.AuthUpdate) bo · 49 unmodified lines
163 164
164// DispatchPersistedAuthUpdate forwards already-persisted file auth updates.165// DispatchPersistedAuthUpdate forwards already-persisted file auth updates.
165func (w *WatcherWrapper) DispatchPersistedAuthUpdate(update watcher.AuthUpdate) bool {166func (w *WatcherWrapper) DispatchPersistedAuthUpdate(update watcher.AuthUpdate) bool {
166 if w == nil || w.dispatchPersistedAuth == nil {167 if w == nil {
167 return false168 return false
168 }169 }
169 return w.dispatchPersistedAuth(update)170 if w.dispatchPersistedAuthWithRev != nil {
171 ok, _ := w.dispatchPersistedAuthWithRev(&update)
172 return ok
173 }
174 if w.dispatchPersistedAuth != nil {
175 return w.dispatchPersistedAuth(update)
176 }
177 return false
178}
179
180// DispatchPersistedAuthUpdateWithRevision forwards already-persisted file auth updates
181// and returns whether it was enqueued along with its assigned watcher revision.
182func (w *WatcherWrapper) DispatchPersistedAuthUpdateWithRevision(update *watcher.AuthUpdate) (bool, uint64) {
183 if w == nil {
184 return false, 0
185 }
186 if w.dispatchPersistedAuthWithRev != nil {
187 return w.dispatchPersistedAuthWithRev(update)
188 }
189 if w.dispatchPersistedAuth != nil && update != nil {
190 ok := w.dispatchPersistedAuth(*update)
191 return ok, update.Revision()
192 }
193 return false, 0
170}194}
171 195
172// SetClients updates the watcher file-backed clients registry.196// SetClients updates the watcher file-backed clients registry.
sdk/cliproxy/watcher.go+3−0
@@ -34,6 +34,9 @@ func defaultWatcherFactory(configPath, authDir string, reload func(*config.Confi · 33 unmodified lines
34 dispatchPersistedAuth: func(update watcher.AuthUpdate) bool {34 dispatchPersistedAuth: func(update watcher.AuthUpdate) bool {
35 return w.DispatchPersistedAuthUpdate(update)35 return w.DispatchPersistedAuthUpdate(update)
36 },36 },
37 dispatchPersistedAuthWithRev: func(update *watcher.AuthUpdate) (bool, uint64) {
38 return w.DispatchPersistedAuthUpdateWithRevision(update)
39 },
37 setPluginAuthParser: func(parser PluginAuthParser) {40 setPluginAuthParser: func(parser PluginAuthParser) {
38 w.SetPluginAuthParser(parser)41 w.SetPluginAuthParser(parser)
39 },42 },