Serialize Codex probe pairs and preserve final service tiers

.downstream/native-domains.md

@@ -129,11 +129,20 @@ timer. `/prepare` and `/verify` retain their interfaces. Same-revision, same-wir
 qualifications share work; cancelling one caller does not cancel other waiters,
 and the final departing waiter cancels abandoned work. Distinct targets no longer
 fail because another target is validating. Outbound observations share the existing
-ten execution slots and serialize by account. Preparation has a ten-minute bound,
-continues after a target failure, and reports partial readiness after rechecking
+ten execution slots and serialize by account. The two target shots retain one
+account lease from mint through continuation, so a different model or client
+window cannot interleave another probe. Source preparation finishes before that
+lease is acquired; nested shots reuse it without taking a second permit.
+Preparation has a ten-minute bound, continues after a target failure, and reports
+partial readiness after rechecking
 expiry. Policy, credential/model-map and exit changes invalidate displayed evidence;
 actual dispatch still verifies the finalized request fingerprint.
 
+The effective service tier travels from the finalized HTTP body or WS frame in
+local request context into both qualification and its cache key. Explicit
+`default`, `auto` and `scale` remain distinct even though their routing hints are
+model-only. No internal tier header is sent upstream.
+
 Status adds `setup`, `observed_since` and `recent_usage`. The last actual dispatch
 and in-process counters are retained per configured target/model/protocol/purpose;
 configuration publication resets this index. HTTP/SSE, pooled WS and native WS

backend/internal/service/codex_gateway_borrow.go

@@ -802,6 +802,7 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6
 func borrowRequestFingerprint(req *http.Request, model, proxy string, profile *tlsfingerprint.Profile, cookie string) string {
 	tlsJSON, _ := json.Marshal(profile)
 	values := []string{cookie, model, proxy, req.URL.EscapedPath(), req.Header.Get("Authorization"), req.Header.Get("ChatGPT-Account-ID"), req.Header.Get("User-Agent"), req.Header.Get("Originator"), req.Header.Get("Version"), req.Header.Get("X-Codex-Turn-State"), string(tlsJSON)}
+	values = append(values, codexBorrowRequestServiceTier(req))
 	// Include the native identity and routing policy preserved by the probe.
 	// Per-request tracing IDs and the probe's random legacy session_id are not
 	// qualification identities and must not defeat reuse or failure cooldowns.

backend/internal/service/codex_gateway_borrow_diagnose_test.go

@@ -175,6 +175,12 @@ func TestCodexBorrowDiagnoseWSCountsConfiguredPrewarm(t *testing.T) {
 }
 
 func TestCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T) {
+	for _, tier := range []string{"priority", "default", "flex", "ultrafast", ""} {
+		t.Run("tier_"+tier, func(t *testing.T) { testCodexBorrowDiagnoseHTTPToolContinuation(t, tier) })
+	}
+}
+
+func testCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T, tier string) {
 	account := borrowCoreAccount(2)
 	generator, repo := newPelicanGeneratorForTest(account, &pelicanGeneratorUpstream{})
 	gateway := generator.openaiGatewayService
@@ -183,7 +189,7 @@ func TestCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T) {
 		business++
 		body := borrowCoreBody(t, req)
 		assert.Equal(t, "gpt-6.1-sol", gjson.GetBytes(body, "model").String())
-		assert.Equal(t, "priority", gjson.GetBytes(body, "service_tier").String())
+		assert.Equal(t, tier, gjson.GetBytes(body, "service_tier").String())
 		assert.NotEmpty(t, req.Header.Get("session-id"))
 		assert.Contains(t, req.Header.Get("Cookie"), "__oailb=synthetic-borrowed-cookie")
 		var output []any
@@ -210,15 +216,15 @@ func TestCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T) {
 	})
 	s := NewCodexGatewayBorrowService(nil, repo, gateway, borrowCoreProbe(func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
 		probes++
-		assert.Equal(t, "priority", gjson.GetBytes(borrowCoreBody(t, req), "service_tier").String())
+		assert.Equal(t, tier, gjson.GetBytes(borrowCoreBody(t, req), "service_tier").String())
 		return borrowCoreResponse("gpt-6.1-sol", "OK", "probe-state"), nil
 	}), nil)
 	s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6.1-sol"}}, false)
 	defer s.Stop()
 	borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
 	var results []CodexBorrowDiagnosticResult
 	var count int32
-	err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{AccountID: 2, Model: "gpt-6.1-sol", Transport: "http", Scenario: "codex_session", ServiceTier: "priority"}, func(e CodexBorrowDiagnosticEvent) {
+	err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{AccountID: 2, Model: "gpt-6.1-sol", Transport: "http", Scenario: "codex_session", ServiceTier: tier}, func(e CodexBorrowDiagnosticEvent) {
 		if e.Result != nil {
 			results = append(results, *e.Result)
 		}

backend/internal/service/codex_gateway_borrow_http.go

@@ -105,6 +105,9 @@ func (s *OpenAIGatewayService) prepareCodexGatewayBorrowHTTP(ctx context.Context
 	prepared.service.mu.Lock()
 	prepared.revision = prepared.service.revision
 	prepared.service.mu.Unlock()
+	if tier, ok := req.Context().Value(codexBorrowServiceTierContextKey{}).(string); ok {
+		ctx = withCodexBorrowServiceTier(ctx, tier)
+	}
 	borrowed, application, err := prepared.service.Apply(req.WithContext(ctx), account, model, proxy, nil, false)
 	if err != nil {
 		tracker.blocked(err)

backend/internal/service/codex_gateway_borrow_probe.go

@@ -207,7 +207,18 @@ func borrowCandidateFromCookies(cookies []*http.Cookie, path string, now time.Ti
 
 func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, candidate *codexGatewayBorrowCandidate) CodexGatewayBorrowVerification {
 	expires := candidate.expires
-	result := CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), ExpiresAt: &expires, Reason: "target_probe_failed", RequestShape: "ranxi_full", ServiceTier: borrowProbeServiceTier(template.Header)}
+	result := CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), ExpiresAt: &expires, Reason: "target_probe_failed", RequestShape: "ranxi_full", ServiceTier: codexBorrowRequestServiceTier(template)}
+	ctx = withCodexBorrowServiceTier(ctx, result.ServiceTier)
+	// A two-shot STATE verdict is one observation. Releasing account capacity
+	// after mint would let another model/window mint a ticket before continuation.
+	// Keep the shared account lease across both shots; source preparation already
+	// finished, so this cannot hold a target permit while waiting for a source.
+	ctx, release, err := acquireCodexBorrowObservation(ctx, account.ID)
+	if err != nil {
+		result.Error = err.Error()
+		return result
+	}
+	defer release()
 	cookies := []*http.Cookie{{Name: "__oailb", Value: candidate.cookie.Value}}
 	fire := func(state string, cookies []*http.Cookie) (codexGatewayBorrowObservation, error) {
 		shotCtx, cancel := context.WithTimeout(ctx, codexGatewayBorrowShotTimeout)
@@ -323,7 +334,8 @@ func borrowProbeServiceTier(headers http.Header) string {
 
 func borrowObservationRequest(ctx context.Context, headers http.Header, model, prompt, effort, state string, cookies []*http.Cookie, purpose HTTPUpstreamProfile) (*http.Request, error) {
 	ctx = WithCodexGatewayBorrowObservation(WithHTTPUpstreamRedirectsDisabled(WithHTTPUpstreamProfile(ctx, purpose)))
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(borrowObservationPayload(model, prompt, effort, borrowProbeServiceTier(headers))))
+	tier := codexBorrowRequestServiceTier((&http.Request{Header: headers}).WithContext(ctx))
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(borrowObservationPayload(model, prompt, effort, tier)))
 	if err != nil {
 		return nil, err
 	}
@@ -366,14 +378,7 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account *
 	if err := consumeBorrowDiagnosticRequest(req.Context()); err != nil {
 		return shot, err
 	}
-	if pelicanExecutionFromContext(req.Context()) == nil {
-		release, err := sharedPelicanExecution.acquire(req.Context(), account.ID)
-		if err != nil {
-			return shot, err
-		}
-		defer release()
-	}
-	leasedCtx, release, acquireErr := AcquirePelicanExecution(req.Context(), account.ID)
+	leasedCtx, release, acquireErr := acquireCodexBorrowObservation(req.Context(), account.ID)
 	if acquireErr != nil {
 		shot.errorText = acquireErr.Error()
 		return shot, acquireErr
@@ -446,6 +451,35 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account *
 	return shot, nil
 }
 
+// Both public-request preparation and manual diagnostics use the same account
+// and global permits. A sequential target pair lends its active lease to each
+// shot; a standalone source observation acquires and releases its own permit.
+func acquireCodexBorrowObservation(ctx context.Context, accountID int64) (context.Context, func(), error) {
+	if pelicanExecutionFromContext(ctx) != nil {
+		return AcquirePelicanExecution(ctx, accountID)
+	}
+	if err := ctx.Err(); err != nil {
+		return ctx, nil, err
+	}
+	if lease, _ := ctx.Value(pelicanExecutionLeaseContextKey{}).(*pelicanExecutionLease); lease != nil && lease.active.Load() {
+		if lease.accountID != accountID {
+			return ctx, nil, errors.New("borrow observation cannot acquire another account while holding a probe lease")
+		}
+		return ctx, func() {}, nil
+	}
+	release, err := sharedPelicanExecution.acquire(ctx, accountID)
+	if err != nil {
+		return ctx, nil, err
+	}
+	lease := &pelicanExecutionLease{accountID: accountID}
+	lease.active.Store(true)
+	return context.WithValue(ctx, pelicanExecutionLeaseContextKey{}, lease), func() {
+		if lease.active.Swap(false) {
+			release()
+		}
+	}, nil
+}
+
 // A Responses terminal event is authoritative even when the HTTP connection
 // remains open. Collect the original bytes, without waiting for EOF or reading
 // beyond that terminal payload; sendObservation closes the body immediately.

backend/internal/service/codex_gateway_borrow_probe_serial_test.go

@@ -0,0 +1,116 @@
+package service
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCodexBorrowProbePairCannotInterleaveOnSameAccount(t *testing.T) {
+	for _, manual := range []bool{false, true} {
+		name := "business"
+		if manual {
+			name = "diagnostic"
+		}
+		t.Run(name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer cancel()
+			if manual {
+				state := newPelicanExecutionState(ctx, sharedPelicanExecution, time.Minute, time.Now)
+				defer state.finish()
+				ctx = context.WithValue(state.parent, pelicanExecutionContextKey{}, state)
+			}
+			account := borrowCoreAccount(92001)
+			started, release := make(chan struct{}), make(chan struct{})
+			var mu sync.Mutex
+			var order []string
+			s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+				id := req.Header.Get("X-Codex-Window-ID")
+				phase := "mint"
+				if req.Header.Get("X-Codex-Turn-State") != "" {
+					phase = "continue"
+				}
+				mu.Lock()
+				order = append(order, id+":"+phase)
+				mu.Unlock()
+				if id == "first" && phase == "mint" {
+					close(started)
+					select {
+					case <-release:
+					case <-req.Context().Done():
+						return nil, req.Context().Err()
+					}
+				}
+				return borrowCoreResponse("gpt-6-astra", "OK", id), nil
+			}, account)
+			borrowCoreCandidate(s, time.Now().Add(time.Minute))
+			run := func(id string, done chan<- CodexGatewayBorrowVerification) {
+				template := &http.Request{Header: http.Header{"X-Codex-Window-Id": []string{id}}}
+				done <- s.probeTarget(ctx, template, account, "gpt-6-astra", "", nil, s.candidate)
+			}
+			first, second := make(chan CodexGatewayBorrowVerification, 1), make(chan CodexGatewayBorrowVerification, 1)
+			go run("first", first)
+			select {
+			case <-started:
+			case <-ctx.Done():
+				t.Fatal("first probe did not start")
+			}
+			go run("second", second)
+			queued := false
+			deadline := time.Now().Add(time.Second)
+			for !queued && time.Now().Before(deadline) {
+				sharedPelicanExecution.mu.Lock()
+				for _, waiter := range sharedPelicanExecution.waiting {
+					queued = queued || waiter.accountID == account.ID
+				}
+				sharedPelicanExecution.mu.Unlock()
+				if !queued {
+					time.Sleep(time.Millisecond)
+				}
+			}
+			close(release)
+			require.True(t, queued, "second probe must wait on the same account")
+			require.True(t, (<-first).Success)
+			require.True(t, (<-second).Success)
+			mu.Lock()
+			defer mu.Unlock()
+			require.Equal(t, []string{"first:mint", "first:continue", "second:mint", "second:continue"}, order)
+		})
+	}
+}
+
+func TestCodexBorrowProbeLeaseCancellationDoesNotBlockOtherAccounts(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	leased, release, err := acquireCodexBorrowObservation(ctx, 92002)
+	require.NoError(t, err)
+	defer release()
+	_, nestedRelease, err := acquireCodexBorrowObservation(leased, 92002)
+	require.NoError(t, err)
+	nestedRelease() // A nested shot cannot release the whole pair's lease.
+	_, releaseOther, err := acquireCodexBorrowObservation(ctx, 92003)
+	require.NoError(t, err, "another account can use remaining global capacity")
+	releaseOther()
+	_, _, err = acquireCodexBorrowObservation(leased, 92003)
+	require.Error(t, err, "a pair cannot wait for a source while retaining its target lease")
+	queued, cancelQueued := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() {
+		_, releaseQueued, queuedErr := acquireCodexBorrowObservation(queued, 92002)
+		if releaseQueued != nil {
+			releaseQueued()
+		}
+		done <- queuedErr
+	}()
+	cancelQueued()
+	require.ErrorIs(t, <-done, context.Canceled)
+	release()
+	_, releaseAgain, err := acquireCodexBorrowObservation(ctx, 92002)
+	require.NoError(t, err, "cancellation and nested releases must not leak account capacity")
+	releaseAgain()
+}

backend/internal/service/codex_gateway_borrow_request_context.go

@@ -0,0 +1,22 @@
+package service
+
+import (
+	"context"
+	"net/http"
+)
+
+type codexBorrowServiceTierContextKey struct{}
+
+// The finalized body owns service_tier. The routing hint intentionally omits
+// default/auto/scale, so it cannot reconstruct every actual request condition.
+// Keep the exact effective tier locally; never add an internal transport header.
+func withCodexBorrowServiceTier(ctx context.Context, tier string) context.Context {
+	return context.WithValue(ctx, codexBorrowServiceTierContextKey{}, normalizedOpenAIServiceTierValue(tier))
+}
+
+func codexBorrowRequestServiceTier(req *http.Request) string {
+	if tier, ok := req.Context().Value(codexBorrowServiceTierContextKey{}).(string); ok {
+		return tier
+	}
+	return borrowProbeServiceTier(req.Header) // Compatibility for existing manual templates.
+}

backend/internal/service/codex_gateway_borrow_request_context_test.go

@@ -0,0 +1,39 @@
+package service
+
+import (
+	"context"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
+)
+
+func TestCodexBorrowQualificationSeparatesModelOnlyRoutingTiers(t *testing.T) {
+	account := borrowCoreAccount(2)
+	var received []string
+	s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+		tier := gjson.GetBytes(borrowCoreBody(t, req), "service_tier").String()
+		received = append(received, tier)
+		assert.Equal(t, "model=gpt-6.1-sol", req.Header.Get(openAICodexRoutingHintHeader))
+		return borrowCoreResponse("gpt-6.1-sol", "OK", "stable"), nil
+	}, account)
+	borrowCoreCandidate(s, time.Now().Add(time.Minute))
+	_, req, _, err := s.accountTemplate(context.Background(), account.ID, "gpt-6.1-sol")
+	require.NoError(t, err)
+	for _, tier := range []string{"default", "auto", "scale", ""} {
+		wire := req.WithContext(withCodexBorrowServiceTier(req.Context(), tier))
+		_, _, err := s.Apply(wire, account, "gpt-6.1-sol", "", nil, true)
+		require.Error(t, err, "identical routing hints must not share another tier's proof")
+		_, applied, err := s.Apply(wire, account, "gpt-6.1-sol", "", nil, false)
+		require.NoError(t, err)
+		require.Equal(t, tier, applied.Verification.ServiceTier)
+	}
+	require.Equal(t, []string{"default", "default", "auto", "auto", "scale", "scale", "", ""}, received)
+	_, _, err = s.Apply(req.WithContext(withCodexBorrowServiceTier(req.Context(), "default")), account, "gpt-6.1-sol", "", nil, true)
+	require.NoError(t, err, "the first tier's independent successful proof remains reusable")
+	require.Len(t, received, 8)
+}

backend/internal/service/openai_gateway_forward.go

@@ -1631,6 +1631,7 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.
 	// x-codex-beta-features：按真实 Codex 的会话级行为补注（在账号级覆写之后，
 	// 保证不被覆盖丢失）。
 	applyOpenAICodexBetaFeatures(c, account, req.Header)
+	req = req.WithContext(withCodexBorrowServiceTier(req.Context(), gjson.GetBytes(body, "service_tier").String()))
 	if err := s.finalizeCodexOutboundHeaders(req.Context(), c, account, req.Header, gjson.GetBytes(body, "model").String(), gjson.GetBytes(body, "service_tier").String()); err != nil {
 		return nil, err
 	}

backend/internal/service/openai_gateway_messages.go

@@ -386,6 +386,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
 		// 清除身份头。真正发送前恢复完整 Codex 身份，避免 ChatGPT Codex 上游因缺失
 		// originator/OpenAI-Beta 返回 404（issue #3901）。
 		ensureCodexIdentityHeaders(upstreamReq.Header)
+		upstreamReq = upstreamReq.WithContext(withCodexBorrowServiceTier(upstreamReq.Context(), gjson.GetBytes(responsesBody, "service_tier").String()))
 		if err := s.finalizeCodexOutboundHeaders(upstreamReq.Context(), c, account, upstreamReq.Header, upstreamModel, gjson.GetBytes(responsesBody, "service_tier").String()); err != nil {
 			return nil, err
 		}

backend/internal/service/openai_gateway_passthrough.go

@@ -792,6 +792,7 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough(
 	// x-codex-beta-features：按真实 Codex 的会话级行为补注（在账号级覆写之后，
 	// 保证不被覆盖丢失）。
 	applyOpenAICodexBetaFeatures(c, account, req.Header)
+	req = req.WithContext(withCodexBorrowServiceTier(req.Context(), gjson.GetBytes(body, "service_tier").String()))
 	if err := s.finalizeCodexOutboundHeaders(req.Context(), c, account, req.Header, gjson.GetBytes(body, "model").String(), gjson.GetBytes(body, "service_tier").String()); err != nil {
 		return nil, err
 	}

backend/internal/service/openai_ws_forwarder_ingress.go

@@ -1804,6 +1804,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 			return fmt.Errorf("prepare websocket request: %w", prepareErr)
 		}
 		wireFields := gjson.GetManyBytes(wirePayload, "prompt_cache_key", "model", "service_tier")
+		ctx = withCodexBorrowServiceTier(ctx, wireFields[2].String())
 		finalHeaders, _, headerErr := s.buildOpenAIWSHeaders(ctx, c, account, token, wsDecision, isCodexCLI, turnState,
 			strings.TrimSpace(c.GetHeader(openAIWSTurnMetadataHeader)), wireFields[0].String(), wireFields[1].String(), wireFields[2].String())
 		if headerErr != nil {

backend/internal/service/openai_ws_forwarder_v2.go

@@ -175,6 +175,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
 	storeDisabledConnMode := s.openAIWSStoreDisabledConnMode()
 	forceNewConnByPolicy := shouldForceNewConnOnStoreDisabled(storeDisabledConnMode, lastFailureReason)
 	forceNewConn := forceNewConnByPolicy && storeDisabled && previousResponseID == "" && sessionHash != "" && preferredConnID == ""
+	ctx = withCodexBorrowServiceTier(ctx, openAIWSPayloadString(payload, "service_tier"))
 	wsHeaders, sessionResolution, buildHdrErr := s.buildOpenAIWSHeaders(
 		ctx,
 		c,

backend/internal/service/openai_ws_v2_passthrough_adapter.go

@@ -888,6 +888,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2PassthroughAttempt(
 		turnState = strings.TrimSpace(c.GetHeader(openAIWSTurnStateHeader))
 		turnMetadata = strings.TrimSpace(c.GetHeader(openAIWSTurnMetadataHeader))
 	}
+	ctx = withCodexBorrowServiceTier(ctx, gjson.GetBytes(firstClientMessage, "service_tier").String())
 	headers, _, buildHdrErr := s.buildOpenAIWSHeaders(
 		ctx,
 		c,