fix(acceptance): 解除八任务准入瓶颈

线上 P24 验收证明 Worker 注册容量为 48，但异步 admission 始终只有 8 条活跃租约。将单事务调度窗口覆盖当前 2 x P32 的 64 槽，避免多 Worker 在同一 FIFO 小批次上竞争形成隐性全局上限。\n\nGemini 流式请求补充可回放 GetBody 与精确 Content-Length，使带幂等键的 POST 遇到 HTTP/2 可重试传输错误时可安全重放；同时提前部署验收回调收集器，保证旧 Run side effects 能在接管前收口。\n\n已通过 Go 全量测试、go vet、gofmt、Shell bash -n、ShellCheck、验收与人工发布脚本测试、迁移安全检查，以及独立 PostgreSQL 48 条 admission/租约/River job 原子集成测试。

apps/api/cmd/acceptance-load/main.go

@@ -37,6 +37,11 @@ const (
 	upstreamHeader = "X-EasyAI-Acceptance-Upstream"
 )
 
+const (
+	geminiRequestPrefix = `{"contents":[{"role":"user","parts":[{"text":"将参考图编辑为蓝色赛博朋克风格，保留主体结构"},{"inlineData":{"mimeType":"image/png","data":"`
+	geminiRequestSuffix = `"}}]}],"generationConfig":{"responseModalities":["IMAGE"]}}`
+)
+
 var reportURLPattern = regexp.MustCompile(`(?i)(?:https?|postgres(?:ql)?):\/\/[^\s"'<>]+`)
 
 type options struct {
@@ -407,16 +412,8 @@ func runGemini(
 			defer func() { <-slots }()
 			requestStarted := time.Now()
 			endpoint := opts.gateways[logicalIndex%len(opts.gateways)] + "/v1beta/models/" + url.PathEscape(opts.geminiModel) + ":generateContent"
-			requestBody := streamGeminiRequestBody(paddedPNGVariant(inputBytes, geminiInputVariant(opts.runID+"/"+opts.executionID, logicalIndex)))
+			input := paddedPNGVariant(inputBytes, geminiInputVariant(opts.runID+"/"+opts.executionID, logicalIndex))
-			req, err := http.NewRequestWithContext(
+			req, err := newGeminiRequest(ctx, endpoint, input)
-				ctx,
-				http.MethodPost,
-				endpoint,
-				requestBody,
-			)
-			if err != nil {
-				_ = requestBody.Close()
-			}
 			if err == nil {
 				opts.setHeaders(req, logicalIndex, realUpstream)
 				req.Header.Set("Content-Type", "application/json")
@@ -481,7 +478,7 @@ func streamGeminiRequestBody(input []byte) io.ReadCloser {
 		closeWithError := func(err error) {
 			_ = writer.CloseWithError(err)
 		}
-		if _, err := io.WriteString(writer, `{"contents":[{"role":"user","parts":[{"text":"将参考图编辑为蓝色赛博朋克风格，保留主体结构"},{"inlineData":{"mimeType":"image/png","data":"`); err != nil {
+		if _, err := io.WriteString(writer, geminiRequestPrefix); err != nil {
 			closeWithError(err)
 			return
 		}
@@ -494,7 +491,7 @@ func streamGeminiRequestBody(input []byte) io.ReadCloser {
 			closeWithError(err)
 			return
 		}
-		if _, err := io.WriteString(writer, `"}}]}],"generationConfig":{"responseModalities":["IMAGE"]}}`); err != nil {
+		if _, err := io.WriteString(writer, geminiRequestSuffix); err != nil {
 			closeWithError(err)
 			return
 		}
@@ -503,6 +500,21 @@ func streamGeminiRequestBody(input []byte) io.ReadCloser {
 	return reader
 }
 
+func newGeminiRequest(ctx context.Context, endpoint string, input []byte) (*http.Request, error) {
+	getBody := func() (io.ReadCloser, error) {
+		return streamGeminiRequestBody(input), nil
+	}
+	body, _ := getBody()
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, body)
+	if err != nil {
+		_ = body.Close()
+		return nil, err
+	}
+	request.GetBody = getBody
+	request.ContentLength = int64(len(geminiRequestPrefix) + base64.StdEncoding.EncodedLen(len(input)) + len(geminiRequestSuffix))
+	return request, nil
+}
+
 func runVideo(
 	ctx context.Context,
 	client *http.Client,

apps/api/cmd/acceptance-load/main_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -82,6 +83,37 @@ func TestStreamGeminiRequestBodyPreservesInput(t *testing.T) {
 	}
 }
 
+func TestGeminiRequestBodyCanBeReplayedAfterHTTP2Failure(t *testing.T) {
+	input := paddedPNGVariant(256<<10, 91)
+	request, err := newGeminiRequest(t.Context(), "https://gateway.example/v1beta/models/test:generateContent", input)
+	if err != nil {
+		t.Fatalf("new Gemini request: %v", err)
+	}
+	if request.GetBody == nil {
+		t.Fatal("streaming Gemini request does not provide GetBody")
+	}
+	first, err := io.ReadAll(request.Body)
+	if err != nil {
+		t.Fatalf("read first request body: %v", err)
+	}
+	_ = request.Body.Close()
+	replayedBody, err := request.GetBody()
+	if err != nil {
+		t.Fatalf("replay Gemini request body: %v", err)
+	}
+	replayed, err := io.ReadAll(replayedBody)
+	if err != nil {
+		t.Fatalf("read replayed request body: %v", err)
+	}
+	_ = replayedBody.Close()
+	if !bytes.Equal(first, replayed) {
+		t.Fatal("replayed Gemini request body differs from the original")
+	}
+	if request.ContentLength != int64(len(first)) {
+		t.Fatalf("content length=%d, want %d", request.ContentLength, len(first))
+	}
+}
+
 func TestVideoCombinationsProvide128UniqueInputs(t *testing.T) {
 	images := make([]string, 16)
 	for index := range images {

apps/api/internal/runner/admission.go

@@ -33,7 +33,11 @@ const (
 	asyncWorkerCapacityScopeKey = "global"
 	asyncWorkerQueueLimit       = 10000
 	asyncWorkerMaxWaitSeconds   = 24 * 60 * 60
-	asyncAdmissionBatchMax      = 8
+	// One transaction must be able to fill the certified 2 x P32 global
+	// capacity. Smaller windows let multiple Worker dispatchers repeatedly race
+	// on the same FIFO head and can turn a batch size into an accidental cluster
+	// concurrency ceiling.
+	asyncAdmissionBatchMax      = 64
 	acceptanceQueueLimit        = 10000
 	acceptanceQueueMaxWait      = 15 * 60
 	synchronousAdmissionWakeMax = 256
@@ -737,13 +741,7 @@ func (s *Service) dispatchWaitingAsyncAdmissions(ctx context.Context) {
 		case <-ticker.C:
 		}
 		for {
-			batchLimit := s.cfg.AsyncWorkerHardLimit
+			batchLimit := asyncAdmissionBatchLimit(s.cfg.AsyncWorkerHardLimit)
-			if batchLimit <= 0 {
-				batchLimit = 64
-			}
-			if batchLimit > asyncAdmissionBatchMax {
-				batchLimit = asyncAdmissionBatchMax
-			}
 			taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit)
 			if err != nil {
 				s.logger.Warn("list waiting async admissions failed", "error", err)
@@ -789,6 +787,13 @@ func (s *Service) dispatchWaitingAsyncAdmissions(ctx context.Context) {
 	}
 }
 
+func asyncAdmissionBatchLimit(globalHardLimit int) int {
+	if globalHardLimit <= 0 || globalHardLimit > asyncAdmissionBatchMax {
+		return asyncAdmissionBatchMax
+	}
+	return globalHardLimit
+}
+
 func (s *Service) reapExpiredTaskAdmissions(ctx context.Context) {
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()

apps/api/internal/runner/admission_test.go

@@ -62,6 +62,22 @@ func TestAcceptanceAdmissionScopesLeaveProductionPolicyUnchanged(t *testing.T) {
 	}
 }
 
+func TestAsyncAdmissionBatchCoversCertifiedGlobalCapacity(t *testing.T) {
+	for _, testCase := range []struct {
+		globalHardLimit int
+		want            int
+	}{
+		{globalHardLimit: 48, want: 48},
+		{globalHardLimit: 64, want: 64},
+		{globalHardLimit: 128, want: 64},
+		{globalHardLimit: 0, want: 64},
+	} {
+		if got := asyncAdmissionBatchLimit(testCase.globalHardLimit); got != testCase.want {
+			t.Fatalf("async admission batch limit for %d = %d, want %d", testCase.globalHardLimit, got, testCase.want)
+		}
+	}
+}
+
 func TestPinCandidatesToTaskAdmissionPreservesAdmittedCandidate(t *testing.T) {
 	input := []store.RuntimeModelCandidate{
 		{PlatformID: "platform-a", PlatformModelID: "model-a"},

apps/api/internal/store/admission_queue_integration_test.go

@@ -506,13 +506,16 @@ WHERE id = $1::uuid`, recoveryGraceTask.ID); err != nil {
 		ScopeType:       "worker_capacity",
 		ScopeKey:        "batch-" + suffix,
 		ScopeName:       "batch capacity",
-		ConcurrentLimit: 3,
+		ConcurrentLimit: 48,
 		Amount:          1,
 		LeaseTTLSeconds: 120,
 		QueueLimit:      100,
 		MaxWaitSeconds:  600,
 	}
-	batchTasks := []GatewayTask{createTask(true), createTask(true), createTask(true)}
+	batchTasks := make([]GatewayTask, 48)
+	for index := range batchTasks {
+		batchTasks[index] = createTask(true)
+	}
 	batchInputs := make([]TaskAdmissionInput, 0, len(batchTasks))
 	batchJobIDs := make(map[string]int64, len(batchTasks))
 	for index, task := range batchTasks {
@@ -544,6 +547,10 @@ WHERE id = $1::uuid`, input.TaskID, batchJobIDs[input.TaskID])
 			t.Fatalf("batch outcome %d=%+v", index, outcome)
 		}
 	}
+	batchTaskIDs := make([]string, 0, len(batchTasks))
+	for _, task := range batchTasks {
+		batchTaskIDs = append(batchTaskIDs, task.ID)
+	}
 	var batchAdmissions, batchLeases, batchRiverJobs int
 	if err := first.pool.QueryRow(ctx, `
 SELECT
@@ -556,16 +563,19 @@ SELECT
   (SELECT count(*)
    FROM gateway_tasks
    WHERE id = ANY($1::uuid[]) AND river_job_id IS NOT NULL)`,
-		[]string{batchTasks[0].ID, batchTasks[1].ID, batchTasks[2].ID},
+		batchTaskIDs,
 	).Scan(&batchAdmissions, &batchLeases, &batchRiverJobs); err != nil {
 		t.Fatalf("read committed batch admission: %v", err)
 	}
-	if batchAdmissions != 3 || batchLeases != 3 || batchRiverJobs != 3 {
+	if batchAdmissions != len(batchTasks) || batchLeases != len(batchTasks) || batchRiverJobs != len(batchTasks) {
 		t.Fatalf(
-			"batch admissions=%d leases=%d River jobs=%d, want 3/3/3",
+			"batch admissions=%d leases=%d River jobs=%d, want %d/%d/%d",
 			batchAdmissions,
 			batchLeases,
 			batchRiverJobs,
+			len(batchTasks),
+			len(batchTasks),
+			len(batchTasks),
 		)
 	}
 	for _, task := range batchTasks {

scripts/cluster/run-production-acceptance.sh

@@ -3547,6 +3547,7 @@ ensure_acceptance_identity_shards
 ensure_acceptance_real_images
 select_acceptance_models
 ensure_acceptance_model_access
+deploy_protocol_emulator
 create_and_activate_run
 wait_for_existing_tasks_to_drain
 snapshot_pre_acceptance_capacity
@@ -3569,8 +3570,6 @@ if ! verify_resource_preconditions; then
   mark_run_failed "$failure_reason"
   exit 1
 fi
-deploy_protocol_emulator
-
 for profile in P24 P28 P32; do
   if ! apply_capacity_profile "$profile"; then
     failure_gate_id=capacity_profile_apply

tests/release/production-acceptance-script-test.sh

@@ -144,4 +144,8 @@ fi
 wait "$pressure_pid" >/dev/null 2>&1 || true
 [[ -z $active_load_pid ]]
 
+deploy_call_line=$(grep -n '^deploy_protocol_emulator$' "$script" | cut -d: -f1)
+activate_call_line=$(grep -n '^create_and_activate_run$' "$script" | cut -d: -f1)
+[[ -n $deploy_call_line && -n $activate_call_line && $deploy_call_line -lt $activate_call_line ]]
+
 echo 'production_acceptance_script_tests=PASS'