Compare Codex probe request shapes on the same candidate

.downstream/native-domains.md

@@ -188,6 +188,20 @@ items when the final successful envelope has empty output; a successful terminal
 is still required. This matches native streaming continuation instead of assuming
 all output is repeated inside `response.completed`.
 
+The HTTP-only `scenario: probe_contract` compares a fixed candidate under three
+request bodies: pinned ranxi field order without a tier, that same body with the
+effective tier, and the current body with the same tier. With no requested tier,
+the redundant middle pair is skipped. Native identity, routing hint, model,
+proxy and cookie stay fixed; each shot still gets its own legacy `session_id`.
+The original probe selects Astra; the result explicitly marks Sol as a downstream
+model extension. `probe_only: true` results carry per-shot completion and STATE
+facts under `verification`, and their common template/cookie and plaintext-body
+fingerprints under `probe_context`. Top-level business dispatch, application and
+completion stay false. These comparisons never publish business qualifications
+or replace real usage observations. Source acquisition and all shots share the
+existing eight-request cap; the snapshot's expiry cancels remaining comparisons
+without silently selecting a different candidate. Page reads do not invoke it.
+
 Ranxi reference: v2.10.2 (`d3e43f2de33af9e987cffa511d76dfabbcd749da`). Its five
 fingerprint, gateway-cookie, target-probe and automatic-setup core files are
 unchanged from the retained v2.10.0 reference. OAuth search-history compatibility

backend/internal/service/codex_gateway_borrow_diagnose.go

@@ -82,6 +82,8 @@ type CodexBorrowDiagnosticRequest struct {
 }
 
 type CodexBorrowDiagnosticResult struct {
+	ProbeOnly     bool                            `json:"probe_only,omitempty"`
+	ProbeContext  *CodexBorrowProbeContext        `json:"probe_context,omitempty"`
 	Scenario      string                          `json:"scenario,omitempty"`
 	FailureStage  string                          `json:"failure_stage,omitempty"`
 	FailureReason string                          `json:"failure_reason,omitempty"`
@@ -114,15 +116,18 @@ type CodexBorrowDiagnosticEvent struct {
 // Diagnose runs fixed short questions through the normal account sender. The
 // internal observation purpose isolates health/billing; it is never user input.
 func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexBorrowDiagnosticRequest, emit func(CodexBorrowDiagnosticEvent)) error {
-	if request.Scenario != "" && request.Scenario != "codex_session" {
-		return errors.New("scenario must be empty or codex_session")
+	if request.Scenario != "" && request.Scenario != "codex_session" && request.Scenario != "probe_contract" {
+		return errors.New("scenario must be empty, codex_session or probe_contract")
 	}
 	if request.Mode != "" && request.Mode != "ordinary" && request.Mode != "borrowed" {
 		return errors.New("mode must be empty, ordinary or borrowed")
 	}
 	if request.Scenario == "codex_session" && request.Transport != "http" {
 		return errors.New("codex_session uses HTTP; existing WS diagnostics already cover continuation")
 	}
+	if request.Scenario == "probe_contract" && (request.Transport != "http" || request.Mode == "ordinary") {
+		return errors.New("probe_contract requires borrowed HTTP observations")
+	}
 	if request.ServiceTier != "" && request.ServiceTier != "default" && request.ServiceTier != OpenAIFastTierPriority && request.ServiceTier != OpenAIFastTierFlex && request.ServiceTier != OpenAIFastTierUltrafast {
 		return errors.New("unsupported service_tier")
 	}
@@ -152,9 +157,13 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB
 	}
 	s.mu.Lock()
 	revisionCtx := s.revisionCtx
+	revision := s.revision
 	s.mu.Unlock()
 	ctx, cancel := borrowRevisionContext(WithAccountObservation(ctx), revisionCtx, 10*time.Minute)
 	defer cancel()
+	if request.Scenario == "probe_contract" {
+		return s.diagnoseProbeContract(ctx, revision, revisionCtx, request, limit, emit)
+	}
 	var requests atomic.Int32
 	modes := []bool{false, true}
 	if request.Mode == "borrowed" || (request.Scenario == "codex_session" && request.Mode == "") {

backend/internal/service/codex_gateway_borrow_diagnose_contract.go

@@ -0,0 +1,137 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync/atomic"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/tidwall/gjson"
+)
+
+type codexBorrowProbeShapeContextKey struct{}
+
+const (
+	codexBorrowProbeRanxiLiteral = "ranxi_literal_without_tier"
+	codexBorrowProbeRanxiTier    = "ranxi_literal_with_final_tier"
+	codexBorrowProbeCurrent      = "current_final_tier"
+)
+
+// These are observations of one fixed candidate, never qualifications usable
+// by business requests. The original pinned route probe selects Astra; rendering
+// a selected Sol model here is explicitly the downstream model extension.
+type CodexBorrowProbeContext struct {
+	SourceAccountID          int64     `json:"source_account_id"`
+	TargetProxyID            *int64    `json:"target_proxy_id"`
+	CookieFingerprint        string    `json:"cookie_fingerprint"`
+	CandidateExpiresAt       time.Time `json:"candidate_expires_at"`
+	TemplateFingerprint      string    `json:"template_fingerprint"`
+	RequestedTier            string    `json:"requested_service_tier"`
+	FinalTier                string    `json:"final_service_tier"`
+	RoutingHint              string    `json:"routing_hint"`
+	PlaintextBodyFingerprint string    `json:"plaintext_body_fingerprint"`
+	OriginalAstraModel       bool      `json:"original_astra_model"`
+}
+
+// Literal request field order from ranxi2001/sub2api d3e43f2, LGPL-3.0,
+// openai_codex_state_probe.go:320-347. Tier is the only semantic addition in
+// the second variant. Model always stays the explicitly selected real model.
+func borrowRanxiLiteralPayload(model, tier string) []byte {
+	modelJSON, _ := json.Marshal(model)
+	body := `{"model":` + string(modelJSON) + `,"instructions":"Reply with OK.","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"Reply with OK."}]}],"stream":true,"store":false,"parallel_tool_calls":true,"include":["reasoning.encrypted_content"]`
+	if tier != "" {
+		tierJSON, _ := json.Marshal(tier)
+		body += `,"service_tier":` + string(tierJSON)
+	}
+	return []byte(body + `}`)
+}
+
+func (s *CodexGatewayBorrowService) diagnoseProbeContract(ctx context.Context, revision uint64, revisionCtx context.Context, request CodexBorrowDiagnosticRequest, limit int, emit func(CodexBorrowDiagnosticEvent)) error {
+	var requests atomic.Int32
+	d := &codexBorrowDiagnostic{limit: int32(limit), requests: &requests, borrow: true, session: uuid.NewString(), serviceTier: request.ServiceTier,
+		onRequest: func(count int32) {
+			emit(CodexBorrowDiagnosticEvent{Type: "request", Mode: "borrowed", Requests: count, Limit: limit})
+		}}
+	ctx = context.WithValue(WithCodexGatewayBorrowObservation(ctx), codexBorrowDiagnosticContextKey{}, d)
+	emit(CodexBorrowDiagnosticEvent{Type: "phase", Mode: "borrowed", Limit: limit})
+	account, template, proxy, err := s.accountTemplate(ctx, request.AccountID, request.Model)
+	if err != nil {
+		return err
+	}
+	policyBody, err := s.gateway.applyOpenAIFastPolicyToBody(ctx, account, request.Model, borrowObservationPayload(request.Model, "Reply with OK.", "", request.ServiceTier))
+	if err != nil {
+		return err // A blocked local tier must not consume a source or probe request.
+	}
+	finalTier := gjson.GetBytes(policyBody, "service_tier").String()
+	s.mu.Lock()
+	candidate := s.candidate
+	current := s.candidateCurrentLocked(revision, candidate)
+	s.mu.Unlock()
+	if !current {
+		if err := s.prepareSource(ctx, revision, revisionCtx); err != nil {
+			return err
+		}
+		s.mu.Lock()
+		candidate = s.candidate
+		current = s.candidateCurrentLocked(revision, candidate)
+		s.mu.Unlock()
+	}
+	if !current {
+		return ErrCodexGatewayBorrowChanged
+	}
+	// The snapshot cannot be silently replaced halfway through the comparison.
+	ctx, cancel := context.WithDeadline(ctx, candidate.expires)
+	defer cancel()
+	d.prepareHTTPSessionTurn(template.Header, make(map[string]any))
+	credential, err := resolveCredentialAccount(ctx, s.accounts, account)
+	if err != nil {
+		return err
+	}
+	enforceCodexIdentityHeadersForAccount(template.Header, credential, s.gateway.codexIdentityOverrideUA(account))
+	setOpenAICodexRoutingHint(template.Header, account, request.Model, finalTier)
+	template = template.WithContext(withCodexBorrowServiceTier(ctx, finalTier))
+	identity := borrowRequestFingerprint(template, request.Model, proxy, nil, candidate.cookie.Value)
+	shapes := []string{codexBorrowProbeRanxiLiteral, codexBorrowProbeRanxiTier, codexBorrowProbeCurrent}
+	if finalTier == "" {
+		shapes = []string{codexBorrowProbeRanxiLiteral, codexBorrowProbeCurrent}
+	}
+	for _, shape := range shapes {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if requests.Load() >= int32(limit) {
+			return ErrCodexBorrowDiagnosticBudget
+		}
+		tier := finalTier
+		if shape == codexBorrowProbeRanxiLiteral {
+			tier = ""
+		}
+		operation := withCodexBorrowServiceTier(context.WithValue(ctx, codexBorrowProbeShapeContextKey{}, shape), tier)
+		variant := template.WithContext(operation)
+		started := time.Now()
+		verification := s.probeTarget(operation, variant, account, request.Model, proxy, nil, candidate)
+		body := borrowRanxiLiteralPayload(request.Model, tier)
+		if shape == codexBorrowProbeCurrent {
+			body = borrowObservationPayload(request.Model, "Reply with OK.", "", tier)
+		}
+		result := CodexBorrowDiagnosticResult{Scenario: "probe_contract", Mode: "borrowed", ProbeOnly: true,
+			Verification: &verification, DurationMS: time.Since(started).Milliseconds(), ReportedModel: verification.ReportedModel,
+			ProbeContext: &CodexBorrowProbeContext{SourceAccountID: candidate.sourceID, TargetProxyID: account.ProxyID,
+				CookieFingerprint: borrowHash(candidate.cookie.Value), CandidateExpiresAt: candidate.expires,
+				TemplateFingerprint: identity, RequestedTier: request.ServiceTier, FinalTier: finalTier, RoutingHint: template.Header.Get(openAICodexRoutingHintHeader),
+				PlaintextBodyFingerprint: borrowHash(string(body)), OriginalAstraModel: request.Model == codexGatewayBorrowSourceModel}}
+		if !verification.Success {
+			result.FailureStage, result.FailureReason, result.Error = "target_validation", verification.Reason, verification.Error
+		}
+		// No business send occurred. Completed/applied/dispatched remain false;
+		// the verification contains each probe's actual terminal and STATE facts.
+		emit(CodexBorrowDiagnosticEvent{Type: "result", Mode: "borrowed", Result: &result, Requests: requests.Load(), Limit: limit})
+		if errors.Is(ctx.Err(), context.Canceled) {
+			return ctx.Err()
+		}
+	}
+	emit(CodexBorrowDiagnosticEvent{Type: "done", Requests: requests.Load(), Limit: limit})
+	return ctx.Err()
+}

backend/internal/service/codex_gateway_borrow_diagnose_contract_test.go

@@ -0,0 +1,148 @@
+package service
+
+import (
+	"bytes"
+	"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 TestCodexBorrowProbeContractKeepsOneCandidateAndDoesNotQualifyBusiness(t *testing.T) {
+	for _, model := range []string{"gpt-6-astra", "gpt-6.1-sol"} {
+		t.Run(model, func(t *testing.T) {
+			account := borrowCoreAccount(2)
+			var bodies [][]byte
+			var headers []http.Header
+			s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+				assert.EqualValues(t, 2, id)
+				bodies = append(bodies, borrowCoreBody(t, req))
+				headers = append(headers, req.Header.Clone())
+				state := "minted-state"
+				if req.Header.Get("X-Codex-Turn-State") != "" {
+					state = "changed-state" // A failed verdict must not end the comparison.
+				}
+				return borrowCoreResponse(model, "OK", state, "__cflb=target-route; Secure; Path=/"), nil
+			}, account)
+			borrowCoreCandidate(s, time.Now().Add(time.Minute))
+			var results []*CodexBorrowDiagnosticResult
+			var used int32
+			err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{Scenario: "probe_contract", AccountID: 2, Model: model, Transport: "http", ServiceTier: "priority"}, func(e CodexBorrowDiagnosticEvent) {
+				used = e.Requests
+				if e.Result != nil {
+					results = append(results, e.Result)
+				}
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 3)
+			require.EqualValues(t, 6, used)
+			require.Len(t, bodies, 6)
+			for i, result := range results {
+				require.True(t, result.ProbeOnly)
+				require.False(t, result.Dispatched)
+				require.False(t, result.Applied)
+				require.False(t, result.Completed)
+				require.Equal(t, "target_state_changed", result.Verification.Reason)
+				require.True(t, result.Verification.MintCompleted && result.Verification.ContinueCompleted)
+				require.Equal(t, model == "gpt-6-astra", result.ProbeContext.OriginalAstraModel)
+				require.Equal(t, results[0].ProbeContext.TemplateFingerprint, result.ProbeContext.TemplateFingerprint)
+				require.Equal(t, results[0].ProbeContext.CookieFingerprint, result.ProbeContext.CookieFingerprint)
+				require.Equal(t, model, gjson.GetBytes(bodies[i*2], "model").String())
+				require.Equal(t, headers[0].Get("session-id"), headers[i*2].Get("session-id"))
+				require.NotEqual(t, headers[i*2].Get("session_id"), headers[i*2+1].Get("session_id"))
+				require.Equal(t, "minted-state", headers[i*2+1].Get("X-Codex-Turn-State"))
+				require.Contains(t, headers[i*2].Get("Cookie"), "__oailb=synthetic-borrowed-cookie")
+				require.Contains(t, headers[i*2+1].Get("Cookie"), "__cflb=target-route")
+			}
+			require.True(t, bytes.HasPrefix(bodies[0], []byte(`{"model":`)))
+			require.False(t, gjson.GetBytes(bodies[0], "service_tier").Exists())
+			require.True(t, bytes.HasPrefix(bodies[2], []byte(`{"model":`)))
+			require.Equal(t, "priority", gjson.GetBytes(bodies[2], "service_tier").String())
+			require.JSONEq(t, string(bodies[2]), string(bodies[4]))
+			require.NotEqual(t, string(bodies[2]), string(bodies[4]), "the final comparison changes field order only")
+			require.Empty(t, s.targets)
+			require.Empty(t, s.qualifications)
+			require.Empty(t, s.Status().RecentUsage, "probe-only results cannot replace real request evidence")
+		})
+	}
+}
+
+func TestCodexBorrowProbeContractRequestBudgetAndCancellation(t *testing.T) {
+	for _, cancelled := range []bool{false, true} {
+		account := borrowCoreAccount(2)
+		var calls int
+		ctx, cancel := context.WithCancel(context.Background())
+		s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+			calls++
+			if cancelled {
+				cancel()
+			}
+			return borrowCoreResponse("gpt-6-astra", "OK", "state"), nil
+		}, account)
+		borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+		err := s.Diagnose(ctx, CodexBorrowDiagnosticRequest{Scenario: "probe_contract", AccountID: 2, Model: "gpt-6-astra", Transport: "http", RequestLimit: 1}, func(CodexBorrowDiagnosticEvent) {})
+		cancel()
+		require.Error(t, err)
+		require.Equal(t, 1, calls)
+		require.Empty(t, s.targets)
+	}
+}
+
+func TestCodexBorrowProbeContractStopsAtFrozenCandidateExpiry(t *testing.T) {
+	var sources, targets int
+	var s *CodexGatewayBorrowService
+	s = newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+		if id == 1 {
+			sources++
+			return borrowCoreResponse("gpt-6-astra", "OK", "source-state", "__oailb=short-candidate; Secure; Path=/; Max-Age=1"), nil
+		}
+		targets++
+		deadline, ok := req.Context().Deadline()
+		assert.True(t, ok)
+		assert.True(t, deadline.Equal(s.candidate.expires), "a comparison cannot outlive its fixed cookie")
+		<-req.Context().Done()
+		return nil, req.Context().Err()
+	}, borrowCoreAccount(1), borrowCoreAccount(2))
+	err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{Scenario: "probe_contract", AccountID: 2, Model: "gpt-6-astra", Transport: "http"}, func(CodexBorrowDiagnosticEvent) {})
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	require.Equal(t, 1, sources)
+	require.Equal(t, 1, targets)
+	require.Empty(t, s.qualifications)
+}
+
+func TestCodexBorrowProbeContractRespectsTierPolicyBeforeModelCalls(t *testing.T) {
+	for _, action := range []string{BetaPolicyActionBlock, BetaPolicyActionFilter} {
+		t.Run(action, func(t *testing.T) {
+			var calls int
+			s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+				calls++
+				assert.False(t, gjson.GetBytes(borrowCoreBody(t, req), "service_tier").Exists())
+				return borrowCoreResponse("gpt-6-astra", "OK", "state"), nil
+			}, borrowCoreAccount(2))
+			policy := openAIFastFilterPriorityPolicy()
+			policy.Rules[0].Action = action
+			s.gateway.settingService = newOpenAIGatewayServiceWithSettings(t, policy).settingService
+			if action == BetaPolicyActionFilter {
+				borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+			}
+			err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{Scenario: "probe_contract", AccountID: 2, Model: "gpt-6-astra", Transport: "http", ServiceTier: "priority"}, func(e CodexBorrowDiagnosticEvent) {
+				if e.Result != nil {
+					assert.Empty(t, e.Result.ProbeContext.FinalTier)
+				}
+			})
+			if action == BetaPolicyActionBlock {
+				var blocked *OpenAIFastBlockedError
+				require.ErrorAs(t, err, &blocked)
+				require.Zero(t, calls)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, 4, calls, "a filtered tier needs no redundant middle comparison")
+			}
+		})
+	}
+}

backend/internal/service/codex_gateway_borrow_probe.go

@@ -208,6 +208,9 @@ 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: codexBorrowRequestServiceTier(template)}
+	if shape, _ := ctx.Value(codexBorrowProbeShapeContextKey{}).(string); shape != "" {
+		result.RequestShape = shape
+	}
 	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.
@@ -335,7 +338,11 @@ 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)))
 	tier := codexBorrowRequestServiceTier((&http.Request{Header: headers}).WithContext(ctx))
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(borrowObservationPayload(model, prompt, effort, tier)))
+	body := borrowObservationPayload(model, prompt, effort, tier)
+	if shape, _ := ctx.Value(codexBorrowProbeShapeContextKey{}).(string); shape == codexBorrowProbeRanxiLiteral || shape == codexBorrowProbeRanxiTier {
+		body = borrowRanxiLiteralPayload(model, tier)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(body))
 	if err != nil {
 		return nil, err
 	}