diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 75dbc31e80f3..05f928193937 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -19,8 +19,14 @@ nothing. While no Codex runtime row is stored, ## Codex fingerprint simulation -Extensions owns `/admin/codex-fingerprint`. General settings and account -create/edit/bulk-edit forms no longer submit fingerprint configuration. +The Extensions sidebar owns one Codex entry. `/admin/codex` combines borrowing +configuration and status; `/admin/codex/identity` shows client identity; and +`/admin/codex/advanced` owns manual UA/version, identifier modes and compression. +The previous runtime, fingerprint, borrowing and status URLs redirect to these +pages. Configured borrowing opens with its editor folded; a new setup opens the +three-step editor. Full saved/effective identity fields remain available in +keyboard-accessible disclosures. General settings and account create/edit/bulk +forms do not submit fingerprint configuration. `GET/PUT /admin/settings/codex-fingerprint` uses `enabled`, `user_agent`, `client_version` and `version_auto_sync_enabled`; all four fields are required on PUT. GET also returns the synced and effective simulation versions and the @@ -71,18 +77,13 @@ orchestration catalogs retain their existing capabilities. ## Codex gateway borrowing -The Extensions sidebar has separate Codex Borrowing and Pelican Tests entries. -Borrowing's in-page navigation switches between settings at -`/admin/codex-gateway-borrow`, route status at -`/admin/codex-gateway-borrow/status`. Pelican generation/history lives at -`/admin/pelican-tests`; `/admin/codex-pelican-comparison` redirects there. -Opening or navigating these pages only reads -local data. Settings drafts contain configuration IDs/options only, persist in -the browser session, and never affect the saved configuration used by status or -generation. Only explicit actions save, prepare, verify or generate. The status -page polls read-only status every five seconds while preparation/verification is -active; local countdowns expire displayed routes without renewing them. Leaving -the Pelican page cancels its queued and active generation. +The unified Codex page separates configuration, route qualification, actual +request usage and upstream completion. Pelican stays at `/admin/pelican-tests`; +its legacy comparison URL redirects there. Page reads and navigation only read +local data. Unsaved configuration stays in browser-session storage. Saving an +enabled configuration starts exactly one finite preparation, shown as queued +before the worker starts. Status polls every five seconds only while work is +active; countdowns expire routes without renewing them. `GET/PUT /admin/codex-gateway-borrow/config` owns one JSON setting, `codex_gateway_borrow_config`: `enabled`, `source_account_ids`, @@ -103,9 +104,43 @@ Source and target probes use separate HTTP/TLS pools. Business WS connections stay in the ordinary account pool, with fixed one-hour continuation anchors. Status/config/history reads do not call models. Saving an enabled config starts -one finite preparation; business cache misses prepare synchronously. There is -no renewal timer. `/prepare` and `/verify` explicitly prepare or revalidate. -All probes only record observations, including failures before business dispatch. +one finite preparation; business misses prepare on demand. There is no renewal +timer. `/prepare` and `/verify` retain their interfaces. Same-revision, same-wire +qualifications share work; cancelling one caller does not cancel other waiters, +and the final departing waiter cancels abandoned work. Distinct targets no longer +fail because another target is validating. Outbound observations share the existing +ten execution slots and serialize by account. Preparation has a ten-minute bound, +continues after a target failure, and reports partial readiness after rechecking +expiry. Policy, credential/model-map and exit changes invalidate displayed evidence; +actual dispatch still verifies the finalized request fingerprint. + +Status adds `setup`, `observed_since` and `recent_usage`. The last actual dispatch +and in-process counters are retained per configured target/model/protocol/purpose; +configuration publication resets this index. HTTP/SSE, pooled WS and native WS +relay observations record application independently from upstream completion. +Diagnostic and business requests are distinguished. This index contains no request +payloads or credential/cookie values; the existing administrator error records remain. + +`POST /admin/codex-gateway-borrow/diagnose` streams an explicit fixed-account +comparison with `account_id`, `model`, `transport` (`http` or `ws`), and optional +`request_limit` (1–8, default 8). Its ordinary and borrowed requests use the normal +sender with observation isolation and a fixed short prompt. WS uses separate +short-lived diagnostic anchors and performs one continuation on each socket; +account/global WS settings must already permit WS. Source/target probes and all +compatibility sends count toward the same hard request limit. Events include +`phase`, `request`, `result`, `done`, and `error`; results contain original output, +completion, actual borrow application and reported model. Cancellation releases +owned work and sockets. The UI retains the current run and can download JSON; +there is no new table or migration. These observations do not prove intelligence +or physical model identity. Normal token refresh remains permitted. + +Ranxi reference: v2.10.2 (`d3e43f2de33af9e987cffa511d76dfabbcd749da`). Its five +fingerprint, gateway-cookie, target-probe and automatic-setup core files are +unchanged from the retained v2.10.0 reference. OAuth search-history compatibility +is supplied by the official v0.2.15 integration. Automatic group membership, +Mihomo exit rotation and removal of harvesting attempt/concurrency bounds are +not adopted; existing account mappings, network routes and protocol switches stay +under their existing owners. The historical `/admin/codex-gateway-borrow/tests` endpoints retain their cache-qualified, 90-second request contract. They share execution admission with diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index abc00084979e..51d944a4d009 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.github/workflows/downstream-verify.yml @@ -47,6 +47,9 @@ jobs: run: | go test -p=1 -tags unit ./internal/service ./internal/repository ./internal/handler ./internal/handler/admin -run '^Test(CodexGatewayBorrow|Pelican|AccountObservation|ObservedAccountLease)' -count=1 go test -p=1 -tags integration ./internal/repository -run '^TestCodexGatewayBorrowTestRepositoryIntegrationLifecycle$' -count=1 + - name: Codex preparation coordination race tests + working-directory: backend + run: go test -p=1 -race ./internal/service -run '^TestCodexBorrow(Flight|DifferentTargets)' -count=1 - name: Upstream automation classifier tests run: | python3 .github/scripts/test_upstream_risk.py @@ -145,6 +148,7 @@ jobs: src/components/admin/codex/__tests__/BorrowPelicanPreview.spec.ts \ src/views/admin/__tests__/CodexGatewayBorrowView.spec.ts \ src/views/admin/__tests__/CodexGatewayBorrowStatusView.spec.ts \ + src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts \ src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts \ src/api/admin/__tests__/pelicanTests.spec.ts \ src/components/admin/__tests__/CodexBorrowNav.spec.ts \ diff --git a/backend/internal/handler/admin/codex_gateway_borrow_handler.go b/backend/internal/handler/admin/codex_gateway_borrow_handler.go index 5f5868fd1d26..5df1d9da04ad 100644 --- a/backend/internal/handler/admin/codex_gateway_borrow_handler.go +++ b/backend/internal/handler/admin/codex_gateway_borrow_handler.go @@ -66,7 +66,7 @@ func (h *CodexGatewayBorrowHandler) SaveConfig(c *gin.Context) { func (h *CodexGatewayBorrowHandler) Status(c *gin.Context) { codexGatewayBorrowPrivateHeaders(c) - response.Success(c, h.core.Status()) + response.Success(c, h.core.CurrentStatus(c.Request.Context())) } func (h *CodexGatewayBorrowHandler) Prepare(c *gin.Context) { @@ -268,3 +268,40 @@ func (h *CodexGatewayBorrowHandler) ServePreview(c *gin.Context) { } c.Data(http.StatusOK, "text/html; charset=utf-8", []byte(html)) } + +// Diagnose streams explicit administrator observations; opening a page never invokes it. +func (h *CodexGatewayBorrowHandler) Diagnose(c *gin.Context) { + codexGatewayBorrowPrivateHeaders(c) + var request service.CodexBorrowDiagnosticRequest + if err := c.ShouldBindJSON(&request); err != nil { + response.BadRequest(c, err.Error()) + return + } + middleware.SetAuditExtra(c, map[string]any{"result": "codex_gateway_borrow_diagnostic", "account_id": request.AccountID, "model": request.Model, "transport": request.Transport}) + c.Header("Content-Type", "text/event-stream") + c.Header("X-Accel-Buffering", "no") + c.Status(http.StatusOK) + var emission sync.Mutex + var observedRequests int32 + emit := func(event service.CodexBorrowDiagnosticEvent) { + emission.Lock() + defer emission.Unlock() + if event.Requests > observedRequests { + observedRequests = event.Requests + } + if event.Type == "error" { + event.Requests = observedRequests + } + if c.Request.Context().Err() != nil { + return + } + raw, err := json.Marshal(event) + if err == nil { + _, _ = fmt.Fprintf(c.Writer, "data: %s\n\n", raw) + c.Writer.Flush() + } + } + if err := h.core.Diagnose(c.Request.Context(), request, emit); err != nil { + emit(service.CodexBorrowDiagnosticEvent{Type: "error", Error: err.Error(), Limit: service.CodexBorrowDiagnosticMaxRequests}) + } +} diff --git a/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go b/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go index d48b7e6181fd..8bd65bacbc36 100644 --- a/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go +++ b/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go @@ -19,6 +19,36 @@ import ( "github.com/stretchr/testify/require" ) +func TestCodexGatewayBorrowDiagnosePreflightNeverDispatches(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + name, body string + status int + }{ + {"disabled", `{"account_id":1,"model":"gpt-6-astra","transport":"http"}`, http.StatusOK}, + {"unsupported transport", `{"account_id":1,"model":"gpt-6-astra","transport":"invalid"}`, http.StatusOK}, + {"malformed", `{`, http.StatusBadRequest}, + } { + t.Run(test.name, func(t *testing.T) { + router := gin.New() + handler := NewCodexGatewayBorrowHandler(nil, nil, nil) + router.POST("/diagnose", handler.Diagnose) + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/diagnose", strings.NewReader(test.body)) + request.Header.Set("Content-Type", "application/json") + router.ServeHTTP(response, request) + require.Equal(t, test.status, response.Code) + require.Equal(t, "private, no-store", response.Header().Get("Cache-Control")) + if test.status == http.StatusOK { + require.Contains(t, response.Header().Get("Content-Type"), "text/event-stream") + require.Contains(t, response.Body.String(), `"type":"error"`) + require.Contains(t, response.Body.String(), `"requests":0`) + require.NotContains(t, response.Body.String(), `"type":"request"`) + } + }) + } +} + type codexGatewayBorrowHandlerTestRepo struct { task *service.CodexGatewayBorrowTestTask result *service.CodexGatewayBorrowTestResult diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index ffaa02daf49f..1565fa988196 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -150,6 +150,7 @@ func registerCodexGatewayBorrowRoutes(admin *gin.RouterGroup, h *handler.Handler borrow.GET("/status", h.Admin.CodexGatewayBorrow.Status) borrow.POST("/prepare", h.Admin.CodexGatewayBorrow.Prepare) borrow.POST("/verify", h.Admin.CodexGatewayBorrow.Verify) + borrow.POST("/diagnose", h.Admin.CodexGatewayBorrow.Diagnose) borrow.POST("/tests", h.Admin.CodexGatewayBorrow.StartTests) borrow.GET("/tests", h.Admin.CodexGatewayBorrow.ListTests) borrow.GET("/tests/:id", h.Admin.CodexGatewayBorrow.GetTest) diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 2aa5233bae3d..f8e23c4f12a8 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -17,13 +17,13 @@ import ( "io" "net/http" "slices" + "sort" "strings" "sync" "time" infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" - "golang.org/x/sync/singleflight" ) const SettingKeyCodexGatewayBorrowConfig = "codex_gateway_borrow_config" @@ -178,15 +178,18 @@ type CodexGatewayBorrowTargetStatus struct { } type CodexGatewayBorrowStatus struct { - Enabled bool `json:"enabled"` - Revision uint64 `json:"revision"` - GeneratedAt time.Time `json:"generated_at"` - Preparing bool `json:"preparing"` - Config CodexGatewayBorrowConfig `json:"config"` - Candidate *CodexGatewayBorrowCandidateStatus `json:"candidate,omitempty"` - Sources []CodexGatewayBorrowSourceStatus `json:"sources"` - Targets []CodexGatewayBorrowTargetStatus `json:"targets"` - ModelEfforts map[string][]string `json:"model_efforts"` + ObservedSince time.Time `json:"observed_since"` + RecentUsage []CodexGatewayBorrowUsage `json:"recent_usage"` + Setup CodexGatewayBorrowSetup `json:"setup"` + Enabled bool `json:"enabled"` + Revision uint64 `json:"revision"` + GeneratedAt time.Time `json:"generated_at"` + Preparing bool `json:"preparing"` + Config CodexGatewayBorrowConfig `json:"config"` + Candidate *CodexGatewayBorrowCandidateStatus `json:"candidate,omitempty"` + Sources []CodexGatewayBorrowSourceStatus `json:"sources"` + Targets []CodexGatewayBorrowTargetStatus `json:"targets"` + ModelEfforts map[string][]string `json:"model_efforts"` } // Application contains only opaque identities. The selected cookie itself is @@ -210,19 +213,25 @@ type codexGatewayBorrowTargetKey struct { model string } type codexGatewayBorrowTargetCheck struct { - key string - cookieKey string - expires time.Time - retryAfter time.Time - validating bool - result CodexGatewayBorrowVerification + validationID uint64 + policyRevision uint64 + identityRevision string + key string + cookieKey string + expires time.Time + retryAfter time.Time + validating bool + result CodexGatewayBorrowVerification } type CodexGatewayBorrowService struct { - settings SettingRepository - accounts AccountRepository - gateway *OpenAIGatewayService - upstream CodexGatewayBorrowProbeUpstream + validationSequence uint64 + sourceSequence uint64 + qualifications map[string]codexGatewayBorrowTargetCheck + settings SettingRepository + accounts AccountRepository + gateway *OpenAIGatewayService + upstream CodexGatewayBorrowProbeUpstream // A profile supplied by a business caller is reused verbatim. Current Codex // HTTP and WS business paths use standard TLS, so manual preparation uses nil. tlsProfiles *TLSFingerprintProfileService @@ -240,8 +249,12 @@ type CodexGatewayBorrowService struct { prepareRuns int prepareFailedUntil time.Time prepareFailure string - sourceFlight singleflight.Group - targetProbeMu sync.Mutex + sourceFlight codexBorrowFlights + targetFlight codexBorrowFlights + setupFlight codexBorrowFlights + setup CodexGatewayBorrowSetup + usage map[codexBorrowUsageKey]CodexGatewayBorrowUsage + observedSince time.Time wsAnchors codexGatewayBorrowWSAnchorStore } @@ -348,8 +361,15 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.revision++ s.config, s.loaded = cloneCodexGatewayBorrowConfig(cfg), true s.candidate = nil + s.setup = CodexGatewayBorrowSetup{} + if saved && cfg.Enabled { + s.setup = CodexGatewayBorrowSetup{State: "queued", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} + } + s.usage = nil + s.observedSince = time.Now() s.sources = map[int64]CodexGatewayBorrowSourceStatus{} s.targets = map[codexGatewayBorrowTargetKey]codexGatewayBorrowTargetCheck{} + s.qualifications = map[string]codexGatewayBorrowTargetCheck{} s.prepareFailedUntil, s.prepareFailure, s.preparing, s.prepareRuns = time.Time{}, "", false, 0 ctx := s.revisionCtx s.mu.Unlock() @@ -422,11 +442,15 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { s.mu.Lock() defer s.mu.Unlock() now := time.Now() - out := CodexGatewayBorrowStatus{Enabled: s.config.Enabled, Revision: s.revision, GeneratedAt: now, Preparing: s.preparing || s.prepareRuns > 0, - Config: cloneCodexGatewayBorrowConfig(s.config), Sources: []CodexGatewayBorrowSourceStatus{}, Targets: []CodexGatewayBorrowTargetStatus{}, ModelEfforts: codexGatewayBorrowModelEfforts()} + out := CodexGatewayBorrowStatus{Enabled: s.config.Enabled, Revision: s.revision, GeneratedAt: now, Preparing: s.preparing || s.prepareRuns > 0 || s.setup.State == "queued", + ObservedSince: s.observedSince, RecentUsage: []CodexGatewayBorrowUsage{}, Setup: s.setup, Config: cloneCodexGatewayBorrowConfig(s.config), Sources: []CodexGatewayBorrowSourceStatus{}, Targets: []CodexGatewayBorrowTargetStatus{}, ModelEfforts: codexGatewayBorrowModelEfforts()} if c := s.candidate; c != nil { out.Candidate = &CodexGatewayBorrowCandidateStatus{SourceAccountID: c.sourceID, ExpiresAt: c.expires, RemainingSeconds: borrowRemaining(c.expires, now), CookieFingerprint: borrowHash(c.cookie.Value)} } + for _, usage := range s.usage { + out.RecentUsage = append(out.RecentUsage, usage) + } + sort.Slice(out.RecentUsage, func(i, j int) bool { return out.RecentUsage[i].StartedAt.After(out.RecentUsage[j].StartedAt) }) for _, id := range s.config.SourceAccountIDs { row, ok := s.sources[id] if !ok { @@ -448,7 +472,7 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { row := CodexGatewayBorrowTargetStatus{CodexGatewayBorrowVerification: CodexGatewayBorrowVerification{AccountID: id, Model: model, Reason: "not_verified"}, State: "waiting"} if c, ok := s.targets[codexGatewayBorrowTargetKey{id, model}]; ok { row.CodexGatewayBorrowVerification = c.result - row.CacheValid = c.result.Success && s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) + row.CacheValid = c.policyRevision == currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success && s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) expires := c.expires if s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && s.candidate.expires.Before(expires) { expires = s.candidate.expires @@ -465,6 +489,12 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { } else if c.result.Success { row.State, row.Reason = "expired", "route_expired" } + if c.result.Success && s.candidate != nil && c.cookieKey != borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) { + row.State, row.Reason = "waiting", "target_route_changed" + } + if c.policyRevision != currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success { + row.State, row.Reason = "waiting", "identity_changed" + } if now.Before(c.retryAfter) && !c.result.Success { retry := c.retryAfter row.RetryAfter = &retry @@ -497,7 +527,7 @@ func borrowRevisionContext(parent, revision context.Context, timeout time.Durati return ctx, func() { stop(); cancel() } } -func (s *CodexGatewayBorrowService) Prepare(ctx context.Context) (status CodexGatewayBorrowStatus, err error) { +func (s *CodexGatewayBorrowService) Prepare(ctx context.Context) (CodexGatewayBorrowStatus, error) { if s == nil { return s.Status(), ErrCodexGatewayBorrowUnavailable } @@ -507,46 +537,92 @@ func (s *CodexGatewayBorrowService) Prepare(ctx context.Context) (status CodexGa if !cfg.Enabled { return s.Status(), ErrCodexGatewayBorrowUnavailable } - s.mu.Lock() - if s.revision != rev { - s.mu.Unlock() - return s.Status(), ErrCodexGatewayBorrowChanged - } - s.prepareRuns++ - s.mu.Unlock() - defer func() { + _, err := s.setupFlight.do(ctx, revisionCtx, fmt.Sprint(rev), 10*time.Minute, func(runCtx context.Context) (any, error) { + progress := CodexGatewayBorrowSetup{State: "running", Phase: "source", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} + s.setSetup(rev, progress) s.mu.Lock() - if s.revision == rev { - s.prepareRuns-- + if s.revision != rev { + s.mu.Unlock() + return nil, ErrCodexGatewayBorrowChanged } + s.prepareRuns++ s.mu.Unlock() - status = s.Status() - }() - if err := s.prepareSource(ctx, rev, revisionCtx); err != nil { - return s.Status(), err - } - for _, id := range cfg.TargetAccountIDs { - for _, model := range cfg.Models { - if revisionCtx.Err() != nil { - return s.Status(), ErrCodexGatewayBorrowChanged + defer func() { + s.mu.Lock() + if s.revision == rev { + s.prepareRuns-- } - a, req, proxy, err := s.accountTemplate(ctx, id, model) + s.mu.Unlock() + }() + finish := func(err error) { + now := time.Now() + progress.FinishedAt = &now + progress.Phase = "complete" if err != nil { - return s.Status(), err + progress.Error = err.Error() + progress.State = "failed" + } else { + ready := 0 + for _, row := range s.Status().Targets { + if row.CacheValid { + ready++ + } + } + progress.State = "failed" + if ready == progress.Total { + progress.State = "ready" + } else if ready > 0 { + progress.State = "partial" + } } - if _, _, err := s.Apply(req, a, model, proxy, nil, false); err != nil { - return s.Status(), err + s.setSetup(rev, progress) + } + if err := s.prepareSource(runCtx, rev, revisionCtx); err != nil { + finish(err) + return nil, err + } + var lastErr error + for _, id := range cfg.TargetAccountIDs { + for _, model := range cfg.Models { + if err := runCtx.Err(); err != nil { + finish(err) + return nil, err + } + progress.Phase, progress.AccountID, progress.Model = "target", id, model + s.setSetup(rev, progress) + a, req, proxy, itemErr := s.accountTemplate(runCtx, id, model) + if itemErr == nil { + _, _, itemErr = s.Apply(req, a, model, proxy, nil, false) + } else { + s.mu.Lock() + if s.revision == rev { + s.targets[codexGatewayBorrowTargetKey{id, model}] = codexGatewayBorrowTargetCheck{result: CodexGatewayBorrowVerification{AccountID: id, Model: model, CheckedAt: time.Now(), Reason: "account_unavailable", Error: itemErr.Error()}} + } + s.mu.Unlock() + } + progress.Completed++ + if itemErr != nil { + progress.Failed++ + lastErr = itemErr + } + s.setSetup(rev, progress) } } - } - return s.Status(), nil + if progress.Failed == progress.Total { + finish(lastErr) + return nil, lastErr + } + finish(nil) + return nil, nil + }) + return s.Status(), err } func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint64, revisionCtx context.Context) error { if s == nil { return ErrCodexGatewayBorrowUnavailable } - result := s.sourceFlight.DoChan(fmt.Sprint(rev), func() (any, error) { + _, err := s.sourceFlight.do(ctx, revisionCtx, fmt.Sprint(rev), codexGatewayBorrowTimeout, func(ctx context.Context) (any, error) { s.mu.Lock() if s.revision != rev || !s.config.Enabled { s.mu.Unlock() @@ -566,19 +642,28 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 } ids := append([]int64{}, s.config.SourceAccountIDs...) s.preparing = true + s.sourceSequence++ + sequence := s.sourceSequence s.mu.Unlock() probeCtx, cancel := borrowRevisionContext(ctx, revisionCtx, codexGatewayBorrowTimeout) - defer cancel() + defer func() { + cancel() + s.mu.Lock() + if s.revision == rev && s.sourceSequence == sequence { + s.preparing = false + } + s.mu.Unlock() + }() var lastErr error = ErrCodexGatewayBorrowUnavailable for _, id := range ids { - if probeCtx.Err() != nil { + if errors.Is(probeCtx.Err(), context.Canceled) { lastErr = probeCtx.Err() break } candidate, detail, err := s.acquireSource(probeCtx, id) now := time.Now() s.mu.Lock() - if s.revision != rev || revisionCtx.Err() != nil { + if s.revision != rev || revisionCtx.Err() != nil || s.sourceSequence != sequence || errors.Is(probeCtx.Err(), context.Canceled) { s.mu.Unlock() return nil, ErrCodexGatewayBorrowChanged } @@ -590,7 +675,7 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 candidate.expires = old.expires } if s.candidate == nil || s.candidate.cookie.Value != candidate.cookie.Value || s.candidate.cookie.Path != candidate.cookie.Path { - s.targets = map[codexGatewayBorrowTargetKey]codexGatewayBorrowTargetCheck{} + s.qualifications = map[string]codexGatewayBorrowTargetCheck{} } s.candidate = candidate row.State, row.Reason, row.Error, row.ExpiresAt = "ready", "source_qualified", "", &candidate.expires @@ -610,7 +695,7 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Unlock() } s.mu.Lock() - if s.revision == rev { + if s.revision == rev && s.sourceSequence == sequence && !errors.Is(probeCtx.Err(), context.Canceled) { s.preparing = false s.prepareFailedUntil = time.Now().Add(codexGatewayBorrowFailureWait) s.prepareFailure = lastErr.Error() @@ -618,20 +703,19 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Unlock() return nil, lastErr }) - select { - case <-ctx.Done(): - return ctx.Err() - case done := <-result: - return done.Err - } + return err } func borrowRequestFingerprint(req *http.Request, model, proxy string, profile *tlsfingerprint.Profile, cookie string) string { tlsJSON, _ := json.Marshal(profile) - identity, _ := json.Marshal([]string{cookie, model, proxy, req.Header.Get("Authorization"), req.Header.Get("ChatGPT-Account-ID"), req.Header.Get("User-Agent"), req.Header.Get("Originator"), req.Header.Get("Version"), req.Header.Get("X-Codex-Turn-State"), string(tlsJSON)}) + identity, _ := json.Marshal([]string{cookie, model, proxy, req.URL.EscapedPath(), req.Header.Get("Authorization"), req.Header.Get("ChatGPT-Account-ID"), req.Header.Get("User-Agent"), req.Header.Get("Originator"), req.Header.Get("Version"), req.Header.Get("X-Codex-Turn-State"), string(tlsJSON)}) return borrowHash(string(identity)) } +func borrowTargetFingerprint(req *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, cookie string) string { + return borrowHash(borrowRequestFingerprint(req, model, proxy, profile, cookie) + fmt.Sprint(currentCodexFingerprintPolicyForAccount(account).revision)) +} + // Apply changes only __oailb on a request clone. The target's auth, STATE, // __cflb, other cookies, body and proxy stay exactly as supplied by its caller. // cachedOnly is used by manual generation: it never acquires or validates. @@ -647,6 +731,9 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m if s == nil || !CodexGatewayBorrowRequestEligible(req) || account == nil { return req, nil, nil } + if d := borrowDiagnosticFromContext(req.Context()); d != nil && !d.borrow { + return req, nil, nil + } s.mu.Lock() if !s.config.Enabled || !slices.Contains(s.config.Models, model) || !slices.Contains(s.config.TargetAccountIDs, account.ID) { s.mu.Unlock() @@ -672,10 +759,10 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m if candidate == nil || !time.Now().Before(candidate.expires) || !borrowCookiePathMatches(req.URL.Path, candidate.cookie.Path) { return req, nil, ErrCodexGatewayBorrowUnavailable } - key := borrowRequestFingerprint(req, model, proxy, profile, candidate.cookie.Value) + policyRevision := currentCodexFingerprintPolicyForAccount(account).revision + key := borrowTargetFingerprint(req, account, model, proxy, profile, candidate.cookie.Value) targetKey := codexGatewayBorrowTargetKey{account.ID, model} - // Cache hits must precede the global validation lock. One cold target must - // not make another already-qualified target fail with "busy". + // Cached proofs do not wait for preparation on another target or identity. if !force { if check, ok, err := s.cachedTarget(rev, targetKey, key, candidate); ok || err != nil { if err != nil { @@ -687,47 +774,84 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m if cachedOnly { return req, nil, ErrCodexGatewayBorrowUnavailable } - if !s.targetProbeMu.TryLock() { - return req, nil, ErrCodexGatewayBorrowBusy - } - defer s.targetProbeMu.Unlock() - if !force { - if check, ok, err := s.cachedTarget(rev, targetKey, key, candidate); ok || err != nil { - if err != nil { - return req, nil, err + value, err := s.targetFlight.do(req.Context(), revisionCtx, fmt.Sprintf("%d:%d:%s:%s", rev, account.ID, model, key), codexGatewayBorrowTimeout, func(operation context.Context) (any, error) { + if !force { + if check, ok, err := s.cachedTarget(rev, targetKey, key, candidate); ok || err != nil { + if err != nil { + return nil, err + } + return check, nil } - return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires), nil } - } - s.mu.Lock() - if !s.candidateCurrentLocked(rev, candidate) { - if s.revision == rev { - delete(s.targets, targetKey) + s.mu.Lock() + if operation.Err() != nil || !s.candidateCurrentLocked(rev, candidate) { + if s.revision == rev { + delete(s.targets, targetKey) + } + s.mu.Unlock() + return nil, ErrCodexGatewayBorrowChanged } + s.validationSequence++ + validationID := s.validationSequence + s.targets[targetKey] = codexGatewayBorrowTargetCheck{validationID: validationID, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), validating: true, + result: CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, Reason: "validating", CheckedAt: time.Now()}} s.mu.Unlock() - return req, nil, ErrCodexGatewayBorrowChanged - } - s.targets[targetKey] = codexGatewayBorrowTargetCheck{key: key, cookieKey: borrowHash(candidate.cookie.Value), validating: true, - result: CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, Reason: "validating", CheckedAt: time.Now()}} - s.mu.Unlock() - probeCtx, cancel := borrowRevisionContext(req.Context(), revisionCtx, codexGatewayBorrowTimeout) - defer cancel() - result := s.probeTarget(probeCtx, req, account, model, proxy, profile, candidate) - s.mu.Lock() - if !s.candidateCurrentLocked(rev, candidate) { - if s.revision == rev { - delete(s.targets, targetKey) + probeCtx, cancel := borrowRevisionContext(operation, revisionCtx, codexGatewayBorrowTimeout) + defer cancel() + credential, credentialErr := resolveCredentialAccount(probeCtx, s.accounts, account) + identityRevision := borrowAccountIdentityRevision(account, credential) + var result CodexGatewayBorrowVerification + if credentialErr != nil { + result = CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), Reason: "account_unavailable", Error: credentialErr.Error()} + } else { + result = s.probeTarget(probeCtx, req, account, model, proxy, profile, candidate) } + s.mu.Lock() + if errors.Is(operation.Err(), context.Canceled) || !s.candidateCurrentLocked(rev, candidate) { + if s.revision == rev && s.targets[targetKey].validationID == validationID { + delete(s.targets, targetKey) + } + s.mu.Unlock() + return nil, ErrCodexGatewayBorrowChanged + } + check := codexGatewayBorrowTargetCheck{validationID: validationID, identityRevision: identityRevision, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: result} + if !result.Success { + check.retryAfter = time.Now().Add(codexGatewayBorrowFailureWait) + } + if s.targets[targetKey].validationID == validationID { + s.targets[targetKey] = check + } + if s.qualifications == nil { + s.qualifications = make(map[string]codexGatewayBorrowTargetCheck) + } + qualificationKey := fmt.Sprintf("%d:%s:%s", account.ID, model, key) + // Bound request-specific STATE/header variants; expired proofs are discarded. + if len(s.qualifications) >= 1024 { + oldestKey := "" + var oldest time.Time + for k, v := range s.qualifications { + if !time.Now().Before(v.expires) { + delete(s.qualifications, k) + } else if oldestKey == "" || v.result.CheckedAt.Before(oldest) { + oldestKey, oldest = k, v.result.CheckedAt + } + } + if len(s.qualifications) >= 1024 { + delete(s.qualifications, oldestKey) + } + } + s.qualifications[qualificationKey] = check s.mu.Unlock() - return req, nil, ErrCodexGatewayBorrowChanged - } - check := codexGatewayBorrowTargetCheck{key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: result} - if !result.Success { - check.retryAfter = time.Now().Add(codexGatewayBorrowFailureWait) + if !result.Success { + return nil, ErrCodexGatewayBorrowUnavailable + } + return check, nil + }) + if err != nil { + return req, nil, err } - s.targets[targetKey] = check - s.mu.Unlock() - if !result.Success { + check, ok := value.(codexGatewayBorrowTargetCheck) + if !ok { return req, nil, ErrCodexGatewayBorrowUnavailable } return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires), nil @@ -743,7 +867,10 @@ func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBor if !s.candidateCurrentLocked(rev, candidate) { return codexGatewayBorrowTargetCheck{}, false, ErrCodexGatewayBorrowChanged } - old, ok := s.targets[key] + old, ok := s.qualifications[fmt.Sprintf("%d:%s:%s", key.accountID, key.model, fingerprint)] + if !ok { + old, ok = s.targets[key] + } if !ok || old.key != fingerprint || old.validating { return old, false, nil } @@ -765,7 +892,7 @@ func borrowApplication(candidate *codexGatewayBorrowCandidate, fingerprint strin } func borrowApplyCookie(req *http.Request, candidate *codexGatewayBorrowCandidate, check codexGatewayBorrowTargetCheck) *http.Request { - clone := req.Clone(req.Context()) + clone := req.Clone(context.WithValue(req.Context(), codexBorrowAppliedContextKey{}, true)) replaceCodexGatewayBorrowCookie(clone.Header, "__oailb", candidate.cookie.Value) return clone } diff --git a/backend/internal/service/codex_gateway_borrow_coordination.go b/backend/internal/service/codex_gateway_borrow_coordination.go new file mode 100644 index 000000000000..7da7e3672ed9 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_coordination.go @@ -0,0 +1,90 @@ +package service + +import ( + "context" + "sync" + "time" +) + +// A preparation belongs to its configuration revision, not to the first caller. +// Cancelling a waiter only cancels the work when no other waiter needs it. +type codexBorrowFlights struct { + mu sync.Mutex + active map[string]*codexBorrowFlight +} + +type codexBorrowFlight struct { + done chan struct{} + cancel context.CancelFunc + waiters int + value any + err error +} + +func (g *codexBorrowFlights) do(ctx, revision context.Context, key string, timeout time.Duration, run func(context.Context) (any, error)) (any, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + g.mu.Lock() + if g.active == nil { + g.active = make(map[string]*codexBorrowFlight) + } + f := g.active[key] + if f == nil { + operation, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + stop := context.AfterFunc(revision, cancel) + f = &codexBorrowFlight{done: make(chan struct{}), cancel: cancel} + g.active[key] = f + go func() { + defer stop() + defer cancel() + f.value, f.err = run(operation) + g.mu.Lock() + if g.active[key] == f { + delete(g.active, key) + } + close(f.done) + g.mu.Unlock() + }() + } + f.waiters++ + g.mu.Unlock() + defer func() { + g.mu.Lock() + f.waiters-- + if f.waiters == 0 { + f.cancel() + if g.active[key] == f { + delete(g.active, key) + } + } + g.mu.Unlock() + }() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-f.done: + return f.value, f.err + } +} + +type CodexGatewayBorrowSetup struct { + State string `json:"state"` + Phase string `json:"phase"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Completed int `json:"completed"` + Total int `json:"total"` + Failed int `json:"failed"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *CodexGatewayBorrowService) setSetup(revision uint64, setup CodexGatewayBorrowSetup) { + s.mu.Lock() + defer s.mu.Unlock() + if s.revision == revision { + s.setup = setup + } +} diff --git a/backend/internal/service/codex_gateway_borrow_coordination_test.go b/backend/internal/service/codex_gateway_borrow_coordination_test.go new file mode 100644 index 000000000000..43eaeded7b86 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_coordination_test.go @@ -0,0 +1,163 @@ +package service + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" +) + +func TestCodexBorrowFlightCancellationIsPerWaiter(t *testing.T) { + var flights codexBorrowFlights + var calls atomic.Int32 + started, finish := make(chan struct{}), make(chan struct{}) + run := func(ctx context.Context) (any, error) { + calls.Add(1) + close(started) + select { + case <-finish: + return "ready", nil + case <-ctx.Done(): + return nil, ctx.Err() + } + } + first, cancel := context.WithCancel(context.Background()) + firstResult, secondResult := make(chan error, 1), make(chan error, 1) + go func() { + _, err := flights.do(first, context.Background(), "same", time.Second, run) + firstResult <- err + }() + <-started + go func() { + value, err := flights.do(context.Background(), context.Background(), "same", time.Second, run) + if err == nil && value != "ready" { + err = errors.New("missing shared value") + } + secondResult <- err + }() + require.Eventually(t, func() bool { flights.mu.Lock(); defer flights.mu.Unlock(); return flights.active["same"].waiters == 2 }, time.Second, time.Millisecond) + cancel() + require.ErrorIs(t, <-firstResult, context.Canceled) + close(finish) + require.NoError(t, <-secondResult) + require.EqualValues(t, 1, calls.Load()) +} + +func TestCodexBorrowFlightRevisionCancelsPreparation(t *testing.T) { + var flights codexBorrowFlights + revision, cancel := context.WithCancel(context.Background()) + started := make(chan struct{}) + result := make(chan error, 1) + go func() { + _, err := flights.do(context.Background(), revision, "old", time.Second, func(ctx context.Context) (any, error) { close(started); <-ctx.Done(); return nil, ctx.Err() }) + result <- err + }() + <-started + cancel() + require.ErrorIs(t, <-result, context.Canceled) +} + +func TestCodexBorrowPreparePreservesPartialSuccess(t *testing.T) { + s := newBorrowCoreTest(t, func(r *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 2 { + return nil, errors.New("target two rejected") + } + return borrowCoreResponse("gpt-6-astra", "OK", "stable-target-state", ""), nil + }, borrowCoreAccount(1), borrowCoreAccount(2), borrowCoreAccount(3)) + s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2, 3}, Models: []string{"gpt-6-astra"}}, false) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + status, err := s.Prepare(context.Background()) + require.NoError(t, err) + require.Equal(t, "partial", status.Setup.State) + require.Equal(t, 2, status.Setup.Completed) + require.Equal(t, 1, status.Setup.Failed) + require.False(t, status.Targets[0].CacheValid) + require.True(t, status.Targets[1].CacheValid) + repo, ok := s.accounts.(*borrowCoreAccounts) + require.True(t, ok) + account := repo.rows[3] + account.Credentials["model_mapping"] = map[string]any{"gpt-6-astra": "gpt-6-luna"} + require.Equal(t, "model_mapping_mismatch", s.CurrentStatus(context.Background()).Targets[1].Reason) +} + +func TestCodexBorrowDifferentTargetsDoNotReturnBusy(t *testing.T) { + started, finish := make(chan struct{}), make(chan struct{}) + var blocked atomic.Bool + s := newBorrowCoreTest(t, func(r *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 2 && blocked.CompareAndSwap(false, true) { + close(started) + select { + case <-finish: + case <-r.Context().Done(): + return nil, r.Context().Err() + } + } + return borrowCoreResponse("gpt-6-astra", "OK", "stable", ""), nil + }, borrowCoreAccount(1), borrowCoreAccount(2), borrowCoreAccount(3)) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + first := make(chan error, 1) + go func() { _, err := s.Verify(context.Background(), 2, "gpt-6-astra"); first <- err }() + <-started + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := s.Verify(ctx, 3, "gpt-6-astra") + close(finish) + require.NoError(t, err) + require.NoError(t, <-first) +} + +func TestCodexBorrowUsagePreservesStreamAndNewestAttempt(t *testing.T) { + s := newBorrowCoreTest(t, nil) + raw := borrowCoreSSE("gpt-6-astra", "OK") + tracker := s.beginUsage(context.Background(), 2, "gpt-6-astra", "http", true) + response, _ := s.trackHTTPResponse(tracker, &http.Response{StatusCode: 200, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(raw))}, nil) + read, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.NoError(t, response.Body.Close()) + require.Equal(t, raw, string(read)) + status := s.Status() + require.Len(t, status.RecentUsage, 1) + require.Equal(t, "completed", status.RecentUsage[0].Outcome) + old := s.beginUsage(context.Background(), 2, "gpt-6-astra", "http", false) + newer := s.beginUsage(context.Background(), 2, "gpt-6-astra", "http", true) + old.finish(errors.New("old failure")) + newer.observe([]byte(`{"type":"response.completed","response":{"id":"latest","model":"gpt-6-astra","status":"completed"}}`)) + newer.finish(nil) + require.Equal(t, "latest", s.Status().RecentUsage[0].RequestID) + require.EqualValues(t, 3, s.Status().RecentUsage[0].Count) +} + +func TestCodexBorrowDiagnosticBudgetIncludesPreparation(t *testing.T) { + var requests atomic.Int32 + ctx := context.WithValue(context.Background(), codexBorrowDiagnosticContextKey{}, &codexBorrowDiagnostic{requests: &requests, limit: 2}) + require.NoError(t, consumeBorrowDiagnosticRequest(ctx)) + require.NoError(t, consumeBorrowDiagnosticRequest(WithCodexGatewayBorrowObservation(ctx))) + require.ErrorIs(t, consumeBorrowDiagnosticRequest(ctx), ErrCodexBorrowDiagnosticBudget) + require.EqualValues(t, 2, requests.Load()) +} + +func TestCodexBorrowDistinctRequestProofsRemainReusable(t *testing.T) { + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) { + return borrowCoreResponse("gpt-6-astra", "OK", "stable", ""), nil + }, a) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + _, first, _, err := s.accountTemplate(context.Background(), 2, "gpt-6-astra") + require.NoError(t, err) + second := first.Clone(first.Context()) + second.Header.Set("X-Codex-Turn-State", "a-different-client-state") + _, _, err = s.Apply(first, a, "gpt-6-astra", "", nil, false) + require.NoError(t, err) + _, _, err = s.Apply(second, a, "gpt-6-astra", "", nil, false) + require.NoError(t, err) + _, applied, err := s.Apply(first, a, "gpt-6-astra", "", nil, true) + require.NoError(t, err) + require.True(t, applied.Applied, "another request's validation must not replace this proof") +} diff --git a/backend/internal/service/codex_gateway_borrow_core_test.go b/backend/internal/service/codex_gateway_borrow_core_test.go index 8bb9e4825bd3..30b9b0351237 100644 --- a/backend/internal/service/codex_gateway_borrow_core_test.go +++ b/backend/internal/service/codex_gateway_borrow_core_test.go @@ -256,11 +256,9 @@ func TestCodexGatewayBorrowTargetCacheModelAndFingerprintIsolation(t *testing.T) require.Contains(t, template.Header.Get("Cookie"), "__oailb=old", "input request is not modified") } require.Equal(t, 4, calls) - s.targetProbeMu.Lock() _, app, err := s.Apply(template, a, "gpt-6.1-sol", "target-exit", nil, false) require.NoError(t, err) require.True(t, app.Applied, "hot cache is read before global target lock") - s.targetProbeMu.Unlock() for _, header := range []string{"Authorization", "ChatGPT-Account-ID", "User-Agent", "Originator", "Version", "X-Codex-Turn-State"} { req := template.Clone(template.Context()) req.Header.Set(header, req.Header.Get(header)+"-changed") @@ -372,8 +370,8 @@ func TestCodexGatewayBorrowGatePreserves100MiBBodyAndOtherRoutes(t *testing.T) { require.NoError(t, err) req.ContentLength = 100 << 20 borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - key := borrowRequestFingerprint(req, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value) - s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{key: key, cookieKey: borrowHash(s.candidate.cookie.Value), expires: s.candidate.expires, result: CodexGatewayBorrowVerification{Success: true}} + key := borrowTargetFingerprint(req, a, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value) + s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{policyRevision: currentCodexFingerprintPolicyForAccount(a).revision, key: key, cookieKey: borrowHash(s.candidate.cookie.Value), expires: s.candidate.expires, result: CodexGatewayBorrowVerification{Success: true}} wire, app, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) require.NoError(t, err) require.True(t, app.Applied) @@ -467,10 +465,8 @@ func TestCodexGatewayBorrowQualificationFailuresRemainTyped(t *testing.T) { require.True(t, IsCodexGatewayBorrowFailure(err)) require.ErrorIs(t, err, ErrCodexGatewayBorrowUnavailable) borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - s.targetProbeMu.Lock() _, _, err = s.Apply(req, a, "gpt-6-astra", "", nil, false) - s.targetProbeMu.Unlock() require.True(t, IsCodexGatewayBorrowFailure(err)) - require.ErrorIs(t, err, ErrCodexGatewayBorrowBusy) + require.ErrorIs(t, err, ErrCodexGatewayBorrowUnavailable) require.False(t, IsCodexGatewayBorrowFailure(errors.New("ordinary business failure"))) } diff --git a/backend/internal/service/codex_gateway_borrow_diagnose.go b/backend/internal/service/codex_gateway_borrow_diagnose.go new file mode 100644 index 000000000000..82c1a75fde92 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_diagnose.go @@ -0,0 +1,285 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync/atomic" + "time" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" +) + +const CodexBorrowDiagnosticMaxRequests = 8 + +var ErrCodexBorrowDiagnosticBudget = errors.New("diagnostic request budget exhausted (including preparation)") + +type codexBorrowDiagnosticContextKey struct{} +type codexBorrowAppliedContextKey struct{} + +type codexBorrowDiagnostic struct { + limit int32 + onRequest func(int32) + requests *atomic.Int32 + borrow bool + transport string + session string + anchors codexGatewayBorrowWSAnchorStore +} + +func borrowDiagnosticFromContext(ctx context.Context) *codexBorrowDiagnostic { + if ctx == nil { + return nil + } + d, _ := ctx.Value(codexBorrowDiagnosticContextKey{}).(*codexBorrowDiagnostic) + return d +} + +func consumeBorrowDiagnosticRequest(ctx context.Context) error { + d := borrowDiagnosticFromContext(ctx) + if d == nil { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + for { + count := d.requests.Load() + limit := d.limit + if limit == 0 { + limit = CodexBorrowDiagnosticMaxRequests + } + if count >= limit { + return ErrCodexBorrowDiagnosticBudget + } + if d.requests.CompareAndSwap(count, count+1) { + if d.onRequest != nil { + d.onRequest(count + 1) + } + return nil + } + } +} + +type CodexBorrowDiagnosticRequest struct { + RequestLimit int `json:"request_limit,omitempty"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Transport string `json:"transport"` +} + +type CodexBorrowDiagnosticResult struct { + Dispatched bool `json:"dispatched"` + Mode string `json:"mode"` + Turn int `json:"turn"` + Applied bool `json:"applied"` + Completed bool `json:"completed"` + ResponseID string `json:"response_id,omitempty"` + ReportedModel string `json:"reported_model,omitempty"` + Answer string `json:"answer"` + RawResponse string `json:"raw_response"` + Error string `json:"error,omitempty"` + DurationMS int64 `json:"duration_ms"` +} + +type CodexBorrowDiagnosticEvent struct { + Type string `json:"type"` + Mode string `json:"mode,omitempty"` + Requests int32 `json:"requests"` + Limit int `json:"limit"` + Result *CodexBorrowDiagnosticResult `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +// Diagnose runs fixed short questions through the normal account sender. The +// internal observation purpose isolates health/billing; it is never user input. +func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexBorrowDiagnosticRequest, emit func(CodexBorrowDiagnosticEvent)) error { + if request.AccountID <= 0 || (request.Transport != "http" && request.Transport != "ws") { + return errors.New("account_id and transport (http or ws) are required") + } + limit := request.RequestLimit + if limit == 0 { + limit = CodexBorrowDiagnosticMaxRequests + } + if limit < 1 || limit > CodexBorrowDiagnosticMaxRequests { + return errors.New("request_limit must be between 1 and 8") + } + cfg := s.ConfigSnapshot() + if !cfg.Enabled || !slices.Contains(cfg.TargetAccountIDs, request.AccountID) || !slices.Contains(cfg.Models, request.Model) { + return errors.New("select a saved, enabled target and model") + } + account, err := s.accounts.GetByID(ctx, request.AccountID) + if err != nil || !codexGatewayBorrowAccountSupported(account) { + return errors.New("selected account is unavailable") + } + if account.Status != StatusActive || !account.IsModelSupported(request.Model) || normalizeOpenAIModelForUpstream(account, account.GetMappedModel(request.Model)) != request.Model { + return errors.New("selected account is inactive or maps the requested model to another model") + } + if request.Transport == "ws" && s.gateway.getOpenAIWSProtocolResolver().Resolve(account).Transport != OpenAIUpstreamTransportResponsesWebsocketV2 { + return errors.New("WebSocket is not enabled for this account; settings were not changed") + } + s.mu.Lock() + revisionCtx := s.revisionCtx + s.mu.Unlock() + ctx, cancel := borrowRevisionContext(WithAccountObservation(ctx), revisionCtx, 10*time.Minute) + defer cancel() + var requests atomic.Int32 + for _, borrowed := range []bool{false, true} { + if err := ctx.Err(); err != nil { + return err + } + mode := "ordinary" + if borrowed { + mode = "borrowed" + } + d := &codexBorrowDiagnostic{limit: int32(limit), onRequest: func(count int32) { + emit(CodexBorrowDiagnosticEvent{Type: "request", Mode: mode, Requests: count, Limit: limit}) + }, requests: &requests, borrow: borrowed, transport: request.Transport, session: uuid.NewString()} + operation := context.WithValue(ctx, codexBorrowDiagnosticContextKey{}, d) + emit(CodexBorrowDiagnosticEvent{Type: "phase", Mode: mode, Requests: requests.Load(), Limit: limit}) + previous := "" + turns := 1 + if request.Transport == "ws" { + turns = 2 + } + for turn := 1; turn <= turns; turn++ { + result := s.diagnoseTurn(operation, account, request.Model, previous) + result.Mode, result.Turn = mode, turn + emit(CodexBorrowDiagnosticEvent{Type: "result", Mode: mode, Result: &result, Requests: requests.Load(), Limit: limit}) + if !result.Completed { + break + } + previous = result.ResponseID + if request.Transport == "ws" && previous == "" { + break + } + } + d.anchors.mu.Lock() + for key, entry := range d.anchors.entries { + if entry.connID != "" { + s.gateway.getOpenAIWSConnPool().evictConn(key.account, entry.connID) + } + } + d.anchors.mu.Unlock() + if requests.Load() >= int32(limit) { + break + } + } + emit(CodexBorrowDiagnosticEvent{Type: "done", Requests: requests.Load(), Limit: limit}) + return ctx.Err() +} + +func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *Account, model, previous string) CodexBorrowDiagnosticResult { + start := time.Now() + result := CodexBorrowDiagnosticResult{} + state := newPelicanExecutionState(ctx, sharedPelicanExecution, 90*time.Second, time.Now) + ctx = context.WithValue(state.parent, pelicanExecutionContextKey{}, state) + defer state.finish() + BeginPelicanPreparation(ctx) + capture := &pelicanGenerationCapture{account: account, concurrency: s.gateway.concurrencyService, openaiGateway: s.gateway, fallbackModel: model} + ctx = context.WithValue(ctx, pelicanGenerationContextKey{}, capture) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + capture.ginContext = c + c.Request = httptest.NewRequest(http.MethodPost, "/internal/admin/codex-diagnostic", nil).WithContext(ctx) + d := borrowDiagnosticFromContext(ctx) + c.Request.Header.Set("session_id", d.session) + if d.transport == "ws" { + SetOpenAIClientTransport(c, OpenAIClientTransportWS) + } else { + SetOpenAIClientTransport(c, OpenAIClientTransportHTTP) + } + payload := map[string]any{"model": model, "stream": true, "store": false, "input": []map[string]any{{"role": "user", "content": []map[string]string{{"type": "input_text", "text": "Reply with exactly OK."}}}}} + if previous != "" { + payload["previous_response_id"] = previous + } + if levels, _ := AccountTestReasoningOptions(account, model); slices.Contains(levels, "low") { + payload["reasoning"] = map[string]string{"effort": "low"} + } + body, _ := json.Marshal(payload) + forward, sendErr := s.gateway.Forward(ctx, c, account, body) + result.DurationMS = time.Since(start).Milliseconds() + if forward != nil { + result.ResponseID = firstNonEmpty(forward.ResponseID, forward.RequestID) + result.ReportedModel = forward.UpstreamResponseModel + } + capture.mu.Lock() + var last *pelicanHTTPAttempt + if len(capture.attempts) > 0 { + last = capture.attempts[len(capture.attempts)-1] + } + capture.mu.Unlock() + if last == nil { + result.RawResponse = w.Body.String() + if sendErr != nil { + result.Error = sendErr.Error() + } else { + result.Error = "no upstream request was dispatched" + } + return result + } + result.Dispatched = last.dispatched + result.RawResponse = last.raw.String() + if last.wsFrames && last.raw.Len() == 0 && sendErr != nil { + var eventErr *openAIWSUpstreamEventError + if errors.As(sendErr, &eventErr) && len(eventErr.payload) > 0 { + result.RawResponse = string(eventErr.payload) + result.Error = string(eventErr.payload) + } else { + result.Error = sendErr.Error() + } + return result + } + + var parseErr error + var limited bool + if last.wsFrames { + result.Answer, limited, result.ReportedModel, parseErr = parsePelicanWSFrames(result.RawResponse) + } else { + result.Answer, limited, result.ReportedModel, parseErr = parsePelicanTextResponse(last.protocol, result.RawResponse) + } + state.mu.Lock() + result.Applied = state.invocation.BorrowApplied + state.mu.Unlock() + result.Completed = sendErr == nil && parseErr == nil && !limited && last.err == nil && last.status < 400 && strings.TrimSpace(result.Answer) != "" + if sendErr != nil { + result.Error = sendErr.Error() + } else if parseErr != nil { + result.Error = parseErr.Error() + } else if !result.Completed { + result.Error = "upstream did not complete with visible text" + } + return result +} + +func (d *codexBorrowDiagnostic) prepareWS(ctx context.Context, s *CodexGatewayBorrowService, req *http.Request, account *Account, model, previous, proxy string) (*codexGatewayBorrowWSTurn, error) { + key := codexGatewayBorrowWSAnchorKey{account: account.ID, model: model, scope: "diagnostic:" + d.session} + turn, err := d.anchors.begin(key, previous, time.Now()) + if err != nil { + return nil, err + } + turn.diagnostic, turn.diagnosticApplied = true, d.borrow + turn.headers = cloneHeader(req.Header) + if previous == "" && d.borrow { + borrowed, application, err := s.Apply(req, account, model, proxy, nil, false) + if err != nil { + turn.finish(nil, nil, err) + return nil, err + } + if application == nil || !application.Applied { + turn.finish(nil, nil, ErrCodexGatewayBorrowUnavailable) + return nil, ErrCodexGatewayBorrowUnavailable + } + turn.headers = cloneHeader(borrowed.Header) + } + return turn, nil +} + +func (turn *codexGatewayBorrowWSTurn) borrowed() bool { + return turn != nil && (!turn.diagnostic || turn.diagnosticApplied) +} diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_test.go new file mode 100644 index 000000000000..d9af9417cae4 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_diagnose_test.go @@ -0,0 +1,106 @@ +//go:build unit + +package service + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" +) + +func TestCodexBorrowDiagnoseHTTPUsesRealSenderAndIsolatesModes(t *testing.T) { + account := borrowCoreAccount(2) + upstream := &pelicanGeneratorUpstream{body: pelicanResponsesSSE("gpt-6-astra", "completed", "")} + generator, repo := newPelicanGeneratorForTest(account, upstream) + var probes int + s := NewCodexGatewayBorrowService(nil, repo, generator.openaiGatewayService, borrowCoreProbe(func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) { + probes++ + return borrowCoreResponse("gpt-6-astra", "OK", "stable", ""), nil + }), nil) + cfg := CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6-astra"}} + s.publishConfig(cfg, false) + defer s.Stop() + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + var events []CodexBorrowDiagnosticEvent + err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{AccountID: 2, Model: "gpt-6-astra", Transport: "http"}, func(event CodexBorrowDiagnosticEvent) { events = append(events, event) }) + require.NoError(t, err) + var results []CodexBorrowDiagnosticResult + for _, event := range events { + if event.Result != nil { + results = append(results, *event.Result) + } + } + require.Len(t, results, 2) + require.True(t, results[0].Completed, results[0].Error) + require.False(t, results[0].Applied) + require.True(t, results[1].Completed, results[1].Error) + require.True(t, results[1].Applied) + require.Len(t, upstream.requests, 2) + require.NotContains(t, upstream.requests[0].Header.Get("Cookie"), "__oailb") + require.Contains(t, upstream.requests[1].Header.Get("Cookie"), "__oailb=synthetic-borrowed-cookie") + require.Equal(t, cfg, s.ConfigSnapshot()) + require.EqualValues(t, len(upstream.requests)+probes, events[len(events)-1].Requests) + require.LessOrEqual(t, events[len(events)-1].Requests, int32(8)) + require.Equal(t, "diagnostic", s.Status().RecentUsage[0].Origin) +} + +func testCodexBorrowDiagnoseWS(t *testing.T, prewarm bool) { + gateway, account, borrow, probes := borrowWSFixture(t, 2) + gateway.cfg.Gateway.MaxLineSize = defaultMaxLineSize + gateway.cfg.Gateway.OpenAIWS.PrewarmGenerateEnabled = prewarm + repo := &borrowCoreAccounts{rows: map[int64]*Account{account.ID: account}} + gateway.accountRepo = repo + borrow.accounts = repo + borrow.upstream = borrowCoreProbe(func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) { + return borrowCoreResponse("gpt-6-astra", "OK", "stable", ""), nil + }) + completed := func(id string) []byte { + raw, _ := json.Marshal(map[string]any{"type": "response.completed", "response": map[string]any{"id": id, "model": "gpt-6-astra", "status": "completed", "output": []any{map[string]any{"type": "message", "role": "assistant", "content": []any{map[string]any{"type": "output_text", "text": "OK"}}}}, "usage": map[string]any{"input_tokens": 2, "output_tokens": 1}}}) + return raw + } + ordinary := &openAIWSCaptureConn{events: [][]byte{completed("ordinary-1"), completed("ordinary-2")}} + borrowed := &openAIWSCaptureConn{events: [][]byte{completed("borrowed-1"), completed("borrowed-2")}} + if prewarm { + ordinary.events = append([][]byte{completed("ordinary-prewarm")}, ordinary.events...) + borrowed.events = append([][]byte{completed("borrowed-prewarm")}, borrowed.events...) + } + dialer := &borrowWSDialer{conns: []openAIWSClientConn{ordinary, borrowed}} + gateway.openaiWSPool.setClientDialerForTest(dialer) + var requests int32 + var results []CodexBorrowDiagnosticResult + err := borrow.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{AccountID: account.ID, Model: "gpt-6-astra", Transport: "ws"}, func(event CodexBorrowDiagnosticEvent) { + requests = event.Requests + if event.Result != nil { + results = append(results, *event.Result) + } + }) + require.NoError(t, err) + require.Len(t, results, 4) + for _, result := range results { + require.True(t, result.Completed, result.Error) + } + require.False(t, results[0].Applied) + require.True(t, results[2].Applied) + require.Len(t, dialer.headers, 2) + expected := 2 + if prewarm { + expected = 3 + require.EqualValues(t, 8, requests) + } + require.Len(t, ordinary.writes, expected) + require.Len(t, borrowed.writes, expected) + require.Equal(t, "ordinary-1", ordinary.writes[len(ordinary.writes)-1]["previous_response_id"]) + require.Equal(t, "borrowed-1", borrowed.writes[len(borrowed.writes)-1]["previous_response_id"]) + require.Empty(t, probes.requests) + require.Empty(t, borrow.wsAnchors.entries, "diagnostics cannot create client anchors") +} + +func TestCodexBorrowDiagnoseWSContinuesOwnSockets(t *testing.T) { testCodexBorrowDiagnoseWS(t, false) } +func TestCodexBorrowDiagnoseWSCountsConfiguredPrewarm(t *testing.T) { + testCodexBorrowDiagnoseWS(t, true) +} diff --git a/backend/internal/service/codex_gateway_borrow_http.go b/backend/internal/service/codex_gateway_borrow_http.go index f073bdad44ba..1ea93ee99e20 100644 --- a/backend/internal/service/codex_gateway_borrow_http.go +++ b/backend/internal/service/codex_gateway_borrow_http.go @@ -91,6 +91,9 @@ func (s *OpenAIGatewayService) codexGatewayBorrowHTTPConfigured(account *Account func (s *OpenAIGatewayService) prepareCodexGatewayBorrowHTTP(ctx context.Context, req *http.Request, account *Account, model, proxy string) (*http.Request, time.Duration, error) { // Ordinary requests retain their exact identity: continuation diagnostics // bind an immutable body snapshot to this pointer after GetBody is disabled. + if d := borrowDiagnosticFromContext(ctx); d != nil && !d.borrow { + return req, 0, nil + } if !CodexGatewayBorrowRequestEligible(req) || !s.codexGatewayBorrowHTTPConfigured(account, model) { return req, 0, nil } diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go index de7f25a291e3..819bab8de441 100644 --- a/backend/internal/service/codex_gateway_borrow_probe.go +++ b/backend/internal/service/codex_gateway_borrow_probe.go @@ -301,6 +301,16 @@ func (s *CodexGatewayBorrowService) fireObservation(ctx context.Context, headers func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account *Account, proxy string, profile *tlsfingerprint.Profile, expectedModel string, maxBody int64) (codexGatewayBorrowObservation, error) { var shot codexGatewayBorrowObservation + if err := consumeBorrowDiagnosticRequest(req.Context()); err != nil { + return shot, err + } + if pelicanExecutionFromContext(req.Context()) == nil { + release, err := sharedPelicanExecution.acquire(req.Context(), account.ID) + if err != nil { + return shot, err + } + defer release() + } leasedCtx, release, acquireErr := AcquirePelicanExecution(req.Context(), account.ID) if acquireErr != nil { shot.errorText = acquireErr.Error() diff --git a/backend/internal/service/codex_gateway_borrow_probe_stream_test.go b/backend/internal/service/codex_gateway_borrow_probe_stream_test.go index 336f57fbcac4..701aa308666b 100644 --- a/backend/internal/service/codex_gateway_borrow_probe_stream_test.go +++ b/backend/internal/service/codex_gateway_borrow_probe_stream_test.go @@ -108,8 +108,8 @@ func TestCodexGatewayBorrowObservationStopsAtTerminal(t *testing.T) { borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) _, template, _, err := s.accountTemplate(ctx, 2, "gpt-6.1-sol") require.NoError(t, err) - key := borrowRequestFingerprint(template, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value) - s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{key: key, cookieKey: borrowHash(s.candidate.cookie.Value), expires: s.candidate.expires, result: CodexGatewayBorrowVerification{Success: true}} + key := borrowTargetFingerprint(template, target, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value) + s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{policyRevision: currentCodexFingerprintPolicyForAccount(target).revision, key: key, cookieKey: borrowHash(s.candidate.cookie.Value), expires: s.candidate.expires, result: CodexGatewayBorrowVerification{Success: true}} result, err := s.GeneratePelican(ctx, 2, "gpt-6.1-sol", "high") require.NoError(t, err) require.Equal(t, "complete", result.Status) diff --git a/backend/internal/service/codex_gateway_borrow_status.go b/backend/internal/service/codex_gateway_borrow_status.go new file mode 100644 index 000000000000..4fd85d000911 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_status.go @@ -0,0 +1,66 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" +) + +// Read local records without refreshing credentials or probing an upstream. +// The send path still checks the finalized wire identity immediately at dispatch. +func (s *CodexGatewayBorrowService) CurrentStatus(ctx context.Context) CodexGatewayBorrowStatus { + status := s.Status() + if s == nil || s.accounts == nil { + return status + } + accounts := make(map[int64]*Account) + for i := range status.Targets { + row := &status.Targets[i] + account, loaded := accounts[row.AccountID] + if !loaded { + account, _ = s.accounts.GetByID(ctx, row.AccountID) + accounts[row.AccountID] = account + } + reason := "" + if !codexGatewayBorrowAccountSupported(account) { + reason = "account_unavailable" + } else if account.Status != StatusActive { + reason = "account_inactive" + } else if !account.IsModelSupported(row.Model) { + reason = "model_not_allowed" + } else if normalizeOpenAIModelForUpstream(account, account.GetMappedModel(row.Model)) != row.Model { + reason = "model_mapping_mismatch" + } else if row.CacheValid { + credential, err := resolveCredentialAccount(ctx, s.accounts, account) + if err != nil || credential == nil { + reason = "account_unavailable" + } else { + s.mu.Lock() + check := s.targets[codexGatewayBorrowTargetKey{row.AccountID, row.Model}] + s.mu.Unlock() + if check.identityRevision != "" && check.identityRevision != borrowAccountIdentityRevision(account, credential) { + reason = "identity_changed" + } + } + } + if reason != "" { + row.CacheValid = false + row.State = "waiting" + row.Reason = reason + row.Error = fmt.Sprintf("account %d: %s", row.AccountID, reason) + } + } + return status +} + +func borrowAccountIdentityRevision(account, credential *Account) string { + if account == nil || credential == nil { + return "" + } + proxy := "" + if account.Proxy != nil { + proxy = account.Proxy.URL() + } + value, _ := json.Marshal([]any{credential.Credentials, credential.Extra["codex_client_identity"], credential.Extra["codex_fingerprint_seed"], account.Extra["codex_fingerprint_mode"], account.Credentials["model_mapping"], proxy, account.ProxyID}) + return borrowHash(string(value)) +} diff --git a/backend/internal/service/codex_gateway_borrow_usage.go b/backend/internal/service/codex_gateway_borrow_usage.go new file mode 100644 index 000000000000..a76bbd6ae844 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_usage.go @@ -0,0 +1,181 @@ +package service + +import ( + "context" + "io" + "net/http" + "slices" + "strings" + "sync" + "time" + + "github.com/tidwall/gjson" +) + +type CodexGatewayBorrowUsage struct { + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Transport string `json:"transport"` + Origin string `json:"origin"` + Applied bool `json:"applied"` + Reason string `json:"reason"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + Outcome string `json:"outcome"` + RequestID string `json:"request_id,omitempty"` + ReportedModel string `json:"reported_model,omitempty"` + Count uint64 `json:"count"` + AppliedCount uint64 `json:"applied_count"` +} + +type codexBorrowUsageKey struct { + account int64 + model, transport, origin string +} + +// Only configured pairs are retained, one last observation per protocol/purpose. +// Payloads, credentials and routing cookies are never copied into this index. +func (s *CodexGatewayBorrowService) beginUsage(ctx context.Context, accountID int64, model, transport string, applied bool) *codexBorrowUsageTracker { + if s == nil { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + if !slices.Contains(s.config.TargetAccountIDs, accountID) || !slices.Contains(s.config.Models, model) { + return nil + } + origin := "business" + if IsPelicanGeneration(ctx) || borrowDiagnosticFromContext(ctx) != nil { + origin = "diagnostic" + } + key := codexBorrowUsageKey{accountID, model, transport, origin} + previous := s.usage[key] + row := CodexGatewayBorrowUsage{AccountID: accountID, Model: model, Transport: transport, Origin: origin, Applied: applied, StartedAt: time.Now(), Outcome: "sent", Count: previous.Count + 1, AppliedCount: previous.AppliedCount} + row.Reason = "not_applied" + if applied { + row.Reason = "borrow_applied" + row.AppliedCount++ + } else if !s.config.Enabled { + row.Reason = "disabled" + } + if s.usage == nil { + s.usage = make(map[codexBorrowUsageKey]CodexGatewayBorrowUsage) + } + s.usage[key] = row + return &codexBorrowUsageTracker{service: s, key: key, row: row, revision: s.revision} +} + +type codexBorrowUsageTracker struct { + revision uint64 + service *CodexGatewayBorrowService + key codexBorrowUsageKey + row CodexGatewayBorrowUsage + once sync.Once +} + +func (t *codexBorrowUsageTracker) observe(message []byte) { + if t == nil { + return + } + typeName := gjson.GetBytes(message, "type").String() + if id := gjson.GetBytes(message, "response.id").String(); id != "" { + t.row.RequestID = id + } + if model := gjson.GetBytes(message, "response.model").String(); model != "" { + t.row.ReportedModel = model + } + switch typeName { + case "response.completed", "response.done": + if gjson.GetBytes(message, "response.status").String() == "completed" && !gjson.GetBytes(message, "response.error").IsObject() { + t.row.Outcome = "completed" + } else if gjson.GetBytes(message, "response.status").String() != "" || gjson.GetBytes(message, "response.error").IsObject() { + t.row.Outcome = "upstream_error" + } else { + t.row.Outcome = "unobserved" + } + case "error", "response.failed", "response.incomplete", "response.cancelled", "response.canceled": + t.row.Outcome = "upstream_error" + } +} + +func (t *codexBorrowUsageTracker) finish(err error) { + if t == nil { + return + } + t.once.Do(func() { + if err != nil && t.row.Outcome != "completed" { + t.row.Outcome = "transport_error" + } else if t.row.Outcome == "sent" { + t.row.Outcome = "incomplete" + } + now := time.Now() + t.row.FinishedAt = &now + t.service.mu.Lock() + defer t.service.mu.Unlock() + // An older completion must not overwrite a newer attempt or cleared config. + if current, ok := t.service.usage[t.key]; ok && t.service.revision == t.revision && current.Count == t.row.Count { + t.service.usage[t.key] = t.row + } + }) +} + +type codexBorrowUsageBody struct { + io.ReadCloser + tracker *codexBorrowUsageTracker + line []byte + err error +} + +func (b *codexBorrowUsageBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + for _, ch := range p[:n] { + if ch == '\n' { + if strings.HasPrefix(string(b.line), "data:") { + b.tracker.observe(b.line[5:]) + } + b.line = b.line[:0] + } else if len(b.line) < 1<<20 { + b.line = append(b.line, ch) + } + } + if err != nil { + if len(b.line) > 5 && strings.HasPrefix(string(b.line), "data:") { + b.tracker.observe(b.line[5:]) + b.line = nil + } + if err != io.EOF { + b.err = err + } + b.tracker.finish(b.err) + } + return n, err +} + +func (b *codexBorrowUsageBody) Close() error { + err := b.ReadCloser.Close() + if b.err == nil { + b.err = err + } + b.tracker.finish(b.err) + return err +} + +func (s *CodexGatewayBorrowService) trackHTTPResponse(t *codexBorrowUsageTracker, resp *http.Response, err error) (*http.Response, error) { + if t == nil { + return resp, err + } + if resp == nil { + t.finish(err) + return resp, err + } + t.row.RequestID = resp.Header.Get("X-Request-ID") + if resp.StatusCode >= 400 { + t.row.Outcome = "upstream_error" + } + if err != nil || resp.Body == nil { + t.finish(err) + } else { + resp.Body = &codexBorrowUsageBody{ReadCloser: resp.Body, tracker: t} + } + return resp, err +} diff --git a/backend/internal/service/codex_gateway_borrow_usage_ws.go b/backend/internal/service/codex_gateway_borrow_usage_ws.go new file mode 100644 index 000000000000..963282bd12e5 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_usage_ws.go @@ -0,0 +1,57 @@ +package service + +import ( + "context" + "sync" + + openaiwsv2 "github.com/Wei-Shaw/sub2api/internal/service/openai_ws_v2" + coderws "github.com/coder/websocket" + "github.com/tidwall/gjson" +) + +// Wrap the final native WS relay, after policy rewriting. Handshakes and +// blocked client frames never enter this observer. +type codexBorrowUsageFrameConn struct { + openaiwsv2.FrameConn + service *CodexGatewayBorrowService + accountID int64 + applied bool + mu sync.Mutex + current *codexBorrowUsageTracker +} + +func (c *codexBorrowUsageFrameConn) WriteFrame(ctx context.Context, kind coderws.MessageType, payload []byte) error { + c.mu.Lock() + if gjson.GetBytes(payload, "type").String() == "response.create" { + c.current.finish(nil) + c.current = c.service.beginUsage(ctx, c.accountID, gjson.GetBytes(payload, "model").String(), "ws", c.applied) + } + tracker := c.current + c.mu.Unlock() + err := c.FrameConn.WriteFrame(ctx, kind, payload) + if err != nil { + c.mu.Lock() + tracker.finish(err) + c.mu.Unlock() + } + return err +} + +func (c *codexBorrowUsageFrameConn) ReadFrame(ctx context.Context) (coderws.MessageType, []byte, error) { + kind, payload, err := c.FrameConn.ReadFrame(ctx) + c.mu.Lock() + c.current.observe(payload) + if err != nil { + c.current.finish(err) + } else if c.current != nil && c.current.row.Outcome != "sent" { + c.current.finish(nil) + } + c.mu.Unlock() + return kind, payload, err +} + +func (c *codexBorrowUsageFrameConn) finish() { + c.mu.Lock() + defer c.mu.Unlock() + c.current.finish(nil) +} diff --git a/backend/internal/service/codex_gateway_borrow_ws_anchor.go b/backend/internal/service/codex_gateway_borrow_ws_anchor.go index c1bdc5c07283..baa1a98cd63b 100644 --- a/backend/internal/service/codex_gateway_borrow_ws_anchor.go +++ b/backend/internal/service/codex_gateway_borrow_ws_anchor.go @@ -47,6 +47,8 @@ type codexGatewayBorrowWSTurn struct { qualified bool finished sync.Once pelicanOneShot bool + diagnostic bool + diagnosticApplied bool } // Passthrough owns a direct socket for the entire client session. Its anchor is @@ -227,6 +229,9 @@ func (s *OpenAIGatewayService) prepareCodexGatewayBorrowWSTurn(ctx context.Conte if err != nil { return nil, err } + if d := borrowDiagnosticFromContext(ctx); d != nil { + return d.prepareWS(ctx, s.gatewayBorrow, req, account, model, previousID, proxyURL) + } selected := s.codexGatewayBorrowWSSelected(account, model, req) if IsPelicanGeneration(ctx) { if !selected { diff --git a/backend/internal/service/codex_gateway_borrow_ws_passthrough_test.go b/backend/internal/service/codex_gateway_borrow_ws_passthrough_test.go index 09f837ab6b89..1855954e4d84 100644 --- a/backend/internal/service/codex_gateway_borrow_ws_passthrough_test.go +++ b/backend/internal/service/codex_gateway_borrow_ws_passthrough_test.go @@ -64,8 +64,6 @@ func TestCodexGatewayBorrowWS_PassthroughFinalModelAndContinuationUseOneDirectDi borrow.candidate.expires = time.Now().Add(-time.Second) borrow.targets = nil borrow.mu.Unlock() - require.True(t, borrow.targetProbeMu.TryLock()) - defer borrow.targetProbeMu.Unlock() require.NoError(t, client.Write(ctx, coderws.MessageText, []byte(`{"type":"response.create","model":"public-sol-alias","previous_response_id":"resp_direct_1","input":"next","store":false}`))) secondWire := requirePassthroughUpstreamWrite(t, upstream, time.Second) require.Equal(t, "resp_direct_1", gjson.GetBytes(secondWire, "previous_response_id").String()) diff --git a/backend/internal/service/codex_gateway_borrow_ws_test.go b/backend/internal/service/codex_gateway_borrow_ws_test.go index 01f0699fc950..82e3f9e16908 100644 --- a/backend/internal/service/codex_gateway_borrow_ws_test.go +++ b/backend/internal/service/codex_gateway_borrow_ws_test.go @@ -88,8 +88,8 @@ func seedBorrowWSTarget(t *testing.T, svc *OpenAIGatewayService, borrow *CodexGa req, err := codexGatewayBorrowWSRequest(context.Background(), wsURL, headers) require.NoError(t, err) candidate := borrow.candidate - key := borrowRequestFingerprint(req, model, account.Proxy.URL(), nil, candidate.cookie.Value) - borrow.targets[codexGatewayBorrowTargetKey{account.ID, model}] = codexGatewayBorrowTargetCheck{key: key, + key := borrowTargetFingerprint(req, account, model, account.Proxy.URL(), nil, candidate.cookie.Value) + borrow.targets[codexGatewayBorrowTargetKey{account.ID, model}] = codexGatewayBorrowTargetCheck{policyRevision: currentCodexFingerprintPolicyForAccount(account).revision, key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: CodexGatewayBorrowVerification{Success: true}} } @@ -144,8 +144,6 @@ func TestCodexGatewayBorrowWS_TwoModelsContinueOriginalConnectionsAfterCookieExp borrow.config.Enabled = false borrow.targets = nil borrow.mu.Unlock() - require.True(t, borrow.targetProbeMu.TryLock()) - defer borrow.targetProbeMu.Unlock() _, err = runBorrowWSV2(t, svc, account, "gpt-6.1-sol", "resp_astra_1") require.Error(t, err, "disabling borrowing must not let a different model take over an existing response") continued, err := runBorrowWSV2(t, svc, account, "gpt-6-astra", "resp_astra_1") diff --git a/backend/internal/service/openai_codex_request_compression.go b/backend/internal/service/openai_codex_request_compression.go index 7f703372b944..d426ea0a8b18 100644 --- a/backend/internal/service/openai_codex_request_compression.go +++ b/backend/internal/service/openai_codex_request_compression.go @@ -157,5 +157,11 @@ func (s *OpenAIGatewayService) doOpenAICodexUpstream(req *http.Request, account if err != nil { return nil, err } - return s.httpUpstream.Do(wire, proxyURL, account.ID, account.Concurrency) + applied := wire.Context().Value(codexBorrowAppliedContextKey{}) == true + if err := consumeBorrowDiagnosticRequest(wire.Context()); err != nil { + return nil, err + } + tracker := s.gatewayBorrow.beginUsage(wire.Context(), account.ID, model, "http", applied) + resp, sendErr := s.httpUpstream.Do(wire, proxyURL, account.ID, account.Concurrency) + return s.gatewayBorrow.trackHTTPResponse(tracker, resp, sendErr) } diff --git a/backend/internal/service/openai_ws_forwarder_support.go b/backend/internal/service/openai_ws_forwarder_support.go index 8790a0550aa6..9e0d6ee99347 100644 --- a/backend/internal/service/openai_ws_forwarder_support.go +++ b/backend/internal/service/openai_ws_forwarder_support.go @@ -78,6 +78,9 @@ func (s *OpenAIGatewayService) performOpenAIWSGeneratePrewarm( prewarmPayload["generate"] = false prewarmPayloadJSON := payloadAsJSONBytes(prewarmPayload) + if err := consumeBorrowDiagnosticRequest(ctx); err != nil { + return err + } if err := lease.WriteJSONWithContextTimeout(ctx, prewarmPayload, s.openAIWSWriteTimeout()); err != nil { lease.MarkBroken() logOpenAIWSModeInfo( diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 186f612e9c88..62399ac3fb7c 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -393,12 +393,12 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( var observePelicanFrame func([]byte) error var releasePelican func() pelicanPreparationCtx := ctx - ctx, observePelicanFrame, releasePelican, err = preparePelicanWSSend(ctx, account, wsURL, payload, borrowTurn != nil) + ctx, observePelicanFrame, releasePelican, err = preparePelicanWSSend(ctx, account, wsURL, payload, borrowTurn.borrowed()) if err != nil { return nil, err } defer releasePelican() - if IsPelicanGeneration(ctx) && borrowTurn != nil && !time.Now().Before(borrowTurn.expires) { + if IsPelicanGeneration(ctx) && borrowTurn != nil && !borrowTurn.diagnostic && !time.Now().Before(borrowTurn.expires) { // No prompt has been written. Release the expired route and all execution // resources, then perform the ordinary preparation on this same account. releasePelican() @@ -408,7 +408,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( return s.forwardOpenAIWSV2(pelicanPreparationCtx, c, account, reqBody, clientPromptCacheKey, executionScope, token, decision, isCodexCLI, reqStream, originalModel, mappedModel, startTime, attempt, lastFailureReason, agentTaskRecoveryTried) } - if IsPelicanGeneration(ctx) && borrowTurn != nil { + if IsPelicanGeneration(ctx) && borrowTurn != nil && !borrowTurn.diagnostic { prepared, prepareErr := codexGatewayBorrowWSRequest(ctx, wsURL, wsHeaders) if prepareErr != nil { return nil, codexGatewayBorrowWSPreparationError(c, account, prepareErr) @@ -441,6 +441,18 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( } } + if err := consumeBorrowDiagnosticRequest(ctx); err != nil { + return nil, err + } + if capture := pelicanCaptureFromContext(ctx); capture != nil { + capture.mu.Lock() + if len(capture.attempts) > 0 { + capture.attempts[len(capture.attempts)-1].dispatched = true + } + capture.mu.Unlock() + } + usageTracker := s.gatewayBorrow.beginUsage(ctx, account.ID, mappedModel, "ws", borrowTurn.borrowed()) + defer func() { usageTracker.finish(borrowErr) }() if err := lease.WriteJSONWithContextTimeout(ctx, payload, s.openAIWSWriteTimeout()); err != nil { lease.MarkBroken() logOpenAIWSModeInfo( @@ -666,6 +678,7 @@ readLoop: } message, readErr = lease.ReadMessageWithContextTimeout(upstreamReadCtx, currentReadTimeout) if readErr == nil { + usageTracker.observe(message) if observeErr := observePelicanFrame(message); observeErr != nil { lease.MarkBroken() return nil, observeErr diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 2b102bdec262..0412e0e1ee7b 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -1010,8 +1010,10 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2PassthroughAttempt( if !ok { return errors.New("openai ws passthrough upstream connection does not support frame relay") } + usageConn := &codexBorrowUsageFrameConn{FrameConn: upstreamFrameConn, service: s.gatewayBorrow, accountID: account.ID, applied: borrowTurn.borrowed()} + defer usageConn.finish() relayUpstreamFrameConn := &openAIWSPassthroughFirstOutputFrameConn{ - inner: upstreamFrameConn, + inner: usageConn, activeReadTimeout: s.openAIWSPassthroughIdleTimeout(), deadlineChanged: make(chan struct{}, 1), resolveDeadline: func(payload []byte) openAIWSPassthroughFirstOutputDeadline { diff --git a/backend/internal/service/pelican_account_generator.go b/backend/internal/service/pelican_account_generator.go index da3145672205..d199d36b2dce 100644 --- a/backend/internal/service/pelican_account_generator.go +++ b/backend/internal/service/pelican_account_generator.go @@ -89,6 +89,7 @@ type pelicanGenerationCapture struct { } type pelicanHTTPAttempt struct { + dispatched bool protocol string contentType string status int @@ -188,14 +189,14 @@ func PreparePelicanHTTPRequest(req *http.Request, accountID int64, accountConcur firstOutputPolicy, _ := ctx.Value(pelicanFirstOutputPolicyContextKey{}).(*pelicanFirstOutputPolicy) ctx = firstOutputPolicy.arm(ctx) req = req.WithContext(ctx) - borrowApplied := false + borrowApplied := ctx.Value(codexBorrowAppliedContextKey{}) == true if prepared, ok := ctx.Value(codexGatewayBorrowHTTPPreparationContextKey{}).(codexGatewayBorrowHTTPPreparation); ok { borrowApplied = prepared.application != nil && prepared.application.Applied } invocation := PelicanInvocation{Platform: capture.account.Platform, Model: model, Effort: effort, Endpoint: req.URL.Scheme + "://" + req.URL.Host + req.URL.EscapedPath(), Protocol: protocol, Transport: transport, BorrowApplied: borrowApplied} RecordPelicanInvocation(ctx, invocation) - attempt := &pelicanHTTPAttempt{protocol: protocol, model: model, effort: effort} + attempt := &pelicanHTTPAttempt{protocol: protocol, model: model, effort: effort, dispatched: true} capture.mu.Lock() capture.attempts = append(capture.attempts, attempt) capture.mu.Unlock() diff --git a/frontend/src/api/admin/codexGatewayBorrow.ts b/frontend/src/api/admin/codexGatewayBorrow.ts index 3b4e47ab0bfa..20742638e51f 100644 --- a/frontend/src/api/admin/codexGatewayBorrow.ts +++ b/frontend/src/api/admin/codexGatewayBorrow.ts @@ -32,7 +32,26 @@ export interface BorrowTargetStatus extends BorrowSourceStatus { reported_model?: string } +export interface CodexGatewayBorrowUsage { + account_id: number; model: string; transport: string; origin: string + applied: boolean; reason: string; started_at: string; finished_at?: string + outcome: string; request_id?: string; reported_model?: string; count: number; applied_count: number +} +export interface CodexBorrowDiagnosticResult { + dispatched?: boolean + mode: string; turn: number; applied: boolean; completed: boolean; response_id?: string + reported_model?: string; answer: string; raw_response: string; error?: string; duration_ms: number +} +export interface CodexBorrowDiagnosticEvent { + type: 'phase' | 'request' | 'result' | 'done' | 'error'; mode?: string; requests: number; limit: number + result?: CodexBorrowDiagnosticResult; error?: string +} +export interface CodexBorrowDiagnosticRequest { account_id: number; model: string; transport: 'http' | 'ws'; request_limit?: number } + export interface CodexGatewayBorrowStatus { + setup?: { state: string; phase: string; account_id: number; model: string; completed: number; total: number; failed: number; started_at: string; finished_at?: string; error?: string } + recent_usage?: CodexGatewayBorrowUsage[] + observed_since?: string enabled: boolean revision: number generated_at: string @@ -203,6 +222,71 @@ export async function streamBorrowTests( reader.releaseLock() } } +export async function streamBorrowDiagnostic( + request: CodexBorrowDiagnosticRequest, + onEvent: (event: CodexBorrowDiagnosticEvent) => void, + signal: AbortSignal +): Promise { + const token = localStorage.getItem('auth_token') + const response = await fetch(buildApiUrl(`${basePath}/diagnose`), { + method: 'POST', + headers: { + ...(token ? { Authorization: `Bearer ${token}` } : {}), + 'Content-Type': 'application/json', + Accept: 'text/event-stream', + [ADMIN_UI_REQUEST_HEADER]: '1' + }, + credentials: 'include', + body: JSON.stringify(request), + signal + }) + if (!response.ok) { + const body = await response.text() + throw new Error(`HTTP ${response.status}${body ? `: ${body}` : ''}`) + } + if (!response.body) throw new Error('Borrow test stream has no response body') + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let complete = false + const abort = () => { void reader.cancel().catch(() => {}) } + signal.addEventListener('abort', abort, { once: true }) + + function consume(block: string) { + const data = block.split(/\r?\n/).filter(line => line.startsWith('data:')) + .map(line => line.slice(5).replace(/^ /, '')).join('\n') + if (!data) return + const event = JSON.parse(data) as CodexBorrowDiagnosticEvent + if (event.type === 'error') throw new Error(event.error || data) + onEvent(event) + if (event.type === 'done') complete = true + } + + try { + if (signal.aborted) throw new DOMException('Aborted', 'AbortError') + while (!complete) { + const { value, done } = await reader.read() + if (signal.aborted) throw new DOMException('Aborted', 'AbortError') + buffer += decoder.decode(value ?? new Uint8Array(), { stream: !done }) + let boundary: RegExpExecArray | null + while ((boundary = /\r?\n\r?\n/.exec(buffer))) { + const block = buffer.slice(0, boundary.index) + buffer = buffer.slice(boundary.index + boundary[0].length) + consume(block) + if (complete) break + } + if (done) { + if (buffer.trim() && !complete) consume(buffer) + break + } + } + if (!complete) throw new Error('Borrow test stream ended before done') + } finally { + signal.removeEventListener('abort', abort) + await reader.cancel().catch(() => {}) + reader.releaseLock() + } +} export const codexGatewayBorrowAPI = { getConfig: async (signal?: AbortSignal) => (await apiClient.get(`${basePath}/config`, { signal })).data, @@ -212,7 +296,8 @@ export const codexGatewayBorrowAPI = { verify: async (accountId: number, model: string, signal?: AbortSignal) => (await apiClient.post(`${basePath}/verify`, { account_id: accountId, model }, { signal, timeout: 0 })).data, listTests: async (page = 1, signal?: AbortSignal) => (await apiClient.get(`${basePath}/tests`, { params: { page, size: 12 }, signal })).data, getTest: async (id: string, signal?: AbortSignal) => (await apiClient.get(`${basePath}/tests/${encodeURIComponent(id)}`, { signal })).data, - streamTests: streamBorrowTests + streamTests: streamBorrowTests, + diagnose: streamBorrowDiagnostic } export default codexGatewayBorrowAPI diff --git a/frontend/src/components/admin/codex/CodexBorrowActivity.vue b/frontend/src/components/admin/codex/CodexBorrowActivity.vue new file mode 100644 index 000000000000..ec2ebaf7878a --- /dev/null +++ b/frontend/src/components/admin/codex/CodexBorrowActivity.vue @@ -0,0 +1,81 @@ + + + diff --git a/frontend/src/components/admin/codex/CodexLayout.vue b/frontend/src/components/admin/codex/CodexLayout.vue new file mode 100644 index 000000000000..50a57b4118fd --- /dev/null +++ b/frontend/src/components/admin/codex/CodexLayout.vue @@ -0,0 +1,29 @@ + + + diff --git a/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts b/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts new file mode 100644 index 000000000000..6549ce6fbfca --- /dev/null +++ b/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts @@ -0,0 +1,58 @@ +import { describe, expect, it, vi, beforeEach } from 'vitest' +import { flushPromises, mount } from '@vue/test-utils' +import { createI18n } from 'vue-i18n' +import CodexBorrowActivity from '../CodexBorrowActivity.vue' +import type { CodexBorrowDiagnosticEvent, CodexGatewayBorrowStatus } from '@/api/admin/codexGatewayBorrow' + +const { diagnose } = vi.hoisted(() => ({ diagnose: vi.fn() })) +vi.mock('@/api/admin/codexGatewayBorrow', () => ({ codexGatewayBorrowAPI: { diagnose } })) +const status: CodexGatewayBorrowStatus = { + enabled: true, revision: 1, generated_at: new Date().toISOString(), preparing: false, + config: { enabled: true, source_account_ids: [1], target_account_ids: [2], models: ['gpt-6-astra'] }, sources: [], targets: [], recent_usage: [] +} +function open(locale = 'zh') { + return mount(CodexBorrowActivity, { props: { status, accountName: id => `Account ${id}` }, global: { + plugins: [createI18n({ legacy: false, locale, messages: { zh: {}, en: {} } })], + stubs: { RouterLink: { template: '' } } + } }) +} +beforeEach(() => { diagnose.mockReset() }) + +describe('Codex actual-request diagnostics', () => { + it('reads existing evidence without automatically issuing requests', () => { + const wrapper = open() + expect(diagnose).not.toHaveBeenCalled() + expect(wrapper.get('[data-test="borrow-no-usage"]').text()).toContain('还没有实际发送记录') + expect(wrapper.get('[data-test="diagnose-selection"]').element.tagName).toBe('SELECT') + wrapper.unmount() + }) + + it('requires an explicit action and shows applied separately from completed', async () => { + diagnose.mockImplementation(async (_request, emit: (event: CodexBorrowDiagnosticEvent) => void) => { + emit({ type: 'result', requests: 3, limit: 8, result: { mode: 'borrowed', turn: 1, applied: true, completed: false, reported_model: 'gpt-6-astra', answer: '', raw_response: '{"error":"raw upstream failure"}', error: 'raw upstream failure', duration_ms: 10 } }) + emit({ type: 'done', requests: 3, limit: 8 }) + }) + const wrapper = open('en') + await wrapper.get('[data-test="diagnose-start"]').trigger('click') + await flushPromises() + expect(diagnose).toHaveBeenCalledOnce() + expect(diagnose.mock.calls[0][0]).toEqual({ account_id: 2, model: 'gpt-6-astra', transport: 'http' }) + const result = wrapper.get('[data-test="diagnose-result"]') + expect(result.text()).toContain('not completed') + expect(result.text()).toContain('Borrow was applied') + expect(result.text()).toContain('raw upstream failure') + expect(wrapper.emitted('refresh')).toHaveLength(1) + wrapper.unmount() + }) + + it('cancels its explicit diagnostic when the page is closed', async () => { + let signal: AbortSignal | undefined + diagnose.mockImplementation((_request, _emit, abort: AbortSignal) => { signal = abort; return new Promise((_resolve, reject) => abort.addEventListener('abort', () => reject(new Error('aborted')), { once: true })) }) + const wrapper = open() + await wrapper.get('[data-test="diagnose-start"]').trigger('click') + wrapper.unmount() + await flushPromises() + expect(signal?.aborted).toBe(true) + expect(wrapper.emitted('refresh')).toBeUndefined() + }) +}) diff --git a/frontend/src/components/layout/AppSidebar.vue b/frontend/src/components/layout/AppSidebar.vue index 3c731140e274..0c38b4479b63 100644 --- a/frontend/src/components/layout/AppSidebar.vue +++ b/frontend/src/components/layout/AppSidebar.vue @@ -817,9 +817,7 @@ const personalNavItems = computed((): NavItem[] => finalizeNav(buildSelfNavItems const adminExtensionNavItems = computed((): NavItem[] => [ { path: '/admin/system-prompts', label: t('nav.systemPrompts'), icon: DocumentIcon }, - { path: '/admin/codex-runtime', label: t('nav.codexRuntime'), icon: PluginIcon }, - { path: '/admin/codex-fingerprint', label: t('nav.codexFingerprint'), icon: ShieldIcon }, - { path: '/admin/codex-gateway-borrow', label: t('nav.codexGatewayBorrow'), icon: GatewayBorrowIcon }, + { path: '/admin/codex', label: 'Codex', icon: GatewayBorrowIcon }, { path: '/admin/pelican-tests', label: t('nav.pelicanTests'), icon: GatewayBorrowIcon }, ]) @@ -963,7 +961,7 @@ function isActive(path: string): boolean { } function isExtensionActive(path: string): boolean { - if (path === '/admin/codex-gateway-borrow') { + if (path === '/admin/codex') { return isActive(path) } return route.path === path diff --git a/frontend/src/components/layout/__tests__/AppSidebar.simple-mode.spec.ts b/frontend/src/components/layout/__tests__/AppSidebar.simple-mode.spec.ts index 874378f9553c..86ebfbbcce13 100644 --- a/frontend/src/components/layout/__tests__/AppSidebar.simple-mode.spec.ts +++ b/frontend/src/components/layout/__tests__/AppSidebar.simple-mode.spec.ts @@ -110,31 +110,29 @@ describe('AppSidebar simple mode extensions', () => { expect(extensionLinks(wrapper)).toEqual([ '/admin/system-prompts', - '/admin/codex-runtime', - '/admin/codex-fingerprint', - '/admin/codex-gateway-borrow', + '/admin/codex', '/admin/pelican-tests', ]) if (simple) expect(wrapper.text()).not.toContain('nav.myAccount') wrapper.unmount() const regularUser = await renderSidebar({ simple }) - expect(regularUser.find('a[href="/admin/codex-gateway-borrow"]').exists()).toBe(false) - expect(regularUser.find('a[href="/admin/codex-gateway-borrow/status"]').exists()).toBe(false) + expect(regularUser.find('a[href="/admin/codex"]').exists()).toBe(false) + expect(regularUser.find('a[href="/admin/codex/identity"]').exists()).toBe(false) expect(regularUser.find('a[href="/admin/codex-pelican-comparison"]').exists()).toBe(false) expect(regularUser.find('a[href="/admin/pelican-tests"]').exists()).toBe(false) regularUser.unmount() }) it.each([ - '/admin/codex-gateway-borrow', - '/admin/codex-gateway-borrow/status', + '/admin/codex', + '/admin/codex/identity', ])('keeps the single borrowing entry active on %s', async path => { const wrapper = await renderSidebar({ admin: true, path }) const section = wrapper.get('[data-testid="sidebar-extensions"]') - expect(section.get('a[href="/admin/codex-gateway-borrow"]').classes()).toContain('sidebar-link-active') - expect(section.find('a[href="/admin/codex-gateway-borrow/status"]').exists()).toBe(false) + expect(section.get('a[href="/admin/codex"]').classes()).toContain('sidebar-link-active') + expect(section.find('a[href="/admin/codex/identity"]').exists()).toBe(false) expect(section.find('a[href="/admin/codex-pelican-comparison"]').exists()).toBe(false) expect(section.findAll('.sidebar-link-active')).toHaveLength(1) wrapper.unmount() @@ -144,7 +142,7 @@ describe('AppSidebar simple mode extensions', () => { const wrapper = await renderSidebar({ admin: true, path: '/admin/pelican-tests' }) const section = wrapper.get('[data-testid="sidebar-extensions"]') expect(section.get('a[href="/admin/pelican-tests"]').classes()).toContain('sidebar-link-active') - expect(section.get('a[href="/admin/codex-gateway-borrow"]').classes()).not.toContain('sidebar-link-active') + expect(section.get('a[href="/admin/codex"]').classes()).not.toContain('sidebar-link-active') expect(section.findAll('.sidebar-link-active')).toHaveLength(1) wrapper.unmount() }) diff --git a/frontend/src/components/layout/__tests__/AppSidebar.spec.ts b/frontend/src/components/layout/__tests__/AppSidebar.spec.ts index 2c4afa255ae5..774406c4f396 100644 --- a/frontend/src/components/layout/__tests__/AppSidebar.spec.ts +++ b/frontend/src/components/layout/__tests__/AppSidebar.spec.ts @@ -66,7 +66,7 @@ describe('AppSidebar header styles', () => { describe('AppSidebar extensions section', () => { it('keeps downstream tools in a dedicated section', () => { expect(componentSource).toContain("{{ t('nav.extensions') }}") - expect(componentSource).toContain("{ path: '/admin/codex-runtime'") + expect(componentSource).toContain("{ path: '/admin/codex'") }) }) diff --git a/frontend/src/composables/useCodexBorrowUI.ts b/frontend/src/composables/useCodexBorrowUI.ts index 5898e1369c0b..976b68b69e18 100644 --- a/frontend/src/composables/useCodexBorrowUI.ts +++ b/frontend/src/composables/useCodexBorrowUI.ts @@ -85,6 +85,7 @@ export function borrowTargetState(target: BorrowTargetStatus | undefined, now: n } const knownReasons = new Set([ + 'account_unavailable', 'account_inactive', 'model_not_allowed', 'model_mapping_mismatch', 'identity_changed', 'not_prepared', 'not_verified', 'not_configured', 'disabled', 'route_expired', 'source_qualified', 'source_probe_failed', 'source_response_not_qualified', 'source_routing_cookie_unavailable', 'target_probe_passed', 'target_probe_failed', 'target_route_changed', 'target_state_missing', 'target_state_changed', 'validating', diff --git a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts index 9b86467b7185..0707a248462e 100644 --- a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts @@ -142,6 +142,12 @@ export default { rawResponse: 'Raw upstream response', downloadAnswer: 'Download raw answer', reasons: { + account_unavailable: 'Account or credential source unavailable; check account settings', + account_inactive: 'Account is inactive; resolve its status on the account page', + model_not_allowed: 'This model is not allowed by the account', + model_mapping_mismatch: 'This model maps to another model; validation cannot be reused', + identity_changed: 'Identity or exit settings changed; validate again', + not_prepared: 'Borrow routes have not been prepared yet', not_verified: 'This account and model have not completed route validation', not_configured: 'This account and model are absent from the saved configuration', diff --git a/frontend/src/i18n/locales/en/common.ts b/frontend/src/i18n/locales/en/common.ts index 449f7cfd4154..7ce9efa9ec73 100644 --- a/frontend/src/i18n/locales/en/common.ts +++ b/frontend/src/i18n/locales/en/common.ts @@ -188,6 +188,7 @@ export default { subscriptions: 'Subscriptions', accounts: 'Accounts', plugins: 'Plugins', + codex: 'Codex', codexRuntime: 'Codex runtime settings', codexFingerprint: 'Codex fingerprint', proxies: 'Proxies', diff --git a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts index 6a3e966ff1b5..7d4520fba7d2 100644 --- a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts @@ -142,6 +142,12 @@ export default { rawResponse: '上游响应原文', downloadAnswer: '下载回答原文', reasons: { + account_unavailable: '账号或凭据来源不可用，请检查账号设置', + account_inactive: '账号未启用，请先在账号页处理状态', + model_not_allowed: '账号未允许此模型，请检查模型列表', + model_mapping_mismatch: '账号把此模型映射到其他模型，不能复用验证', + identity_changed: '身份或出口配置已改变，请重新验证', + not_prepared: '尚未准备借用线路', not_verified: '此账号与模型尚未完成线路验证', not_configured: '此账号与模型未包含在已保存配置中', diff --git a/frontend/src/i18n/locales/zh/common.ts b/frontend/src/i18n/locales/zh/common.ts index 76586cb0374e..38bb792f0a47 100644 --- a/frontend/src/i18n/locales/zh/common.ts +++ b/frontend/src/i18n/locales/zh/common.ts @@ -188,6 +188,7 @@ export default { subscriptions: '订阅管理', accounts: '账号管理', plugins: '插件管理', + codex: 'Codex', codexRuntime: 'Codex 运行设置', codexFingerprint: 'Codex 指纹', proxies: 'IP管理', diff --git a/frontend/src/router/__tests__/codex-borrow-routes.spec.ts b/frontend/src/router/__tests__/codex-borrow-routes.spec.ts index 447cd83c887d..79ba29152c3c 100644 --- a/frontend/src/router/__tests__/codex-borrow-routes.spec.ts +++ b/frontend/src/router/__tests__/codex-borrow-routes.spec.ts @@ -56,8 +56,9 @@ vi.mock('@/composables/useRoutePrefetch', () => ({ vi.mock('@/router/title', () => ({ resolveRouteDocumentTitle: () => 'Sub2API' })) const borrowRoutes = [ - { path: '/admin/codex-gateway-borrow', name: 'AdminCodexGatewayBorrow' }, - { path: '/admin/codex-gateway-borrow/status', name: 'AdminCodexGatewayBorrowStatus' }, + { path: '/admin/codex', name: 'AdminCodex' }, + { path: '/admin/codex/identity', name: 'AdminCodexIdentity' }, + { path: '/admin/codex/advanced', name: 'AdminCodexAdvanced' }, { path: '/admin/pelican-tests', name: 'AdminPelicanTests' }, ] @@ -77,6 +78,12 @@ describe('Codex borrow page routes', () => { authStore.isSimpleMode = false }) + it.each([['/admin/codex-runtime', '/admin/codex/advanced'], ['/admin/codex-fingerprint', '/admin/codex/identity'], ['/admin/codex-gateway-borrow', '/admin/codex'], ['/admin/codex-gateway-borrow/status', '/admin/codex#status']])('redirects %s to the unified console', (path, destination) => { + const record = harness.routes.find(route => route.path === path) + expect(record?.redirect).toBe(destination) + expect(record?.meta).toMatchObject({ requiresAuth: true, requiresAdmin: true }) + }) + it('redirects the bookmarked comparison address to the independent Pelican page', () => { const record = harness.routes.find(route => route.path === '/admin/codex-pelican-comparison') expect(record?.redirect).toBe('/admin/pelican-tests') diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 421c77b51bc3..c48b8ee38c07 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -538,29 +538,27 @@ const routes: RouteRecordRaw[] = [ } }, { - path: '/admin/codex-runtime', - name: 'AdminCodexRuntime', - component: () => import('@/views/admin/CodexRuntimeView.vue'), - meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex runtime settings', titleKey: 'nav.codexRuntime' } + path: '/admin/codex', + name: 'AdminCodex', + component: () => import('@/views/admin/CodexGatewayBorrowView.vue'), + meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex', titleKey: 'nav.codex' } }, { - path: '/admin/codex-fingerprint', - name: 'AdminCodexFingerprint', + path: '/admin/codex/identity', + name: 'AdminCodexIdentity', component: () => import('@/views/admin/CodexFingerprintView.vue'), - meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex fingerprint', titleKey: 'nav.codexFingerprint' } - }, - { - path: '/admin/codex-gateway-borrow', - name: 'AdminCodexGatewayBorrow', - component: () => import('@/views/admin/CodexGatewayBorrowView.vue'), - meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex Borrow Settings', titleKey: 'admin.codexGatewayBorrow.title' } + meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex', titleKey: 'nav.codex' } }, { - path: '/admin/codex-gateway-borrow/status', - name: 'AdminCodexGatewayBorrowStatus', - component: () => import('@/views/admin/CodexGatewayBorrowStatusView.vue'), - meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex Borrow Status', titleKey: 'nav.codexGatewayBorrowStatus' } + path: '/admin/codex/advanced', + name: 'AdminCodexAdvanced', + component: () => import('@/views/admin/CodexRuntimeView.vue'), + meta: { requiresAuth: true, requiresAdmin: true, title: 'Codex', titleKey: 'nav.codex' } }, + { path: '/admin/codex-runtime', redirect: '/admin/codex/advanced', meta: { requiresAuth: true, requiresAdmin: true } }, + { path: '/admin/codex-fingerprint', redirect: '/admin/codex/identity', meta: { requiresAuth: true, requiresAdmin: true } }, + { path: '/admin/codex-gateway-borrow', redirect: '/admin/codex', meta: { requiresAuth: true, requiresAdmin: true } }, + { path: '/admin/codex-gateway-borrow/status', redirect: '/admin/codex#status', meta: { requiresAuth: true, requiresAdmin: true } }, { path: '/admin/pelican-tests', name: 'AdminPelicanTests', diff --git a/frontend/src/views/admin/CodexFingerprintView.vue b/frontend/src/views/admin/CodexFingerprintView.vue index e59320c66a6d..2d0d1c3eabb8 100644 --- a/frontend/src/views/admin/CodexFingerprintView.vue +++ b/frontend/src/views/admin/CodexFingerprintView.vue @@ -1,8 +1,8 @@