diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index d1f9262bbc81..770168bf81c4 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -156,6 +156,10 @@ use the same local resolver, including the already integrated Cline/Command Code senders. No new provider protocol is introduced. `POST /admin/pelican-tests/tasks` returns 202 and starts a server-owned job. +Optional `repeat_count` defaults to one; each unique account/model target produces +that many independent results. A task permits at most 5000 expanded results. +The composer shows combinations x repeats and preserves one effort per model. +Each result carries `repeat_index` and `repeat_count` in summaries and details. The GET list, snapshot (`/:id`), paginated summaries (`/:id/results`), full single result (`/:id/results/:result_id`), and SSE (`/:id/events`) never start generation. Only `POST /:id/cancel` cancels a background job. Disconnecting observers or @@ -171,7 +175,10 @@ complete administrator evidence. The page supports 2–4 same/cross-task compari filters, enlargement, downloads and parameter copying. Whole-task, single-result and failed-item retests directly create a new task with original model/effort parameters and budget. Cross-task selected retries create one task per source, -preserving each source budget. New submissions retain UUID replay protection; +preserving each source budget. Retests copy each selected result once, preserving +its sample index and scene, with a fresh verification code; they never multiply +by the task repeat count again. Parameter copying merges repeated model rows +and restores the saved repeat count. New submissions retain UUID replay protection; history, browsing, reconnecting and comparison are read-only. New manual tasks select Pelican and/or candy (the UI defaults to both). Each result saves its chosen scene and a fresh visible verification code. The fixed @@ -188,9 +195,12 @@ image input is unreviewed. Only all selected checks passing gives a passed verdi execution status is independent. Stage evidence retains raw bytes as Base64 in detail JSON. A single on-demand Chromium process renders network-blocked SVGs. -All batches share ten model execution slots, with one request per account. -Ordinary selected-account sends participate in normal business -account capacity. Generation starts after preparation and admission; its default +All batches share ten model request execution slots. Different models on one +account may run concurrently, subject to the unchanged business account capacity. +A process-wide account/model turn serializes the entire selected test flow, +including repeats in sample order and matching models in other tasks. Waiting +for a model turn, account capacity, or rendering holds no outbound execution slot. +If account admission fails, its global slot is released before retrying admission. 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 @@ -200,7 +210,10 @@ usage and billing writes are suppressed, including background callbacks. Submission 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. +old records and raw BYTEA output. Migration 281 adds sample indexes, defaults +existing rows to one, and replaces the account/model unique key with a sample key. +Single-run hashes remain compatible with existing submissions; omitted and +explicit one-repeat requests are equivalent. 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. Administrators can explicitly delete a finished task or one of its results through `DELETE /tasks/:id` and diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index e497eec04839..a7ad7f0fff6a 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.github/workflows/downstream-verify.yml @@ -55,10 +55,10 @@ jobs: run: | python3 ../.github/scripts/prepare_test_images.py go test -p=1 -tags unit ./internal/service ./internal/repository ./internal/handler ./internal/handler/admin ./internal/server/routes -run '^Test(Pelican|AccountObservation|ObservedAccountLease)' -count=1 - go test -p=1 -tags integration ./internal/repository -run '^TestPelican(TestRepositoryIntegrationLifecycle|DeletionIntegration)$' -count=1 + go test -p=1 -tags integration ./internal/repository -run '^TestPelican(TestRepositoryIntegrationLifecycle|DeletionIntegration|RepeatSamplesIntegration)$' -count=1 - name: Pelican cancellation race test working-directory: backend - run: go test -p=1 -race -tags unit ./internal/service -run '^TestPelicanRunnerCancellationStopsQueuedAndInflightAndRetainsRawOutput$' -count=1 + run: go test -p=1 -race -tags unit ./internal/service -run '^TestPelican(RunnerCancellationStopsQueuedAndInflightAndRetainsRawOutput|RepeatModel.*|AccountWaitReleases.*)$' -count=1 - name: Account test keepalive lifecycle race tests working-directory: backend run: go test -p=1 -race -tags unit ./internal/service -run '^TestAccountTestKeepalive' -count=1 diff --git a/backend/internal/repository/pelican_background_repo.go b/backend/internal/repository/pelican_background_repo.go index a0b733d8ee1a..743e56601d81 100644 --- a/backend/internal/repository/pelican_background_repo.go +++ b/backend/internal/repository/pelican_background_repo.go @@ -57,7 +57,7 @@ func (r *pelicanTestRepository) GetTaskSnapshot(ctx context.Context, id string) const pelicanResultSummaryColumns = `r.id,r.task_id,r.ordinal,r.account_id,r.account_name,r.platform,r.model_id,r.effort,r.status, (octet_length(r.raw_svg)>0 OR r.status='complete' AND octet_length(r.html)>0), (r.status='incomplete' AND t.error=convert_to('process stopped before the test finished; this task will not be replayed','UTF8')), - r.started_at,r.finished_at,r.generation_started_at,r.duration_ms,r.queue_duration_ms,r.preparation_duration_ms,r.generation_duration_ms,t.expires_at,r.evaluation` + r.started_at,r.finished_at,r.generation_started_at,r.duration_ms,r.queue_duration_ms,r.preparation_duration_ms,r.generation_duration_ms,t.expires_at,r.evaluation,r.repeat_index,COALESCE((t.test_config->>'repeat_count')::integer,1)` func (r *pelicanTestRepository) ListTaskResults(ctx context.Context, id string, filter service.PelicanResultFilter) (*service.PelicanResultPage, error) { if filter.Page < 1 { @@ -122,7 +122,7 @@ func (r *pelicanTestRepository) ListTaskResults(ctx context.Context, id string, item := &service.PelicanResultSummary{} var evaluation []byte if err := rows.Scan(&item.ID, &item.TaskID, &item.Ordinal, &item.AccountID, &item.AccountName, &item.Platform, &item.ModelID, &item.Effort, &item.Status, - &item.HasPreview, &item.Interrupted, &item.StartedAt, &item.FinishedAt, &item.GenerationStartedAt, &item.DurationMS, &item.QueueDurationMS, &item.PreparationDurationMS, &item.GenerationDurationMS, &item.ExpiresAt, &evaluation); err != nil { + &item.HasPreview, &item.Interrupted, &item.StartedAt, &item.FinishedAt, &item.GenerationStartedAt, &item.DurationMS, &item.QueueDurationMS, &item.PreparationDurationMS, &item.GenerationDurationMS, &item.ExpiresAt, &evaluation, &item.RepeatIndex, &item.RepeatCount); err != nil { return nil, err } if len(evaluation) > 0 { diff --git a/backend/internal/repository/pelican_repeats_repo_integration_test.go b/backend/internal/repository/pelican_repeats_repo_integration_test.go new file mode 100644 index 000000000000..2d847ae80a01 --- /dev/null +++ b/backend/internal/repository/pelican_repeats_repo_integration_test.go @@ -0,0 +1,67 @@ +//go:build integration + +package repository + +import ( + "context" + "encoding/json" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "testing" + "time" +) + +func TestPelicanRepeatSamplesIntegration(t *testing.T) { + ctx := context.Background() + db := pelicanPrivateIntegrationDB(t, ctx) + repo := &pelicanTestRepository{db: db} + task := pelicanIntegrationTask(t, 42) + count := 3 + task.TestConfig.RepeatCount = &count + first := task.Results[0] + first.RepeatIndex = 1 + for i := 2; i <= 3; i++ { + row := *first + row.ID = uuid.NewString() + row.Ordinal = i + row.RepeatIndex = i + task.Results = append(task.Results, &row) + } + task.Total = 3 + _, _, err := repo.Create(ctx, task) + require.NoError(t, err) + for _, row := range task.Results { + row.Status = "complete" + row.RawResponse = "RAW-SAMPLE-CANARY" + require.NoError(t, repo.SaveResult(ctx, row)) + } + page, err := repo.ListTaskResults(ctx, task.ID, service.PelicanResultFilter{Page: 1, Size: 2}) + require.NoError(t, err) + require.EqualValues(t, 3, page.Total) + require.Len(t, page.Items, 2) + require.Equal(t, 1, page.Items[0].RepeatIndex) + require.Equal(t, 2, page.Items[1].RepeatIndex) + require.Equal(t, 3, page.Items[0].RepeatCount) + encoded, err := json.Marshal(page) + require.NoError(t, err) + require.NotContains(t, string(encoded), "CANARY") + require.NotContains(t, string(encoded), "raw_response") + next, err := repo.ListTaskResults(ctx, task.ID, service.PelicanResultFilter{Page: 2, Size: 2}) + require.NoError(t, err) + require.Equal(t, 3, next.Items[0].RepeatIndex) + full, err := repo.GetResult(ctx, task.Results[1].ID) + require.NoError(t, err) + require.Equal(t, 2, full.RepeatIndex) + require.Equal(t, 3, full.RepeatCount) + require.Equal(t, "RAW-SAMPLE-CANARY", full.RawResponse) + _, err = db.ExecContext(ctx, "INSERT INTO codex_gateway_borrow_test_results (id,task_id,ordinal,account_id,model_id,effort,status,repeat_index) VALUES ($1,$2,4,42,'gpt-6-astra','high','pending',2)", uuid.NewString(), task.ID) + require.Error(t, err, "one model cannot have the same sample number twice") + require.NoError(t, repo.FinishTask(ctx, task.ID, "complete", "", time.Now())) + require.NoError(t, repo.DeleteTaskResult(ctx, task.ID, full.ID)) + _, err = repo.GetResult(ctx, full.ID) + require.ErrorIs(t, err, service.ErrPelicanTestNotFound) + page, err = repo.ListTaskResults(ctx, task.ID, service.PelicanResultFilter{}) + require.NoError(t, err) + require.EqualValues(t, 2, page.Total) +} diff --git a/backend/internal/repository/pelican_test_repo.go b/backend/internal/repository/pelican_test_repo.go index 0cf77f1b52f4..52e7c3aaf8f9 100644 --- a/backend/internal/repository/pelican_test_repo.go +++ b/backend/internal/repository/pelican_test_repo.go @@ -23,7 +23,7 @@ const pelicanTestResultColumns = `r.id, r.task_id, r.ordinal, r.account_id, r.ac 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.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, - r.evaluation,r.evidence,r.raw_svg,r.review_image` + r.evaluation,r.evidence,r.raw_svg,r.review_image,r.repeat_index,COALESCE((t.test_config->>'repeat_count')::integer,1)` type pelicanTestScanner interface{ Scan(...any) error } @@ -61,7 +61,7 @@ func scanPelicanTestResult(row pelicanTestScanner) (*service.PelicanTestResult, &result.UpstreamModel, &result.Effort, &result.Status, &answer, &rawResponse, &rawHTML, &html, &rawError, &started, &finished, &result.DurationMS, &result.ExpiresAt, &result.Platform, &result.ActualEndpoint, &result.ActualProtocol, &result.ActualTransport, &result.BorrowApplied, &result.QueueDurationMS, &result.PreparationDurationMS, &result.GenerationDurationMS, &generationStarted, - &evaluation, &evidence, &svg, &result.ReviewImage); err != nil { + &evaluation, &evidence, &svg, &result.ReviewImage, &result.RepeatIndex, &result.RepeatCount); 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) @@ -135,13 +135,20 @@ func (r *pelicanTestRepository) Create(ctx context.Context, task *service.Pelica return nil, false, err } for _, result := range task.Results { + if result.RepeatIndex == 0 { + result.RepeatIndex = 1 + } + result.RepeatCount = 1 + if task.TestConfig.RepeatCount != nil { + result.RepeatCount = *task.TestConfig.RepeatCount + } evaluation, marshalErr := json.Marshal(result.Evaluation) if marshalErr != nil { return nil, false, marshalErr } if _, err = tx.ExecContext(ctx, `INSERT INTO codex_gateway_borrow_test_results - (id,task_id,ordinal,account_id,account_name,model_id,effort,status,platform,evaluation) - VALUES ($1,$2,$3,$4,$5,$6,$7,'pending',$8,$9)`, result.ID, stored.ID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.Effort, result.Platform, evaluation); err != nil { + (id,task_id,ordinal,account_id,account_name,model_id,effort,status,platform,evaluation,repeat_index) + VALUES ($1,$2,$3,$4,$5,$6,$7,'pending',$8,$9,$10)`, result.ID, stored.ID, result.Ordinal, result.AccountID, result.AccountName, result.ModelID, result.Effort, result.Platform, evaluation, result.RepeatIndex); err != nil { return nil, false, err } } diff --git a/backend/internal/repository/pelican_test_repo_integration_test.go b/backend/internal/repository/pelican_test_repo_integration_test.go index 047d40925a1e..caacb27b6024 100644 --- a/backend/internal/repository/pelican_test_repo_integration_test.go +++ b/backend/internal/repository/pelican_test_repo_integration_test.go @@ -113,6 +113,13 @@ func pelicanPrivateIntegrationDB(t *testing.T, ctx context.Context) *sql.DB { require.NoError(t, err) _, err = privateDB.ExecContext(ctx, string(evaluationDDL)) require.NoError(t, err) + repeatDDL, err := migrations.FS.ReadFile("281_pelican_repeat_samples.sql") + require.NoError(t, err) + _, err = privateDB.ExecContext(ctx, string(repeatDDL)) + require.NoError(t, err) + var legacyIndex int + require.NoError(t, privateDB.QueryRowContext(ctx, "SELECT repeat_index FROM codex_gateway_borrow_test_results WHERE id=$1", legacyResult).Scan(&legacyIndex)) + require.Equal(t, 1, legacyIndex) var executionMode string var timeout int var retention int64 diff --git a/backend/internal/repository/pelican_test_repo_test.go b/backend/internal/repository/pelican_test_repo_test.go index 43c7311de1bd..21699e9f252a 100644 --- a/backend/internal/repository/pelican_test_repo_test.go +++ b/backend/internal/repository/pelican_test_repo_test.go @@ -29,9 +29,9 @@ func pelicanTaskTestRows(task *service.PelicanTestTask) *sqlmock.Rows { } func pelicanResultTestRows(result *service.PelicanTestResult) *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", "platform", "actual_endpoint", "actual_protocol", "actual_transport", "borrow_applied", "queue_duration_ms", "preparation_duration_ms", "generation_duration_ms", "generation_started_at", "evaluation", "evidence", "raw_svg", "review_image"}).AddRow( + 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", "evaluation", "evidence", "raw_svg", "review_image", "repeat_index", "repeat_count"}).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, nil, []byte{}, []byte{}, []byte{}) + result.Platform, result.ActualEndpoint, result.ActualProtocol, result.ActualTransport, result.BorrowApplied, result.QueueDurationMS, result.PreparationDurationMS, result.GenerationDurationMS, result.GenerationStartedAt, nil, []byte{}, []byte{}, []byte{}, result.RepeatIndex, result.RepeatCount) } func pelicanRepoFixture(t *testing.T) *service.PelicanTestTask { @@ -41,7 +41,7 @@ func pelicanRepoFixture(t *testing.T) *service.PelicanTestTask { require.NoError(t, err) task := &service.PelicanTestTask{ID: uuid.NewString(), ClientTaskID: clientID.String(), CreatedBy: 7, RequestHash: "hash", Status: "pending", Prompt: service.PelicanPrompt, CreatedAt: now, ExpiresAt: now.Add(service.PelicanTestTTL), Total: 1, GenerationTimeoutSeconds: service.PelicanDefaultGenerationTimeoutSeconds, ExecutionMode: service.PelicanExecutionModeAccount} - task.Results = []*service.PelicanTestResult{{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, ReviewImage: []byte{}}} + task.Results = []*service.PelicanTestResult{{ID: uuid.NewString(), TaskID: task.ID, Ordinal: 1, RepeatIndex: 1, RepeatCount: 1, AccountID: 42, AccountName: "paused target", Platform: "openai", ModelID: "gpt-6-astra", Effort: "high", Status: "pending", ExpiresAt: task.ExpiresAt, ReviewImage: []byte{}}} return task } @@ -51,7 +51,7 @@ func TestPelicanTestRepositoryCreateAtomicAndReplayReadsWithoutWrites(t *testing 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, task.GenerationTimeoutSeconds, task.ExecutionMode, []byte("{}")).WillReturnRows(pelicanTaskTestRows(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, result.Platform, []byte("null")).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, []byte("null"), 1).WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() stored, replayed, err := repo.Create(context.Background(), task) require.NoError(t, err) diff --git a/backend/internal/service/pelican_account_generator.go b/backend/internal/service/pelican_account_generator.go index 1e0baf0f496d..c1edf410950f 100644 --- a/backend/internal/service/pelican_account_generator.go +++ b/backend/internal/service/pelican_account_generator.go @@ -128,17 +128,10 @@ func PreparePelicanHTTPRequest(req *http.Request, accountID int64, accountConcur if capture == nil || capture.account.ID != accountID { return req, finish, nil } - ctx, releaseExecution, acquireErr := AcquirePelicanExecution(req.Context(), accountID) + ctx, release, acquireErr := acquirePelicanRequestSlots(req.Context(), capture.concurrency, accountID, accountConcurrency) if acquireErr != nil { return nil, finish, acquireErr } - releaseAccount, acquireErr := acquirePelicanBusinessAccountSlot(ctx, capture.concurrency, accountID, accountConcurrency) - if acquireErr != nil { - releaseExecution() - return nil, finish, acquireErr - } - var releaseOnce sync.Once - release := func() { releaseOnce.Do(func() { releaseAccount(); releaseExecution() }) } req = req.WithContext(ctx) fallbackModel := capture.fallbackModel if capture.ginContext != nil { @@ -200,27 +193,39 @@ func PreparePelicanHTTPRequest(req *http.Request, accountID int64, accountConcur return req, finish, nil } -func acquirePelicanBusinessAccountSlot(ctx context.Context, concurrency *ConcurrencyService, accountID int64, limit int) (func(), error) { - if concurrency == nil || limit <= 0 { - return func() {}, nil - } - finishQueueWait := BeginPelicanQueueWait(ctx) - defer finishQueueWait() - ticker := time.NewTicker(100 * time.Millisecond) - defer ticker.Stop() +// Failed account admission releases global capacity before waiting. This keeps +// one saturated account from occupying all ten outbound slots. +func acquirePelicanRequestSlots(ctx context.Context, concurrency *ConcurrencyService, accountID int64, limit int) (context.Context, func(), error) { for { - lease, acquired, err := concurrency.AcquireObservedAccountLease(ctx, accountID, limit, func(cause error) { CancelPelicanExecution(ctx, cause) }) + leaseCtx, releaseExecution, err := AcquirePelicanExecution(ctx, accountID) + if err != nil { + return ctx, nil, err + } + if concurrency == nil || limit <= 0 { + return leaseCtx, releaseExecution, nil + } + finishWait := BeginPelicanQueueWait(ctx) + lease, acquired, err := concurrency.AcquireObservedAccountLease(leaseCtx, accountID, limit, func(cause error) { CancelPelicanExecution(leaseCtx, cause) }) + finishWait() if err != nil { - return nil, fmt.Errorf("acquire account concurrency capacity: %w", err) + releaseExecution() + return ctx, nil, fmt.Errorf("acquire account concurrency capacity: %w", err) } if acquired { - return lease.Release, nil + var once sync.Once + return leaseCtx, func() { once.Do(func() { lease.Release(); releaseExecution() }) }, nil } + releaseExecution() + finishWait = BeginPelicanQueueWait(ctx) + timer := time.NewTimer(100 * time.Millisecond) select { case <-ctx.Done(): - return nil, ctx.Err() - case <-ticker.C: + timer.Stop() + finishWait() + return ctx, nil, ctx.Err() + case <-timer.C: } + finishWait() } } @@ -233,17 +238,10 @@ func preparePelicanWSSend(ctx context.Context, account *Account, endpoint string if capture == nil || account == nil || capture.account.ID != account.ID { return ctx, observe, func() {}, nil } - ctx, releaseExecution, err := AcquirePelicanExecution(ctx, account.ID) - if err != nil { - return ctx, observe, func() {}, err - } - releaseAccount, err := acquirePelicanBusinessAccountSlot(ctx, capture.concurrency, account.ID, account.Concurrency) + ctx, release, err := acquirePelicanRequestSlots(ctx, capture.concurrency, account.ID, account.Concurrency) if err != nil { - releaseExecution() return ctx, observe, func() {}, err } - var once sync.Once - release := func() { once.Do(func() { releaseAccount(); releaseExecution() }) } model := openAIWSPayloadString(payload, "model") effort := gjson.GetBytes(payloadAsJSONBytes(payload), "reasoning.effort").String() RecordPelicanInvocation(ctx, PelicanInvocation{Platform: account.Platform, Model: model, Effort: effort, diff --git a/backend/internal/service/pelican_admission_test.go b/backend/internal/service/pelican_admission_test.go new file mode 100644 index 000000000000..35fe5c2983ae --- /dev/null +++ b/backend/internal/service/pelican_admission_test.go @@ -0,0 +1,112 @@ +//go:build unit + +package service + +import ( + "context" + "github.com/stretchr/testify/require" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" +) + +type pelicanAdmissionCache struct { + ConcurrencyCache + mu sync.Mutex + active map[int64]int + attempted chan int64 +} + +func (c *pelicanAdmissionCache) AcquireAccountSlot(_ context.Context, id int64, limit int, _ string) (bool, error) { + c.mu.Lock() + defer c.mu.Unlock() + select { + case c.attempted <- id: + default: + } + if c.active[id] >= limit { + return false, nil + } + c.active[id]++ + return true, nil +} +func (c *pelicanAdmissionCache) ReleaseAccountSlot(_ context.Context, id int64, _ string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.active[id]-- + return nil +} +func (c *pelicanAdmissionCache) RefreshAccountSlot(context.Context, int64, string) (bool, error) { + return true, nil +} + +func TestPelicanAccountWaitReleasesGlobalCapacityHTTPAndWS(t *testing.T) { + for _, transport := range []string{"http", "websocket"} { + t.Run(transport, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + coordinator := newPelicanExecutionCoordinator(1) + cache := &pelicanAdmissionCache{active: map[int64]int{1: 1}, attempted: make(chan int64, 100)} + concurrency := NewConcurrencyService(cache) + pending := make(chan error, 10) + for range 10 { + go func() { + state := newPelicanExecutionState(ctx, coordinator, time.Minute, time.Now) + defer state.finish() + requestCtx := context.WithValue(ctx, pelicanExecutionContextKey{}, state) + _, release, err := acquirePelicanRequestSlots(requestCtx, concurrency, 1, 1) + if release != nil { + release() + } + pending <- err + }() + } + for range 10 { + select { + case <-cache.attempted: + case <-ctx.Done(): + t.Fatal("blocked account attempts did not yield global capacity") + } + } + account := pelicanFixtureAccount(2) + account.Concurrency = 1 + state := newPelicanExecutionState(ctx, coordinator, time.Minute, time.Now) + defer state.finish() + requestCtx := context.WithValue(ctx, pelicanExecutionContextKey{}, state) + requestCtx = context.WithValue(requestCtx, pelicanGenerationContextKey{}, &pelicanGenerationCapture{account: account, concurrency: concurrency, fallbackModel: "model"}) + if transport == "http" { + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, "https://synthetic.invalid/v1/responses", strings.NewReader("{\"model\":\"model\",\"input\":\"fixture\"}")) + require.NoError(t, err) + _, finish, err := PreparePelicanHTTPRequest(req, 2, 1, "http") + require.NoError(t, err) + response, err := finish(&http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader("fixture"))}, nil) + require.NoError(t, err) + coordinator.mu.Lock() + active := coordinator.active + coordinator.mu.Unlock() + require.Equal(t, 1, active) + require.NoError(t, response.Body.Close()) + require.NoError(t, response.Body.Close()) + } else { + _, _, release, err := preparePelicanWSSend(requestCtx, account, "wss://synthetic.invalid/responses", map[string]any{"model": "model"}) + require.NoError(t, err) + release() + release() + } + cancel() + for range 10 { + require.ErrorIs(t, <-pending, context.Canceled) + } + coordinator.mu.Lock() + require.Zero(t, coordinator.active) + coordinator.mu.Unlock() + cache.mu.Lock() + require.Zero(t, cache.active[2]) + require.Equal(t, 1, cache.active[1]) + cache.mu.Unlock() + }) + } +} diff --git a/backend/internal/service/pelican_background.go b/backend/internal/service/pelican_background.go index b2cb91970988..5ca3fd8021b2 100644 --- a/backend/internal/service/pelican_background.go +++ b/backend/internal/service/pelican_background.go @@ -11,6 +11,8 @@ import ( ) type PelicanResultSummary struct { + RepeatIndex int `json:"repeat_index"` + RepeatCount int `json:"repeat_count"` Evaluation *PelicanEvaluation `json:"evaluation,omitempty"` PreviewURL string `json:"preview_url,omitempty"` ID string `json:"id"` @@ -97,7 +99,7 @@ type pelicanBackgroundTask struct { const pelicanInterruptedMessage = "process stopped before the test finished; this task will not be replayed" func pelicanResultSummary(result *PelicanTestResult, phase string) *PelicanResultSummary { - return &PelicanResultSummary{ID: result.ID, TaskID: result.TaskID, Ordinal: result.Ordinal, + return &PelicanResultSummary{ID: result.ID, TaskID: result.TaskID, Ordinal: result.Ordinal, RepeatIndex: result.RepeatIndex, RepeatCount: result.RepeatCount, Evaluation: clonePelicanEvaluation(result.Evaluation), AccountID: result.AccountID, AccountName: result.AccountName, Platform: result.Platform, ModelID: result.ModelID, Effort: result.Effort, Status: result.Status, Phase: phase, diff --git a/backend/internal/service/pelican_evaluation.go b/backend/internal/service/pelican_evaluation.go index 821ab6fc66a5..201c2ff5b008 100644 --- a/backend/internal/service/pelican_evaluation.go +++ b/backend/internal/service/pelican_evaluation.go @@ -42,9 +42,10 @@ motion：按时间顺序比较多帧，脚随踏板协调运动、车轮转动 {"checks":{"pelican":true,"bicycle":true,"riding":true,"motion":true}}` type PelicanTestConfig struct { - RetestKey string `json:"retest_key,omitempty"` - TestItems []string `json:"test_items,omitempty"` - Scene string `json:"scene,omitempty"` + RepeatCount *int `json:"repeat_count,omitempty"` + RetestKey string `json:"retest_key,omitempty"` + TestItems []string `json:"test_items,omitempty"` + Scene string `json:"scene,omitempty"` } type PelicanEvaluation struct { @@ -97,7 +98,19 @@ func clonePelicanEvaluation(value *PelicanEvaluation) *PelicanEvaluation { return © } +func (c PelicanTestConfig) repetitions() int { + if c.RepeatCount == nil { + return 1 + } + return *c.RepeatCount +} + func normalizePelicanConfig(c PelicanTestConfig) (PelicanTestConfig, error) { + count := c.repetitions() + if count < 1 || count > pelicanTestMaxTargets { + return c, fmt.Errorf("%w: repeat_count must be between 1 and %d", ErrPelicanTestInvalidRequest, pelicanTestMaxTargets) + } + c.RepeatCount = &count if len(c.TestItems) == 0 { return c, nil } // Original /tests contract. diff --git a/backend/internal/service/pelican_execution.go b/backend/internal/service/pelican_execution.go index a25267ebc47b..d97619b190c6 100644 --- a/backend/internal/service/pelican_execution.go +++ b/backend/internal/service/pelican_execution.go @@ -25,45 +25,68 @@ type pelicanExecutionContextKey struct{} type pelicanExecutionLeaseContextKey struct{} type pelicanExecutionWaiter struct { + ready chan struct{} + granted bool +} + +type pelicanModelKey struct { accountID int64 - ready chan struct{} - granted bool + modelID string } type pelicanExecutionCoordinator struct { - mu sync.Mutex - limit int - active int - accounts map[int64]bool - waiting []*pelicanExecutionWaiter + mu sync.Mutex + limit int + active int + waiting []*pelicanExecutionWaiter + models map[pelicanModelKey]chan struct{} } func newPelicanExecutionCoordinator(limit int) *pelicanExecutionCoordinator { - return &pelicanExecutionCoordinator{limit: limit, accounts: make(map[int64]bool)} + return &pelicanExecutionCoordinator{limit: limit, models: make(map[pelicanModelKey]chan struct{})} } -// scheduleLocked grants account and global capacity atomically. No waiter holds -// one resource while waiting for the other, and each account keeps FIFO order. -func (c *pelicanExecutionCoordinator) scheduleLocked() { - for i := 0; i < len(c.waiting) && c.active < c.limit; { - waiter := c.waiting[i] - if c.accounts[waiter.accountID] { - i++ - continue +// A model group owns its turn across the entire selected test flow. It never +// holds outbound capacity while waiting, rendering, or between requests. +func (c *pelicanExecutionCoordinator) acquireModel(ctx context.Context, accountID int64, modelID string) (func(), error) { + key := pelicanModelKey{accountID, modelID} + for { + if err := ctx.Err(); err != nil { + return nil, err + } + c.mu.Lock() + previous, busy := c.models[key] + if !busy { + done := make(chan struct{}) + c.models[key] = done + c.mu.Unlock() + var once sync.Once + return func() { once.Do(func() { c.mu.Lock(); delete(c.models, key); close(done); c.mu.Unlock() }) }, nil } - c.waiting = append(c.waiting[:i], c.waiting[i+1:]...) - c.accounts[waiter.accountID] = true + c.mu.Unlock() + select { + case <-previous: + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + +func (c *pelicanExecutionCoordinator) scheduleLocked() { + for len(c.waiting) > 0 && c.active < c.limit { + waiter := c.waiting[0] + c.waiting = c.waiting[1:] c.active++ waiter.granted = true close(waiter.ready) } } -func (c *pelicanExecutionCoordinator) acquire(ctx context.Context, accountID int64) (func(), error) { +func (c *pelicanExecutionCoordinator) acquire(ctx context.Context, _ int64) (func(), error) { if err := ctx.Err(); err != nil { return nil, err } - waiter := &pelicanExecutionWaiter{accountID: accountID, ready: make(chan struct{})} + waiter := &pelicanExecutionWaiter{ready: make(chan struct{})} c.mu.Lock() c.waiting = append(c.waiting, waiter) c.scheduleLocked() @@ -75,7 +98,6 @@ func (c *pelicanExecutionCoordinator) acquire(ctx context.Context, accountID int c.mu.Lock() if err := ctx.Err(); err != nil { if waiter.granted { - delete(c.accounts, accountID) c.active-- } else { for i, pending := range c.waiting { @@ -91,15 +113,7 @@ func (c *pelicanExecutionCoordinator) acquire(ctx context.Context, accountID int } c.mu.Unlock() var once sync.Once - return func() { - once.Do(func() { - c.mu.Lock() - delete(c.accounts, accountID) - c.active-- - c.scheduleLocked() - c.mu.Unlock() - }) - }, nil + return func() { once.Do(func() { c.mu.Lock(); c.active--; c.scheduleLocked(); c.mu.Unlock() }) }, nil } type pelicanExecutionLease struct { diff --git a/backend/internal/service/pelican_gallery.go b/backend/internal/service/pelican_gallery.go index 45b4e66f23e4..4a4180bf7b31 100644 --- a/backend/internal/service/pelican_gallery.go +++ b/backend/internal/service/pelican_gallery.go @@ -65,6 +65,7 @@ func (r *PelicanTestRunner) Retest(ctx context.Context, createdBy int64, id stri if len(parameters.TestItems) == 0 { parameters.TestItems = []string{"pelican"} } + seenModels := make(map[pelicanModelKey]bool) for page := 1; ; page++ { rows, readErr := r.TaskResults(ctx, id, PelicanResultFilter{Page: page, Size: 50}) if readErr != nil { @@ -85,13 +86,22 @@ func (r *PelicanTestRunner) Retest(ctx context.Context, createdBy int64, id stri if row.Evaluation != nil { target.Scene = row.Evaluation.Scene } - parameters.Targets = append(parameters.Targets, target) + key := pelicanModelKey{row.AccountID, row.ModelID} + if !seenModels[key] { + parameters.Targets = append(parameters.Targets, target) + seenModels[key] = true + } + index := row.RepeatIndex + if index == 0 { + index = 1 + } + parameters.samples = append(parameters.samples, pelicanTestSample{PelicanTestTarget: target, RepeatIndex: index}) } if page*50 >= int(rows.Total) { break } } - if len(request.ResultIDs) > 0 && len(parameters.Targets) != len(request.ResultIDs) { + if len(request.ResultIDs) > 0 && len(parameters.samples) != len(request.ResultIDs) { return nil, ErrPelicanTestInvalidRequest } return r.StartBackground(ctx, createdBy, parameters, catalog) diff --git a/backend/internal/service/pelican_gallery_test.go b/backend/internal/service/pelican_gallery_test.go index 9ee01790e81d..014b188be379 100644 --- a/backend/internal/service/pelican_gallery_test.go +++ b/backend/internal/service/pelican_gallery_test.go @@ -37,3 +37,44 @@ func TestPelicanRetestPreservesSceneBudgetAndReplayIdentity(t *testing.T) { _, err = runner.Retest(context.Background(), 7, source.ID, request, nil) require.ErrorIs(t, err, ErrPelicanTestReplayConflict) } + +func TestPelicanRepeatRetestsDoNotMultiplySamples(t *testing.T) { + for _, mode := range []string{"all", "failed", "selected"} { + t.Run(mode, func(t *testing.T) { + runner, repo, catalog := newPelicanBackgroundFixture(t, func(context.Context, int64, string, string) (*PelicanResult, error) { + return &PelicanResult{Status: "complete"}, nil + }) + count := 3 + source := &PelicanTestTask{ID: "source", Status: "complete", ExecutionMode: PelicanExecutionModeAccount, GenerationTimeoutSeconds: 600, ExpiresAt: time.Now().Add(time.Hour), TestConfig: PelicanTestConfig{RepeatCount: &count, TestItems: []string{"pelican"}}} + for i := 1; i <= 3; i++ { + verdict := "passed" + if i == 2 { + verdict = "invalid" + } + source.Results = append(source.Results, &PelicanTestResult{ID: string(rune('a' + i)), TaskID: source.ID, Ordinal: i, AccountID: 1, ModelID: "gpt-6.1-sol", Status: "complete", RepeatIndex: i, RepeatCount: 3, Evaluation: &PelicanEvaluation{TestItems: []string{"pelican"}, Scene: PelicanScenes[i], Nonce: "OLD", Verdict: verdict}}) + } + repo.tasks[source.ID] = source + request := PelicanRetestRequest{ClientTaskID: pelicanV7(t)} + expected := []int{1, 2, 3} + if mode == "failed" { + request.FailedOnly = true + expected = []int{2} + } + if mode == "selected" { + request.ResultIDs = []string{source.Results[0].ID, source.Results[2].ID} + expected = []int{1, 3} + } + created, err := runner.Retest(context.Background(), 7, source.ID, request, catalog) + require.NoError(t, err) + require.Equal(t, len(expected), created.Total) + repo.mu.Lock() + defer repo.mu.Unlock() + for i, row := range repo.tasks[created.ID].Results { + require.Equal(t, expected[i], row.RepeatIndex) + require.Equal(t, 3, row.RepeatCount) + require.Equal(t, PelicanScenes[expected[i]], row.Evaluation.Scene) + require.NotEqual(t, "OLD", row.Evaluation.Nonce) + } + }) + } +} diff --git a/backend/internal/service/pelican_repeats_test.go b/backend/internal/service/pelican_repeats_test.go new file mode 100644 index 000000000000..1f7a78fb6359 --- /dev/null +++ b/backend/internal/service/pelican_repeats_test.go @@ -0,0 +1,213 @@ +//go:build unit + +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestPelicanRepeatExpansionValidationAndReplay(t *testing.T) { + runner := newPelicanTestRunner(newPelicanMemoryTestRepo(), nil, nil) + t.Cleanup(runner.Stop) + count := 3 + request := PelicanTestRequest{ClientTaskID: pelicanV7(t), PelicanTestConfig: PelicanTestConfig{RepeatCount: &count, TestItems: []string{"pelican", "candy"}}, Targets: []PelicanTestTarget{{AccountID: 1, ModelID: "a"}, {AccountID: 1, ModelID: "b"}}} + task, replayed, err := runner.Create(context.Background(), 7, request) + require.NoError(t, err) + require.False(t, replayed) + require.Equal(t, 6, task.Total) + ids, nonces := map[string]bool{}, map[string]bool{} + for i, row := range task.Results { + require.Equal(t, i%3+1, row.RepeatIndex) + require.Equal(t, 3, row.RepeatCount) + require.False(t, ids[row.ID]) + ids[row.ID] = true + require.False(t, nonces[row.Evaluation.Nonce]) + nonces[row.Evaluation.Nonce] = true + require.Contains(t, PelicanScenes, row.Evaluation.Scene) + } + _, replayed, err = runner.Create(context.Background(), 7, request) + require.NoError(t, err) + require.True(t, replayed) + changed := 2 + request.RepeatCount = &changed + _, _, err = runner.Create(context.Background(), 7, request) + require.ErrorIs(t, err, ErrPelicanTestReplayConflict) + for _, bad := range []int{0, -1, 5001, 2501} { + request.RepeatCount = &bad + _, err = normalizePelicanTestRequest(request, time.Now()) + require.ErrorIs(t, err, ErrPelicanTestInvalidRequest) + } + boundary := 2500 + request.RepeatCount = &boundary + _, err = normalizePelicanTestRequest(request, time.Now()) + require.NoError(t, err) + legacy := PelicanTestRequest{ClientTaskID: pelicanV7(t), Targets: []PelicanTestTarget{{AccountID: 1, ModelID: "model"}}} + normalized, err := normalizePelicanTestRequest(legacy, time.Now()) + require.NoError(t, err) + oldPayload := "{\"targets\":[{\"account_id\":1,\"model_id\":\"model\",\"effort\":\"\"}],\"generation_timeout_seconds\":600,\"execution_mode\":\"account\"}" + digest := sha256.Sum256([]byte(oldPayload)) + require.Equal(t, hex.EncodeToString(digest[:]), pelicanTestRequestHash(normalized)) + one := 1 + legacy.RepeatCount = &one + normalizedOne, err := normalizePelicanTestRequest(legacy, time.Now()) + require.NoError(t, err) + require.Equal(t, pelicanTestRequestHash(normalized), pelicanTestRequestHash(normalizedOne)) +} + +type pelicanParallelEvaluator struct { + call func(context.Context, int64, string, PelicanModelInput) (*PelicanResult, error) +} + +func (g pelicanParallelEvaluator) GeneratePelican(context.Context, int64, string, string) (*PelicanResult, error) { + return nil, fmt.Errorf("legacy generator unused") +} +func (g pelicanParallelEvaluator) GeneratePelicanInput(ctx context.Context, id int64, model, _ string, input PelicanModelInput) (*PelicanResult, error) { + return g.call(ctx, id, model, input) +} +func (g pelicanParallelEvaluator) PelicanImageCapability(context.Context, int64, string) (bool, error) { + return true, nil +} + +func TestPelicanRepeatModelFlowsParallelAndOrdered(t *testing.T) { + var mu sync.Mutex + stages := map[string][]string{} + started := make(chan string, 2) + gate := make(chan struct{}) + evaluator := pelicanParallelEvaluator{call: func(ctx context.Context, id int64, model string, input PelicanModelInput) (*PelicanResult, error) { + ctx, release, err := AcquirePelicanExecution(ctx, id) + if err != nil { + return nil, err + } + defer release() + _ = BeginPelicanGeneration(ctx) + stage := "generating" + answer := "" + if input.Text == PelicanCandyPrompt { + stage = "candy" + answer = "21" + } else if len(input.Images) > 0 { + stage = "reviewing" + answer = "{\"checks\":{\"pelican\":true,\"bicycle\":true,\"riding\":true,\"motion\":true}}" + } else { + nonce := strings.Split(strings.Split(input.Text, "本次校验码：")[1], "。")[0] + answer = fmt.Sprintf("%s", nonce) + } + mu.Lock() + first := len(stages[model]) == 0 + stages[model] = append(stages[model], stage) + mu.Unlock() + if first { + started <- model + select { + case <-gate: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return &PelicanResult{Status: "complete", RawAnswer: answer}, nil + }} + runner := newPelicanTestRunner(newPelicanMemoryTestRepo(), evaluator, nil) + t.Cleanup(runner.Stop) + runner.renderer = pelicanFixedRenderer{} + count := 2 + task, _, err := runner.Create(context.Background(), 1, PelicanTestRequest{ClientTaskID: pelicanV7(t), PelicanTestConfig: PelicanTestConfig{RepeatCount: &count, TestItems: []string{"pelican", "candy"}}, Targets: []PelicanTestTarget{{AccountID: 1, ModelID: "a"}, {AccountID: 1, ModelID: "b"}}}) + require.NoError(t, err) + done := make(chan error, 1) + go func() { done <- runner.Run(context.Background(), task, nil) }() + t.Cleanup(func() { + select { + case <-gate: + default: + close(gate) + } + }) + for range 2 { + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("different models on one account did not overlap") + } + } + close(gate) + require.NoError(t, <-done) + for _, model := range []string{"a", "b"} { + require.Equal(t, []string{"candy", "generating", "reviewing", "candy", "generating", "reviewing"}, stages[model]) + } + for _, row := range task.Results { + require.Equal(t, "passed", row.Evaluation.Verdict) + require.NotEmpty(t, row.SVG) + } +} + +func TestPelicanRepeatFailureContinuesAndCancellationPersistsTail(t *testing.T) { + for _, stop := range []bool{false, true} { + t.Run(fmt.Sprint(stop), func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls := 0 + generator := pelicanFakeGenerator{generate: func(context.Context, int64, string, string) (*PelicanResult, error) { + calls++ + if stop { + cancel() + } + return &PelicanResult{Status: "failed", RawResponse: "retained"}, nil + }} + runner := newPelicanTestRunner(newPelicanMemoryTestRepo(), generator, nil) + t.Cleanup(runner.Stop) + count := 3 + task, _, err := runner.Create(context.Background(), 1, PelicanTestRequest{ClientTaskID: pelicanV7(t), PelicanTestConfig: PelicanTestConfig{RepeatCount: &count}, Targets: []PelicanTestTarget{{AccountID: 1, ModelID: "a"}}}) + require.NoError(t, err) + require.NoError(t, runner.Run(ctx, task, nil)) + if stop { + require.Equal(t, 1, calls) + for _, row := range task.Results[1:] { + require.Equal(t, "cancelled", row.Status) + } + } else { + require.Equal(t, 3, calls) + } + require.Equal(t, "retained", task.Results[0].RawResponse) + require.Equal(t, 3, task.Completed) + }) + } +} + +func TestPelicanRepeatModelTurnCancellationAcrossTasks(t *testing.T) { + coordinator := newPelicanExecutionCoordinator(10) + release, err := coordinator.acquireModel(context.Background(), 1, "a") + require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + blocked := make(chan error, 1) + go func() { + next, err := coordinator.acquireModel(ctx, 1, "a") + if next != nil { + next() + } + blocked <- err + }() + parallel, err := coordinator.acquireModel(context.Background(), 1, "b") + require.NoError(t, err) + parallel() + coordinator.mu.Lock() + require.Zero(t, coordinator.active, "model turns never hold outbound slots") + coordinator.mu.Unlock() + cancel() + require.ErrorIs(t, <-blocked, context.Canceled) + release() + release() + next, err := coordinator.acquireModel(context.Background(), 1, "a") + require.NoError(t, err) + next() + coordinator.mu.Lock() + require.Empty(t, coordinator.models) + coordinator.mu.Unlock() +} diff --git a/backend/internal/service/pelican_test_runner.go b/backend/internal/service/pelican_test_runner.go index ba7b9457bef5..473a6778efdc 100644 --- a/backend/internal/service/pelican_test_runner.go +++ b/backend/internal/service/pelican_test_runner.go @@ -37,7 +37,13 @@ type PelicanTestTarget struct { Effort string `json:"effort"` } +type pelicanTestSample struct { + PelicanTestTarget + RepeatIndex int +} + type PelicanTestRequest struct { + samples []pelicanTestSample PelicanTestConfig ClientTaskID string `json:"client_task_id"` GenerationTimeoutSeconds int `json:"generation_timeout_seconds"` @@ -45,6 +51,8 @@ type PelicanTestRequest struct { } type PelicanTestResult struct { + RepeatIndex int `json:"repeat_index"` + RepeatCount int `json:"repeat_count"` Evaluation *PelicanEvaluation `json:"evaluation,omitempty"` Evidence map[string]*PelicanStageEvidence `json:"evidence,omitempty"` SVG string `json:"raw_svg,omitempty"` @@ -134,7 +142,7 @@ type pelicanAccountReader interface { } // All account-test batches share ten outbound execution slots. -// Account leases surround actual requests rather than preparation. +// Model turns cover whole flows; outbound leases cover only actual requests. type PelicanTestRunner struct { renderer pelicanFrameRenderer submissionMu sync.Mutex @@ -270,19 +278,30 @@ func normalizePelicanTestRequest(request PelicanTestRequest, now time.Time) (Pel if len(targets) == 0 || len(targets) > pelicanTestMaxTargets { return request, fmt.Errorf("%w: targets must contain 1 to %d unique account/model pairs", ErrPelicanTestInvalidRequest, pelicanTestMaxTargets) } + if len(request.samples) == 0 && request.repetitions() > pelicanTestMaxTargets/len(targets) { + return request, fmt.Errorf("%w: repeat_count allows at most %d for these combinations (5000 total tests)", ErrPelicanTestInvalidRequest, pelicanTestMaxTargets/len(targets)) + } + if len(request.samples) > pelicanTestMaxTargets { + return request, ErrPelicanTestInvalidRequest + } request.Targets = targets return request, nil } func pelicanTestRequestHash(request PelicanTestRequest) string { + repeatCount := request.repetitions() + if repeatCount == 1 { + repeatCount = 0 + } // Keep existing single-run request hashes valid. raw, _ := json.Marshal(struct { + RepeatCount int `json:"repeat_count,omitempty"` Targets []PelicanTestTarget `json:"targets"` GenerationTimeoutSeconds int `json:"generation_timeout_seconds"` ExecutionMode string `json:"execution_mode"` TestItems []string `json:"test_items,omitempty"` Scene string `json:"scene,omitempty"` RetestKey string `json:"retest_key,omitempty"` - }{request.Targets, request.GenerationTimeoutSeconds, PelicanExecutionModeAccount, request.TestItems, request.Scene, request.RetestKey}) + }{repeatCount, request.Targets, request.GenerationTimeoutSeconds, PelicanExecutionModeAccount, request.TestItems, request.Scene, request.RetestKey}) hash := sha256.Sum256(raw) return hex.EncodeToString(hash[:]) } @@ -297,15 +316,23 @@ func (r *PelicanTestRunner) Create(ctx context.Context, createdBy int64, request if err = r.initialize(ctx); err != nil { return nil, false, err } + samples := request.samples + if len(samples) == 0 { + for _, target := range request.Targets { + for index := 1; index <= request.repetitions(); index++ { + samples = append(samples, pelicanTestSample{PelicanTestTarget: target, RepeatIndex: index}) + } + } + } now := r.now().UTC() task := &PelicanTestTask{ID: uuid.NewString(), ClientTaskID: request.ClientTaskID, TestConfig: request.PelicanTestConfig, CreatedBy: createdBy, RequestHash: pelicanTestRequestHash(request), Status: "pending", GenerationTimeoutSeconds: request.GenerationTimeoutSeconds, ExecutionMode: PelicanExecutionModeAccount, Prompt: PelicanPrompt, CreatedAt: now, ExpiresAt: now.Add(PelicanTestTTL), - Total: len(request.Targets), Results: make([]*PelicanTestResult, 0, len(request.Targets))} + Total: len(samples), Results: make([]*PelicanTestResult, 0, len(samples))} accountInfo := make(map[int64]*Account) - for i, target := range request.Targets { + for i, target := range samples { account, found := accountInfo[target.AccountID] if !found && r.accounts != nil { if read, readErr := r.accounts.GetByID(ctx, target.AccountID); readErr == nil { @@ -318,11 +345,11 @@ func (r *PelicanTestRunner) Create(ctx context.Context, createdBy int64, request name, platform = account.Name, account.Platform } task.Results = append(task.Results, &PelicanTestResult{ID: uuid.NewString(), TaskID: task.ID, - Ordinal: i + 1, AccountID: target.AccountID, AccountName: name, Platform: platform, ModelID: target.ModelID, + Ordinal: i + 1, RepeatIndex: target.RepeatIndex, RepeatCount: request.repetitions(), AccountID: target.AccountID, AccountName: name, Platform: platform, ModelID: target.ModelID, Effort: target.Effort, Status: "pending", ExpiresAt: task.ExpiresAt, PreviewUnavailable: "test has not completed"}) if len(request.TestItems) > 0 { - evaluation, evaluationErr := newPelicanEvaluation(request.PelicanTestConfig, target) + evaluation, evaluationErr := newPelicanEvaluation(request.PelicanTestConfig, target.PelicanTestTarget) if evaluationErr != nil { return nil, false, evaluationErr } @@ -414,72 +441,95 @@ func (r *PelicanTestRunner) Run(parent context.Context, task *PelicanTestTask, e var wg sync.WaitGroup var errorsMu sync.Mutex var persistenceErrors []string - for _, result := range task.Results { - result := result - wg.Add(1) - go func() { - defer wg.Done() - if result.Evaluation != nil { - r.runEvaluation(ctx, result, time.Duration(task.GenerationTimeoutSeconds)*time.Second, emit) - if r.stopCtx.Err() != nil && result.Status != "complete" { - result.Status = "incomplete" - result.Error = pelicanInterruptedMessage + "\n" + result.Error - } - writeCtx, cancelWrite := context.WithTimeout(context.WithoutCancel(parent), 10*time.Second) - defer cancelWrite() - if err := r.repo.SaveResult(writeCtx, result); err != nil { - errorsMu.Lock() - persistenceErrors = append(persistenceErrors, err.Error()) - errorsMu.Unlock() - } - if emit != nil { - emit(PelicanTestEvent{Type: "result_complete", TaskID: task.ID, Result: result}) - } - return + runResult := func(result *PelicanTestResult) { + modelQueueMS := nonnegativePelicanDuration(r.now().Sub(started)).Milliseconds() + result.QueueDurationMS = modelQueueMS + if result.Evaluation != nil { + r.runEvaluation(ctx, result, time.Duration(task.GenerationTimeoutSeconds)*time.Second, emit) + if r.stopCtx.Err() != nil && result.Status != "complete" { + result.Status = "incomplete" + result.Error = pelicanInterruptedMessage + "\n" + result.Error } - budget := task.GenerationTimeoutSeconds - if budget == 0 { - budget = PelicanDefaultGenerationTimeoutSeconds + writeCtx, cancelWrite := context.WithTimeout(context.WithoutCancel(parent), 10*time.Second) + defer cancelWrite() + if err := r.repo.SaveResult(writeCtx, result); err != nil { + errorsMu.Lock() + persistenceErrors = append(persistenceErrors, err.Error()) + errorsMu.Unlock() } - 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(PelicanTestEvent{Type: "result_phase", Phase: phase, TaskID: task.ID, Result: ©}) - } + if emit != nil { + emit(PelicanTestEvent{Type: "result_complete", TaskID: task.ID, Result: result}) } - testCtx := context.WithValue(state.parent, pelicanExecutionContextKey{}, state) + return + } + 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 - emit(PelicanTestEvent{Type: "result_phase", Phase: "queued", TaskID: task.ID, Result: ©}) - } - r.runOne(testCtx, result, r.generator, emit) - 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" + applyPelicanExecutionSnapshot(©, snapshot) + emit(PelicanTestEvent{Type: "result_phase", Phase: phase, TaskID: task.ID, Result: ©}) } - if r.stopCtx.Err() != nil && result.Status != "complete" { - result.Status, result.Error, result.PreviewUnavailable = "incomplete", pelicanInterruptedMessage+"\n"+result.Error, "test interrupted" + } + testCtx := context.WithValue(state.parent, pelicanExecutionContextKey{}, state) + if emit != nil { + copy := *result + emit(PelicanTestEvent{Type: "result_phase", Phase: "queued", TaskID: task.ID, Result: ©}) + } + r.runOne(testCtx, result, r.generator, emit) + snapshot := state.finish() + applyPelicanExecutionSnapshot(result, snapshot) + result.QueueDurationMS += modelQueueMS + if snapshot.generationTimedOut && ctx.Err() == nil && result.Status != "skipped" { + result.Status = "incomplete" + result.PreviewUnavailable = "test incomplete" + if result.Error != "" { + result.Error += "\n" } - writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(parent), 10*time.Second) - writeErr := r.repo.SaveResult(writeCtx, result) - writeCancel() - if writeErr != nil { - errorsMu.Lock() - persistenceErrors = append(persistenceErrors, writeErr.Error()) - errorsMu.Unlock() - cancel() + result.Error += "generation budget exhausted" + } + if r.stopCtx.Err() != nil && result.Status != "complete" { + result.Status, result.Error, result.PreviewUnavailable = "incomplete", pelicanInterruptedMessage+"\n"+result.Error, "test interrupted" + } + writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(parent), 10*time.Second) + writeErr := r.repo.SaveResult(writeCtx, result) + writeCancel() + if writeErr != nil { + errorsMu.Lock() + persistenceErrors = append(persistenceErrors, writeErr.Error()) + errorsMu.Unlock() + cancel() + } + if emit != nil { + emit(PelicanTestEvent{Type: "result_complete", TaskID: task.ID, Result: result}) + } + } + groups := make(map[pelicanModelKey][]*PelicanTestResult) + var order []pelicanModelKey + for _, result := range task.Results { + key := pelicanModelKey{result.AccountID, result.ModelID} + if _, exists := groups[key]; !exists { + order = append(order, key) + } + groups[key] = append(groups[key], result) + } + for _, key := range order { + results := groups[key] + slices.SortStableFunc(results, func(a, b *PelicanTestResult) int { return a.RepeatIndex - b.RepeatIndex }) + wg.Add(1) + go func() { + defer wg.Done() + release, err := r.coordinator.acquireModel(ctx, key.accountID, key.modelID) + if err == nil { + defer release() } - if emit != nil { - emit(PelicanTestEvent{Type: "result_complete", TaskID: task.ID, Result: result}) + // Cancelled groups still persist every pending sample without dispatching. + for _, result := range results { + runResult(result) } }() } diff --git a/backend/internal/service/pelican_test_runner_test.go b/backend/internal/service/pelican_test_runner_test.go index 8a3bad42993a..6109421cfe4b 100644 --- a/backend/internal/service/pelican_test_runner_test.go +++ b/backend/internal/service/pelican_test_runner_test.go @@ -176,15 +176,15 @@ func TestPelicanRunnerRejectsExpiredV7AfterRowsAreGone(t *testing.T) { require.ErrorIs(t, err, ErrPelicanTestInvalidRequest) } -func TestPelicanRunnerSharesTenSlotsAndSerializesAccountsAcrossBatches(t *testing.T) { +func TestPelicanRunnerSharesTenSlotsAndSerializesModelsAcrossBatches(t *testing.T) { repo := newPelicanMemoryTestRepo() var mu sync.Mutex active, maxActive, calls := 0, 0, 0 - accountActive := map[int64]int{} - maxAccountActive := map[int64]int{} + accountActive := map[pelicanModelKey]int{} + maxAccountActive := map[pelicanModelKey]int{} started := make(chan struct{}, 16) gate := make(chan struct{}) - gen := pelicanFakeGenerator{generate: func(ctx context.Context, id int64, _, effort string) (*PelicanResult, error) { + gen := pelicanFakeGenerator{generate: func(ctx context.Context, id int64, model, effort string) (*PelicanResult, error) { ctx, release, acquireErr := AcquirePelicanExecution(ctx, id) if acquireErr != nil { return nil, acquireErr @@ -194,12 +194,12 @@ func TestPelicanRunnerSharesTenSlotsAndSerializesAccountsAcrossBatches(t *testin mu.Lock() calls++ active++ - accountActive[id]++ + accountActive[pelicanModelKey{id, model}]++ if active > maxActive { maxActive = active } - if accountActive[id] > maxAccountActive[id] { - maxAccountActive[id] = accountActive[id] + if accountActive[pelicanModelKey{id, model}] > maxAccountActive[pelicanModelKey{id, model}] { + maxAccountActive[pelicanModelKey{id, model}] = accountActive[pelicanModelKey{id, model}] } mu.Unlock() started <- struct{}{} @@ -209,7 +209,7 @@ func TestPelicanRunnerSharesTenSlotsAndSerializesAccountsAcrossBatches(t *testin } mu.Lock() active-- - accountActive[id]-- + accountActive[pelicanModelKey{id, model}]-- mu.Unlock() return &PelicanResult{Status: "skipped", Effort: effort, Error: "selected account unavailable"}, nil }} @@ -221,6 +221,7 @@ func TestPelicanRunnerSharesTenSlotsAndSerializesAccountsAcrossBatches(t *testin secondTargets[i] = PelicanTestTarget{AccountID: int64(i + 7), ModelID: "gpt-6.1-sol"} } secondTargets[0].AccountID = 1 + secondTargets[1] = PelicanTestTarget{AccountID: 2, ModelID: "gpt-6-astra"} first, _, err := runner.Create(context.Background(), 1, PelicanTestRequest{ClientTaskID: pelicanV7(t), Targets: firstTargets}) require.NoError(t, err) second, _, err := runner.Create(context.Background(), 1, PelicanTestRequest{ClientTaskID: pelicanV7(t), Targets: secondTargets}) @@ -450,7 +451,7 @@ func TestPelicanExecutionIsScopedAndRejectsIncorrectCrossAccountNesting(t *testi release() coordinator.mu.Lock() require.Zero(t, coordinator.active) - require.Empty(t, coordinator.accounts) + require.Empty(t, coordinator.models) require.Empty(t, coordinator.waiting) coordinator.mu.Unlock() state.finish() diff --git a/backend/migrations/281_pelican_repeat_samples.sql b/backend/migrations/281_pelican_repeat_samples.sql new file mode 100644 index 000000000000..8c4a1d4016fa --- /dev/null +++ b/backend/migrations/281_pelican_repeat_samples.sql @@ -0,0 +1,18 @@ +-- Preserve existing observations as sample one and permit independent repeats. +ALTER TABLE codex_gateway_borrow_test_results + ADD COLUMN repeat_index INTEGER NOT NULL DEFAULT 1 CHECK (repeat_index > 0); + +-- PostgreSQL truncates generated identifiers; locate this exact legacy key by +-- its columns rather than guessing its truncated name. +DO $$ +DECLARE legacy_key TEXT; +BEGIN + SELECT conname INTO STRICT legacy_key FROM pg_constraint + WHERE conrelid = 'codex_gateway_borrow_test_results'::regclass + AND contype = 'u' + AND pg_get_constraintdef(oid) = 'UNIQUE (task_id, account_id, model_id)'; + EXECUTE format('ALTER TABLE codex_gateway_borrow_test_results DROP CONSTRAINT %I', legacy_key); +END $$; + +ALTER TABLE codex_gateway_borrow_test_results + ADD CONSTRAINT pelican_result_sample_unique UNIQUE (task_id, account_id, model_id, repeat_index); diff --git a/frontend/src/api/admin/pelicanTests.ts b/frontend/src/api/admin/pelicanTests.ts index f650901e5a3c..7461daab84e4 100644 --- a/frontend/src/api/admin/pelicanTests.ts +++ b/frontend/src/api/admin/pelicanTests.ts @@ -100,12 +100,18 @@ export interface PelicanOptions { export type PelicanPhase = 'queued' | 'preparing' | 'generating' | 'candy' | 'rendering' | 'reviewing' export const PELICAN_SCENES = ['海边木栈道', '秋日林道', '春日草坡', '雨后湿地', '黄昏公路', '湖畔风车', '热带海岛', '雪山谷地'] -export interface PelicanTestConfig { test_items?: string[]; scene?: string } +export interface PelicanTestConfig { + repeat_count?: number + test_items?: string[] + scene?: string +} export interface PelicanEvaluation { test_items: string[]; scene: string; nonce: string; generation: string; candy: string; review: string; verdict: string; checks?: Record; review_checks?: Record; frame_times?: number[] } export interface PelicanStageEvidence { prompt: string; raw_answer: string; raw_response: string; error: string; status: string; duration_ms: number; invocation: Record } export interface PelicanRetestRequest { client_task_id: string; result_ids?: string[]; failed_only?: boolean } export interface PelicanTestResult extends PelicanResultBase { + repeat_index?: number + repeat_count?: number task_id?: string platform?: string actual_endpoint?: string @@ -215,6 +221,7 @@ export async function streamPelicanTests( } export interface PelicanResultSummary { + repeat_index?: number; repeat_count?: number evaluation?: PelicanEvaluation; preview_url?: string id: string; task_id: string; ordinal: number; account_id: number; account_name: string platform: string; model_id: string; effort: string; status: string; phase?: PelicanPhase diff --git a/frontend/src/components/admin/codex/PelicanResultCard.vue b/frontend/src/components/admin/codex/PelicanResultCard.vue index 3834a3ea871f..35ac6c070569 100644 --- a/frontend/src/components/admin/codex/PelicanResultCard.vue +++ b/frontend/src/components/admin/codex/PelicanResultCard.vue @@ -12,6 +12,7 @@

{{ text('账号', 'Account') }}：{{ result.account_name }} #{{ result.account_id }}

{{ text('场景', 'Scene') }}：{{ result.evaluation.scene }}

{{ result.model_id }} · {{ result.effort || t('admin.pelicanTests.normalDefault') }} · {{ (result.duration_ms / 1000).toFixed(1) }}s

+

{{ repeatLabel }}

{{ text('手动检测', 'Manual test') }}

@@ -29,10 +30,11 @@ import { useI18n } from 'vue-i18n' import type { PelicanResultSummary } from '@/api/admin/pelicanTests' import PelicanPreview from './PelicanPreview.vue' import PelicanEvaluationBadges from './PelicanEvaluationBadges.vue' -import { pelicanLabel, pelicanTone } from '@/utils/pelicanDisplay' +import { pelicanLabel, pelicanTone, pelicanRepeatLabel } from '@/utils/pelicanDisplay' const props = defineProps<{ result: PelicanResultSummary; selected: boolean; selectionFull: boolean; now: number; canDelete?: boolean; busy?: boolean }>() defineEmits<{ select: [PelicanResultSummary]; open: [PelicanResultSummary]; enlarge: [PelicanResultSummary]; retest: [PelicanResultSummary]; delete: [PelicanResultSummary] }>() const { t, locale } = useI18n() +const repeatLabel = computed(() => pelicanRepeatLabel(props.result, locale.value)) const text = (zh: string, en: string) => locale.value.startsWith('zh') ? zh : en const expired = computed(() => Date.parse(props.result.expires_at) <= props.now) const unfinished = computed(() => ['pending', 'running'].includes(props.result.status)) diff --git a/frontend/src/components/admin/codex/PelicanTaskComposer.vue b/frontend/src/components/admin/codex/PelicanTaskComposer.vue index e886cae15212..7fbf22901bb0 100644 --- a/frontend/src/components/admin/codex/PelicanTaskComposer.vue +++ b/frontend/src/components/admin/codex/PelicanTaskComposer.vue @@ -31,6 +31,11 @@ @@ -100,7 +105,11 @@ const bulkModel = ref(''), bulkEffort = ref(''), effortLevels = ['none', 'minima let key = 0, pageRequest: AbortController | undefined, debounce: ReturnType | undefined const duplicates = computed(() => pelicanDuplicateKeys(rows.value)) const testItems = ref(props.initial?.test_items || ['pelican', 'candy']), scene = ref(props.initial?.scene || '') -const valid = computed(() => testItems.value.length > 0 && rows.value.length > 0 && rows.value.length <= 5000 && !optionsLoading.value && !rows.value.some(row => issue(row))) +const repeatCount = ref(props.initial?.repeat_count ?? 1) +const repeatMaximum = computed(() => Math.floor(5000 / Math.max(1, rows.value.length))) +const repeatValid = computed(() => Number.isInteger(repeatCount.value) && repeatCount.value >= 1 && repeatCount.value <= repeatMaximum.value) +const totalTests = computed(() => rows.value.length * repeatCount.value) +const valid = computed(() => repeatValid.value && testItems.value.length > 0 && rows.value.length > 0 && rows.value.length <= 5000 && !optionsLoading.value && !rows.value.some(row => issue(row))) const budgetValid = computed(() => Number.isInteger(budgetMinutes.value) && budgetMinutes.value >= 1 && budgetMinutes.value <= 30) function modelOption(row: PelicanCombination) { return options.value[row.account_id]?.models.find(model => model.id === row.model_id.trim()) } function modelLabel(model: PelicanModelOption) { return model.display_name && model.display_name !== model.id ? `${model.display_name} (${model.id})` : model.id } @@ -173,7 +182,7 @@ function addRow(id: number) { } function removeRow(rowKey: number) { const id = rows.value.find(row => row.key === rowKey)?.account_id; rows.value = rows.value.filter(row => row.key !== rowKey); if (id && !rows.value.some(row => row.account_id === id)) selected.value = selected.value.filter(value => value !== id) } function applyBulk() { const result = applyPelicanBulk(rows.value, options.value, bulkModel.value, bulkEffort.value); rows.value = result.rows; bulkMessage.value = result.duplicate ? text('会产生重复组合，未应用；请修改已有行。', 'Would create duplicates; edit the existing rows.') : `${text('已保留不兼容账号的原参数', 'Incompatible accounts kept their parameters')}: ${[...new Set(result.skipped)].join(', ') || '0'}` } -function submit() { if (valid.value && budgetValid.value) emit('submit', { generation_timeout_seconds: budgetMinutes.value * 60, test_items: testItems.value, scene: scene.value, targets: rows.value.map(row => ({ account_id: row.account_id, model_id: row.model_id.trim(), effort: row.effort })) }) } +function submit() { if (valid.value && budgetValid.value) emit('submit', { repeat_count: repeatCount.value, generation_timeout_seconds: budgetMinutes.value * 60, test_items: testItems.value, scene: scene.value, targets: rows.value.map(row => ({ account_id: row.account_id, model_id: row.model_id.trim(), effort: row.effort })) }) } watch(selected, ids => { rows.value = rows.value.filter(row => ids.includes(row.account_id)) for (const id of ids) if (options.value[id] && !rows.value.some(row => row.account_id === id)) rows.value.push({ key: ++key, account_id: id, model_id: pelicanDefaultModel(options.value[id]), effort: '' }) diff --git a/frontend/src/components/admin/codex/PelicanWorkDetail.vue b/frontend/src/components/admin/codex/PelicanWorkDetail.vue index ba7772adac36..f80d0f06b559 100644 --- a/frontend/src/components/admin/codex/PelicanWorkDetail.vue +++ b/frontend/src/components/admin/codex/PelicanWorkDetail.vue @@ -3,6 +3,7 @@
+
{{ text('测试次数', 'Test run') }}
{{ repeatLabel }}
{{ text('账号与模型', 'Account and model') }}
{{ title }} · {{ result.effort || text('模型默认', 'Model default') }}
{{ text('耗时', 'Duration') }}
{{ ((result.duration_ms || 0) / 1000).toFixed(1) }}s · {{ result.started_at ? new Date(result.started_at).toLocaleString() : '—' }}
{{ text('排队', 'Queue') }} {{ ((result.queue_duration_ms || 0) / 1000).toFixed(1) }}s · {{ text('准备', 'Preparation') }} {{ ((result.preparation_duration_ms || 0) / 1000).toFixed(1) }}s · {{ text('模型调用', 'Model calls') }} {{ ((result.generation_duration_ms || 0) / 1000).toFixed(1) }}s
@@ -36,11 +37,12 @@ import { computed, ref } from 'vue' import { useI18n } from 'vue-i18n' import type { PelicanTestResult } from '@/api/admin/pelicanTests' -import { decodePelicanEvidence, pelicanLabel } from '@/utils/pelicanDisplay' +import { decodePelicanEvidence, pelicanLabel, pelicanRepeatLabel } from '@/utils/pelicanDisplay' import PelicanPreview from './PelicanPreview.vue' import PelicanEvaluationBadges from './PelicanEvaluationBadges.vue' const props = defineProps<{ result: PelicanTestResult }>() const { locale } = useI18n() +const repeatLabel = computed(() => pelicanRepeatLabel(props.result, locale.value)) const text = (zh: string, en: string) => locale.value.startsWith('zh') ? zh : en const label = (value: string) => pelicanLabel(value, locale.value) const source = computed(() => props.result.raw_svg || props.result.raw_html || props.result.html) diff --git a/frontend/src/components/admin/codex/__tests__/PelicanTaskComposer.spec.ts b/frontend/src/components/admin/codex/__tests__/PelicanTaskComposer.spec.ts index 1f2a184420ad..6eae2d556704 100644 --- a/frontend/src/components/admin/codex/__tests__/PelicanTaskComposer.spec.ts +++ b/frontend/src/components/admin/codex/__tests__/PelicanTaskComposer.spec.ts @@ -40,7 +40,7 @@ describe('Pelican three-step composer', () => { expect(wrapper.findAll('[data-test^="pelican-effort-"]').every(effort => effort.element.value === '')).toBe(true) await wrapper.get('[data-test="pelican-next"]').trigger('click') await wrapper.get('[data-test="pelican-test-start"]').trigger('click') - expect(wrapper.emitted('submit')?.[0]?.[0]).toEqual({ generation_timeout_seconds: 600, test_items: ['pelican', 'candy'], scene: '', targets: [{ account_id: 1, model_id: 'gpt-6.1-sol', effort: '' }, { account_id: 51, model_id: 'gpt-6.1-sol', effort: '' }] }) + expect(wrapper.emitted('submit')?.[0]?.[0]).toEqual({ repeat_count: 1, generation_timeout_seconds: 600, test_items: ['pelican', 'candy'], scene: '', targets: [{ account_id: 1, model_id: 'gpt-6.1-sol', effort: '' }, { account_id: 51, model_id: 'gpt-6.1-sol', effort: '' }] }) }) it('requires correction of unsupported and duplicate combinations instead of dropping them', async () => { const wrapper = await make({ initial: { generation_timeout_seconds: 600, targets: [{ account_id: 1, model_id: 'codex-auto-review', effort: '' }] } }) @@ -78,3 +78,23 @@ describe('Pelican three-step composer', () => { expect(wrapper.emitted('submit')).toBeUndefined() }) }) + +it('validates uniform repeats and previews the expanded total for multiple models', async () => { + const wrapper = await make({ initial: { repeat_count: 3, generation_timeout_seconds: 600, targets: [{ account_id: 1, model_id: 'gpt-6.1-sol', effort: 'high' }, { account_id: 1, model_id: 'gpt-6-astra', effort: '' }] } }) + const count = wrapper.get('[data-test="pelican-repeat-count"]') + expect(count.element.value).toBe('3') + expect(count.attributes('max')).toBe('2500') + for (const invalid of [0, -1, 1.5, 2501]) { + await count.setValue(invalid) + expect(count.attributes('aria-invalid')).toBe('true') + expect(wrapper.get('[data-test="pelican-next"]').element.disabled).toBe(true) + } + await count.setValue(2500) + expect(wrapper.get('[data-test="pelican-next"]').element.disabled).toBe(false) + await count.setValue(3) + await wrapper.get('[data-test="pelican-next"]').trigger('click') + expect(wrapper.get('[data-test="pelican-total-tests"]').text()).toContain('2 个账号／模型组合 × 3 次 = 6 条测试') + expect(wrapper.text()).toContain('同账号不同模型可并行') + await wrapper.get('[data-test="pelican-test-start"]').trigger('click') + expect(wrapper.emitted('submit')?.[0]?.[0]).toEqual(expect.objectContaining({ repeat_count: 3, targets: [{ account_id: 1, model_id: 'gpt-6.1-sol', effort: 'high' }, { account_id: 1, model_id: 'gpt-6-astra', effort: '' }] })) +}) diff --git a/frontend/src/utils/pelicanDisplay.ts b/frontend/src/utils/pelicanDisplay.ts index c0f1006a8184..97a2e63ef29d 100644 --- a/frontend/src/utils/pelicanDisplay.ts +++ b/frontend/src/utils/pelicanDisplay.ts @@ -18,3 +18,8 @@ export function pelicanTone(value: string) { } export function pelicanFailed(item: PelicanResultSummary) { return ['failed', 'incomplete', 'skipped'].includes(item.status) || !!item.evaluation && ['invalid', 'error'].includes(item.evaluation.verdict) } export function decodePelicanEvidence(value?: string) { if (!value) return ''; return new TextDecoder().decode(Uint8Array.from(atob(value), c => c.charCodeAt(0))) } + +export function pelicanRepeatLabel(result: { repeat_index?: number; repeat_count?: number }, locale: string): string { + const index = result.repeat_index || 1, count = result.repeat_count || 1 + return locale.startsWith('zh') ? `第 ${index} / ${count} 次` : `Run ${index} / ${count}` +} diff --git a/frontend/src/views/admin/CodexPelicanComparisonView.vue b/frontend/src/views/admin/CodexPelicanComparisonView.vue index 84be3629752d..249a57138481 100644 --- a/frontend/src/views/admin/CodexPelicanComparisonView.vue +++ b/frontend/src/views/admin/CodexPelicanComparisonView.vue @@ -26,7 +26,7 @@

{{ connection === 'retrying' ? text('观察连接中断，正在重连；后台任务继续。', 'Reconnecting; the background task continues.') : text('后台任务运行中，离开页面不会取消。', 'Running in the background; leaving this page does not cancel it.') }}

{{ activeTask.error }}
-
{{ text('跨任务比较', 'Compare across tasks') }} {{ comparison.length }} / 4
+
{{ text('跨任务比较', 'Compare across tasks') }} {{ comparison.length }} / 4

{{ text('鹈鹕骑行画廊', 'Pelican riding gallery') }}

{{ results.total }} {{ text('条记录', 'observations') }} · {{ loading ? text('读取中…', 'Loading…') : text('每页 24 条', '24 per page') }}
@@ -51,10 +51,11 @@ import PelicanTaskComposer from '@/components/admin/codex/PelicanTaskComposer.vu import PelicanResultCard from '@/components/admin/codex/PelicanResultCard.vue' import PelicanWorkDetail from '@/components/admin/codex/PelicanWorkDetail.vue' import { pelicanTestsAPI, PELICAN_SCENES, createPelicanClientTaskId, type PelicanTaskSnapshot, type PelicanTestRequest, type PelicanRetestRequest, type PelicanResultSummary, type PelicanResultPage, type PelicanTestResult, type PelicanTestHistory } from '@/api/admin/pelicanTests' -import { pelicanFailed, pelicanLabel } from '@/utils/pelicanDisplay' +import { pelicanFailed, pelicanLabel, pelicanRepeatLabel } from '@/utils/pelicanDisplay' import { extractApiErrorMessage } from '@/utils/apiError' const { locale } = useI18n() const text = (zh: string, en: string) => locale.value.startsWith('zh') ? zh : en +const repeatLabel = (item: PelicanResultSummary) => pelicanRepeatLabel(item, locale.value) const label = (value: string) => pelicanLabel(value, locale.value) const formatTime = (value: string) => new Date(value).toLocaleString() const lifecycle = new AbortController() @@ -135,7 +136,7 @@ async function retestItems(items: PelicanResultSummary[]) { } async function copyTask() { const task = activeTask.value; if (!task || busy.value) return; copying.value = true - try { const items: PelicanResultSummary[] = []; for (let next = 1; ; next++) { const rows = await pelicanTestsAPI.getResults(task.id, { page: next, page_size: 50 }, lifecycle.signal); items.push(...rows.items); if (next * 50 >= rows.total) break } draft.value = { generation_timeout_seconds: task.generation_timeout_seconds || 600, test_items: task.test_config?.test_items?.length ? task.test_config.test_items : ['pelican'], scene: task.test_config?.scene || '', targets: items.map(item => ({ account_id: item.account_id, model_id: item.model_id, effort: item.effort, scene: item.evaluation?.scene })) }; composerOpen.value = true } + try { const items: PelicanResultSummary[] = []; for (let next = 1; ; next++) { const rows = await pelicanTestsAPI.getResults(task.id, { page: next, page_size: 50 }, lifecycle.signal); items.push(...rows.items); if (next * 50 >= rows.total) break } draft.value = { repeat_count: task.test_config?.repeat_count ?? 1, generation_timeout_seconds: task.generation_timeout_seconds || 600, test_items: task.test_config?.test_items?.length ? task.test_config.test_items : ['pelican'], scene: task.test_config?.scene || '', targets: [...new Map(items.map(item => [JSON.stringify([item.account_id, item.model_id]), { account_id: item.account_id, model_id: item.model_id, effort: item.effort }])).values()] }; composerOpen.value = true } catch (value) { report(value) } finally { copying.value = false } } async function stopTask() { if (!activeTask.value || stopping.value) return; stopping.value = true; try { activeTask.value = await pelicanTestsAPI.cancelTask(activeTask.value.id) } catch (value) { stopping.value = false; report(value) } } diff --git a/frontend/src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts b/frontend/src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts index 0b26a3f56d42..2c4b7b035f3b 100644 --- a/frontend/src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts +++ b/frontend/src/views/admin/__tests__/CodexPelicanComparisonView.spec.ts @@ -92,3 +92,20 @@ describe('Pelican gallery workbench', () => { expect(mocks.retestTask).not.toHaveBeenCalled() }) }) + +it('copies each model once with its configured repeat count and displays sample numbers', async () => { + const source = task(); source.test_config = { test_items: ['pelican', 'candy'], repeat_count: 3 } + mocks.getTask.mockResolvedValue(source) + const second = { ...result('r2'), repeat_index: 2, repeat_count: 3 } + mocks.gallery.mockResolvedValue({ items: [second], total: 1, page: 1, page_size: 24 }) + mocks.getResults.mockResolvedValue({ items: [result('r1'), second, { ...result('r3'), model_id: 'gpt-6-astra' }], total: 3, page: 1, page_size: 50 }) + mocks.getResult.mockResolvedValue({ ...second, raw_answer: '', raw_response: '', raw_html: '', html: '', error: '', preview_unavailable: '' }) + const wrapper = await make(); await selectTask(wrapper) + expect(wrapper.get('[data-test="pelican-repeat-label"]').text()).toBe('第 2 / 3 次') + await wrapper.get('[data-test="pelican-result-r2"] button').trigger('click'); await flushPromises() + expect(wrapper.get('[data-test="pelican-repeat-detail"]').text()).toContain('第 2 / 3 次') + const copy = wrapper.findAll('button').find(button => button.text() === '复制参数')! + await copy.trigger('click'); await flushPromises() + expect(wrapper.getComponent('[data-test="composer-stub"]').props('initial')).toEqual(expect.objectContaining({ repeat_count: 3, targets: [{ account_id: 1, model_id: 'gpt-6.1-sol', effort: 'medium' }, { account_id: 1, model_id: 'gpt-6-astra', effort: 'medium' }] })) + expect(mocks.startTask).not.toHaveBeenCalled() +})