重构鹈鹕后台任务与分页比较工作台

.downstream/native-domains.md

@@ -213,9 +213,10 @@ candidate owner. `probe_context.subject_role` distinguishes source self-checks
 from target checks. Self-check results do not qualify business traffic or mutate
 the source's health/status.
 
-Ranxi reference: v2.10.2 (`d3e43f2de33af9e987cffa511d76dfabbcd749da`). Its five
+Ranxi reference: v2.10.3 (`fd1b5ee4eeb20961fbb783fa6f136a1704271e90`). Its
 fingerprint, gateway-cookie, target-probe and automatic-setup core files are
-unchanged from the retained v2.10.0 reference. OAuth search-history compatibility
+unchanged from v2.10.2. The scoped update and retained host adaptations are
+listed in [the review](upstream-review-ranxi-v2.10.3.md). OAuth search-history compatibility
 is supplied by the official v0.2.15 integration. Automatic group membership,
 Mihomo exit rotation and removal of harvesting attempt/concurrency bounds are
 not adopted; existing account mappings, network routes and protocol switches stay
@@ -227,15 +228,35 @@ 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.
+`/admin/pelican-tests` is a task/result workbench. The new-task composer loads
+accounts in pages of 50, retains selection across pages, and offers explicit
+current-page/all-matching scopes. Models use locally saved mappings and catalogs;
+valid backend defaults are respected and effort defaults to the model default.
+Known unsupported choices and duplicate account/model rows must be corrected or
+removed before submission; manual unverified models remain explicitly marked.
+Bulk configuration preserves incompatible rows. Text capability and generation
+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.
+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
+closing the page leaves it running. Repeated client UUIDs with identical normalized
+parameters return the existing snapshot; conflicting parameters fail. Restart and
+shutdown preserve completed works, mark unfinished work interrupted, and never
+replay uncertain model calls. No schema migration is added.
+
+The original `/tests` SSE endpoint retains request-bound cancellation and its
+request/response contract. Both APIs use the same runner, history and execution
+capacity. Summary SQL does not load raw responses, answers or HTML; details retain
+complete administrator evidence. The page supports 2–4 same/cross-task comparisons,
+filters, enlargement, downloads, parameter copying and failed-item retries into a
+new reviewed task. History, browsing, reconnecting and comparison are read-only.
+The fixed Pelican prompt is preserved. The actual sending path, credentials,
+identity, model mapping, proxy, TLS, protocol and configured borrowing remain
+owned by each account. No API Key, group policy, model score or fallback account
+is manufactured for a test.
 
 All batches share ten model execution slots, with one request per account.
 Nested borrowing probes participate in the same coordination and normal business

backend/internal/handler/admin/pelican_background_handler.go

@@ -0,0 +1,155 @@
+package admin
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"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"
+)
+
+func (h *PelicanTestHandler) StartTask(c *gin.Context) {
+	codexGatewayBorrowPrivateHeaders(c)
+	var request service.CodexGatewayBorrowTestRequest
+	if err := c.ShouldBindJSON(&request); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	var createdBy int64
+	if subject, ok := middleware.GetAuthSubjectFromContext(c); ok {
+		createdBy = subject.UserID
+	}
+	task, err := h.runner.StartBackground(c.Request.Context(), createdBy, request, h.catalog)
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	middleware.SetAuditExtra(c, map[string]any{"result": "pelican_background_test", "task_id": task.ID, "replayed": task.Replayed, "targets": task.Total})
+	response.Accepted(c, task)
+}
+
+func pelicanTaskID(c *gin.Context) (string, bool) {
+	codexGatewayBorrowPrivateHeaders(c)
+	id, err := uuid.Parse(strings.TrimSpace(c.Param("id")))
+	if err != nil {
+		response.BadRequest(c, "task ID must be a UUID")
+		return "", false
+	}
+	return id.String(), true
+}
+
+func (h *PelicanTestHandler) GetTask(c *gin.Context) {
+	id, ok := pelicanTaskID(c)
+	if !ok {
+		return
+	}
+	task, err := h.runner.TaskSnapshot(c.Request.Context(), id)
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	response.Success(c, task)
+}
+
+func (h *PelicanTestHandler) GetTaskResults(c *gin.Context) {
+	id, ok := pelicanTaskID(c)
+	if !ok {
+		return
+	}
+	page, size := response.ParsePagination(c)
+	accountID, _ := strconv.ParseInt(c.Query("account_id"), 10, 64)
+	result, err := h.runner.TaskResults(c.Request.Context(), id, service.PelicanResultFilter{Page: page, Size: size, AccountID: accountID, Model: c.Query("model"), Status: c.Query("status")})
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	response.Success(c, result)
+}
+
+func (h *PelicanTestHandler) GetTaskResult(c *gin.Context) {
+	id, ok := pelicanTaskID(c)
+	if !ok {
+		return
+	}
+	resultID, err := uuid.Parse(c.Param("result_id"))
+	if err != nil {
+		response.BadRequest(c, "result ID must be a UUID")
+		return
+	}
+	result, err := h.runner.TaskResult(c.Request.Context(), id, resultID.String())
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	response.Success(c, h.preview.DecorateResult(result))
+}
+
+func (h *PelicanTestHandler) CancelTask(c *gin.Context) {
+	id, ok := pelicanTaskID(c)
+	if !ok {
+		return
+	}
+	task, err := h.runner.CancelBackground(c.Request.Context(), id)
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	middleware.SetAuditExtra(c, map[string]any{"result": "pelican_cancel_requested", "task_id": task.ID})
+	response.Success(c, task)
+}
+
+// Read-only progress stream. Disconnecting never owns or cancels execution;
+// reconnect starts with a durable snapshot, not a POST or model replay.
+func (h *PelicanTestHandler) ObserveTask(c *gin.Context) {
+	id, ok := pelicanTaskID(c)
+	if !ok {
+		return
+	}
+	task, err := h.runner.TaskSnapshot(c.Request.Context(), id)
+	if err != nil {
+		codexGatewayBorrowTestError(c, err)
+		return
+	}
+	c.Header("Content-Type", "text/event-stream")
+	c.Header("Connection", "keep-alive")
+	c.Header("X-Accel-Buffering", "no")
+	c.Status(http.StatusOK)
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+	for {
+		kind := "snapshot"
+		if task.Status != "pending" && task.Status != "running" {
+			kind = "task_complete"
+		}
+		payload, err := json.Marshal(struct {
+			Type string                       `json:"type"`
+			Task *service.PelicanTaskSnapshot `json:"task"`
+		}{kind, task})
+		if err != nil {
+			return
+		}
+		if _, err = fmt.Fprintf(c.Writer, "data: %s\n\n", payload); err != nil {
+			return
+		}
+		c.Writer.Flush()
+		if kind == "task_complete" {
+			return
+		}
+		select {
+		case <-c.Request.Context().Done():
+			return
+		case <-ticker.C:
+		}
+		task, err = h.runner.TaskSnapshot(c.Request.Context(), id)
+		if err != nil {
+			return
+		}
+	}
+}

backend/internal/handler/admin/pelican_background_handler_test.go

@@ -0,0 +1,60 @@
+//go:build unit
+
+package admin
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+type pelicanSummaryHandlerRepo struct {
+	*codexGatewayBorrowHandlerTestRepo
+}
+
+func (r *pelicanSummaryHandlerRepo) GetTaskSnapshot(context.Context, string) (*service.PelicanTaskSnapshot, error) {
+	copy := *r.task
+	copy.Results = nil
+	return &service.PelicanTaskSnapshot{CodexGatewayBorrowTestTask: &copy, Counts: map[string]int{"complete": 1}}, nil
+}
+func (r *pelicanSummaryHandlerRepo) ListTaskResults(context.Context, string, service.PelicanResultFilter) (*service.PelicanResultPage, error) {
+	return &service.PelicanResultPage{Items: []*service.PelicanResultSummary{{ID: r.result.ID, TaskID: r.task.ID, Status: "complete", HasPreview: true}}, Total: 1, Page: 1, PageSize: 24}, nil
+}
+func TestPelicanBackgroundReadEndpointsNeverGenerateOrSendRawInProgress(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	legacy, base := newCodexGatewayBorrowHandlerFixture(t)
+	repo := &pelicanSummaryHandlerRepo{base}
+	runner := service.NewCodexGatewayBorrowTestRunner(repo, nil, nil)
+	t.Cleanup(runner.Stop)
+	h := NewPelicanTestHandler(nil, runner, legacy.preview)
+	router := gin.New()
+	router.GET("/tasks/:id", h.GetTask)
+	router.GET("/tasks/:id/results", h.GetTaskResults)
+	router.GET("/tasks/:id/results/:result_id", h.GetTaskResult)
+	router.GET("/tasks/:id/events", h.ObserveTask)
+	for _, suffix := range []string{"", "/results", "/events"} {
+		w := httptest.NewRecorder()
+		router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/tasks/"+base.task.ID+suffix, nil))
+		require.Equal(t, http.StatusOK, w.Code)
+		require.NotContains(t, w.Body.String(), "raw_answer")
+		require.NotContains(t, w.Body.String(), "admin-visible")
+		require.Equal(t, "private, no-store", w.Header().Get("Cache-Control"))
+	}
+	full := httptest.NewRecorder()
+	router.ServeHTTP(full, httptest.NewRequest(http.MethodGet, "/tasks/"+base.task.ID+"/results/"+base.result.ID, nil))
+	require.Equal(t, http.StatusOK, full.Code)
+	require.Contains(t, full.Body.String(), "admin-visible")
+	base.task.Status = "running"
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	disconnected := httptest.NewRecorder()
+	router.ServeHTTP(disconnected, httptest.NewRequest(http.MethodGet, "/tasks/"+base.task.ID+"/events", nil).WithContext(ctx))
+	require.Equal(t, "running", base.task.Status)
+	require.Zero(t, base.creates)
+	require.Zero(t, base.writes)
+}

backend/internal/repository/pelican_background_repo.go

@@ -0,0 +1,108 @@
+package repository
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+)
+
+var _ service.PelicanTaskReader = (*codexGatewayBorrowTestRepository)(nil)
+
+func (r *codexGatewayBorrowTestRepository) GetTaskSnapshot(ctx context.Context, id string) (*service.PelicanTaskSnapshot, error) {
+	task, err := scanCodexGatewayBorrowTestTask(r.db.QueryRowContext(ctx, `SELECT `+codexGatewayBorrowTestTaskColumns+`
+ FROM codex_gateway_borrow_test_tasks WHERE scope='manual_pelican' AND (id=$1 OR client_task_id=$1) AND expires_at>CURRENT_TIMESTAMP`, id))
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, service.ErrCodexGatewayBorrowTestNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	snapshot := &service.PelicanTaskSnapshot{CodexGatewayBorrowTestTask: task, Counts: make(map[string]int)}
+	rows, err := r.db.QueryContext(ctx, `SELECT status, COUNT(*), COUNT(*) FILTER (WHERE status='complete' AND octet_length(html)=0)
+ FROM codex_gateway_borrow_test_results WHERE task_id=$1 GROUP BY status`, task.ID)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	for rows.Next() {
+		var status string
+		var count, noPreview int
+		if err := rows.Scan(&status, &count, &noPreview); err != nil {
+			return nil, err
+		}
+		snapshot.Counts[status] = count
+		snapshot.Counts["no_preview"] += noPreview
+	}
+	snapshot.Completed = 0
+	for status, count := range snapshot.Counts {
+		if status != "pending" && status != "running" && status != "no_preview" {
+			snapshot.Completed += count
+		}
+	}
+	return snapshot, rows.Err()
+}
+
+// Keep this projection independent from codexGatewayBorrowTestResultColumns:
+// no raw_answer, raw_response, raw_html, html or complete error enters a page.
+const pelicanResultSummaryColumns = `r.id,r.task_id,r.ordinal,r.account_id,r.account_name,r.platform,r.model_id,r.effort,r.status,
+ (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`
+
+func (r *codexGatewayBorrowTestRepository) ListTaskResults(ctx context.Context, id string, filter service.PelicanResultFilter) (*service.PelicanResultPage, error) {
+	if filter.Page < 1 {
+		filter.Page = 1
+	}
+	if filter.Size < 1 {
+		filter.Size = 24
+	}
+	if filter.Size > 50 {
+		filter.Size = 50
+	}
+	args := []any{id}
+	where := []string{"t.scope='manual_pelican'", "t.id=$1", "t.expires_at>CURRENT_TIMESTAMP"}
+	add := func(clause string, value any) {
+		args = append(args, value)
+		where = append(where, fmt.Sprintf(clause, len(args)))
+	}
+	if filter.AccountID > 0 {
+		add("r.account_id=$%d", filter.AccountID)
+	}
+	if filter.Model != "" {
+		add("r.model_id=$%d", filter.Model)
+	}
+	switch filter.Status {
+	case "":
+	case "no_preview":
+		where = append(where, "r.status='complete' AND octet_length(r.html)=0")
+	case "pending", "running", "complete", "failed", "incomplete", "cancelled", "skipped":
+		add("r.status=$%d", filter.Status)
+	default:
+		return nil, service.ErrCodexGatewayBorrowTestInvalidRequest
+	}
+	from := ` FROM codex_gateway_borrow_test_results r JOIN codex_gateway_borrow_test_tasks t ON t.id=r.task_id WHERE ` + strings.Join(where, " AND ")
+	page := &service.PelicanResultPage{Items: []*service.PelicanResultSummary{}, Page: filter.Page, PageSize: filter.Size}
+	if err := r.db.QueryRowContext(ctx, "SELECT COUNT(*)"+from, args...).Scan(&page.Total); err != nil {
+		return nil, err
+	}
+	args = append(args, filter.Size, (filter.Page-1)*filter.Size)
+	query := "SELECT " + pelicanResultSummaryColumns + from + fmt.Sprintf(" ORDER BY r.ordinal LIMIT $%d OFFSET $%d", len(args)-1, len(args))
+	rows, err := r.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	for rows.Next() {
+		item := &service.PelicanResultSummary{}
+		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); err != nil {
+			return nil, err
+		}
+		page.Items = append(page.Items, item)
+	}
+	return page, rows.Err()
+}