diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md
index 50ec349c4738..9e3bf1934061 100644
--- a/.downstream/native-domains.md
+++ b/.downstream/native-domains.md
@@ -19,17 +19,18 @@ nothing. While no Codex runtime row is stored,
## Codex gateway borrowing
-The Extensions sidebar has one Codex Borrowing entry. Its in-page navigation
-switches between three independently addressable administrator pages: settings at
+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`, and manual comparison/history at
-`/admin/codex-pelican-comparison`. Opening or navigating these pages only reads
+`/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 comparison page cancels its active generation.
+the Pelican page cancels its queued and active generation.
`GET/PUT /admin/codex-gateway-borrow/config` owns one JSON setting,
`codex_gateway_borrow_config`: `enabled`, `source_account_ids`,
@@ -54,17 +55,44 @@ 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.
-`POST /admin/codex-gateway-borrow/tests` accepts a timestamped UUIDv7 and explicit
-account/model pairs. It sends the fixed pelican prompt once per pair, only with
-matching unexpired qualification, using at most three shared generation slots
-and serializing each account. It does not prepare missing routes. An expired
-or future UUID is rejected; replaying a saved UUID cannot dispatch again.
-Migration 275 adds only this feature's task/result tables. Records expire 24
-hours after task creation, disappear from reads immediately, and are removed
-by scoped minute cleanup. Output/errors retain their original bytes.
+The historical `/admin/codex-gateway-borrow/tests` endpoints retain their
+cache-qualified, 90-second request contract. They share execution admission with
+the independent tests described below.
+
+## Independent Pelican tests
+
+`/admin/pelican-tests` provides local account/model options, explicit generation
+with SSE progress, and stored history/detail. Candidates include all platforms,
+account types and account states, using local mappings, saved catalogs and
+platform defaults. Option reads never refresh credentials or request an upstream
+model directory. Each selected account/model receives the fixed Pelican prompt
+once. The server chooses its actual existing account sending path, model mapping,
+credentials, identity headers, proxy, TLS, protocol and borrowing configuration;
+it creates no API Key, group policy or client handshake, and never switches to a
+different account on failure. Unsupported text capabilities retain a skip reason.
+
+All batches share ten model execution slots, with one request per account.
+Nested borrowing probes participate in the same coordination and normal business
+account capacity. Generation starts after preparation and admission; its default
+budget is 600 seconds, configurable from 60 to 1800 seconds. Queue, preparation
+and generation durations are recorded separately. Observed business account slots
+renew every 30 seconds; lost capacity cancels the operation. Existing credential
+refresh remains available, while account health, quota, scheduling, affinity,
+usage and billing writes are suppressed, including background callbacks.
+
+Task IDs are timestamped UUIDv7 values. Expired/future IDs are rejected and replay
+cannot dispatch again. Migration 275 owns the task/result tables; migration 276
+adds budgets, execution phases and actual invocation metadata in place, preserving
+old records and raw BYTEA output. Existing records retain their 90-second budget.
+Records expire 24 hours after task creation, immediately disappear from reads,
+and are removed by scoped minute cleanup. Output/errors retain original bytes;
+binary upstream envelopes use lossless Base64. Truncated or incomplete output is
+recorded without another generation or automatic repair.
Result previews use short random capability URLs under
-`/codex-gateway-borrow/preview/`, an opaque script-enabled sandbox and their own
+`/pelican-tests/preview/` (the historical borrowing path remains an alias),
+expire within 30 minutes and the task's remaining lifetime, and use an opaque
+script-enabled sandbox and their own
response CSP. Inline animation and external scripts/styles/fonts/images are
allowed. Preview documents receive no administrator token or UI bridge.
Adaptation sources and retained notices are in
diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml
index e3255bae73a5..5e9664d2d378 100644
--- a/.github/workflows/downstream-verify.yml
+++ b/.github/workflows/downstream-verify.yml
@@ -44,7 +44,7 @@ jobs:
- name: Native gateway borrowing contracts
working-directory: backend
run: |
- go test -p=1 -tags unit ./internal/service ./internal/repository ./internal/handler ./internal/handler/admin -run '^TestCodexGatewayBorrow' -count=1
+ 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: Upstream automation classifier tests
run: |
@@ -145,6 +145,7 @@ jobs:
src/views/admin/__tests__/CodexGatewayBorrowView.spec.ts \
src/views/admin/__tests__/CodexGatewayBorrowStatusView.spec.ts \
src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts \
+ src/api/admin/__tests__/pelicanTests.spec.ts \
src/components/admin/__tests__/CodexBorrowNav.spec.ts \
src/router/__tests__/codex-borrow-routes.spec.ts \
src/components/account/__tests__/EditAccountModal.spec.ts \
diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go
index bfec3a57bc70..abfb409a7db2 100644
--- a/backend/cmd/server/wire_gen.go
+++ b/backend/cmd/server/wire_gen.go
@@ -310,14 +310,16 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
return nil, err
}
codexGatewayBorrowTestRepository := repository.NewCodexGatewayBorrowTestRepository(db)
- codexGatewayBorrowTestRunner := service.NewCodexGatewayBorrowTestRunner(codexGatewayBorrowTestRepository, codexGatewayBorrowService, accountRepository)
+ codexGatewayBorrowTestRunner := service.ProvidePelicanTestRunner(codexGatewayBorrowTestRepository, codexGatewayBorrowService, accountRepository, accountTestService, gatewayService, geminiMessagesCompatService, concurrencyService)
codexGatewayBorrowPreviewService := service.NewCodexGatewayBorrowPreviewService(codexGatewayBorrowTestRepository)
codexGatewayBorrowHandler := admin.NewCodexGatewayBorrowHandler(codexGatewayBorrowService, codexGatewayBorrowTestRunner, codexGatewayBorrowPreviewService)
+ pelicanTestCatalog := service.NewPelicanTestCatalog(accountRepository)
+ pelicanTestHandler := admin.NewPelicanTestHandler(pelicanTestCatalog, codexGatewayBorrowTestRunner, codexGatewayBorrowPreviewService)
upstreamBillingProbeService := service.ProvideUpstreamBillingProbeService(accountRepository, accountTestService, settingService, leaderLockCache, db)
openCodeGoUsageService := service.ProvideOpenCodeGoUsageService(accountRepository, httpUpstream, settingService, leaderLockCache, db)
idempotencyCoordinator := service.ProvideIdempotencyCoordinator(idempotencyRepository, configConfig)
claudeResetCreditService := service.ProvideClaudeResetCreditService(accountRepository, claudeTokenProvider, proxyRepository, settingService, idempotencyCoordinator, leaderLockCache)
- adminHandlers := handler.ProvideAdminHandlers(dashboardHandler, adminUserHandler, groupHandler, accountHandler, accountJobHandler, adminAnnouncementHandler, dataManagementHandler, backupHandler, oAuthHandler, openAIOAuthHandler, geminiOAuthHandler, antigravityOAuthHandler, grokOAuthHandler, cnProviderHandler, proxyHandler, adminRedeemHandler, promoHandler, settingHandler, opsHandler, systemHandler, adminSubscriptionHandler, adminUsageHandler, userAttributeHandler, errorPassthroughHandler, tlsFingerprintProfileHandler, pluginHandler, adminAPIKeyHandler, scheduledTestHandler, channelHandler, channelMonitorHandler, channelMonitorRequestTemplateHandler, contentModerationHandler, promptAdminHandler, paymentHandler, affiliateHandler, complianceHandler, auditLogHandler, systemPromptHandler, reasoningRecoveryHandler, codexGatewayBorrowHandler, upstreamBillingProbeService, ollamaCloudUsageService, openCodeGoUsageService, claudeResetCreditService)
+ adminHandlers := handler.ProvideAdminHandlers(dashboardHandler, adminUserHandler, groupHandler, accountHandler, accountJobHandler, adminAnnouncementHandler, dataManagementHandler, backupHandler, oAuthHandler, openAIOAuthHandler, geminiOAuthHandler, antigravityOAuthHandler, grokOAuthHandler, cnProviderHandler, proxyHandler, adminRedeemHandler, promoHandler, settingHandler, opsHandler, systemHandler, adminSubscriptionHandler, adminUsageHandler, userAttributeHandler, errorPassthroughHandler, tlsFingerprintProfileHandler, pluginHandler, adminAPIKeyHandler, scheduledTestHandler, channelHandler, channelMonitorHandler, channelMonitorRequestTemplateHandler, contentModerationHandler, promptAdminHandler, paymentHandler, affiliateHandler, complianceHandler, auditLogHandler, systemPromptHandler, reasoningRecoveryHandler, codexGatewayBorrowHandler, pelicanTestHandler, upstreamBillingProbeService, ollamaCloudUsageService, openCodeGoUsageService, claudeResetCreditService)
usageRecordWorkerPool := service.NewUsageRecordWorkerPool(configConfig)
userMsgQueueCache := repository.NewUserMsgQueueCache(redisClient)
userMessageQueueService := service.ProvideUserMessageQueueService(userMsgQueueCache, rpmCache, configConfig)
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 7d1380a29527..d48b7e6181fd 100644
--- a/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go
+++ b/backend/internal/handler/admin/codex_gateway_borrow_handler_test.go
@@ -84,6 +84,7 @@ func TestCodexGatewayBorrowHandlerPreviewUsesOnlyCapabilityAndOverridesCSP(t *te
r := gin.New()
r.Use(middleware.SecurityHeaders(config.CSPConfig{Enabled: true}, nil))
r.GET("/api/v1/codex-gateway-borrow/preview/:cap/index.html", h.ServePreview)
+ r.GET("/api/v1/pelican-tests/preview/:cap/index.html", h.ServePreview)
r.GET("/", func(c *gin.Context) { c.String(http.StatusOK, "main app") })
w := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodGet, decorated.PreviewURL, nil)
diff --git a/backend/internal/handler/admin/pelican_test_handler.go b/backend/internal/handler/admin/pelican_test_handler.go
new file mode 100644
index 000000000000..71b5611d3daf
--- /dev/null
+++ b/backend/internal/handler/admin/pelican_test_handler.go
@@ -0,0 +1,218 @@
+package admin
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/Wei-Shaw/sub2api/internal/pkg/response"
+ "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+ "github.com/Wei-Shaw/sub2api/internal/service"
+ "github.com/gin-gonic/gin"
+ "github.com/google/uuid"
+)
+
+// PelicanTestHandler owns the independent administrator entry point. Borrowing
+// configuration and preparation remain in CodexGatewayBorrowHandler; this entry
+// selects only an account, a model, an effort, and a generation budget.
+type PelicanTestHandler struct {
+ catalog *service.PelicanTestCatalog
+ runner *service.CodexGatewayBorrowTestRunner
+ preview *service.CodexGatewayBorrowPreviewService
+}
+
+func NewPelicanTestHandler(catalog *service.PelicanTestCatalog, runner *service.CodexGatewayBorrowTestRunner, preview *service.CodexGatewayBorrowPreviewService) *PelicanTestHandler {
+ return &PelicanTestHandler{catalog: catalog, runner: runner, preview: preview}
+}
+
+func (h *PelicanTestHandler) GetOptions(c *gin.Context) {
+ h.options(c, nil)
+}
+
+func (h *PelicanTestHandler) GetAccountOptions(c *gin.Context) {
+ codexGatewayBorrowPrivateHeaders(c)
+ var request struct {
+ AccountIDs []int64 `json:"account_ids"`
+ }
+ if err := c.ShouldBindJSON(&request); err != nil {
+ response.BadRequest(c, err.Error())
+ return
+ }
+ h.options(c, request.AccountIDs)
+}
+
+func (h *PelicanTestHandler) options(c *gin.Context, accountIDs []int64) {
+ codexGatewayBorrowPrivateHeaders(c)
+ options, err := h.catalog.Options(c.Request.Context(), accountIDs)
+ if err != nil {
+ if errors.Is(err, service.ErrPelicanTestOptionsInvalidRequest) {
+ response.BadRequest(c, err.Error())
+ } else {
+ response.Error(c, http.StatusInternalServerError, err.Error())
+ }
+ return
+ }
+ response.Success(c, options)
+}
+
+func (h *PelicanTestHandler) StartTests(c *gin.Context) {
+ codexGatewayBorrowPrivateHeaders(c)
+ var request service.CodexGatewayBorrowTestRequest
+ if err := c.ShouldBindJSON(&request); err != nil {
+ response.BadRequest(c, err.Error())
+ return
+ }
+ if _, err := uuid.Parse(strings.TrimSpace(request.ClientTaskID)); err != nil {
+ response.BadRequest(c, "client_task_id must be a UUID")
+ return
+ }
+ if len(request.Targets) == 0 {
+ response.BadRequest(c, "targets is required")
+ return
+ }
+ // Only this server-side entry point can select account execution. A JSON
+ // field cannot change the legacy borrowing handler's existing contract.
+ request.Standalone = true
+ var createdBy int64
+ if subject, ok := middleware.GetAuthSubjectFromContext(c); ok {
+ createdBy = subject.UserID
+ }
+ task, replayed, err := h.runner.Create(c.Request.Context(), createdBy, request)
+ if err != nil {
+ codexGatewayBorrowTestError(c, err)
+ return
+ }
+ middleware.SetAuditExtra(c, map[string]any{"result": "pelican_manual_test", "task_id": task.ID, "replayed": replayed, "targets": task.Total})
+ ctx, cancel := context.WithCancel(c.Request.Context())
+ defer cancel()
+ c.Header("Content-Type", "text/event-stream")
+ c.Header("Connection", "keep-alive")
+ c.Header("X-Accel-Buffering", "no")
+ c.Status(http.StatusOK)
+ var writeMu sync.Mutex
+ emit := func(event service.CodexGatewayBorrowTestEvent) {
+ if ctx.Err() != nil {
+ return
+ }
+ if event.Task != nil {
+ event.Task = h.preview.DecorateTask(event.Task)
+ }
+ if event.Result != nil {
+ event.Result = h.preview.DecorateResult(event.Result)
+ }
+ payload, marshalErr := json.Marshal(event)
+ if marshalErr != nil {
+ cancel()
+ return
+ }
+ writeMu.Lock()
+ defer writeMu.Unlock()
+ if ctx.Err() != nil {
+ return
+ }
+ if _, writeErr := fmt.Fprintf(c.Writer, "data: %s\n\n", payload); writeErr != nil {
+ cancel()
+ return
+ }
+ c.Writer.Flush()
+ }
+ emit(service.CodexGatewayBorrowTestEvent{Type: "task_start", Task: task})
+ if replayed {
+ emit(service.CodexGatewayBorrowTestEvent{Type: "task_complete", Task: task})
+ return
+ }
+ keepAliveDone, keepAliveStopped := make(chan struct{}), make(chan struct{})
+ go func() {
+ defer close(keepAliveStopped)
+ ticker := time.NewTicker(15 * time.Second)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-keepAliveDone:
+ return
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ writeMu.Lock()
+ if ctx.Err() == nil {
+ if _, writeErr := fmt.Fprint(c.Writer, ": keep-alive\n\n"); writeErr != nil {
+ cancel()
+ } else {
+ c.Writer.Flush()
+ }
+ }
+ writeMu.Unlock()
+ }
+ }
+ }()
+ runErr := h.runner.Run(ctx, task, emit)
+ close(keepAliveDone)
+ <-keepAliveStopped
+ if runErr != nil && ctx.Err() == nil {
+ task.Status, task.Error = "incomplete", runErr.Error()
+ emit(service.CodexGatewayBorrowTestEvent{Type: "task_complete", Task: task})
+ }
+}
+
+func (h *PelicanTestHandler) ListTests(c *gin.Context) {
+ codexGatewayBorrowPrivateHeaders(c)
+ page, size := response.ParsePagination(c)
+ if rawSize := c.Query("size"); rawSize != "" {
+ if parsed, err := strconv.Atoi(rawSize); err == nil && parsed > 0 {
+ size = parsed
+ }
+ }
+ list, err := h.runner.List(c.Request.Context(), page, size)
+ if err != nil {
+ codexGatewayBorrowTestError(c, err)
+ return
+ }
+ response.Success(c, list)
+}
+
+func (h *PelicanTestHandler) GetTest(c *gin.Context) {
+ codexGatewayBorrowPrivateHeaders(c)
+ id := strings.TrimSpace(c.Param("id"))
+ if id == "last" {
+ list, err := h.runner.List(c.Request.Context(), 1, 1)
+ if err != nil {
+ codexGatewayBorrowTestError(c, err)
+ return
+ }
+ if len(list.Items) == 0 {
+ response.NotFound(c, "no unexpired pelican test")
+ return
+ }
+ id = list.Items[0].ID
+ }
+ parsed, err := uuid.Parse(id)
+ if err != nil {
+ response.BadRequest(c, "test ID must be a UUID or last")
+ return
+ }
+ task, err := h.runner.Get(c.Request.Context(), parsed.String())
+ if err != nil {
+ codexGatewayBorrowTestError(c, err)
+ return
+ }
+ response.Success(c, h.preview.DecorateTask(task))
+}
+
+func (h *PelicanTestHandler) ServePreview(c *gin.Context) {
+ codexGatewayBorrowPrivateHeaders(c)
+ c.Header("Content-Security-Policy", service.CodexGatewayBorrowPreviewCSP)
+ c.Header("X-Frame-Options", "SAMEORIGIN")
+ c.Header("Cross-Origin-Resource-Policy", "cross-origin")
+ html, err := h.preview.Resolve(c.Request.Context(), strings.TrimSpace(c.Param("cap")))
+ if err != nil {
+ c.Status(http.StatusGone)
+ return
+ }
+ c.Data(http.StatusOK, "text/html; charset=utf-8", []byte(html))
+}
diff --git a/backend/internal/handler/admin/pelican_test_handler_test.go b/backend/internal/handler/admin/pelican_test_handler_test.go
new file mode 100644
index 000000000000..b3c5f7abfc5f
--- /dev/null
+++ b/backend/internal/handler/admin/pelican_test_handler_test.go
@@ -0,0 +1,160 @@
+//go:build unit
+
+package admin
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ "github.com/Wei-Shaw/sub2api/internal/service"
+ "github.com/gin-gonic/gin"
+ "github.com/stretchr/testify/require"
+)
+
+type pelicanTestHandlerLocalAccountReader struct {
+ service.AccountRepository
+ reads int
+}
+
+type pelicanTestHandlerReplayCaptureRepo struct {
+ *codexGatewayBorrowHandlerTestRepo
+ incoming *service.CodexGatewayBorrowTestTask
+}
+
+func (r *pelicanTestHandlerReplayCaptureRepo) Create(ctx context.Context, task *service.CodexGatewayBorrowTestTask) (*service.CodexGatewayBorrowTestTask, bool, error) {
+ r.incoming = task
+ return r.codexGatewayBorrowHandlerTestRepo.Create(ctx, task)
+}
+
+func (r *pelicanTestHandlerLocalAccountReader) GetByIDs(_ context.Context, _ []int64) ([]*service.Account, error) {
+ r.reads++
+ return []*service.Account{{ID: 42, Name: "paused account", Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey, Status: service.StatusDisabled}}, nil
+}
+
+func TestPelicanTestHandlerOptionsDefaultsAndSelectedLocalModels(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+ accounts := &pelicanTestHandlerLocalAccountReader{}
+ h := NewPelicanTestHandler(service.NewPelicanTestCatalog(accounts), nil, nil)
+ router := gin.New()
+ router.GET("/options", h.GetOptions)
+ router.POST("/options", h.GetAccountOptions)
+ request := func(method, body string) *httptest.ResponseRecorder {
+ w := httptest.NewRecorder()
+ r := httptest.NewRequest(method, "/options", strings.NewReader(body))
+ r.Header.Set("Content-Type", "application/json")
+ router.ServeHTTP(w, r)
+ return w
+ }
+ defaults := request(http.MethodGet, "")
+ require.Equal(t, http.StatusOK, defaults.Code)
+ require.Zero(t, accounts.reads)
+ require.Contains(t, defaults.Body.String(), `"default_generation_timeout_seconds":600`)
+ require.Equal(t, "private, no-store", defaults.Header().Get("Cache-Control"))
+ selected := request(http.MethodPost, `{"account_ids":[42,999]}`)
+ require.Equal(t, http.StatusOK, selected.Code)
+ var envelope struct {
+ Data service.PelicanTestOptions `json:"data"`
+ }
+ require.NoError(t, json.Unmarshal(selected.Body.Bytes(), &envelope))
+ require.Equal(t, 1, accounts.reads)
+ require.Len(t, envelope.Data.Accounts, 2)
+ require.Equal(t, service.StatusDisabled, envelope.Data.Accounts[0].Status)
+ require.Equal(t, "missing", envelope.Data.Accounts[1].Status)
+ ids := make([]string, 0, len(envelope.Data.Accounts[0].Models))
+ for _, model := range envelope.Data.Accounts[0].Models {
+ ids = append(ids, model.ID)
+ }
+ require.Contains(t, ids, "gpt-6-astra")
+ require.Contains(t, ids, "gpt-6.1-sol")
+ invalid := request(http.MethodPost, `{"account_ids":[0]}`)
+ require.Equal(t, http.StatusBadRequest, invalid.Code)
+ require.Equal(t, 1, accounts.reads)
+}
+
+func TestPelicanTestHandlerSSEReplayAndHistoryNeverGenerateAndPreserveRawContent(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+ legacy, repo := newCodexGatewayBorrowHandlerFixture(t)
+ h := NewPelicanTestHandler(service.NewPelicanTestCatalog(nil), legacy.runner, legacy.preview)
+ router := gin.New()
+ router.GET("/tests/:id", h.GetTest)
+ router.GET("/tests", h.ListTests)
+ router.POST("/tests", h.StartTests)
+ router.GET("/api/v1/pelican-tests/preview/:cap/index.html", h.ServePreview)
+ history := httptest.NewRecorder()
+ router.ServeHTTP(history, httptest.NewRequest(http.MethodGet, "/tests/last", nil))
+ require.Equal(t, http.StatusOK, history.Code)
+ var envelope struct {
+ Data service.CodexGatewayBorrowTestTask `json:"data"`
+ }
+ require.NoError(t, json.Unmarshal(history.Body.Bytes(), &envelope))
+ require.Equal(t, repo.result.Error, envelope.Data.Results[0].Error)
+ require.Equal(t, repo.result.RawAnswer, envelope.Data.Results[0].RawAnswer)
+ require.True(t, strings.HasPrefix(envelope.Data.Results[0].PreviewURL, "/api/v1/pelican-tests/preview/"))
+ preview := httptest.NewRecorder()
+ router.ServeHTTP(preview, httptest.NewRequest(http.MethodGet, envelope.Data.Results[0].PreviewURL, nil))
+ require.Equal(t, http.StatusOK, preview.Code)
+ require.Equal(t, repo.result.HTML, preview.Body.String())
+ require.Equal(t, service.CodexGatewayBorrowPreviewCSP, preview.Header().Get("Content-Security-Policy"))
+ stream := httptest.NewRecorder()
+ body := fmt.Sprintf(`{"client_task_id":%q,"generation_timeout_seconds":600,"targets":[{"account_id":42,"model_id":"gpt-6-astra"}]}`, repo.task.ClientTaskID)
+ request := httptest.NewRequest(http.MethodPost, "/tests", strings.NewReader(body))
+ request.Header.Set("Content-Type", "application/json")
+ router.ServeHTTP(stream, request)
+ require.Equal(t, http.StatusOK, stream.Code)
+ require.Equal(t, "text/event-stream", stream.Header().Get("Content-Type"))
+ require.Contains(t, stream.Body.String(), `"type":"task_start"`)
+ require.Contains(t, stream.Body.String(), `"type":"task_complete"`)
+ require.NotContains(t, stream.Body.String(), `"type":"result_started"`)
+ require.Contains(t, stream.Body.String(), "original upstream error token=admin-visible")
+ require.Equal(t, 1, repo.creates)
+ require.Zero(t, repo.writes)
+}
+
+func TestPelicanTestHandlerForcesAccountModeAndFixedPromptBeforeReplay(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+ _, existing := newCodexGatewayBorrowHandlerFixture(t)
+ repo := &pelicanTestHandlerReplayCaptureRepo{codexGatewayBorrowHandlerTestRepo: existing}
+ runner := service.NewCodexGatewayBorrowTestRunner(repo, nil, nil)
+ preview := service.NewCodexGatewayBorrowPreviewService(repo)
+ t.Cleanup(runner.Stop)
+ t.Cleanup(preview.Stop)
+ h := NewPelicanTestHandler(service.NewPelicanTestCatalog(nil), runner, preview)
+ router := gin.New()
+ router.POST("/tests", h.StartTests)
+ body := fmt.Sprintf(`{"client_task_id":%q,"standalone":false,"execution_mode":"legacy_cache","prompt":"client modified prompt","targets":[{"account_id":42,"model_id":"gpt-6-astra"}]}`, existing.task.ClientTaskID)
+ request := httptest.NewRequest(http.MethodPost, "/tests", strings.NewReader(body))
+ request.Header.Set("Content-Type", "application/json")
+ w := httptest.NewRecorder()
+ router.ServeHTTP(w, request)
+ require.Equal(t, http.StatusOK, w.Code)
+ require.NotNil(t, repo.incoming)
+ require.Equal(t, service.PelicanExecutionModeAccount, repo.incoming.ExecutionMode)
+ require.Equal(t, service.CodexGatewayBorrowPelicanPrompt, repo.incoming.Prompt)
+ require.Equal(t, 600, repo.incoming.GenerationTimeoutSeconds)
+ require.Equal(t, int64(42), repo.incoming.Results[0].AccountID)
+ require.Empty(t, repo.incoming.Results[0].Effort, "the generator applies only the selected model's supported default")
+ require.Zero(t, existing.writes, "replay must not dispatch the forced-account execution")
+}
+
+func TestPelicanTestHandlerRejectsGenerationBudgetBeforeCreate(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+ legacy, repo := newCodexGatewayBorrowHandlerFixture(t)
+ h := NewPelicanTestHandler(service.NewPelicanTestCatalog(nil), legacy.runner, legacy.preview)
+ router := gin.New()
+ router.POST("/tests", h.StartTests)
+ for _, seconds := range []int{59, 1801} {
+ w := httptest.NewRecorder()
+ body := fmt.Sprintf(`{"client_task_id":%q,"generation_timeout_seconds":%d,"targets":[{"account_id":42,"model_id":"gpt-6-astra"}]}`, repo.task.ClientTaskID, seconds)
+ r := httptest.NewRequest(http.MethodPost, "/tests", strings.NewReader(body))
+ r.Header.Set("Content-Type", "application/json")
+ router.ServeHTTP(w, r)
+ require.Equal(t, http.StatusBadRequest, w.Code)
+ }
+ require.Zero(t, repo.creates, "invalid budget must not leave a pending task")
+ require.Zero(t, repo.writes)
+}
diff --git a/backend/internal/handler/handler.go b/backend/internal/handler/handler.go
index 3b493f1c7450..6d01386b2010 100644
--- a/backend/internal/handler/handler.go
+++ b/backend/internal/handler/handler.go
@@ -47,6 +47,7 @@ type AdminHandlers struct {
SystemPrompt *admin.SystemPromptHandler
ReasoningRecovery *admin.ReasoningRecoveryHandler
CodexGatewayBorrow *admin.CodexGatewayBorrowHandler
+ PelicanTest *admin.PelicanTestHandler
}
// Handlers contains all HTTP handlers
diff --git a/backend/internal/handler/wire.go b/backend/internal/handler/wire.go
index 90a4e357ad52..2f027c6003d9 100644
--- a/backend/internal/handler/wire.go
+++ b/backend/internal/handler/wire.go
@@ -51,6 +51,7 @@ func ProvideAdminHandlers(
systemPromptHandler *admin.SystemPromptHandler,
reasoningRecoveryHandler *admin.ReasoningRecoveryHandler,
codexGatewayBorrowHandler *admin.CodexGatewayBorrowHandler,
+ pelicanTestHandler *admin.PelicanTestHandler,
upstreamBillingProbe *service.UpstreamBillingProbeService,
ollamaCloudUsage *service.OllamaCloudUsageService,
opencodeGoUsage *service.OpenCodeGoUsageService,
@@ -101,6 +102,7 @@ func ProvideAdminHandlers(
SystemPrompt: systemPromptHandler,
ReasoningRecovery: reasoningRecoveryHandler,
CodexGatewayBorrow: codexGatewayBorrowHandler,
+ PelicanTest: pelicanTestHandler,
}
}
@@ -302,6 +304,7 @@ var ProviderSet = wire.NewSet(
admin.NewSystemPromptHandler,
admin.NewReasoningRecoveryHandler,
admin.NewCodexGatewayBorrowHandler,
+ admin.NewPelicanTestHandler,
ProvideAccountJobRuntime,
// AdminHandlers and Handlers constructors
diff --git a/backend/internal/repository/codex_gateway_borrow_test_repo.go b/backend/internal/repository/codex_gateway_borrow_test_repo.go
index ddf8f78a6fe1..b154fcbcd6a3 100644
--- a/backend/internal/repository/codex_gateway_borrow_test_repo.go
+++ b/backend/internal/repository/codex_gateway_borrow_test_repo.go
@@ -16,11 +16,12 @@ func NewCodexGatewayBorrowTestRepository(db *sql.DB) service.CodexGatewayBorrowT
}
const codexGatewayBorrowTestTaskColumns = `id, client_task_id, created_by, request_hash, status, prompt,
- created_at, expires_at, started_at, finished_at, total, completed, error`
+ created_at, expires_at, started_at, finished_at, total, completed, error, generation_timeout_seconds, execution_mode`
const codexGatewayBorrowTestResultColumns = `r.id, r.task_id, r.ordinal, r.account_id, r.account_name, r.model_id,
r.upstream_model, r.effort, r.status, r.raw_answer, r.raw_response, r.raw_html, r.html, r.error,
- r.started_at, r.finished_at, r.duration_ms, t.expires_at`
+ r.started_at, r.finished_at, r.duration_ms, t.expires_at, r.platform, r.actual_endpoint, r.actual_protocol,
+ r.actual_transport, r.borrow_applied, r.queue_duration_ms, r.preparation_duration_ms, r.generation_duration_ms, r.generation_started_at`
type codexGatewayBorrowTestScanner interface{ Scan(...any) error }
@@ -29,7 +30,7 @@ func scanCodexGatewayBorrowTestTask(row codexGatewayBorrowTestScanner) (*service
var started, finished sql.NullTime
var rawError []byte
if err := row.Scan(&task.ID, &task.ClientTaskID, &task.CreatedBy, &task.RequestHash, &task.Status, &task.Prompt,
- &task.CreatedAt, &task.ExpiresAt, &started, &finished, &task.Total, &task.Completed, &rawError); err != nil {
+ &task.CreatedAt, &task.ExpiresAt, &started, &finished, &task.Total, &task.Completed, &rawError, &task.GenerationTimeoutSeconds, &task.ExecutionMode); err != nil {
return nil, err
}
if started.Valid {
@@ -45,10 +46,11 @@ func scanCodexGatewayBorrowTestTask(row codexGatewayBorrowTestScanner) (*service
func scanCodexGatewayBorrowTestResult(row codexGatewayBorrowTestScanner) (*service.CodexGatewayBorrowTestResult, error) {
result := &service.CodexGatewayBorrowTestResult{}
var answer, rawResponse, rawHTML, html, rawError []byte
- var started, finished sql.NullTime
+ var started, finished, generationStarted sql.NullTime
if err := row.Scan(&result.ID, &result.TaskID, &result.Ordinal, &result.AccountID, &result.AccountName, &result.ModelID,
&result.UpstreamModel, &result.Effort, &result.Status, &answer, &rawResponse, &rawHTML, &html, &rawError,
- &started, &finished, &result.DurationMS, &result.ExpiresAt); err != nil {
+ &started, &finished, &result.DurationMS, &result.ExpiresAt, &result.Platform, &result.ActualEndpoint, &result.ActualProtocol,
+ &result.ActualTransport, &result.BorrowApplied, &result.QueueDurationMS, &result.PreparationDurationMS, &result.GenerationDurationMS, &generationStarted); err != nil {
return nil, err
}
result.RawAnswer, result.RawResponse, result.RawHTML, result.HTML, result.Error = string(answer), string(rawResponse), string(rawHTML), string(html), string(rawError)
@@ -58,6 +60,9 @@ func scanCodexGatewayBorrowTestResult(row codexGatewayBorrowTestScanner) (*servi
if finished.Valid {
result.FinishedAt = &finished.Time
}
+ if generationStarted.Valid {
+ result.GenerationStartedAt = &generationStarted.Time
+ }
if result.Status != "complete" {
result.PreviewUnavailable = "test " + result.Status
} else if result.HTML == "" {
@@ -75,10 +80,10 @@ func (r *codexGatewayBorrowTestRepository) Create(ctx context.Context, task *ser
}
defer func() { _ = tx.Rollback() }()
stored, err := scanCodexGatewayBorrowTestTask(tx.QueryRowContext(ctx, `INSERT INTO codex_gateway_borrow_test_tasks
- (id,client_task_id,created_by,request_hash,status,prompt,total,created_at,expires_at)
- VALUES ($1,$2,$3,$4,'pending',$5,$6,$7,$8)
+ (id,client_task_id,created_by,request_hash,status,prompt,total,created_at,expires_at,generation_timeout_seconds,execution_mode)
+ VALUES ($1,$2,$3,$4,'pending',$5,$6,$7,$8,$9,$10)
ON CONFLICT (client_task_id) DO NOTHING RETURNING `+codexGatewayBorrowTestTaskColumns,
- task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Prompt, task.Total, task.CreatedAt, task.ExpiresAt))
+ task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Prompt, task.Total, task.CreatedAt, task.ExpiresAt, task.GenerationTimeoutSeconds, task.ExecutionMode))
if errors.Is(err, sql.ErrNoRows) {
stored, err = scanCodexGatewayBorrowTestTask(tx.QueryRowContext(ctx, `SELECT `+codexGatewayBorrowTestTaskColumns+`
FROM codex_gateway_borrow_test_tasks WHERE scope='manual_pelican' AND client_task_id=$1 FOR UPDATE`, task.ClientTaskID))
@@ -102,8 +107,8 @@ func (r *codexGatewayBorrowTestRepository) Create(ctx context.Context, task *ser
}
for _, result := range task.Results {
if _, err = tx.ExecContext(ctx, `INSERT INTO codex_gateway_borrow_test_results
- (id,task_id,ordinal,account_id,account_name,model_id,effort,status)
- VALUES ($1,$2,$3,$4,$5,$6,$7,'pending')`, result.ID, stored.ID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.Effort); err != nil {
+ (id,task_id,ordinal,account_id,account_name,model_id,effort,status,platform)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,'pending',$8)`, result.ID, stored.ID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.Effort, result.Platform); err != nil {
return nil, false, err
}
}
@@ -221,12 +226,15 @@ func (r *codexGatewayBorrowTestRepository) SaveResult(ctx context.Context, resul
defer func() { _ = tx.Rollback() }()
if err = codexGatewayBorrowRequireAffected(tx.ExecContext(ctx, `UPDATE codex_gateway_borrow_test_results r
SET account_name=$3,upstream_model=$4,effort=$5,status=$6,raw_answer=$7,raw_response=$8,raw_html=$9,html=$10,error=$11,
- started_at=$12,finished_at=$13,duration_ms=$14,updated_at=CURRENT_TIMESTAMP
+ started_at=$12,finished_at=$13,duration_ms=$14,platform=$15,actual_endpoint=$16,actual_protocol=$17,
+ actual_transport=$18,borrow_applied=$19,queue_duration_ms=$20,preparation_duration_ms=$21,
+ generation_duration_ms=$22,generation_started_at=$23,updated_at=CURRENT_TIMESTAMP
FROM codex_gateway_borrow_test_tasks t
WHERE r.id=$1 AND r.task_id=$2 AND t.id=r.task_id AND t.scope='manual_pelican' AND t.expires_at>CURRENT_TIMESTAMP`,
result.ID, result.TaskID, result.AccountName, result.UpstreamModel, result.Effort, result.Status,
[]byte(result.RawAnswer), []byte(result.RawResponse), []byte(result.RawHTML), []byte(result.HTML), []byte(result.Error),
- result.StartedAt, result.FinishedAt, result.DurationMS)); err != nil {
+ result.StartedAt, result.FinishedAt, result.DurationMS, result.Platform, result.ActualEndpoint, result.ActualProtocol,
+ result.ActualTransport, result.BorrowApplied, result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS, result.GenerationStartedAt)); err != nil {
return err
}
if _, err = tx.ExecContext(ctx, `UPDATE codex_gateway_borrow_test_tasks t
diff --git a/backend/internal/repository/codex_gateway_borrow_test_repo_integration_test.go b/backend/internal/repository/codex_gateway_borrow_test_repo_integration_test.go
index ff93f6f8fbbd..e277e70c1353 100644
--- a/backend/internal/repository/codex_gateway_borrow_test_repo_integration_test.go
+++ b/backend/internal/repository/codex_gateway_borrow_test_repo_integration_test.go
@@ -63,7 +63,42 @@ func codexGatewayBorrowPrivateIntegrationDB(t *testing.T, ctx context.Context) *
ddl, err := migrations.FS.ReadFile("275_codex_gateway_borrow_tests.sql")
require.NoError(t, err)
_, err = privateDB.ExecContext(ctx, string(ddl))
- require.NoError(t, err, "apply the actual new migration inside the private fixture")
+ require.NoError(t, err, "apply the original observation schema inside the private fixture")
+ // Put one legacy record in the original schema before applying the new
+ // migration. Its existing output and retention must survive unchanged.
+ legacyTask, legacyResult := uuid.NewString(), uuid.NewString()
+ legacyClient, err := uuid.NewV7()
+ require.NoError(t, err)
+ created := time.Now().UTC().Truncate(time.Microsecond)
+ legacyAnswer := []byte("legacy original answer\x00tail")
+ _, err = privateDB.ExecContext(ctx, `INSERT INTO codex_gateway_borrow_test_tasks
+ (id,client_task_id,created_by,request_hash,status,prompt,total,created_at,expires_at)
+ VALUES ($1,$2,7,'legacy-hash','complete',$3,1,$4,$5)`, legacyTask, legacyClient.String(), service.CodexGatewayBorrowPelicanPrompt, created, created.Add(service.CodexGatewayBorrowTestTTL))
+ require.NoError(t, err)
+ _, err = privateDB.ExecContext(ctx, `INSERT INTO codex_gateway_borrow_test_results
+ (id,task_id,ordinal,account_id,model_id,effort,status,raw_answer,error)
+ VALUES ($1,$2,1,42,'gpt-6-astra','high','complete',$3,$3)`, legacyResult, legacyTask, legacyAnswer)
+ require.NoError(t, err)
+ ddl, err = migrations.FS.ReadFile("276_pelican_test_execution.sql")
+ require.NoError(t, err)
+ _, err = privateDB.ExecContext(ctx, string(ddl))
+ require.NoError(t, err, "apply the actual additive execution migration")
+ var answer, originalError []byte
+ var executionMode string
+ var timeout int
+ var retention int64
+ err = privateDB.QueryRowContext(ctx, `SELECT r.raw_answer,r.error,t.execution_mode,t.generation_timeout_seconds,
+ EXTRACT(EPOCH FROM (t.expires_at-t.created_at))::BIGINT
+ FROM codex_gateway_borrow_test_results r JOIN codex_gateway_borrow_test_tasks t ON t.id=r.task_id
+ WHERE r.id=$1`, legacyResult).Scan(&answer, &originalError, &executionMode, &timeout, &retention)
+ require.NoError(t, err)
+ require.Equal(t, legacyAnswer, answer)
+ require.Equal(t, legacyAnswer, originalError)
+ require.Equal(t, service.PelicanExecutionModeLegacyCache, executionMode)
+ require.Equal(t, 90, timeout, "historical observations retain their actual legacy budget")
+ require.Equal(t, int64(service.CodexGatewayBorrowTestTTL/time.Second), retention)
+ _, err = privateDB.ExecContext(ctx, `DELETE FROM codex_gateway_borrow_test_tasks WHERE id=$1`, legacyTask)
+ require.NoError(t, err)
return privateDB
}
@@ -76,10 +111,11 @@ func codexGatewayBorrowIntegrationTask(t *testing.T, accountID int64) *service.C
ID: uuid.NewString(), ClientTaskID: clientID.String(), CreatedBy: 7,
RequestHash: strings.Repeat("a", 64), Status: "pending", Prompt: service.CodexGatewayBorrowPelicanPrompt,
CreatedAt: created, ExpiresAt: created.Add(service.CodexGatewayBorrowTestTTL), Total: 1,
+ GenerationTimeoutSeconds: service.PelicanDefaultGenerationTimeoutSeconds, ExecutionMode: service.PelicanExecutionModeAccount,
}
task.Results = []*service.CodexGatewayBorrowTestResult{{
ID: uuid.NewString(), TaskID: task.ID, Ordinal: 1, AccountID: accountID,
- AccountName: "private integration fixture", ModelID: "gpt-6-astra", Effort: "high",
+ AccountName: "private integration fixture", Platform: "openai", ModelID: "gpt-6-astra", Effort: "high",
Status: "pending", ExpiresAt: task.ExpiresAt,
}}
return task
@@ -100,6 +136,8 @@ func TestCodexGatewayBorrowTestRepositoryIntegrationLifecycle(t *testing.T) {
require.Equal(t, task.ID, stored.ID)
require.Equal(t, "pending", stored.Status)
require.Len(t, stored.Results, 1)
+ require.Equal(t, service.PelicanDefaultGenerationTimeoutSeconds, stored.GenerationTimeoutSeconds)
+ require.Equal(t, service.PelicanExecutionModeAccount, stored.ExecutionMode)
require.Equal(t, service.CodexGatewayBorrowTestTTL, stored.ExpiresAt.Sub(stored.CreatedAt))
started := time.Now().UTC().Truncate(time.Microsecond)
@@ -108,6 +146,10 @@ func TestCodexGatewayBorrowTestRepositoryIntegrationLifecycle(t *testing.T) {
finished := started.Add(250 * time.Millisecond)
result.Status, result.UpstreamModel = "complete", "gpt-6-astra-reported"
result.StartedAt, result.FinishedAt, result.DurationMS = &started, &finished, 250
+ generationStarted := started.Add(50 * time.Millisecond)
+ result.GenerationStartedAt = &generationStarted
+ result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS = 20, 30, 200
+ result.ActualEndpoint, result.ActualProtocol, result.ActualTransport, result.BorrowApplied = "https://local.test/responses", "responses", "http", true
result.RawAnswer = "```svg\n\n```"
result.RawResponse = "data: {\"original\":\"full stream\"}\n\n\x00end"
result.RawHTML = ""
@@ -127,6 +169,15 @@ func TestCodexGatewayBorrowTestRepositoryIntegrationLifecycle(t *testing.T) {
require.Equal(t, "gpt-6-astra", readResult.ModelID)
require.Equal(t, result.UpstreamModel, readResult.UpstreamModel)
require.Equal(t, result.DurationMS, readResult.DurationMS)
+ require.Equal(t, result.Platform, readResult.Platform)
+ require.Equal(t, result.ActualEndpoint, readResult.ActualEndpoint)
+ require.Equal(t, result.ActualProtocol, readResult.ActualProtocol)
+ require.Equal(t, result.ActualTransport, readResult.ActualTransport)
+ require.True(t, readResult.BorrowApplied)
+ require.Equal(t, result.QueueDurationMS, readResult.QueueDurationMS)
+ require.Equal(t, result.PreparationDurationMS, readResult.PreparationDurationMS)
+ require.Equal(t, result.GenerationDurationMS, readResult.GenerationDurationMS)
+ require.True(t, readResult.GenerationStartedAt.Equal(generationStarted))
require.True(t, readResult.StartedAt.Equal(started))
require.True(t, readResult.FinishedAt.Equal(finished))
require.True(t, readResult.ExpiresAt.Equal(task.ExpiresAt))
diff --git a/backend/internal/repository/codex_gateway_borrow_test_repo_test.go b/backend/internal/repository/codex_gateway_borrow_test_repo_test.go
index 32d3a5b1fc9e..f06f0d55a0cd 100644
--- a/backend/internal/repository/codex_gateway_borrow_test_repo_test.go
+++ b/backend/internal/repository/codex_gateway_borrow_test_repo_test.go
@@ -24,13 +24,14 @@ func newCodexGatewayBorrowRepoTest(t *testing.T) (*codexGatewayBorrowTestReposit
}
func codexGatewayBorrowTaskTestRows(task *service.CodexGatewayBorrowTestTask) *sqlmock.Rows {
- return sqlmock.NewRows([]string{"id", "client_task_id", "created_by", "request_hash", "status", "prompt", "created_at", "expires_at", "started_at", "finished_at", "total", "completed", "error"}).AddRow(
- task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Status, task.Prompt, task.CreatedAt, task.ExpiresAt, nil, nil, task.Total, task.Completed, []byte(task.Error))
+ return sqlmock.NewRows([]string{"id", "client_task_id", "created_by", "request_hash", "status", "prompt", "created_at", "expires_at", "started_at", "finished_at", "total", "completed", "error", "generation_timeout_seconds", "execution_mode"}).AddRow(
+ task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Status, task.Prompt, task.CreatedAt, task.ExpiresAt, nil, nil, task.Total, task.Completed, []byte(task.Error), task.GenerationTimeoutSeconds, task.ExecutionMode)
}
func codexGatewayBorrowResultTestRows(result *service.CodexGatewayBorrowTestResult) *sqlmock.Rows {
- return sqlmock.NewRows([]string{"id", "task_id", "ordinal", "account_id", "account_name", "model_id", "upstream_model", "effort", "status", "raw_answer", "raw_response", "raw_html", "html", "error", "started_at", "finished_at", "duration_ms", "expires_at"}).AddRow(
- result.ID, result.TaskID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.UpstreamModel, result.Effort, result.Status, []byte(result.RawAnswer), []byte(result.RawResponse), []byte(result.RawHTML), []byte(result.HTML), []byte(result.Error), nil, nil, result.DurationMS, result.ExpiresAt)
+ return sqlmock.NewRows([]string{"id", "task_id", "ordinal", "account_id", "account_name", "model_id", "upstream_model", "effort", "status", "raw_answer", "raw_response", "raw_html", "html", "error", "started_at", "finished_at", "duration_ms", "expires_at", "platform", "actual_endpoint", "actual_protocol", "actual_transport", "borrow_applied", "queue_duration_ms", "preparation_duration_ms", "generation_duration_ms", "generation_started_at"}).AddRow(
+ result.ID, result.TaskID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.UpstreamModel, result.Effort, result.Status, []byte(result.RawAnswer), []byte(result.RawResponse), []byte(result.RawHTML), []byte(result.HTML), []byte(result.Error), result.StartedAt, result.FinishedAt, result.DurationMS, result.ExpiresAt,
+ result.Platform, result.ActualEndpoint, result.ActualProtocol, result.ActualTransport, result.BorrowApplied, result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS, result.GenerationStartedAt)
}
func codexGatewayBorrowRepoFixture(t *testing.T) *service.CodexGatewayBorrowTestTask {
@@ -38,8 +39,9 @@ func codexGatewayBorrowRepoFixture(t *testing.T) *service.CodexGatewayBorrowTest
now := time.Now().UTC()
clientID, err := uuid.NewV7()
require.NoError(t, err)
- task := &service.CodexGatewayBorrowTestTask{ID: uuid.NewString(), ClientTaskID: clientID.String(), CreatedBy: 7, RequestHash: "hash", Status: "pending", Prompt: service.CodexGatewayBorrowPelicanPrompt, CreatedAt: now, ExpiresAt: now.Add(service.CodexGatewayBorrowTestTTL), Total: 1}
- task.Results = []*service.CodexGatewayBorrowTestResult{{ID: uuid.NewString(), TaskID: task.ID, Ordinal: 1, AccountID: 42, AccountName: "paused target", ModelID: "gpt-6-astra", Effort: "high", Status: "pending", ExpiresAt: task.ExpiresAt}}
+ task := &service.CodexGatewayBorrowTestTask{ID: uuid.NewString(), ClientTaskID: clientID.String(), CreatedBy: 7, RequestHash: "hash", Status: "pending", Prompt: service.CodexGatewayBorrowPelicanPrompt, CreatedAt: now, ExpiresAt: now.Add(service.CodexGatewayBorrowTestTTL), Total: 1,
+ GenerationTimeoutSeconds: service.PelicanDefaultGenerationTimeoutSeconds, ExecutionMode: service.PelicanExecutionModeAccount}
+ task.Results = []*service.CodexGatewayBorrowTestResult{{ID: uuid.NewString(), TaskID: task.ID, Ordinal: 1, AccountID: 42, AccountName: "paused target", Platform: "openai", ModelID: "gpt-6-astra", Effort: "high", Status: "pending", ExpiresAt: task.ExpiresAt}}
return task
}
@@ -47,15 +49,17 @@ func TestCodexGatewayBorrowTestRepositoryCreateAtomicAndReplayReadsWithoutWrites
repo, mock := newCodexGatewayBorrowRepoTest(t)
task := codexGatewayBorrowRepoFixture(t)
mock.ExpectBegin()
- mock.ExpectQuery("INSERT INTO codex_gateway_borrow_test_tasks").WithArgs(task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Prompt, task.Total, task.CreatedAt, task.ExpiresAt).WillReturnRows(codexGatewayBorrowTaskTestRows(task))
+ mock.ExpectQuery("INSERT INTO codex_gateway_borrow_test_tasks").WithArgs(task.ID, task.ClientTaskID, task.CreatedBy, task.RequestHash, task.Prompt, task.Total, task.CreatedAt, task.ExpiresAt, task.GenerationTimeoutSeconds, task.ExecutionMode).WillReturnRows(codexGatewayBorrowTaskTestRows(task))
result := task.Results[0]
- mock.ExpectExec("INSERT INTO codex_gateway_borrow_test_results").WithArgs(result.ID, task.ID, 1, result.AccountID, result.AccountName, result.ModelID, result.Effort).WillReturnResult(sqlmock.NewResult(0, 1))
+ mock.ExpectExec("INSERT INTO codex_gateway_borrow_test_results").WithArgs(result.ID, task.ID, 1, result.AccountID, result.AccountName, result.ModelID, result.Effort, result.Platform).WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectCommit()
stored, replayed, err := repo.Create(context.Background(), task)
require.NoError(t, err)
require.False(t, replayed)
require.Equal(t, task.ID, stored.ID)
require.Len(t, stored.Results, 1)
+ require.Equal(t, task.GenerationTimeoutSeconds, stored.GenerationTimeoutSeconds)
+ require.Equal(t, task.ExecutionMode, stored.ExecutionMode)
// A repeated client UUID gets the original full observation, not another child.
task.Status = "complete"
result.Status, result.RawAnswer, result.RawResponse, result.Error = "complete", "answer\x00tail", "full raw stream\x00end", "untruncated upstream diagnostic"
@@ -71,6 +75,7 @@ func TestCodexGatewayBorrowTestRepositoryCreateAtomicAndReplayReadsWithoutWrites
require.Equal(t, result.RawAnswer, stored.Results[0].RawAnswer)
require.Equal(t, result.RawResponse, stored.Results[0].RawResponse)
require.Equal(t, result.Error, stored.Results[0].Error)
+ require.Equal(t, result.Platform, stored.Results[0].Platform)
require.NoError(t, mock.ExpectationsWereMet())
}
@@ -121,6 +126,11 @@ func TestCodexGatewayBorrowTestRepositoryExpiredReadsAndScopedCleanup(t *testing
require.Contains(t, string(data), "client_task_id UUID NOT NULL UNIQUE")
require.Contains(t, string(data), "expires_at - created_at = INTERVAL '24 hours'")
require.Contains(t, string(data), "raw_answer BYTEA NOT NULL")
+ data, err = migrations.FS.ReadFile("276_pelican_test_execution.sql")
+ require.NoError(t, err)
+ require.Contains(t, string(data), "generation_timeout_seconds BETWEEN 60 AND 1800")
+ require.Contains(t, string(data), "DEFAULT 'legacy_cache'")
+ require.NotContains(t, string(data), "DROP")
}
func TestCodexGatewayBorrowTestRepositorySavesFullRawValuesAndMarksRestartIncomplete(t *testing.T) {
@@ -128,11 +138,22 @@ func TestCodexGatewayBorrowTestRepositorySavesFullRawValuesAndMarksRestartIncomp
task := codexGatewayBorrowRepoFixture(t)
result := task.Results[0]
result.Status, result.UpstreamModel, result.RawAnswer, result.RawResponse, result.RawHTML, result.HTML, result.Error = "incomplete", "reported", "answer\x00tail", "SSE\x00raw", "source\x00raw", "html\x00raw", "API returned 403: entire original body\x00tail"
+ started := time.Now().UTC()
+ result.ActualEndpoint, result.ActualProtocol, result.ActualTransport, result.BorrowApplied = "https://upstream.test/responses", "responses", "http", true
+ result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS, result.GenerationStartedAt = 1200, 2300, 3400, &started
mock.ExpectBegin()
- mock.ExpectExec("UPDATE codex_gateway_borrow_test_results r.*t.expires_at>CURRENT_TIMESTAMP").WithArgs(result.ID, result.TaskID, result.AccountName, result.UpstreamModel, result.Effort, result.Status, []byte(result.RawAnswer), []byte(result.RawResponse), []byte(result.RawHTML), []byte(result.HTML), []byte(result.Error), nil, nil, int64(0)).WillReturnResult(sqlmock.NewResult(0, 1))
+ mock.ExpectExec("UPDATE codex_gateway_borrow_test_results r.*t.expires_at>CURRENT_TIMESTAMP").WithArgs(result.ID, result.TaskID, result.AccountName, result.UpstreamModel, result.Effort, result.Status, []byte(result.RawAnswer), []byte(result.RawResponse), []byte(result.RawHTML), []byte(result.HTML), []byte(result.Error), nil, nil, int64(0),
+ result.Platform, result.ActualEndpoint, result.ActualProtocol, result.ActualTransport, result.BorrowApplied, result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS, result.GenerationStartedAt).WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec("UPDATE codex_gateway_borrow_test_tasks t.*completed=.*status NOT IN").WithArgs(task.ID).WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectCommit()
require.NoError(t, repo.SaveResult(context.Background(), result))
+ mock.ExpectQuery("WHERE t.scope='manual_pelican' AND r.id=\\$1 AND t.expires_at>CURRENT_TIMESTAMP").WithArgs(result.ID).WillReturnRows(codexGatewayBorrowResultTestRows(result))
+ stored, err := repo.GetResult(context.Background(), result.ID)
+ require.NoError(t, err)
+ result.PreviewUnavailable = "test incomplete"
+ require.Equal(t, result, stored, "actual invocation and phase durations round-trip alongside untruncated raw observations")
+ // The reader derives the preview reason from the terminal status.
+ require.Equal(t, "test incomplete", stored.PreviewUnavailable)
mock.ExpectBegin()
mock.ExpectExec("UPDATE codex_gateway_borrow_test_results r.*status='incomplete'.*t.scope='manual_pelican'.*r.status IN").WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec("UPDATE codex_gateway_borrow_test_tasks t.*status='incomplete'.*t.scope='manual_pelican'").WillReturnResult(sqlmock.NewResult(0, 1))
diff --git a/backend/internal/repository/concurrency_account_lease_test.go b/backend/internal/repository/concurrency_account_lease_test.go
new file mode 100644
index 000000000000..156d236a75cb
--- /dev/null
+++ b/backend/internal/repository/concurrency_account_lease_test.go
@@ -0,0 +1,96 @@
+//go:build unit
+
+package repository
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/Wei-Shaw/sub2api/internal/service"
+ "github.com/alicebob/miniredis/v2"
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+)
+
+func TestObservedAccountLeaseCacheRefreshRetainsThirtyMinuteRequestAndActiveIndex(t *testing.T) {
+ server := miniredis.RunT(t)
+ client := redis.NewClient(&redis.Options{Addr: server.Addr()})
+ t.Cleanup(func() { require.NoError(t, client.Close()) })
+ cache := NewConcurrencyCache(client, 0, 0)
+ refresher, ok := cache.(service.AccountSlotRefreshCache)
+ require.True(t, ok)
+ ctx := context.Background()
+ now := time.Now().UTC().Truncate(time.Second)
+ server.SetTime(now)
+ acquired, err := cache.AcquireAccountSlot(ctx, 42, 1, "observed-request")
+ require.NoError(t, err)
+ require.True(t, acquired)
+ for i := 0; i < 60; i++ {
+ now = now.Add(30 * time.Second)
+ server.FastForward(30 * time.Second)
+ server.SetTime(now)
+ owned, refreshErr := refresher.RefreshAccountSlot(ctx, 42, "observed-request")
+ require.NoError(t, refreshErr)
+ require.True(t, owned)
+ count, countErr := cache.GetAccountConcurrency(ctx, 42)
+ require.NoError(t, countErr)
+ require.Equal(t, 1, count)
+ acquired, err = cache.AcquireAccountSlot(ctx, 42, 1, "business-must-wait")
+ require.NoError(t, err)
+ require.False(t, acquired, "the same regular account capacity remains reserved for the entire 30-minute observation")
+ }
+ score, err := client.ZScore(ctx, accountSlotKey(42), "observed-request").Result()
+ require.NoError(t, err)
+ require.Equal(t, float64(now.Unix()), score)
+ indexExpiry, err := client.ZScore(ctx, accountActiveIndexKey, "42").Result()
+ require.NoError(t, err)
+ require.Equal(t, float64(now.Unix()+15*60), indexExpiry)
+ require.Equal(t, 15*time.Minute, server.TTL(accountSlotKey(42)))
+ require.NoError(t, cache.ReleaseAccountSlot(ctx, 42, "observed-request"))
+ _, err = client.ZScore(ctx, accountSlotKey(42), "observed-request").Result()
+ require.ErrorIs(t, err, redis.Nil)
+ _, err = client.ZScore(ctx, accountActiveIndexKey, "42").Result()
+ require.ErrorIs(t, err, redis.Nil)
+ acquired, err = cache.AcquireAccountSlot(ctx, 42, 1, "business-after-release")
+ require.NoError(t, err)
+ require.True(t, acquired)
+}
+
+func TestObservedAccountLeaseCacheCannotRecreateExpiredOrDeletedOwner(t *testing.T) {
+ for _, expired := range []bool{false, true} {
+ t.Run(map[bool]string{false: "deleted", true: "expired"}[expired], func(t *testing.T) {
+ server := miniredis.RunT(t)
+ client := redis.NewClient(&redis.Options{Addr: server.Addr()})
+ t.Cleanup(func() { require.NoError(t, client.Close()) })
+ cache := NewConcurrencyCache(client, 1, 60)
+ refresher := cache.(service.AccountSlotRefreshCache)
+ ctx := context.Background()
+ now := time.Now().UTC().Truncate(time.Second)
+ server.SetTime(now)
+ acquired, err := cache.AcquireAccountSlot(ctx, 42, 1, "lost-owner")
+ require.NoError(t, err)
+ require.True(t, acquired)
+ if expired {
+ now = now.Add(time.Minute)
+ server.FastForward(time.Minute)
+ server.SetTime(now)
+ } else {
+ require.NoError(t, client.ZRem(ctx, accountSlotKey(42), "lost-owner").Err())
+ }
+ acquired, err = cache.AcquireAccountSlot(ctx, 42, 1, "new-business-owner")
+ require.NoError(t, err)
+ require.True(t, acquired)
+ owned, err := refresher.RefreshAccountSlot(ctx, 42, "lost-owner")
+ require.NoError(t, err)
+ require.False(t, owned, "refresh must not silently reacquire beyond the account's business limit")
+ _, err = client.ZScore(ctx, accountSlotKey(42), "lost-owner").Result()
+ require.ErrorIs(t, err, redis.Nil)
+ count, err := cache.GetAccountConcurrency(ctx, 42)
+ require.NoError(t, err)
+ require.Equal(t, 1, count)
+ _, err = client.ZScore(ctx, accountSlotKey(42), "new-business-owner").Result()
+ require.NoError(t, err)
+ })
+ }
+}
diff --git a/backend/internal/repository/concurrency_cache.go b/backend/internal/repository/concurrency_cache.go
index c21268f52611..abaddf8c5c05 100644
--- a/backend/internal/repository/concurrency_cache.go
+++ b/backend/internal/repository/concurrency_cache.go
@@ -106,6 +106,23 @@ var (
return {0, now}
`)
+ // Renew one existing regular account-slot member. A removed or expired
+ // request ID is never recreated, even when the account has spare capacity.
+ refreshAccountSlotScript = redis.NewScript(`
+ redis.replicate_commands()
+ local key = KEYS[1]
+ local ttl = tonumber(ARGV[1])
+ local requestID = ARGV[2]
+ local now = tonumber(redis.call('TIME')[1])
+ redis.call('ZREMRANGEBYSCORE', key, '-inf', now - ttl)
+ if redis.call('ZSCORE', key, requestID) == false then
+ return {0, now}
+ end
+ redis.call('ZADD', key, now, requestID)
+ redis.call('EXPIRE', key, ttl)
+ return {1, now}
+ `)
+
// getCountScript 统计有序集合中的槽位数量并清理过期条目
// 使用 Redis TIME 命令获取服务器时间
// KEYS[1] = 普通槽位键,KEYS[2] = 对应 Live 槽位键
@@ -652,6 +669,19 @@ func (c *concurrencyCache) ReleaseAccountSlot(ctx context.Context, accountID int
return nil
}
+func (c *concurrencyCache) RefreshAccountSlot(ctx context.Context, accountID int64, requestID string) (bool, error) {
+ result, now, err := runScriptInt64Pair(ctx, c.rdb, refreshAccountSlotScript, []string{accountSlotKey(accountID)}, c.slotTTLSeconds, requestID)
+ if err != nil {
+ return false, err
+ }
+ if result == 1 {
+ c.touchActiveIndexAt(ctx, accountActiveIndexKey, accountID, now+int64(c.slotTTLSeconds))
+ } else {
+ c.refreshAccountActiveIndex(ctx, accountID)
+ }
+ return result == 1, nil
+}
+
func (c *concurrencyCache) GetAccountConcurrency(ctx context.Context, accountID int64) (int, error) {
key := accountSlotKey(accountID)
// 时间戳在 Lua 脚本内使用 Redis TIME 命令获取
diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go
index 757607d97fc5..771c0c72adc0 100644
--- a/backend/internal/repository/http_upstream.go
+++ b/backend/internal/repository/http_upstream.go
@@ -203,11 +203,18 @@ func NewHTTPUpstream(cfg *config.Config) service.HTTPUpstream {
// 注意:
// - 调用方必须关闭 resp.Body,否则会导致 inFlight 计数泄漏
// - inFlight > 0 的客户端不会被淘汰,确保活跃请求不被中断
-func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID int64, accountConcurrency int) (*http.Response, error) {
+func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID int64, accountConcurrency int) (response *http.Response, requestErr error) {
applyGrokCLIProxyHeaders(req)
if err := s.validateRequestHost(req); err != nil {
return nil, err
}
+ var finishPelican func(*http.Response, error) (*http.Response, error)
+ var err error
+ req, finishPelican, err = service.PreparePelicanHTTPRequest(req, accountID, accountConcurrency, "http")
+ if err != nil {
+ return nil, err
+ }
+ defer func() { response, requestErr = finishPelican(response, requestErr) }()
profile := service.HTTPUpstreamProfileDefault
if req != nil {
profile = service.HTTPUpstreamProfileFromContext(req.Context())
@@ -246,7 +253,7 @@ func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID i
//
// profile 为 nil 时不启用 TLS 指纹,行为与 Do 方法相同。
// profile 非 nil 时使用指定的 Profile 进行 TLS 指纹伪装。
-func (s *httpUpstreamService) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, profile *tlsfingerprint.Profile) (*http.Response, error) {
+func (s *httpUpstreamService) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, profile *tlsfingerprint.Profile) (response *http.Response, requestErr error) {
if profile == nil {
return s.Do(req, proxyURL, accountID, accountConcurrency)
}
@@ -274,6 +281,13 @@ func (s *httpUpstreamService) DoWithTLS(req *http.Request, proxyURL string, acco
if err := s.validateRequestHost(req); err != nil {
return nil, err
}
+ var finishPelican func(*http.Response, error) (*http.Response, error)
+ var err error
+ req, finishPelican, err = service.PreparePelicanHTTPRequest(req, accountID, accountConcurrency, "http")
+ if err != nil {
+ return nil, err
+ }
+ defer func() { response, requestErr = finishPelican(response, requestErr) }()
entry, err := s.acquireClientWithTLS(proxyURL, accountID, accountConcurrency, profile, upstreamProfile)
if err != nil {
diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go
index 1dd14476a020..4e2007bd58e5 100644
--- a/backend/internal/server/routes/admin.go
+++ b/backend/internal/server/routes/admin.go
@@ -23,6 +23,7 @@ func RegisterAdminRoutes(
// 插件 UI 使用短时能力 URL,仅提供经过安装校验的静态资源。
v1.GET("/plugin-ui/:token/*path", h.Admin.Plugin.ServeUIAsset)
v1.GET("/codex-gateway-borrow/preview/:cap/index.html", h.Admin.CodexGatewayBorrow.ServePreview)
+ v1.GET("/pelican-tests/preview/:cap/index.html", h.Admin.PelicanTest.ServePreview)
admin := v1.Group("/admin")
admin.Use(gin.HandlerFunc(adminAuth))
@@ -133,6 +134,7 @@ func RegisterAdminRoutes(
// 推理恢复全局开关
registerReasoningRecoveryRoutes(admin, h)
registerCodexGatewayBorrowRoutes(admin, h)
+ registerPelicanTestRoutes(admin, h)
// 邀请返利(专属用户管理)
registerAffiliateRoutes(admin, h)
diff --git a/backend/internal/server/routes/pelican_tests.go b/backend/internal/server/routes/pelican_tests.go
new file mode 100644
index 000000000000..492d580f167f
--- /dev/null
+++ b/backend/internal/server/routes/pelican_tests.go
@@ -0,0 +1,15 @@
+package routes
+
+import (
+ "github.com/Wei-Shaw/sub2api/internal/handler"
+ "github.com/gin-gonic/gin"
+)
+
+func registerPelicanTestRoutes(admin *gin.RouterGroup, h *handler.Handlers) {
+ pelican := admin.Group("/pelican-tests")
+ pelican.GET("/options", h.Admin.PelicanTest.GetOptions)
+ pelican.POST("/options", h.Admin.PelicanTest.GetAccountOptions)
+ pelican.POST("/tests", h.Admin.PelicanTest.StartTests)
+ pelican.GET("/tests", h.Admin.PelicanTest.ListTests)
+ pelican.GET("/tests/:id", h.Admin.PelicanTest.GetTest)
+}
diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go
index b9691c94a697..980025e7fe67 100644
--- a/backend/internal/service/account_test_service.go
+++ b/backend/internal/service/account_test_service.go
@@ -164,6 +164,10 @@ type AccountTestService struct {
settingService *SettingService
tlsFPProfileService *TLSFingerprintProfileService
openaiGatewayService *OpenAIGatewayService
+ pelicanGatewayService *GatewayService
+ pelicanGeminiService *GeminiMessagesCompatService
+ pelicanConcurrencyService *ConcurrencyService
+ pelicanBorrowService *CodexGatewayBorrowService
modelMetadataRegistryMu sync.Mutex
modelMetadataRegistry map[string]modelsDevProvider
modelMetadataRegistryAt time.Time
@@ -194,7 +198,36 @@ func (s *AccountTestService) SetPluginManager(pluginManager *PluginManager) {
func (s *AccountTestService) SetOpenAIGatewayService(gateway *OpenAIGatewayService) {
if s != nil {
s.openaiGatewayService = gateway
+ if gateway != nil && s.pelicanBorrowService != nil {
+ gateway.SetCodexGatewayBorrowService(s.pelicanBorrowService)
+ }
+ }
+}
+
+func (s *AccountTestService) SetCodexGatewayBorrowService(borrow *CodexGatewayBorrowService) {
+ if s != nil {
+ s.pelicanBorrowService = borrow
+ if s.openaiGatewayService != nil {
+ s.openaiGatewayService.SetCodexGatewayBorrowService(borrow)
+ }
+ }
+}
+
+// SetPelicanBusinessServices binds the same account senders used by ordinary
+// gateway requests. Dependencies are installed once during application wiring.
+func (s *AccountTestService) SetPelicanBusinessServices(gateway *GatewayService, gemini *GeminiMessagesCompatService, concurrency *ConcurrencyService) {
+ if s != nil {
+ s.pelicanGatewayService = gateway
+ s.pelicanGeminiService = gemini
+ s.pelicanConcurrencyService = concurrency
+ }
+}
+
+func (s *AccountTestService) setAccountTestError(ctx context.Context, accountID int64, message string) error {
+ if IsAccountObservation(ctx) {
+ return nil
}
+ return s.accountRepo.SetError(ctx, accountID, message)
}
// FetchOpenAIAccountModels projects shared discovery data into the account UI
@@ -430,10 +463,13 @@ func (s *AccountTestService) TestAccountConnection(c *gin.Context, accountID int
}
// Route to platform-specific test method
- ctx, finishObservation := s.quotaActivity.Attach(ctx)
- c.Request = c.Request.WithContext(ctx)
- ObserveQuotaAccount(ctx, account.ID)
- defer finishObservation()
+ if !IsAccountObservation(ctx) {
+ var finishObservation func()
+ ctx, finishObservation = s.quotaActivity.Attach(ctx)
+ c.Request = c.Request.WithContext(ctx)
+ ObserveQuotaAccount(ctx, account.ID)
+ defer finishObservation()
+ }
if account.IsOpenCodeGo() {
return s.testOpenCodeGoConnection(c, account, modelID, prompt)
}
@@ -633,7 +669,7 @@ func (s *AccountTestService) testClaudeAccountConnection(c *gin.Context, account
// 403 表示账号被上游封禁,标记为 error 状态
if resp.StatusCode == http.StatusForbidden {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
@@ -701,7 +737,7 @@ func (s *AccountTestService) testClaudeVertexServiceAccountConnection(c *gin.Con
body, _ := io.ReadAll(resp.Body)
errMsg := fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body))
if resp.StatusCode == http.StatusForbidden {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
}
@@ -993,7 +1029,7 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
}
defer func() { _ = resp.Body.Close() }()
- if isOAuth && !account.IsShadow() && s.accountRepo != nil {
+ if isOAuth && !account.IsShadow() && s.accountRepo != nil && !IsAccountObservation(ctx) {
if updates, err := extractOpenAICodexProbeUpdates(resp); err == nil && len(updates) > 0 {
_ = s.accountRepo.UpdateExtra(ctx, account.ID, updates)
mergeAccountExtra(account, updates)
@@ -1018,7 +1054,7 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
// 401 Unauthorized: 标记账号为永久错误
if resp.StatusCode == http.StatusUnauthorized && !budgetExceeded && s.accountRepo != nil {
errMsg := fmt.Sprintf("Authentication failed (401): %s", string(body))
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body)))
}
@@ -1191,6 +1227,9 @@ func (s *AccountTestService) applyGrokTestRequestHeaders(req *http.Request, acco
}
func (s *AccountTestService) observeGrokTestResponse(ctx context.Context, account *Account, resp *http.Response) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if resp == nil {
return
}
@@ -2211,7 +2250,7 @@ func (s *AccountTestService) testOpenAIChatCompletionsConnection(
}
if resp.StatusCode == http.StatusUnauthorized && !budgetExceeded && s.accountRepo != nil {
errMsg := fmt.Sprintf("Chat Completions authentication failed (401): %s", string(body))
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, fmt.Sprintf("Chat Completions API (/v1/chat/completions) returned %d: %s", resp.StatusCode, string(body)))
}
@@ -2335,7 +2374,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
resp, err := s.doOpenAIAccountTestUpstream(req, proxyURL, account, true)
if err != nil {
- if s.accountRepo != nil {
+ if s.accountRepo != nil && !IsAccountObservation(ctx) {
updates := buildOpenAICompactProbeExtraUpdates(nil, nil, err, false, time.Now())
_ = s.accountRepo.UpdateExtra(ctx, account.ID, updates)
mergeAccountExtra(account, updates)
@@ -2357,7 +2396,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
}
compactionFound := openAICompactProbeFoundCompactionItem(body)
- if s.accountRepo != nil {
+ if s.accountRepo != nil && !IsAccountObservation(ctx) {
updates := buildOpenAICompactProbeExtraUpdates(resp, body, nil, compactionFound, time.Now())
if !account.IsShadow() {
if codexUpdates, err := extractOpenAICodexProbeUpdates(resp); err == nil && len(codexUpdates) > 0 {
@@ -2377,7 +2416,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
if resp.StatusCode != http.StatusOK {
if resp.StatusCode == http.StatusUnauthorized && !budgetExceeded && s.accountRepo != nil {
errMsg := fmt.Sprintf("Authentication failed (401): %s", string(body))
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body)))
}
@@ -2392,7 +2431,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
}
func (s *AccountTestService) reconcileOpenAI429State(ctx context.Context, account *Account, headers http.Header, body []byte) {
- if s == nil || s.accountRepo == nil || account == nil {
+ if IsAccountObservation(ctx) || s == nil || s.accountRepo == nil || account == nil {
return
}
// Spark quota is observed through its own /wham scope, not global probe 429s.
@@ -2463,7 +2502,7 @@ func (s *AccountTestService) markOpenAIBudgetExceededFromTest(ctx context.Contex
(statusCode != http.StatusOK || !isOpenAIBudgetExceededTerminalEvent(body)) {
return false
}
- if s != nil {
+ if s != nil && !IsAccountObservation(ctx) {
s.openaiGatewayService.handleOpenAIBudgetExceeded(ctx, account, body)
}
return true
diff --git a/backend/internal/service/account_test_service_cn_adaptive.go b/backend/internal/service/account_test_service_cn_adaptive.go
index d73665bdc0d6..079ac6857ccd 100644
--- a/backend/internal/service/account_test_service_cn_adaptive.go
+++ b/backend/internal/service/account_test_service_cn_adaptive.go
@@ -96,7 +96,7 @@ func (s *AccountTestService) testCNProviderAdaptiveAnthropicConnection(c *gin.Co
body, _ := io.ReadAll(resp.Body)
errMsg := fmt.Sprintf("Adaptive Anthropic endpoint returned %d: %s", resp.StatusCode, string(body))
if resp.StatusCode == http.StatusUnauthorized && s.accountRepo != nil {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
}
@@ -150,7 +150,7 @@ func (s *AccountTestService) testCNProviderAdaptiveResponsesConnection(c *gin.Co
body, _ := io.ReadAll(resp.Body)
errMsg := fmt.Sprintf("Adaptive Responses endpoint returned %d: %s", resp.StatusCode, string(body))
if resp.StatusCode == http.StatusUnauthorized && s.accountRepo != nil {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
}
@@ -240,7 +240,7 @@ func (s *AccountTestService) testCNProviderAnthropicConnection(c *gin.Context, a
body, _ := io.ReadAll(resp.Body)
errMsg := fmt.Sprintf("Anthropic endpoint returned %d: %s", resp.StatusCode, string(body))
if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && s.accountRepo != nil {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
}
diff --git a/backend/internal/service/account_test_service_typesafe.go b/backend/internal/service/account_test_service_typesafe.go
index 27d1093cec1e..a55a16ee575b 100644
--- a/backend/internal/service/account_test_service_typesafe.go
+++ b/backend/internal/service/account_test_service_typesafe.go
@@ -83,7 +83,7 @@ func (s *AccountTestService) testTypeSafeAccountConnection(c *gin.Context, accou
errMsg := fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body))
// 401/403 表示 API Key 无效或被上游拒绝,标记为 error 状态。
if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && s.accountRepo != nil {
- _ = s.accountRepo.SetError(ctx, account.ID, errMsg)
+ _ = s.setAccountTestError(ctx, account.ID, errMsg)
}
return s.sendErrorAndEnd(c, errMsg)
}
diff --git a/backend/internal/service/antigravity_credits_overages.go b/backend/internal/service/antigravity_credits_overages.go
index ec3650859ce6..10b45f6174ef 100644
--- a/backend/internal/service/antigravity_credits_overages.go
+++ b/backend/internal/service/antigravity_credits_overages.go
@@ -59,6 +59,9 @@ func (a *Account) isCreditsExhausted() bool {
// setCreditsExhausted 标记账号积分耗尽:写入 model_rate_limits["AICredits"] + 更新缓存。
func (s *AntigravityGatewayService) setCreditsExhausted(ctx context.Context, account *Account) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if account == nil || account.ID == 0 {
return
}
@@ -74,6 +77,9 @@ func (s *AntigravityGatewayService) setCreditsExhausted(ctx context.Context, acc
// clearCreditsExhausted 清除账号的 AICredits 限流 key。
func (s *AntigravityGatewayService) clearCreditsExhausted(ctx context.Context, account *Account) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if account == nil || account.ID == 0 || account.Extra == nil {
return
}
diff --git a/backend/internal/service/antigravity_gateway_retry.go b/backend/internal/service/antigravity_gateway_retry.go
index ff3862569097..6ca62db41465 100644
--- a/backend/internal/service/antigravity_gateway_retry.go
+++ b/backend/internal/service/antigravity_gateway_retry.go
@@ -710,7 +710,7 @@ urlFallbackLoop:
}
}
- if resp != nil && resp.StatusCode < 400 && usedBaseURL != "" {
+ if !IsAccountObservation(p.ctx) && resp != nil && resp.StatusCode < 400 && usedBaseURL != "" {
antigravity.DefaultURLAvailability.MarkSuccess(usedBaseURL)
}
@@ -807,6 +807,9 @@ const googleConfigErrorCooldown = 1 * time.Minute
// tempUnscheduleGoogleConfigError 对服务端配置类 400 错误触发临时封禁,
// 避免短时间内反复调度到同一个有问题的账号。
func tempUnscheduleGoogleConfigError(ctx context.Context, repo AccountRepository, accountID int64, logPrefix string) {
+ if IsAccountObservation(ctx) {
+ return
+ }
until := time.Now().Add(googleConfigErrorCooldown)
reason := "400: invalid project resource name (auto temp-unschedule 1m)"
if err := repo.SetTempUnschedulable(ctx, accountID, until, reason); err != nil {
@@ -822,6 +825,9 @@ const emptyResponseCooldown = 1 * time.Minute
// tempUnscheduleEmptyResponse 对空流式响应触发临时封禁,
// 避免短时间内反复调度到同一个返回空响应的账号。
func tempUnscheduleEmptyResponse(ctx context.Context, repo AccountRepository, accountID int64, logPrefix string) {
+ if IsAccountObservation(ctx) {
+ return
+ }
until := time.Now().Add(emptyResponseCooldown)
reason := "empty stream response (auto temp-unschedule 1m)"
if err := repo.SetTempUnschedulable(ctx, accountID, until, reason); err != nil {
@@ -867,6 +873,9 @@ func isSingleAccountRetry(ctx context.Context) bool {
// 直接使用上游返回的模型 ID(如 claude-sonnet-4-5)作为限流 key
// 返回是否已成功设置(若模型名为空或 repo 为 nil 将返回 false)
func setModelRateLimitByModelName(ctx context.Context, repo AccountRepository, accountID int64, modelName, prefix string, statusCode int, resetAt time.Time, afterSmartRetry bool) bool {
+ if IsAccountObservation(ctx) {
+ return false
+ }
if repo == nil || modelName == "" {
return false
}
@@ -903,6 +912,9 @@ func (s *AntigravityGatewayService) setAntigravityModelRateLimits(ctx context.Co
}
func (s *AntigravityGatewayService) clearStickySession(ctx context.Context, groupID int64, sessionHash string) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if s == nil || s.cache == nil || strings.TrimSpace(sessionHash) == "" {
return
}
@@ -1159,6 +1171,9 @@ func (s *AntigravityGatewayService) handleModelRateLimit(p *handleModelRateLimit
// setModelRateLimitAndClearSession 设置模型限流并清除粘性会话
func (s *AntigravityGatewayService) setModelRateLimitAndClearSession(p *handleModelRateLimitParams, info *antigravitySmartRetryInfo) {
+ if IsAccountObservation(p.ctx) {
+ return
+ }
resetAt := time.Now().Add(info.RetryDelay)
logger.LegacyPrintf("service.antigravity_gateway", "%s status=%d model_rate_limited model=%s account=%d reset_in=%v",
p.prefix, p.statusCode, info.ModelName, p.account.ID, info.RetryDelay)
@@ -1173,6 +1188,9 @@ func (s *AntigravityGatewayService) setModelRateLimitAndClearSession(p *handleMo
// updateAccountModelRateLimitInCache 立即更新 Redis 中账号的模型限流状态
func (s *AntigravityGatewayService) updateAccountModelRateLimitInCache(ctx context.Context, account *Account, modelKey string, resetAt time.Time) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if s.schedulerSnapshot == nil || account == nil || modelKey == "" {
return
}
@@ -1205,6 +1223,9 @@ func (s *AntigravityGatewayService) handleUpstreamError(
requestedModel string,
groupID int64, sessionHash string, isStickySession bool,
) *handleModelRateLimitResult {
+ if IsAccountObservation(ctx) {
+ return nil
+ }
// 遵守自定义错误码策略:未命中则跳过所有限流处理
if !account.ShouldHandleErrorCode(statusCode) {
return nil
diff --git a/backend/internal/service/antigravity_internal500_penalty.go b/backend/internal/service/antigravity_internal500_penalty.go
index 747a4d4eb329..1938800fa892 100644
--- a/backend/internal/service/antigravity_internal500_penalty.go
+++ b/backend/internal/service/antigravity_internal500_penalty.go
@@ -35,6 +35,9 @@ func isAntigravityInternalServerError(statusCode int, body []byte) bool {
func (s *AntigravityGatewayService) applyInternal500Penalty(
ctx context.Context, prefix string, account *Account, count int64,
) {
+ if IsAccountObservation(ctx) {
+ return
+ }
switch {
case count >= int64(internal500PenaltyTier3Threshold):
reason := fmt.Sprintf("INTERNAL 500 consecutive failures: %d rounds", count)
@@ -71,6 +74,9 @@ func (s *AntigravityGatewayService) applyInternal500Penalty(
func (s *AntigravityGatewayService) handleInternal500RetryExhausted(
ctx context.Context, prefix string, account *Account,
) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if s.internal500Cache == nil {
return
}
@@ -87,6 +93,9 @@ func (s *AntigravityGatewayService) handleInternal500RetryExhausted(
func (s *AntigravityGatewayService) resetInternal500Counter(
ctx context.Context, prefix string, accountID int64,
) {
+ if IsAccountObservation(ctx) {
+ return
+ }
if s.internal500Cache == nil {
return
}
diff --git a/backend/internal/service/antigravity_token_provider.go b/backend/internal/service/antigravity_token_provider.go
index a0d0e1c26e8a..f7fcdd568d67 100644
--- a/backend/internal/service/antigravity_token_provider.go
+++ b/backend/internal/service/antigravity_token_provider.go
@@ -104,7 +104,9 @@ func (p *AntigravityTokenProvider) GetAccessToken(ctx context.Context, account *
result, err := p.refreshAPI.RefreshIfNeeded(refreshCtx, account, p.executor, antigravityTokenRefreshSkew)
if err != nil {
// 标记账号临时不可调度,避免后续请求继续命中
- p.markTempUnschedulable(account, err)
+ if !IsAccountObservation(ctx) {
+ p.markTempUnschedulable(account, err)
+ }
if p.refreshPolicy.OnRefreshError == ProviderRefreshErrorReturn {
return "", err
}
diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go
index 7ac331caadb0..2aa5233bae3d 100644
--- a/backend/internal/service/codex_gateway_borrow.go
+++ b/backend/internal/service/codex_gateway_borrow.go
@@ -396,7 +396,7 @@ type codexGatewayBorrowObservationContextKey struct{}
// The token provider may refresh credentials normally, but must not attach an
// observation failure to account health, scheduling or quota.
func WithCodexGatewayBorrowObservation(ctx context.Context) context.Context {
- return context.WithValue(ctx, codexGatewayBorrowObservationContextKey{}, true)
+ return context.WithValue(WithAccountObservation(ctx), codexGatewayBorrowObservationContextKey{}, true)
}
func IsCodexGatewayBorrowObservation(ctx context.Context) bool {
return ctx != nil && ctx.Value(codexGatewayBorrowObservationContextKey{}) == true
diff --git a/backend/internal/service/codex_gateway_borrow_http.go b/backend/internal/service/codex_gateway_borrow_http.go
index dc01e0731054..f073bdad44ba 100644
--- a/backend/internal/service/codex_gateway_borrow_http.go
+++ b/backend/internal/service/codex_gateway_borrow_http.go
@@ -188,6 +188,9 @@ func withCodexGatewayBorrowHTTPFirstOutputStart(ctx context.Context, start time.
}
func codexGatewayBorrowHTTPFirstOutputStart(ctx context.Context, fallback time.Time) time.Time {
+ if IsPelicanGeneration(ctx) {
+ return PelicanFirstOutputStart(ctx, fallback)
+ }
if ctx != nil {
if start, ok := ctx.Value(codexGatewayBorrowHTTPFirstOutputContextKey{}).(time.Time); ok {
return start
diff --git a/backend/internal/service/codex_gateway_borrow_preview.go b/backend/internal/service/codex_gateway_borrow_preview.go
index a9a5ea17b93c..1f36381024b9 100644
--- a/backend/internal/service/codex_gateway_borrow_preview.go
+++ b/backend/internal/service/codex_gateway_borrow_preview.go
@@ -196,7 +196,7 @@ func (p *CodexGatewayBorrowPreviewService) DecorateResult(result *CodexGatewayBo
}
p.capabilities[token] = codexGatewayBorrowPreviewCapability{ResultID: decorated.ID, ExpiresAt: expires}
p.mu.Unlock()
- decorated.PreviewURL = "/api/v1/codex-gateway-borrow/preview/" + token + "/index.html"
+ decorated.PreviewURL = "/api/v1/pelican-tests/preview/" + token + "/index.html"
decorated.PreviewExpiresAt, decorated.PreviewUnavailable = &expires, ""
return &decorated
}
diff --git a/backend/internal/service/codex_gateway_borrow_preview_test.go b/backend/internal/service/codex_gateway_borrow_preview_test.go
index 71ba1ce8d109..1f2eed0f0980 100644
--- a/backend/internal/service/codex_gateway_borrow_preview_test.go
+++ b/backend/internal/service/codex_gateway_borrow_preview_test.go
@@ -56,7 +56,8 @@ func TestCodexGatewayBorrowPreviewCapabilityExpiryAndRecordRecheck(t *testing.T)
require.Empty(t, decorated.PreviewUnavailable)
require.Equal(t, result.ExpiresAt, *decorated.PreviewExpiresAt)
require.Empty(t, result.PreviewURL, "minting must not mutate the stored result")
- token := strings.TrimSuffix(strings.TrimPrefix(decorated.PreviewURL, "/api/v1/codex-gateway-borrow/preview/"), "/index.html")
+ require.True(t, strings.HasPrefix(decorated.PreviewURL, "/api/v1/pelican-tests/preview/"))
+ token := strings.TrimSuffix(strings.TrimPrefix(decorated.PreviewURL, "/api/v1/pelican-tests/preview/"), "/index.html")
require.Len(t, token, 43)
html, err := preview.Resolve(context.Background(), token)
require.NoError(t, err)
@@ -68,7 +69,7 @@ func TestCodexGatewayBorrowPreviewCapabilityExpiryAndRecordRecheck(t *testing.T)
result.ExpiresAt = now.Add(time.Hour)
decorated = preview.DecorateResult(result)
require.Equal(t, now.Add(CodexGatewayBorrowPreviewTTL), *decorated.PreviewExpiresAt)
- token = strings.TrimSuffix(strings.TrimPrefix(decorated.PreviewURL, "/api/v1/codex-gateway-borrow/preview/"), "/index.html")
+ token = strings.TrimSuffix(strings.TrimPrefix(decorated.PreviewURL, "/api/v1/pelican-tests/preview/"), "/index.html")
now = now.Add(CodexGatewayBorrowPreviewTTL)
reads := repo.resultReads
_, err = preview.Resolve(context.Background(), token)
diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go
index fcbaf59cd00f..de7f25a291e3 100644
--- a/backend/internal/service/codex_gateway_borrow_probe.go
+++ b/backend/internal/service/codex_gateway_borrow_probe.go
@@ -301,6 +301,27 @@ 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
+ leasedCtx, release, acquireErr := AcquirePelicanExecution(req.Context(), account.ID)
+ if acquireErr != nil {
+ shot.errorText = acquireErr.Error()
+ return shot, acquireErr
+ }
+ defer release()
+ req = req.WithContext(leasedCtx)
+ if pelicanExecutionFromContext(leasedCtx) != nil {
+ var capacity *ConcurrencyService
+ if capture := pelicanCaptureFromContext(leasedCtx); capture != nil {
+ capacity = capture.concurrency
+ } else if s.gateway != nil {
+ capacity = s.gateway.concurrencyService
+ }
+ releaseCapacity, capacityErr := acquirePelicanBusinessAccountSlot(leasedCtx, capacity, account.ID, account.Concurrency)
+ if capacityErr != nil {
+ shot.errorText = capacityErr.Error()
+ return shot, capacityErr
+ }
+ defer releaseCapacity()
+ }
wire, err := prepareOpenAICodexWireRequest(req, account)
if err != nil {
return shot, err
diff --git a/backend/internal/service/codex_gateway_borrow_test_runner.go b/backend/internal/service/codex_gateway_borrow_test_runner.go
index f42b04b317d2..b6adbfbc5b26 100644
--- a/backend/internal/service/codex_gateway_borrow_test_runner.go
+++ b/backend/internal/service/codex_gateway_borrow_test_runner.go
@@ -17,13 +17,12 @@ import (
const (
CodexGatewayBorrowTestTTL = 24 * time.Hour
CodexGatewayBorrowPelicanPrompt = "创建一个HTML,内容是SVG绘制一个鹈鹕骑自行车的2D动画,你不需要任何测试"
- codexGatewayBorrowTestTimeout = 90 * time.Second
codexGatewayBorrowTestMaxTargets = 5000
)
var (
ErrCodexGatewayBorrowTestNotFound = errors.New("codex gateway borrow test not found or expired")
- ErrCodexGatewayBorrowTestReplayConflict = errors.New("client_task_id was reused with different targets")
+ ErrCodexGatewayBorrowTestReplayConflict = errors.New("client_task_id was reused with different test parameters")
ErrCodexGatewayBorrowTestExpired = errors.New("codex gateway borrow test has expired")
ErrCodexGatewayBorrowTestInvalidRequest = errors.New("invalid manual pelican test request")
)
@@ -35,50 +34,63 @@ type CodexGatewayBorrowTestTarget struct {
}
type CodexGatewayBorrowTestRequest struct {
- ClientTaskID string `json:"client_task_id"`
- Targets []CodexGatewayBorrowTestTarget `json:"targets"`
+ ClientTaskID string `json:"client_task_id"`
+ GenerationTimeoutSeconds int `json:"generation_timeout_seconds"`
+ Targets []CodexGatewayBorrowTestTarget `json:"targets"`
+ Standalone bool `json:"-"`
}
type CodexGatewayBorrowTestResult struct {
- ID string `json:"id"`
- TaskID string `json:"task_id"`
- Ordinal int `json:"ordinal"`
- AccountID int64 `json:"account_id"`
- AccountName string `json:"account_name"`
- ModelID string `json:"model_id"`
- UpstreamModel string `json:"upstream_model"`
- Effort string `json:"effort"`
- Status string `json:"status"`
- RawAnswer string `json:"raw_answer"`
- RawResponse string `json:"raw_response"`
- RawHTML string `json:"raw_html"`
- HTML string `json:"html"`
- Error string `json:"error"`
- StartedAt *time.Time `json:"started_at,omitempty"`
- FinishedAt *time.Time `json:"finished_at,omitempty"`
- DurationMS int64 `json:"duration_ms"`
- ExpiresAt time.Time `json:"expires_at"`
- PreviewURL string `json:"preview_url,omitempty"`
- PreviewExpiresAt *time.Time `json:"preview_expires_at,omitempty"`
- PreviewUnavailable string `json:"preview_unavailable"`
+ ID string `json:"id"`
+ TaskID string `json:"task_id"`
+ Ordinal int `json:"ordinal"`
+ AccountID int64 `json:"account_id"`
+ AccountName string `json:"account_name"`
+ Platform string `json:"platform"`
+ ModelID string `json:"model_id"`
+ UpstreamModel string `json:"upstream_model"`
+ Effort string `json:"effort"`
+ Status string `json:"status"`
+ RawAnswer string `json:"raw_answer"`
+ RawResponse string `json:"raw_response"`
+ RawHTML string `json:"raw_html"`
+ HTML string `json:"html"`
+ Error string `json:"error"`
+ StartedAt *time.Time `json:"started_at,omitempty"`
+ FinishedAt *time.Time `json:"finished_at,omitempty"`
+ DurationMS int64 `json:"duration_ms"`
+ GenerationStartedAt *time.Time `json:"generation_started_at,omitempty"`
+ QueueDurationMS int64 `json:"queue_duration_ms"`
+ PreparationDurationMS int64 `json:"preparation_duration_ms"`
+ GenerationDurationMS int64 `json:"generation_duration_ms"`
+ ActualEndpoint string `json:"actual_endpoint"`
+ ActualProtocol string `json:"actual_protocol"`
+ ActualTransport string `json:"actual_transport"`
+ BorrowApplied bool `json:"borrow_applied"`
+ ExpiresAt time.Time `json:"expires_at"`
+ PreviewURL string `json:"preview_url,omitempty"`
+ PreviewExpiresAt *time.Time `json:"preview_expires_at,omitempty"`
+ PreviewUnavailable string `json:"preview_unavailable"`
}
type CodexGatewayBorrowTestTask struct {
- ID string `json:"id"`
- ClientTaskID string `json:"client_task_id"`
- CreatedBy int64 `json:"created_by"`
- RequestHash string `json:"-"`
- Status string `json:"status"`
- Prompt string `json:"prompt"`
- CreatedAt time.Time `json:"created_at"`
- ExpiresAt time.Time `json:"expires_at"`
- StartedAt *time.Time `json:"started_at,omitempty"`
- FinishedAt *time.Time `json:"finished_at,omitempty"`
- Total int `json:"total"`
- Completed int `json:"completed"`
- Error string `json:"error"`
- Results []*CodexGatewayBorrowTestResult `json:"results,omitempty"`
- Replayed bool `json:"replayed"`
+ ID string `json:"id"`
+ ClientTaskID string `json:"client_task_id"`
+ CreatedBy int64 `json:"created_by"`
+ RequestHash string `json:"-"`
+ GenerationTimeoutSeconds int `json:"generation_timeout_seconds"`
+ ExecutionMode string `json:"execution_mode"`
+ Status string `json:"status"`
+ Prompt string `json:"prompt"`
+ CreatedAt time.Time `json:"created_at"`
+ ExpiresAt time.Time `json:"expires_at"`
+ StartedAt *time.Time `json:"started_at,omitempty"`
+ FinishedAt *time.Time `json:"finished_at,omitempty"`
+ Total int `json:"total"`
+ Completed int `json:"completed"`
+ Error string `json:"error"`
+ Results []*CodexGatewayBorrowTestResult `json:"results,omitempty"`
+ Replayed bool `json:"replayed"`
}
type CodexGatewayBorrowTestList struct {
@@ -111,30 +123,26 @@ type codexGatewayBorrowAccountReader interface {
GetByID(context.Context, int64) (*Account, error)
}
-type codexGatewayBorrowAccountSlot struct {
- slot chan struct{}
- refs int
-}
-
-// All batches share these three slots and account locks. No native test service,
-// recovery service, rate-limit writer or account-state writer is used here.
+// All batches, including legacy cache-only tests, share ten outbound execution
+// slots. Account leases surround actual requests rather than route preparation.
type CodexGatewayBorrowTestRunner struct {
- repo CodexGatewayBorrowTestRepository
- generator codexGatewayBorrowPelicanGenerator
- accounts codexGatewayBorrowAccountReader
- slots chan struct{}
- accountMu sync.Mutex
- accountSlots map[int64]*codexGatewayBorrowAccountSlot
- initMu sync.Mutex
- initialized bool
- stopCtx context.Context
- stop context.CancelFunc
- stopOnce sync.Once
- cleanupDone chan struct{}
- runMu sync.Mutex
- running sync.WaitGroup
- stopped bool
- now func() time.Time
+ repo CodexGatewayBorrowTestRepository
+ generator codexGatewayBorrowPelicanGenerator
+ standaloneGenerator codexGatewayBorrowPelicanGenerator
+ accounts codexGatewayBorrowAccountReader
+ coordinator *pelicanExecutionCoordinator
+ initMu sync.Mutex
+ initialized bool
+ stopCtx context.Context
+ stop context.CancelFunc
+ stopOnce sync.Once
+ cleanupDone chan struct{}
+ runMu sync.Mutex
+ running sync.WaitGroup
+ stopped bool
+ activeTasks map[string]bool
+ now func() time.Time
+ generationContext func(context.Context, time.Duration) (context.Context, context.CancelFunc)
}
func NewCodexGatewayBorrowTestRunner(repo CodexGatewayBorrowTestRepository, core *CodexGatewayBorrowService, accounts AccountRepository) *CodexGatewayBorrowTestRunner {
@@ -144,12 +152,18 @@ func NewCodexGatewayBorrowTestRunner(repo CodexGatewayBorrowTestRepository, core
func newCodexGatewayBorrowTestRunner(repo CodexGatewayBorrowTestRepository, generator codexGatewayBorrowPelicanGenerator, accounts codexGatewayBorrowAccountReader) *CodexGatewayBorrowTestRunner {
ctx, cancel := context.WithCancel(context.Background())
r := &CodexGatewayBorrowTestRunner{repo: repo, generator: generator, accounts: accounts,
- slots: make(chan struct{}, 3), accountSlots: make(map[int64]*codexGatewayBorrowAccountSlot),
- stopCtx: ctx, stop: cancel, cleanupDone: make(chan struct{}), now: time.Now}
+ coordinator: sharedPelicanExecution, activeTasks: make(map[string]bool),
+ stopCtx: ctx, stop: cancel, cleanupDone: make(chan struct{}), now: time.Now, generationContext: context.WithTimeout}
go r.cleanupLoop()
return r
}
+// SetStandaloneGenerator is configured once before serving. Both entry points
+// use the same runner so restart interruption and outbound capacity stay shared.
+func (r *CodexGatewayBorrowTestRunner) SetStandaloneGenerator(generator codexGatewayBorrowPelicanGenerator) {
+ r.standaloneGenerator = generator
+}
+
func (r *CodexGatewayBorrowTestRunner) Stop() {
r.stopOnce.Do(func() {
r.runMu.Lock()
@@ -204,19 +218,26 @@ func normalizeCodexGatewayBorrowTestRequest(request CodexGatewayBorrowTestReques
return request, fmt.Errorf("%w: client_task_id timestamp is in the future", ErrCodexGatewayBorrowTestInvalidRequest)
}
request.ClientTaskID = id.String()
+ if request.GenerationTimeoutSeconds == 0 {
+ request.GenerationTimeoutSeconds = PelicanDefaultGenerationTimeoutSeconds
+ }
+ if request.GenerationTimeoutSeconds < PelicanMinGenerationTimeoutSeconds || request.GenerationTimeoutSeconds > PelicanMaxGenerationTimeoutSeconds {
+ return request, fmt.Errorf("%w: generation_timeout_seconds must be between %d and %d", ErrCodexGatewayBorrowTestInvalidRequest,
+ PelicanMinGenerationTimeoutSeconds, PelicanMaxGenerationTimeoutSeconds)
+ }
targets := make([]CodexGatewayBorrowTestTarget, 0, len(request.Targets))
seen := make(map[string]bool, len(request.Targets))
for _, target := range request.Targets {
target.ModelID = strings.TrimSpace(target.ModelID)
target.Effort = strings.ToLower(strings.TrimSpace(target.Effort))
- if target.Effort == "" {
+ if target.Effort == "" && !request.Standalone {
target.Effort = "high"
}
if target.AccountID <= 0 || target.ModelID == "" || len(target.ModelID) > 512 || strings.ContainsRune(target.ModelID, '\x00') {
return request, fmt.Errorf("%w: each target needs a positive account_id and a valid model_id", ErrCodexGatewayBorrowTestInvalidRequest)
}
switch target.Effort {
- case "none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra":
+ case "", "none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra":
default:
return request, fmt.Errorf("%w: invalid reasoning effort %q", ErrCodexGatewayBorrowTestInvalidRequest, target.Effort)
}
@@ -244,24 +265,43 @@ func (r *CodexGatewayBorrowTestRunner) Create(ctx context.Context, createdBy int
if err = r.initialize(ctx); err != nil {
return nil, false, err
}
+ mode := PelicanExecutionModeLegacyCache
+ if request.Standalone {
+ mode = PelicanExecutionModeAccount
+ }
+ // Default legacy requests retain their existing hash so bookmarks/replays
+ // continue to read records created before this migration. New execution
+ // parameters participate in every standalone request's idempotency check.
raw, _ := json.Marshal(request.Targets)
+ if request.Standalone || request.GenerationTimeoutSeconds != PelicanDefaultGenerationTimeoutSeconds {
+ raw, _ = json.Marshal(struct {
+ Targets []CodexGatewayBorrowTestTarget `json:"targets"`
+ GenerationTimeoutSeconds int `json:"generation_timeout_seconds"`
+ ExecutionMode string `json:"execution_mode"`
+ }{request.Targets, request.GenerationTimeoutSeconds, mode})
+ }
hash := sha256.Sum256(raw)
now := r.now().UTC()
task := &CodexGatewayBorrowTestTask{ID: uuid.NewString(), ClientTaskID: request.ClientTaskID,
CreatedBy: createdBy, RequestHash: hex.EncodeToString(hash[:]), Status: "pending",
+ GenerationTimeoutSeconds: request.GenerationTimeoutSeconds, ExecutionMode: mode,
Prompt: CodexGatewayBorrowPelicanPrompt, CreatedAt: now, ExpiresAt: now.Add(CodexGatewayBorrowTestTTL),
Total: len(request.Targets), Results: make([]*CodexGatewayBorrowTestResult, 0, len(request.Targets))}
- names := make(map[int64]string)
+ accountInfo := make(map[int64]*Account)
for i, target := range request.Targets {
- name, found := names[target.AccountID]
+ account, found := accountInfo[target.AccountID]
if !found && r.accounts != nil {
- if account, readErr := r.accounts.GetByID(ctx, target.AccountID); readErr == nil && account != nil {
- name = account.Name
+ if read, readErr := r.accounts.GetByID(ctx, target.AccountID); readErr == nil {
+ account = read
}
- names[target.AccountID] = name
+ accountInfo[target.AccountID] = account
+ }
+ name, platform := "", ""
+ if account != nil {
+ name, platform = account.Name, account.Platform
}
task.Results = append(task.Results, &CodexGatewayBorrowTestResult{ID: uuid.NewString(), TaskID: task.ID,
- Ordinal: i + 1, AccountID: target.AccountID, AccountName: name, ModelID: target.ModelID,
+ Ordinal: i + 1, AccountID: target.AccountID, AccountName: name, Platform: platform, ModelID: target.ModelID,
Effort: target.Effort, Status: "pending", ExpiresAt: task.ExpiresAt,
PreviewUnavailable: "test has not completed"})
}
@@ -288,62 +328,34 @@ func (r *CodexGatewayBorrowTestRunner) Get(ctx context.Context, id string) (*Cod
type CodexGatewayBorrowTestEvent struct {
Type string `json:"type"`
+ Phase string `json:"phase,omitempty"`
TaskID string `json:"task_id,omitempty"`
Task *CodexGatewayBorrowTestTask `json:"task,omitempty"`
Result *CodexGatewayBorrowTestResult `json:"result,omitempty"`
}
-func (r *CodexGatewayBorrowTestRunner) acquire(ctx context.Context, accountID int64) (func(), bool) {
- r.accountMu.Lock()
- entry := r.accountSlots[accountID]
- if entry == nil {
- entry = &codexGatewayBorrowAccountSlot{slot: make(chan struct{}, 1)}
- r.accountSlots[accountID] = entry
- }
- entry.refs++
- r.accountMu.Unlock()
- dropReference := func() {
- r.accountMu.Lock()
- entry.refs--
- if entry.refs == 0 {
- delete(r.accountSlots, accountID)
- }
- r.accountMu.Unlock()
- }
- select {
- case entry.slot <- struct{}{}:
- case <-ctx.Done():
- dropReference()
- return nil, false
- }
- select {
- case r.slots <- struct{}{}:
- case <-ctx.Done():
- <-entry.slot
- dropReference()
- return nil, false
- }
- if ctx.Err() != nil {
- <-r.slots
- <-entry.slot
- dropReference()
- return nil, false
- }
- return func() { <-r.slots; <-entry.slot; dropReference() }, true
-}
-
// Run is synchronous with the SSE request. Cancellation stops both queued and
// in-flight targets; detached, short writes retain their final observations.
func (r *CodexGatewayBorrowTestRunner) Run(parent context.Context, task *CodexGatewayBorrowTestTask, emit func(CodexGatewayBorrowTestEvent)) error {
r.runMu.Lock()
+ if task.Replayed || task.Status != "pending" || r.activeTasks[task.ID] {
+ r.runMu.Unlock()
+ return nil
+ }
if r.stopped {
r.runMu.Unlock()
return r.finishUnstarted(parent, task, "cancelled", "test runner stopped", emit)
}
r.running.Add(1)
+ r.activeTasks[task.ID] = true
r.runMu.Unlock()
- defer r.running.Done()
- ctx, cancel := context.WithCancel(parent)
+ defer func() {
+ r.runMu.Lock()
+ delete(r.activeTasks, task.ID)
+ r.runMu.Unlock()
+ r.running.Done()
+ }()
+ ctx, cancel := context.WithDeadline(parent, task.ExpiresAt)
defer cancel()
stopLink := context.AfterFunc(r.stopCtx, cancel)
defer stopLink()
@@ -352,6 +364,9 @@ func (r *CodexGatewayBorrowTestRunner) Run(parent context.Context, task *CodexGa
return ErrCodexGatewayBorrowTestExpired
}
if err := r.repo.StartTask(ctx, task.ID, started); err != nil {
+ if errors.Is(err, ErrCodexGatewayBorrowTestNotFound) {
+ return err
+ }
status := "incomplete"
if ctx.Err() != nil {
status = "cancelled"
@@ -367,13 +382,48 @@ func (r *CodexGatewayBorrowTestRunner) Run(parent context.Context, task *CodexGa
wg.Add(1)
go func() {
defer wg.Done()
- if release, ok := r.acquire(ctx, result.AccountID); ok {
- defer release()
- r.runOne(ctx, result, emit)
+ budget := task.GenerationTimeoutSeconds
+ if budget == 0 {
+ budget = PelicanDefaultGenerationTimeoutSeconds
+ }
+ state := newPelicanExecutionState(ctx, r.coordinator, time.Duration(budget)*time.Second, r.now)
+ state.withTimeout = r.generationContext
+ state.onPhase = func(phase string, snapshot pelicanExecutionSnapshot) {
+ if emit != nil {
+ copy := *result
+ applyPelicanExecutionSnapshot(©, snapshot)
+ emit(CodexGatewayBorrowTestEvent{Type: "result_phase", Phase: phase, TaskID: task.ID, Result: ©})
+ }
+ }
+ testCtx := context.WithValue(state.parent, pelicanExecutionContextKey{}, state)
+ if emit != nil {
+ copy := *result
+ emit(CodexGatewayBorrowTestEvent{Type: "result_phase", Phase: "queued", TaskID: task.ID, Result: ©})
+ }
+ if task.ExecutionMode == PelicanExecutionModeAccount {
+ r.runOne(testCtx, result, r.standaloneGenerator, false, emit)
} else {
- finished := r.now().UTC()
- result.Status, result.Error, result.FinishedAt = "cancelled", "test cancelled before dispatch", &finished
- result.PreviewUnavailable = "test cancelled"
+ // The existing cache-only generator never prepares nested routes.
+ // Its outer lease also covers legacy callers without sender hooks.
+ leasedCtx, release, acquireErr := AcquirePelicanExecution(testCtx, result.AccountID)
+ if acquireErr == nil {
+ r.runOne(leasedCtx, result, r.generator, true, emit)
+ release()
+ } else {
+ finished := r.now().UTC()
+ result.Status, result.Error, result.FinishedAt = "cancelled", "test cancelled before dispatch", &finished
+ result.PreviewUnavailable = "test cancelled"
+ }
+ }
+ snapshot := state.finish()
+ applyPelicanExecutionSnapshot(result, snapshot)
+ if snapshot.generationTimedOut && ctx.Err() == nil && result.Status != "skipped" {
+ result.Status = "incomplete"
+ result.PreviewUnavailable = "test incomplete"
+ if result.Error != "" {
+ result.Error += "\n"
+ }
+ result.Error += "generation budget exhausted"
}
writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(parent), 10*time.Second)
writeErr := r.repo.SaveResult(writeCtx, result)
@@ -408,7 +458,8 @@ func (r *CodexGatewayBorrowTestRunner) Run(parent context.Context, task *CodexGa
return nil
}
-func (r *CodexGatewayBorrowTestRunner) runOne(ctx context.Context, result *CodexGatewayBorrowTestResult, emit func(CodexGatewayBorrowTestEvent)) {
+func (r *CodexGatewayBorrowTestRunner) runOne(ctx context.Context, result *CodexGatewayBorrowTestResult, generator codexGatewayBorrowPelicanGenerator, legacy bool, emit func(CodexGatewayBorrowTestEvent)) {
+ callerCtx := ctx
now := r.now().UTC()
if !result.ExpiresAt.After(now) {
result.Status, result.Error, result.FinishedAt = "skipped", "test record expired before dispatch", &now
@@ -432,11 +483,18 @@ func (r *CodexGatewayBorrowTestRunner) runOne(ctx context.Context, result *Codex
result.PreviewUnavailable = "test cancelled"
return
}
- testCtx, cancel := context.WithTimeout(ctx, codexGatewayBorrowTestTimeout)
- generated, err := r.generator.GeneratePelican(testCtx, result.AccountID, result.ModelID, result.Effort)
- cancel()
+ BeginPelicanPreparation(ctx)
+ if generator == nil {
+ result.Status, result.Error, result.FinishedAt = "skipped", "account Pelican generator is unavailable", &now
+ result.PreviewUnavailable = "test skipped"
+ return
+ }
+ if legacy {
+ ctx = BeginPelicanGeneration(ctx)
+ }
+ generated, err := generator.GeneratePelican(ctx, result.AccountID, result.ModelID, result.Effort)
finished := r.now().UTC()
- result.FinishedAt, result.DurationMS = &finished, finished.Sub(now).Milliseconds()
+ result.FinishedAt, result.DurationMS = &finished, nonnegativePelicanDuration(finished.Sub(now)).Milliseconds()
if generated != nil {
result.RawAnswer, result.RawResponse, result.Error = generated.RawAnswer, generated.RawResponse, generated.Error
result.UpstreamModel = generated.ModelID
@@ -454,11 +512,24 @@ func (r *CodexGatewayBorrowTestRunner) runOne(ctx context.Context, result *Codex
if result.Status != "skipped" && result.Status != "incomplete" && result.Status != "failed" {
result.Status = "failed"
}
+ if errors.Is(err, context.DeadlineExceeded) && result.Status != "skipped" {
+ result.Status = "incomplete"
+ }
}
- if ctx.Err() != nil {
+ if callerCtx.Err() != nil {
result.Status = "cancelled"
+ if errors.Is(context.Cause(callerCtx), context.DeadlineExceeded) {
+ result.Status = "incomplete"
+ }
+ if cause := context.Cause(callerCtx); errors.Is(cause, ErrObservedAccountLeaseLost) {
+ result.Status = "incomplete"
+ if result.Error != "" {
+ result.Error += "\n"
+ }
+ result.Error += cause.Error()
+ }
if result.Error == "" {
- result.Error = ctx.Err().Error()
+ result.Error = callerCtx.Err().Error()
}
}
switch result.Status {
diff --git a/backend/internal/service/codex_gateway_borrow_test_runner_test.go b/backend/internal/service/codex_gateway_borrow_test_runner_test.go
index 7a027b5c8579..cc66411f4843 100644
--- a/backend/internal/service/codex_gateway_borrow_test_runner_test.go
+++ b/backend/internal/service/codex_gateway_borrow_test_runner_test.go
@@ -4,6 +4,7 @@ package service
import (
"context"
+ "encoding/json"
"errors"
"sync"
"testing"
@@ -134,6 +135,7 @@ func TestCodexGatewayBorrowRunnerPersistsBeforeDispatchDedupsAndReplays(t *testi
require.NoError(t, err)
require.True(t, replayed)
require.Equal(t, task.ID, replayedTask.ID)
+ require.NoError(t, runner.Run(context.Background(), replayedTask, nil))
require.Equal(t, 1, calls, "reading an existing task must never regenerate")
request.Targets[0].Effort = "medium"
_, _, err = runner.Create(context.Background(), 7, request)
@@ -174,7 +176,7 @@ func TestCodexGatewayBorrowRunnerRejectsExpiredV7AfterRowsAreGone(t *testing.T)
require.ErrorIs(t, err, ErrCodexGatewayBorrowTestInvalidRequest)
}
-func TestCodexGatewayBorrowRunnerSharesThreeSlotsAndSerializesAccountsAcrossBatches(t *testing.T) {
+func TestCodexGatewayBorrowRunnerSharesTenSlotsAndSerializesAccountsAcrossBatches(t *testing.T) {
repo := newCodexGatewayBorrowMemoryTestRepo()
var mu sync.Mutex
active, maxActive, calls := 0, 0, 0
@@ -207,24 +209,30 @@ func TestCodexGatewayBorrowRunnerSharesThreeSlotsAndSerializesAccountsAcrossBatc
}}
runner := newCodexGatewayBorrowTestRunner(repo, gen, nil)
t.Cleanup(runner.Stop)
- first, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Targets: []CodexGatewayBorrowTestTarget{{AccountID: 1, ModelID: "gpt-6-astra"}, {AccountID: 2, ModelID: "gpt-6-astra"}, {AccountID: 3, ModelID: "gpt-6-astra"}}})
+ firstTargets, secondTargets := make([]CodexGatewayBorrowTestTarget, 7), make([]CodexGatewayBorrowTestTarget, 7)
+ for i := range firstTargets {
+ firstTargets[i] = CodexGatewayBorrowTestTarget{AccountID: int64(i + 1), ModelID: "gpt-6-astra"}
+ secondTargets[i] = CodexGatewayBorrowTestTarget{AccountID: int64(i + 7), ModelID: "gpt-6.1-sol"}
+ }
+ secondTargets[0].AccountID = 1
+ first, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Targets: firstTargets})
require.NoError(t, err)
- second, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Targets: []CodexGatewayBorrowTestTarget{{AccountID: 1, ModelID: "gpt-6.1-sol"}, {AccountID: 4, ModelID: "gpt-6-astra"}, {AccountID: 5, ModelID: "gpt-6-astra"}}})
+ second, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Targets: secondTargets})
require.NoError(t, err)
done := make(chan error, 2)
go func() { done <- runner.Run(context.Background(), first, nil) }()
go func() { done <- runner.Run(context.Background(), second, nil) }()
- for i := 0; i < 3; i++ {
+ for i := 0; i < PelicanExecutionConcurrency; i++ {
select {
case <-started:
case <-time.After(2 * time.Second):
- t.Fatal("three dispatches did not start")
+ t.Fatal("ten dispatches did not start")
}
}
mu.Lock()
activeBeforeRelease := active
mu.Unlock()
- require.Equal(t, 3, activeBeforeRelease)
+ require.Equal(t, PelicanExecutionConcurrency, activeBeforeRelease)
close(gate)
for i := 0; i < 2; i++ {
select {
@@ -236,8 +244,8 @@ func TestCodexGatewayBorrowRunnerSharesThreeSlotsAndSerializesAccountsAcrossBatc
}
mu.Lock()
defer mu.Unlock()
- require.Equal(t, 6, calls)
- require.Equal(t, 3, maxActive)
+ require.Equal(t, 14, calls)
+ require.Equal(t, PelicanExecutionConcurrency, maxActive)
for _, max := range maxAccountActive {
require.Equal(t, 1, max)
}
@@ -247,7 +255,7 @@ func TestCodexGatewayBorrowRunnerSharesThreeSlotsAndSerializesAccountsAcrossBatc
func TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRawOutput(t *testing.T) {
repo := newCodexGatewayBorrowMemoryTestRepo()
- started := make(chan struct{}, 4)
+ started := make(chan struct{}, PelicanExecutionConcurrency+1)
gen := codexGatewayBorrowFakeGenerator{generate: func(ctx context.Context, _ int64, _, _ string) (*CodexGatewayBorrowPelicanResult, error) {
started <- struct{}{}
<-ctx.Done()
@@ -255,7 +263,7 @@ func TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRaw
}}
runner := newCodexGatewayBorrowTestRunner(repo, gen, nil)
t.Cleanup(runner.Stop)
- targets := make([]CodexGatewayBorrowTestTarget, 4)
+ targets := make([]CodexGatewayBorrowTestTarget, PelicanExecutionConcurrency+1)
for i := range targets {
targets[i] = CodexGatewayBorrowTestTarget{AccountID: int64(i + 1), ModelID: "gpt-6-astra"}
}
@@ -264,7 +272,7 @@ func TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRaw
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- runner.Run(ctx, task, nil) }()
- for i := 0; i < 3; i++ {
+ for i := 0; i < PelicanExecutionConcurrency; i++ {
select {
case <-started:
case <-time.After(2 * time.Second):
@@ -290,7 +298,287 @@ func TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRaw
require.Contains(t, result.Error, "full upstream failure detail")
}
}
- require.Equal(t, 3, withPartial)
+ require.Equal(t, PelicanExecutionConcurrency, withPartial)
+}
+
+func TestPelicanRequestBudgetBoundsDefaultsAndServerMode(t *testing.T) {
+ base := CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Standalone: true,
+ Targets: []CodexGatewayBorrowTestTarget{{AccountID: 1, ModelID: "local-model"}}}
+ for _, seconds := range []int{0, 60, 600, 1800} {
+ request := base
+ request.GenerationTimeoutSeconds = seconds
+ normalized, err := normalizeCodexGatewayBorrowTestRequest(request, time.Now().UTC())
+ require.NoError(t, err)
+ if seconds == 0 {
+ require.Equal(t, PelicanDefaultGenerationTimeoutSeconds, normalized.GenerationTimeoutSeconds)
+ } else {
+ require.Equal(t, seconds, normalized.GenerationTimeoutSeconds)
+ }
+ require.Empty(t, normalized.Targets[0].Effort, "models without high must retain their normal default")
+ }
+ for _, seconds := range []int{-1, 1, 59, 1801} {
+ request := base
+ request.GenerationTimeoutSeconds = seconds
+ _, err := normalizeCodexGatewayBorrowTestRequest(request, time.Now().UTC())
+ require.ErrorIs(t, err, ErrCodexGatewayBorrowTestInvalidRequest)
+ }
+ var decoded CodexGatewayBorrowTestRequest
+ require.NoError(t, json.Unmarshal([]byte(`{"standalone":true,"client_task_id":"ignored","generation_timeout_seconds":600}`), &decoded))
+ require.False(t, decoded.Standalone, "the legacy endpoint cannot be promoted to account mode by request JSON")
+}
+
+func TestPelicanRunnerReplayIncludesBudgetAndExecutionMode(t *testing.T) {
+ repo := newCodexGatewayBorrowMemoryTestRepo()
+ runner := newCodexGatewayBorrowTestRunner(repo, nil, nil)
+ t.Cleanup(runner.Stop)
+ request := CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Standalone: true,
+ Targets: []CodexGatewayBorrowTestTarget{{AccountID: 1, ModelID: "gpt-6-astra", Effort: "high"}}}
+ task, replayed, err := runner.Create(context.Background(), 1, request)
+ require.NoError(t, err)
+ require.False(t, replayed)
+ require.Equal(t, PelicanExecutionModeAccount, task.ExecutionMode)
+ require.Equal(t, PelicanDefaultGenerationTimeoutSeconds, task.GenerationTimeoutSeconds)
+ request.GenerationTimeoutSeconds = 60
+ _, replayed, err = runner.Create(context.Background(), 1, request)
+ require.True(t, replayed)
+ require.ErrorIs(t, err, ErrCodexGatewayBorrowTestReplayConflict)
+ request.GenerationTimeoutSeconds, request.Standalone = 0, false
+ _, replayed, err = runner.Create(context.Background(), 1, request)
+ require.True(t, replayed)
+ require.ErrorIs(t, err, ErrCodexGatewayBorrowTestReplayConflict)
+}
+
+func TestPelicanRunnerGenerationClockStartsAfterPreparationAndRecordsActualInvocation(t *testing.T) {
+ repo := newCodexGatewayBorrowMemoryTestRepo()
+ runner := newCodexGatewayBorrowTestRunner(repo, nil, nil)
+ t.Cleanup(runner.Stop)
+ now := time.Now().UTC().Add(time.Second)
+ runner.now = func() time.Time { return now }
+ budgetStarts := 0
+ runner.generationContext = func(ctx context.Context, budget time.Duration) (context.Context, context.CancelFunc) {
+ budgetStarts++
+ require.Equal(t, 10*time.Minute, budget)
+ return context.WithCancel(ctx)
+ }
+ runner.SetStandaloneGenerator(codexGatewayBorrowFakeGenerator{generate: func(ctx context.Context, id int64, _, _ string) (*CodexGatewayBorrowPelicanResult, error) {
+ require.Zero(t, budgetStarts, "neither queue nor preparation may start generation's timer")
+ now = now.Add(2 * time.Minute)
+ requestCtx, release, err := AcquirePelicanExecution(ctx, id)
+ require.NoError(t, err)
+ defer release()
+ finishBusinessWait := BeginPelicanQueueWait(requestCtx)
+ now = now.Add(10 * time.Second)
+ finishBusinessWait()
+ finishBusinessWait()
+ requestCtx = BeginPelicanGeneration(requestCtx)
+ require.Equal(t, 1, budgetStarts)
+ RecordPelicanInvocation(requestCtx, PelicanInvocation{Platform: "openai", Model: "gpt-6.1-sol", Effort: "high",
+ Endpoint: "https://upstream.test/backend-api/codex/responses", Protocol: "responses", Transport: "http", BorrowApplied: true})
+ now = now.Add(30 * time.Second)
+ _ = BeginPelicanGeneration(requestCtx)
+ require.Equal(t, 1, budgetStarts, "same-account compatibility retries use the original generation clock")
+ now = now.Add(5 * time.Second)
+ return &CodexGatewayBorrowPelicanResult{Status: "complete", ModelID: "response-reported-model", RawAnswer: "", RawResponse: "original response\x00tail"}, nil
+ }})
+ task, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Standalone: true,
+ Targets: []CodexGatewayBorrowTestTarget{{AccountID: 1, ModelID: "mapped-alias"}}})
+ require.NoError(t, err)
+ var phases []string
+ require.NoError(t, runner.Run(context.Background(), task, func(event CodexGatewayBorrowTestEvent) {
+ if event.Type == "result_phase" {
+ phases = append(phases, event.Phase)
+ }
+ }))
+ result := task.Results[0]
+ require.Equal(t, []string{"queued", "preparing", "generating"}, phases)
+ require.Equal(t, "complete", result.Status)
+ require.Equal(t, "mapped-alias", result.ModelID)
+ require.Equal(t, "gpt-6.1-sol", result.UpstreamModel)
+ require.Equal(t, "openai", result.Platform)
+ require.Equal(t, "high", result.Effort)
+ require.Equal(t, "responses", result.ActualProtocol)
+ require.Equal(t, "http", result.ActualTransport)
+ require.Equal(t, "https://upstream.test/backend-api/codex/responses", result.ActualEndpoint)
+ require.True(t, result.BorrowApplied)
+ require.Equal(t, int64(10000), result.QueueDurationMS)
+ require.Equal(t, int64(120000), result.PreparationDurationMS)
+ require.Equal(t, int64(35000), result.GenerationDurationMS)
+ require.Equal(t, int64(165000), result.DurationMS)
+ require.NotNil(t, result.GenerationStartedAt)
+ require.Equal(t, "original response\x00tail", result.RawResponse)
+}
+
+func TestPelicanRunnerNestedPreparationDoesNotReserveAllExecutionSlots(t *testing.T) {
+ repo := newCodexGatewayBorrowMemoryTestRepo()
+ runner := newCodexGatewayBorrowTestRunner(repo, nil, nil)
+ t.Cleanup(runner.Stop)
+ preparing := make(chan struct{}, PelicanExecutionConcurrency+1)
+ allowPreparation := make(chan struct{})
+ var mu sync.Mutex
+ active, maxActive, probes, generations := 0, 0, 0, 0
+ accountActive := map[int64]int{}
+ maxAccountActive := map[int64]int{}
+ observe := func(ctx context.Context, id int64, generation bool) error {
+ requestCtx, release, err := AcquirePelicanExecution(ctx, id)
+ if err != nil {
+ return err
+ }
+ defer release()
+ // Sender wrappers for this exact account/request are reentrant and do
+ // not take another global permit or wait on their own account lease.
+ _, nestedRelease, err := AcquirePelicanExecution(requestCtx, id)
+ if err != nil {
+ return err
+ }
+ defer nestedRelease()
+ mu.Lock()
+ active++
+ accountActive[id]++
+ if active > maxActive {
+ maxActive = active
+ }
+ if accountActive[id] > maxAccountActive[id] {
+ maxAccountActive[id] = accountActive[id]
+ }
+ if generation {
+ generations++
+ } else {
+ probes++
+ }
+ accountActive[id]--
+ active--
+ mu.Unlock()
+ return nil
+ }
+ runner.SetStandaloneGenerator(codexGatewayBorrowFakeGenerator{generate: func(ctx context.Context, id int64, model, effort string) (*CodexGatewayBorrowPelicanResult, error) {
+ preparing <- struct{}{}
+ select {
+ case <-allowPreparation:
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+ // Every target needs the same source; that source is also selected as
+ // its own Pelican target. None of these outer preparations may retain
+ // the final target's account/global lease while requesting the source.
+ if err := observe(ctx, 1, false); err != nil {
+ return nil, err
+ }
+ if err := observe(ctx, id, true); err != nil {
+ return nil, err
+ }
+ return &CodexGatewayBorrowPelicanResult{Status: "complete", ModelID: model, Effort: effort, RawAnswer: ""}, nil
+ }})
+ targets := make([]CodexGatewayBorrowTestTarget, PelicanExecutionConcurrency+1)
+ for i := range targets {
+ targets[i] = CodexGatewayBorrowTestTarget{AccountID: int64(i + 1), ModelID: "local"}
+ }
+ task, _, err := runner.Create(context.Background(), 1, CodexGatewayBorrowTestRequest{ClientTaskID: codexGatewayBorrowV7(t), Standalone: true, Targets: targets})
+ require.NoError(t, err)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- runner.Run(ctx, task, nil) }()
+ for range targets {
+ select {
+ case <-preparing:
+ case <-time.After(2 * time.Second):
+ cancel()
+ t.Fatal("outer preparations reserved execution permits before their nested source requests")
+ }
+ }
+ close(allowPreparation)
+ select {
+ case err := <-done:
+ require.NoError(t, err)
+ case <-time.After(2 * time.Second):
+ t.Fatal("source/target preparation deadlocked with the selected source's own Pelican generation")
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ require.Equal(t, len(targets), probes)
+ require.Equal(t, len(targets), generations)
+ require.LessOrEqual(t, maxActive, PelicanExecutionConcurrency)
+ for _, maximum := range maxAccountActive {
+ require.Equal(t, 1, maximum)
+ }
+ require.Equal(t, "complete", task.Status)
+}
+
+func TestPelicanExecutionIsScopedAndRejectsIncorrectCrossAccountNesting(t *testing.T) {
+ ordinary := context.WithValue(context.Background(), struct{}{}, "ordinary context")
+ ctx, release, err := AcquirePelicanExecution(ordinary, 1)
+ require.NoError(t, err)
+ require.Equal(t, ordinary, ctx)
+ release()
+ require.Equal(t, ordinary, BeginPelicanGeneration(ordinary))
+ BeginPelicanPreparation(ordinary)
+ RecordPelicanInvocation(ordinary, PelicanInvocation{Model: "ignored"})
+ BeginPelicanQueueWait(ordinary)()
+ coordinator := newPelicanExecutionCoordinator(1)
+ state := newPelicanExecutionState(context.Background(), coordinator, 10*time.Minute, time.Now)
+ ctx = context.WithValue(state.parent, pelicanExecutionContextKey{}, state)
+ leasing, release, err := AcquirePelicanExecution(ctx, 1)
+ require.NoError(t, err)
+ _, nestedRelease, err := AcquirePelicanExecution(leasing, 1)
+ require.NoError(t, err)
+ nestedRelease()
+ coordinator.mu.Lock()
+ active := coordinator.active
+ coordinator.mu.Unlock()
+ require.Equal(t, 1, active)
+ _, _, err = AcquirePelicanExecution(leasing, 2)
+ require.ErrorContains(t, err, "preparation must finish")
+ release()
+ release()
+ coordinator.mu.Lock()
+ require.Zero(t, coordinator.active)
+ require.Empty(t, coordinator.accounts)
+ require.Empty(t, coordinator.waiting)
+ coordinator.mu.Unlock()
+ state.finish()
+}
+
+func TestPelicanRunnerGenerationTimeoutRetainsIncompleteOutput(t *testing.T) {
+ repo := newCodexGatewayBorrowMemoryTestRepo()
+ runner := newCodexGatewayBorrowTestRunner(repo, nil, nil)
+ t.Cleanup(runner.Stop)
+ now := time.Now().UTC().Add(time.Second)
+ runner.now = func() time.Time { return now }
+ var timeout context.CancelCauseFunc
+ runner.generationContext = func(ctx context.Context, budget time.Duration) (context.Context, context.CancelFunc) {
+ require.Equal(t, time.Minute, budget)
+ budgetCtx, cancel := context.WithCancelCause(ctx)
+ timeout = cancel
+ return budgetCtx, func() { cancel(context.Canceled) }
+ }
+ runner.SetStandaloneGenerator(codexGatewayBorrowFakeGenerator{generate: func(ctx context.Context, id int64, _, _ string) (*CodexGatewayBorrowPelicanResult, error) {
+ now = now.Add(90 * time.Second)
+ requestCtx, release, err := AcquirePelicanExecution(ctx, id)
+ require.NoError(t, err)
+ defer release()
+ requestCtx = BeginPelicanGeneration(requestCtx)
+ now = now.Add(time.Minute)
+ timeout(context.DeadlineExceeded)
+ <-requestCtx.Done()
+ <-ctx.Done()
+ require.ErrorIs(t, context.Cause(ctx), context.DeadlineExceeded, "the entire generator, including retries and later preparation, inherits the budget")
+ _, _, err = AcquirePelicanExecution(ctx, id+1)
+ require.ErrorIs(t, err, context.Canceled, "budget expiry cannot start another probe or compatibility retry")
+ return &CodexGatewayBorrowPelicanResult{Status: "incomplete", RawAnswer: "