diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 927094699dbc..1f74540d1c04 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -100,6 +100,15 @@ and has a 90-second budget. Each target/model is validated with two completed HTTP 200 streams, at most 45 seconds each: the first must return STATE, and the second may omit STATE or return the same value. Qualification includes the target identity, actual model, exit, client headers, STATE and TLS profile. +Target probes preserve the template's native session/thread/window headers and +generate only a fresh legacy `session_id` for each shot, matching ranxi's full +probe contract. The effective service tier in the gateway-owned routing hint is +also included in the probe body. Proofs include native identity and routing +headers; per-request tracing IDs do not invalidate them. Manual templates use a +stable synthetic window per account/model so cached-only generation can reuse +their proof without changing actual client windows. The pinned upstream's +HTTP borrowing is Astra-only: Sol is a downstream extension, independently +validated against Sol rather than certified by an Astra pass. Source and target probes use separate HTTP/TLS pools. Business WS connections stay in the ordinary account pool, with fixed one-hour continuation anchors. @@ -125,6 +134,13 @@ cancellation; that detail remains available as `read_error`. Captures are snapsh under a lock, and the usage index preserves the first terminal event. Diagnostic and business requests are distinguished. This index contains no request payloads or credential/cookie values; the existing administrator error records remain. +Attempts now start before preparation: `attempt_count` and `blocked_count` are +separate from existing `count` (dispatches) and `applied_count`. `dispatched`, +`failure_stage`, client and gateway request IDs distinguish a local rejection +from an inference failure. Original target reasons survive failure cooldowns; +Ops classifies local borrowing rejection as gateway/routing with upstream status +zero, retaining the compatible client 503 code. Business evidence is displayed +ahead of diagnostics. Candidate expiry never implies a failed check is usable. `POST /admin/codex-gateway-borrow/diagnose` streams an explicit fixed-account comparison with `account_id`, `model`, `transport` (`http` or `ws`), and optional @@ -138,6 +154,15 @@ completion, actual borrow application and reported model. Cancellation releases owned work and sockets. The UI retains the current run and can download JSON; there is no new table or migration. These observations do not prove intelligence or physical model identity. Normal token refresh remains permitted. +Optional `scenario: codex_session` checks HTTP native client metadata, a fixed +side-effect-free echo tool call, returned STATE and complete-history continuation. +It defaults to the borrowed path to fit preparation and two business turns within +the same eight-request bound. Optional `mode` selects `ordinary` or `borrowed`; +omitting it for the original short scenario retains both-path comparison. +Optional `service_tier` exercises the effective routing tier. Results include the +specific verification (shape, actual model/tier, both completions, STATE lengths +and route replacement), failure stage and tool-round-trip result. HTTP diagnostic +history and STATE live only within that invocation, never in client sessions. Ranxi reference: v2.10.2 (`d3e43f2de33af9e987cffa511d76dfabbcd749da`). Its five fingerprint, gateway-cookie, target-probe and automatic-setup core files are diff --git a/backend/internal/handler/admin/codex_gateway_borrow_handler.go b/backend/internal/handler/admin/codex_gateway_borrow_handler.go index 5df1d9da04ad..86133e954b25 100644 --- a/backend/internal/handler/admin/codex_gateway_borrow_handler.go +++ b/backend/internal/handler/admin/codex_gateway_borrow_handler.go @@ -277,7 +277,7 @@ func (h *CodexGatewayBorrowHandler) Diagnose(c *gin.Context) { response.BadRequest(c, err.Error()) return } - middleware.SetAuditExtra(c, map[string]any{"result": "codex_gateway_borrow_diagnostic", "account_id": request.AccountID, "model": request.Model, "transport": request.Transport}) + middleware.SetAuditExtra(c, map[string]any{"result": "codex_gateway_borrow_diagnostic", "account_id": request.AccountID, "model": request.Model, "transport": request.Transport, "scenario": request.Scenario, "mode": request.Mode, "service_tier": request.ServiceTier}) c.Header("Content-Type", "text/event-stream") c.Header("X-Accel-Buffering", "no") c.Status(http.StatusOK) diff --git a/backend/internal/handler/ops_codex_borrow_test.go b/backend/internal/handler/ops_codex_borrow_test.go new file mode 100644 index 000000000000..49b9c96355d0 --- /dev/null +++ b/backend/internal/handler/ops_codex_borrow_test.go @@ -0,0 +1,29 @@ +package handler + +import ( + "net/http/httptest" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestOpsCodexBorrowPreparationIsNotAnUpstream503(t *testing.T) { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest("POST", "/v1/responses", nil) + err := &service.CodexGatewayBorrowFailure{Cause: service.ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: "target_state_changed", Detail: "full diagnostic detail"} + service.RecordCodexGatewayBorrowPreparationFailure(c, &service.Account{ID: 2, Platform: "openai"}, err) + phase, limited, owner, source := classifyOpsErrorLog(c, "upstream_error", "no qualified Codex gateway borrow route is available", "CODEX_GATEWAY_BORROW_UNAVAILABLE", 503) + require.Equal(t, "routing", phase) + require.False(t, limited, "a platform failure still counts as a failed client request") + require.Equal(t, "platform", owner) + require.Equal(t, "gateway", source) + entry := &service.OpsInsertErrorLogInput{} + applyOpsUpstreamFieldsFromContext(c, entry) + require.NotNil(t, entry.UpstreamStatusCode) + require.Zero(t, *entry.UpstreamStatusCode) + require.Len(t, entry.UpstreamErrors, 1) + require.Equal(t, "target_state_changed", entry.UpstreamErrors[0].Reason) + require.Contains(t, *entry.UpstreamErrorDetail, "full diagnostic detail") +} diff --git a/backend/internal/handler/ops_error_logger.go b/backend/internal/handler/ops_error_logger.go index 7a922acc62a5..4202f020e2df 100644 --- a/backend/internal/handler/ops_error_logger.go +++ b/backend/internal/handler/ops_error_logger.go @@ -1796,7 +1796,7 @@ func applyOpsUpstreamErrorEvents(entry *service.OpsInsertErrorLogInput, events [ entry.UpstreamStatusCode = nil entry.UpstreamErrorMessage = nil entry.UpstreamErrorDetail = nil - if last.Stage == string(service.GatewayFailureStageAccountAuth) { + if last.Stage == string(service.GatewayFailureStageAccountAuth) || last.Kind == string(service.CodexGatewayBorrowPreparationFailureReason) { code := 0 entry.UpstreamStatusCode = &code } else if last.UpstreamStatusCode > 0 { @@ -2272,6 +2272,9 @@ func normalizeOpsErrorType(errType string, code string) string { } func classifyOpsPhase(errType, message, code string) string { + if code == "CODEX_GATEWAY_BORROW_UNAVAILABLE" { + return "routing" + } msg := strings.ToLower(message) // Standardized phases: request|auth|account_auth|routing|upstream|network|internal // Map billing/concurrency/response => request; scheduling => routing. @@ -2324,6 +2327,11 @@ func classifyOpsSeverity(errType string, status int) string { } func classifyOpsErrorLog(c *gin.Context, errType, message, code string, status int) (phase string, isBusinessLimited bool, errorOwner string, errorSource string) { + if code == "CODEX_GATEWAY_BORROW_UNAVAILABLE" { + // A local qualification rejection is a real client failure, but no + // inference request was dispatched and no upstream HTTP 503 was seen. + return "routing", false, "platform", "gateway" + } phase = classifyOpsPhase(errType, message, code) routingCapacityLimited := isOpsRoutingCapacityLimited(c) clientBusinessLimited := service.HasOpsClientBusinessLimited(c) diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index f8e23c4f12a8..176e1b8b427a 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -46,13 +46,27 @@ var ( // A local route qualification failure is distinct from an actual business // request failing upstream. Callers must surface it before ordinary account // failover, stream-timeout or health hooks. Unwrap preserves the specific cause. -type CodexGatewayBorrowFailure struct{ Cause error } +type CodexGatewayBorrowFailure struct { + Cause error + Stage string + Reason string + Detail string + RetryAfter *time.Time + Verification *CodexGatewayBorrowVerification +} func (e *CodexGatewayBorrowFailure) Error() string { if e == nil || e.Cause == nil { return "Codex gateway borrow qualification failed" } - return "Codex gateway borrow qualification failed: " + e.Cause.Error() + detail := e.Cause.Error() + if e.Detail != "" { + detail += ": " + e.Detail + } + if e.Reason != "" { + detail = e.Stage + "/" + e.Reason + ": " + detail + } + return "Codex gateway borrow qualification failed: " + detail } func (e *CodexGatewayBorrowFailure) Unwrap() error { if e == nil { @@ -65,6 +79,33 @@ func IsCodexGatewayBorrowFailure(err error) bool { return errors.As(err, &failure) } +func borrowVerificationFailure(check codexGatewayBorrowTargetCheck) error { + result := check.result + failure := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: result.Reason, Detail: result.Error, Verification: &result} + if !check.retryAfter.IsZero() { + retry := check.retryAfter + failure.RetryAfter = &retry + } + return failure +} + +func codexBorrowFailureStage(err error) (string, string) { + var failure *CodexGatewayBorrowFailure + if errors.As(err, &failure) && failure.Reason != "" { + return failure.Stage, failure.Reason + } + if errors.Is(err, ErrCodexGatewayBorrowChanged) { + return "qualification", "configuration_changed" + } + if errors.Is(err, context.Canceled) { + return "preparation", "cancelled" + } + if errors.Is(err, context.DeadlineExceeded) { + return "preparation", "deadline_exceeded" + } + return "preparation", "borrow_unavailable" +} + // CodexGatewayBorrowProbeUpstream is deliberately a distinct injection type. // Its implementation must own separate clients, pools and TLS caches, including // separation between source acquisition and target observation. @@ -155,18 +196,25 @@ type CodexGatewayBorrowSourceStatus struct { } type CodexGatewayBorrowVerification struct { - AccountID int64 `json:"account_id"` - Model string `json:"model"` - Success bool `json:"success"` - Reason string `json:"reason"` - Error string `json:"error,omitempty"` - MintStatus int `json:"mint_status"` - ContinueStatus int `json:"continue_status"` - Minted bool `json:"minted"` - NewTicket bool `json:"new_ticket"` - ReportedModel string `json:"reported_model,omitempty"` - CheckedAt time.Time `json:"checked_at"` - ExpiresAt *time.Time `json:"expires_at,omitempty"` + RequestShape string `json:"request_shape,omitempty"` + ServiceTier string `json:"service_tier,omitempty"` + MintCompleted bool `json:"mint_completed"` + ContinueCompleted bool `json:"continue_completed"` + MintStateLength int `json:"mint_state_length"` + ContinueStateLength int `json:"continue_state_length"` + RouteChanged bool `json:"route_changed"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Success bool `json:"success"` + Reason string `json:"reason"` + Error string `json:"error,omitempty"` + MintStatus int `json:"mint_status"` + ContinueStatus int `json:"continue_status"` + Minted bool `json:"minted"` + NewTicket bool `json:"new_ticket"` + ReportedModel string `json:"reported_model,omitempty"` + CheckedAt time.Time `json:"checked_at"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` } type CodexGatewayBorrowTargetStatus struct { @@ -195,6 +243,7 @@ type CodexGatewayBorrowStatus struct { // Application contains only opaque identities. The selected cookie itself is // available solely in the cloned outbound request's Cookie header. type CodexGatewayBorrowApplication struct { + Verification CodexGatewayBorrowVerification Applied bool SourceAccountID int64 ExpiresAt time.Time @@ -708,7 +757,14 @@ 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) - identity, _ := json.Marshal([]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 := []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)} + // 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. + for _, name := range []string{"session-id", "thread-id", "x-codex-window-id", "x-codex-installation-id", "x-codex-beta-features", openAICodexRoutingHintHeader} { + values = append(values, req.Header.Get(name)) + } + identity, _ := json.Marshal(values) return borrowHash(string(identity)) } @@ -724,6 +780,21 @@ func (s *CodexGatewayBorrowService) Apply(req *http.Request, account *Account, m if err != nil && !IsCodexGatewayBorrowFailure(err) { err = &CodexGatewayBorrowFailure{Cause: err} } + var diagnostic *codexBorrowDiagnostic + if req != nil { + diagnostic = borrowDiagnosticFromContext(req.Context()) + } + if d := diagnostic; d != nil { + if application != nil { + result := application.Verification + d.verification.Store(&result) + } else { + var failure *CodexGatewayBorrowFailure + if errors.As(err, &failure) { + d.verification.Store(failure.Verification) + } + } + } return wire, application, err } @@ -750,7 +821,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m return req, nil, ErrCodexGatewayBorrowUnavailable } if err := s.prepareSource(req.Context(), rev, revisionCtx); err != nil { - return req, nil, err + return req, nil, &CodexGatewayBorrowFailure{Cause: err, Stage: "source_preparation", Reason: "source_prepare_failed"} } s.mu.Lock() candidate = s.candidate @@ -768,7 +839,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m if err != nil { return req, nil, err } - return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires), nil + return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires, check.result), nil } } if cachedOnly { @@ -843,7 +914,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m s.qualifications[qualificationKey] = check s.mu.Unlock() if !result.Success { - return nil, ErrCodexGatewayBorrowUnavailable + return nil, borrowVerificationFailure(check) } return check, nil }) @@ -854,7 +925,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m if !ok { return req, nil, ErrCodexGatewayBorrowUnavailable } - return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires), nil + return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires, check.result), nil } func (s *CodexGatewayBorrowService) candidateCurrentLocked(rev uint64, candidate *codexGatewayBorrowCandidate) bool { @@ -882,13 +953,17 @@ func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBor return old, true, nil } if !old.result.Success && now.Before(old.retryAfter) { - return old, false, ErrCodexGatewayBorrowUnavailable + return old, false, borrowVerificationFailure(old) } return old, false, nil } -func borrowApplication(candidate *codexGatewayBorrowCandidate, fingerprint string, expires time.Time) *CodexGatewayBorrowApplication { - return &CodexGatewayBorrowApplication{Applied: true, SourceAccountID: candidate.sourceID, ExpiresAt: expires, Fingerprint: fingerprint, CookieFingerprint: borrowHash(candidate.cookie.Value)} +func borrowApplication(candidate *codexGatewayBorrowCandidate, fingerprint string, expires time.Time, verification ...CodexGatewayBorrowVerification) *CodexGatewayBorrowApplication { + app := &CodexGatewayBorrowApplication{Applied: true, SourceAccountID: candidate.sourceID, ExpiresAt: expires, Fingerprint: fingerprint, CookieFingerprint: borrowHash(candidate.cookie.Value)} + if len(verification) > 0 { + app.Verification = verification[0] + } + return app } func borrowApplyCookie(req *http.Request, candidate *codexGatewayBorrowCandidate, check codexGatewayBorrowTargetCheck) *http.Request { diff --git a/backend/internal/service/codex_gateway_borrow_diagnose.go b/backend/internal/service/codex_gateway_borrow_diagnose.go index 633831155402..6eb5c5ac3a19 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose.go @@ -23,13 +23,18 @@ type codexBorrowDiagnosticContextKey struct{} type codexBorrowAppliedContextKey struct{} type codexBorrowDiagnostic struct { - limit int32 - onRequest func(int32) - requests *atomic.Int32 - borrow bool - transport string - session string - anchors codexGatewayBorrowWSAnchorStore + scenario string + serviceTier string + history []any + turnState string + verification atomic.Pointer[CodexGatewayBorrowVerification] + limit int32 + onRequest func(int32) + requests *atomic.Int32 + borrow bool + transport string + session string + anchors codexGatewayBorrowWSAnchorStore } func borrowDiagnosticFromContext(ctx context.Context) *codexBorrowDiagnostic { @@ -67,6 +72,9 @@ func consumeBorrowDiagnosticRequest(ctx context.Context) error { } type CodexBorrowDiagnosticRequest struct { + Scenario string `json:"scenario,omitempty"` + Mode string `json:"mode,omitempty"` + ServiceTier string `json:"service_tier,omitempty"` RequestLimit int `json:"request_limit,omitempty"` AccountID int64 `json:"account_id"` Model string `json:"model"` @@ -74,18 +82,24 @@ type CodexBorrowDiagnosticRequest struct { } type CodexBorrowDiagnosticResult struct { - ReadError string `json:"read_error,omitempty"` - Dispatched bool `json:"dispatched"` - Mode string `json:"mode"` - Turn int `json:"turn"` - Applied bool `json:"applied"` - Completed bool `json:"completed"` - ResponseID string `json:"response_id,omitempty"` - ReportedModel string `json:"reported_model,omitempty"` - Answer string `json:"answer"` - RawResponse string `json:"raw_response"` - Error string `json:"error,omitempty"` - DurationMS int64 `json:"duration_ms"` + Scenario string `json:"scenario,omitempty"` + FailureStage string `json:"failure_stage,omitempty"` + FailureReason string `json:"failure_reason,omitempty"` + Verification *CodexGatewayBorrowVerification `json:"verification,omitempty"` + ToolRoundTrip bool `json:"tool_round_trip,omitempty"` + StateLength int `json:"state_length,omitempty"` + ReadError string `json:"read_error,omitempty"` + Dispatched bool `json:"dispatched"` + Mode string `json:"mode"` + Turn int `json:"turn"` + Applied bool `json:"applied"` + Completed bool `json:"completed"` + ResponseID string `json:"response_id,omitempty"` + ReportedModel string `json:"reported_model,omitempty"` + Answer string `json:"answer"` + RawResponse string `json:"raw_response"` + Error string `json:"error,omitempty"` + DurationMS int64 `json:"duration_ms"` } type CodexBorrowDiagnosticEvent struct { @@ -100,6 +114,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.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.ServiceTier != "" && request.ServiceTier != "default" && request.ServiceTier != OpenAIFastTierPriority && request.ServiceTier != OpenAIFastTierFlex && request.ServiceTier != OpenAIFastTierUltrafast { + return errors.New("unsupported service_tier") + } if request.AccountID <= 0 || (request.Transport != "http" && request.Transport != "ws") { return errors.New("account_id and transport (http or ws) are required") } @@ -130,7 +156,13 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB ctx, cancel := borrowRevisionContext(WithAccountObservation(ctx), revisionCtx, 10*time.Minute) defer cancel() var requests atomic.Int32 - for _, borrowed := range []bool{false, true} { + modes := []bool{false, true} + if request.Mode == "borrowed" || (request.Scenario == "codex_session" && request.Mode == "") { + modes = []bool{true} + } else if request.Mode == "ordinary" { + modes = []bool{false} + } + for _, borrowed := range modes { if err := ctx.Err(); err != nil { return err } @@ -140,12 +172,12 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB } d := &codexBorrowDiagnostic{limit: int32(limit), onRequest: func(count int32) { emit(CodexBorrowDiagnosticEvent{Type: "request", Mode: mode, Requests: count, Limit: limit}) - }, requests: &requests, borrow: borrowed, transport: request.Transport, session: uuid.NewString()} + }, requests: &requests, borrow: borrowed, transport: request.Transport, session: uuid.NewString(), scenario: request.Scenario, serviceTier: request.ServiceTier} operation := context.WithValue(ctx, codexBorrowDiagnosticContextKey{}, d) emit(CodexBorrowDiagnosticEvent{Type: "phase", Mode: mode, Requests: requests.Load(), Limit: limit}) previous := "" turns := 1 - if request.Transport == "ws" { + if request.Transport == "ws" || request.Scenario == "codex_session" { turns = 2 } for turn := 1; turn <= turns; turn++ { @@ -189,6 +221,7 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A capture.ginContext = c c.Request = httptest.NewRequest(http.MethodPost, "/internal/admin/codex-diagnostic", nil).WithContext(ctx) d := borrowDiagnosticFromContext(ctx) + d.verification.Store(nil) c.Request.Header.Set("session_id", d.session) if d.transport == "ws" { SetOpenAIClientTransport(c, OpenAIClientTransportWS) @@ -202,8 +235,23 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A if levels, _ := AccountTestReasoningOptions(account, model); slices.Contains(levels, "low") { payload["reasoning"] = map[string]string{"effort": "low"} } + if d.serviceTier != "" { + payload["service_tier"] = d.serviceTier + } + if d.scenario == "codex_session" { + d.prepareHTTPSessionTurn(c.Request.Header, payload) + } body, _ := json.Marshal(payload) forward, sendErr := s.gateway.Forward(ctx, c, account, body) + result.Scenario, result.Verification = d.scenario, d.verification.Load() + if sendErr != nil { + result.FailureStage, result.FailureReason = codexBorrowFailureStage(sendErr) + if result.Verification != nil && !result.Verification.Success { + result.FailureStage, result.FailureReason = "target_validation", result.Verification.Reason + } else if !IsCodexGatewayBorrowRequestFailure(sendErr) { + result.FailureStage, result.FailureReason = "upstream", "request_failed" + } + } result.DurationMS = time.Since(start).Milliseconds() if forward != nil { result.ResponseID = firstNonEmpty(forward.ResponseID, forward.RequestID) @@ -224,6 +272,15 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A result.ReadError = last.err.Error() } result.RawResponse = last.raw.String() + if d.scenario == "codex_session" && last.dispatched { + state.mu.Lock() + result.Applied = state.invocation.BorrowApplied + state.mu.Unlock() + result.StateLength = len(extractOpenAICodexTurnState(w.Header())) + d.turnState = extractOpenAICodexTurnState(w.Header()) + d.completeHTTPSessionTurn(&result, sendErr, last.status) + return result + } if last.wsFrames && last.raw.Len() == 0 && sendErr != nil { var eventErr *openAIWSUpstreamEventError if errors.As(sendErr, &eventErr) && len(eventErr.payload) > 0 { diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session.go b/backend/internal/service/codex_gateway_borrow_diagnose_session.go new file mode 100644 index 000000000000..a82850d1c3e2 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session.go @@ -0,0 +1,112 @@ +package service + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/google/uuid" + "github.com/tidwall/gjson" +) + +const codexBorrowEchoTool = "codex_borrow_echo" +const codexBorrowEchoValue = "borrow-continuation-check" + +// A fixed, side-effect-free tool round trip exercises native client metadata, +// STATE echo and full-history HTTP continuation without reading a user's chat. +// Each diagnostic owns its history; nothing enters business continuation stores. +func (d *codexBorrowDiagnostic) prepareHTTPSessionTurn(headers http.Header, payload map[string]any) { + headers.Set("User-Agent", CodexCanonicalUserAgent()) + headers.Set("originator", codexTUIOriginator) + headers.Set("session-id", d.session) + headers.Set("thread-id", d.session) + headers.Set("x-client-request-id", uuid.NewString()) + headers.Set("x-codex-window-id", d.session+":0") + headers.Set("x-codex-installation-id", d.session) + metadata, _ := json.Marshal(map[string]string{"installation_id": d.session, "session_id": d.session, "thread_id": d.session, "turn_id": uuid.NewString(), "window_id": d.session + ":0"}) + headers.Set("x-codex-turn-metadata", string(metadata)) + if d.turnState != "" { + headers.Set("x-codex-turn-state", d.turnState) + } + payload["client_metadata"] = map[string]string{"session_id": d.session, "thread_id": d.session, "x-codex-installation-id": d.session, "x-codex-turn-metadata": string(metadata)} + payload["prompt_cache_key"] = d.session + payload["include"] = []string{"reasoning.encrypted_content"} + payload["parallel_tool_calls"] = false + payload["tools"] = []any{map[string]any{"type": "function", "name": codexBorrowEchoTool, "description": "Return value unchanged.", "strict": true, "parameters": map[string]any{"type": "object", "properties": map[string]any{"value": map[string]string{"type": "string"}}, "required": []string{"value"}, "additionalProperties": false}}} + delete(payload, "previous_response_id") // HTTP sends complete history, as the native client does. + if len(d.history) == 0 { + d.history = []any{map[string]any{"role": "user", "content": []any{map[string]string{"type": "input_text", "text": "Call codex_borrow_echo with value borrow-continuation-check. After its result, reply with that value only."}}}} + payload["tool_choice"] = map[string]string{"type": "function", "name": codexBorrowEchoTool} + } else { + payload["tool_choice"] = "none" + } + payload["input"] = d.history +} + +func codexBorrowSessionTerminal(raw string) (gjson.Result, error) { + for _, line := range strings.Split(raw, "\n") { + if !strings.HasPrefix(line, "data:") { + continue + } + event := gjson.Parse(strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + switch event.Get("type").String() { + case "response.completed", "response.done": + response := event.Get("response") + if response.Get("status").String() == "completed" && !response.Get("error").IsObject() { + return response, nil + } + return response, errors.New("upstream terminal was not completed") + case "error", "response.failed", "response.incomplete", "response.cancelled", "response.canceled": + return gjson.Result{}, errors.New(event.Raw) + } + } + return gjson.Result{}, errors.New("upstream SSE ended without a completed response") +} + +func (d *codexBorrowDiagnostic) completeHTTPSessionTurn(result *CodexBorrowDiagnosticResult, sendErr error, status int) { + response, err := codexBorrowSessionTerminal(result.RawResponse) + if sendErr != nil { + result.Error = sendErr.Error() + return + } + if err != nil { + result.Error = err.Error() + return + } + if status < 200 || status >= 300 { + result.Error = "upstream HTTP request did not succeed" + return + } + result.ResponseID, result.ReportedModel = response.Get("id").String(), response.Get("model").String() + if len(d.history) == 1 { + var callID, arguments string + for _, item := range response.Get("output").Array() { + d.history = append(d.history, json.RawMessage(item.Raw)) + if item.Get("type").String() == "function_call" && item.Get("name").String() == codexBorrowEchoTool { + callID, arguments = item.Get("call_id").String(), item.Get("arguments").String() + } + } + if callID == "" || gjson.Get(arguments, "value").String() != codexBorrowEchoValue { + result.Error = "upstream did not return the requested fixed echo tool call" + return + } + d.history = append(d.history, map[string]string{"type": "function_call_output", "call_id": callID, "output": codexBorrowEchoValue}) + result.Completed, result.Answer = true, "codex_borrow_echo(\""+codexBorrowEchoValue+"\")" + return + } + var answer strings.Builder + for _, item := range response.Get("output").Array() { + for _, content := range item.Get("content").Array() { + if content.Get("type").String() == "output_text" { + _, _ = answer.WriteString(content.Get("text").String()) + } + } + } + result.Answer = strings.TrimSpace(answer.String()) + result.Completed = result.Answer == codexBorrowEchoValue + result.ToolRoundTrip = result.Completed + if !result.Completed { + result.Error = "continuation did not return the fixed tool result" + } +} diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_test.go index 6d7aca0fd5d6..4df12b3c9bb8 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_test.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_test.go @@ -12,7 +12,9 @@ import ( "time" "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" ) // A real HTTP stream can deliver its final bytes together with cancellation @@ -171,3 +173,71 @@ func TestCodexBorrowDiagnoseWSContinuesOwnSockets(t *testing.T) { testCodexBorro func TestCodexBorrowDiagnoseWSCountsConfiguredPrewarm(t *testing.T) { testCodexBorrowDiagnoseWS(t, true) } + +func TestCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T) { + account := borrowCoreAccount(2) + generator, repo := newPelicanGeneratorForTest(account, &pelicanGeneratorUpstream{}) + gateway := generator.openaiGatewayService + var business, probes int + gateway.httpUpstream = borrowCoreProbe(func(req *http.Request, _ string, id int64, n int, _ *tlsfingerprint.Profile) (*http.Response, error) { + 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.NotEmpty(t, req.Header.Get("session-id")) + assert.Contains(t, req.Header.Get("Cookie"), "__oailb=synthetic-borrowed-cookie") + var output []any + if business == 1 { + assert.Empty(t, req.Header.Get("X-Codex-Turn-State")) + output = []any{map[string]any{"type": "function_call", "name": codexBorrowEchoTool, "call_id": "echo-call", "arguments": `{"value":"borrow-continuation-check"}`}} + } else { + assert.Equal(t, "business-turn-state", req.Header.Get("X-Codex-Turn-State")) + assert.Equal(t, "none", gjson.GetBytes(body, "tool_choice").String()) + assert.False(t, gjson.GetBytes(body, "previous_response_id").Exists()) + assert.Equal(t, "function_call_output", gjson.GetBytes(body, "input.2.type").String()) + assert.NotEmpty(t, gjson.GetBytes(body, "input.1.call_id").String()) + assert.Equal(t, gjson.GetBytes(body, "input.1.call_id").String(), gjson.GetBytes(body, "input.2.call_id").String()) + output = []any{map[string]any{"type": "message", "role": "assistant", "content": []any{map[string]string{"type": "output_text", "text": codexBorrowEchoValue}}}} + } + raw, _ := json.Marshal(map[string]any{"type": "response.completed", "response": map[string]any{"id": "response-test", "model": "gpt-6.1-sol", "status": "completed", "output": output, "usage": map[string]int{"input_tokens": 5, "output_tokens": 3}}}) + _, finish, err := PreparePelicanHTTPRequest(req, id, n, "http") + if err != nil { + return nil, err + } + headers := http.Header{"Content-Type": {"text/event-stream"}} + headers.Set("X-Codex-Turn-State", "business-turn-state") + return finish(&http.Response{StatusCode: 200, Header: headers, Body: io.NopCloser(strings.NewReader("data: " + string(raw) + "\n\n"))}, nil) + }) + 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()) + 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) { + if e.Result != nil { + results = append(results, *e.Result) + } + count = e.Requests + }) + require.NoError(t, err) + require.Len(t, results, 2) + for _, r := range results { + require.True(t, r.Completed, r.Error) + require.True(t, r.Applied) + require.True(t, r.Dispatched) + require.NotNil(t, r.Verification) + } + require.True(t, results[1].ToolRoundTrip) + require.Equal(t, 2, business) + require.EqualValues(t, business+probes, count) + require.LessOrEqual(t, count, int32(8)) + usage := s.Status().RecentUsage[0] + require.EqualValues(t, 2, usage.AttemptCount) + require.EqualValues(t, 2, usage.Count) + require.Zero(t, usage.BlockedCount) +} diff --git a/backend/internal/service/codex_gateway_borrow_http.go b/backend/internal/service/codex_gateway_borrow_http.go index 1ea93ee99e20..c854477b69df 100644 --- a/backend/internal/service/codex_gateway_borrow_http.go +++ b/backend/internal/service/codex_gateway_borrow_http.go @@ -47,14 +47,15 @@ func RecordCodexGatewayBorrowPreparationFailure(c *gin.Context, account *Account return } detail := err.Error() - setOpsUpstreamError(c, http.StatusServiceUnavailable, detail, detail) + stage, reason := codexBorrowFailureStage(err) + setOpsUpstreamError(c, 0, detail, detail) event := OpsUpstreamErrorEvent{ - UpstreamStatusCode: http.StatusServiceUnavailable, - Kind: string(CodexGatewayBorrowPreparationFailureReason), - Scope: string(GatewayFailureScopeRequest), - Reason: string(CodexGatewayBorrowPreparationFailureReason), - Message: detail, - Detail: detail, + Kind: string(CodexGatewayBorrowPreparationFailureReason), + Scope: string(GatewayFailureScopeRequest), + Stage: stage, + Reason: reason, + Message: detail, + Detail: detail, } if account != nil { event.ProxyID, event.ProxyName = opsUpstreamProxyID(account), opsUpstreamProxyName(account) @@ -98,16 +99,21 @@ func (s *OpenAIGatewayService) prepareCodexGatewayBorrowHTTP(ctx context.Context return req, 0, nil } started := time.Now() + tracker := s.gatewayBorrow.beginAttempt(ctx, account.ID, model, "http") + req = req.WithContext(context.WithValue(req.Context(), codexBorrowUsageContextKey{}, tracker)) prepared := codexGatewayBorrowHTTPPreparation{service: s.gatewayBorrow, accountID: account.ID, model: model, proxy: proxy} prepared.service.mu.Lock() prepared.revision = prepared.service.revision prepared.service.mu.Unlock() borrowed, application, err := prepared.service.Apply(req.WithContext(ctx), account, model, proxy, nil, false) if err != nil { + tracker.blocked(err) return nil, 0, err } if application == nil || !application.Applied { - return nil, 0, &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowChanged} + failure := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowChanged} + tracker.blocked(failure) + return nil, 0, failure } prepared.application = application req = borrowed.WithContext(context.WithValue(req.Context(), codexGatewayBorrowHTTPPreparationContextKey{}, prepared)) @@ -117,12 +123,23 @@ func (s *OpenAIGatewayService) prepareCodexGatewayBorrowHTTP(ctx context.Context // Forward already performed its one preparation before starting the business // clock. At dispatch only matching cached evidence may be used. Expiry, a save // or an identity change must not start another probe underneath that clock. -func (s *OpenAIGatewayService) applyCodexGatewayBorrowHTTP(req *http.Request, account *Account, model, proxy string) (*http.Request, error) { +func (s *OpenAIGatewayService) applyCodexGatewayBorrowHTTP(req *http.Request, account *Account, model, proxy string) (wire *http.Request, failure error) { + defer func() { + if failure != nil { + codexBorrowUsageFromContext(req.Context()).blocked(failure) + } + }() prepared, ok := req.Context().Value(codexGatewayBorrowHTTPPreparationContextKey{}).(codexGatewayBorrowHTTPPreparation) if !ok { if s.gatewayBorrow == nil { return req, nil } + if CodexGatewayBorrowRequestEligible(req) && s.codexGatewayBorrowHTTPConfigured(account, model) && codexBorrowUsageFromContext(req.Context()) == nil { + if d := borrowDiagnosticFromContext(req.Context()); d == nil || d.borrow { + tracker := s.gatewayBorrow.beginAttempt(req.Context(), account.ID, model, "http") + req = req.WithContext(context.WithValue(req.Context(), codexBorrowUsageContextKey{}, tracker)) + } + } borrowed, _, err := s.gatewayBorrow.Apply(req, account, model, proxy, nil, false) return borrowed, err } diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go index 819bab8de441..5c3c1ff870a0 100644 --- a/backend/internal/service/codex_gateway_borrow_probe.go +++ b/backend/internal/service/codex_gateway_borrow_probe.go @@ -83,6 +83,11 @@ func (s *CodexGatewayBorrowService) accountTemplate(ctx context.Context, id int6 req.Header.Set("Authorization", "Bearer "+token) setOpenAIChatGPTAccountHeaders(req.Header, credential) applyOpenAICodexProbeHeaders(req.Header) + // Manual preparation and cached-only generation rebuild this template. + // Give that synthetic window a stable identity, without changing any real + // client's window or the per-shot legacy session_id. A random window here + // would make a freshly validated proof unusable by its own next caller. + req.Header.Set("X-Codex-Window-ID", uuid.NewSHA1(uuid.NameSpaceOID, []byte(fmt.Sprintf("codex-borrow-template:%d:%s", id, model))).String()) enforceCodexIdentityHeadersForAccount(req.Header, credential, s.gateway.codexIdentityOverrideUA(a)) setOpenAICodexRoutingHint(req.Header, a, model, "") req = req.WithContext(WithCodexGatewayBorrowModel(req.Context(), actual)) @@ -169,7 +174,7 @@ 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"} + result := CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), ExpiresAt: &expires, Reason: "target_probe_failed", RequestShape: "ranxi_full", ServiceTier: borrowProbeServiceTier(template.Header)} cookies := []*http.Cookie{{Name: "__oailb", Value: candidate.cookie.Value}} fire := func(state string, cookies []*http.Cookie) (codexGatewayBorrowObservation, error) { shotCtx, cancel := context.WithTimeout(ctx, codexGatewayBorrowShotTimeout) @@ -178,11 +183,13 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h } mint, err := fire("", cookies) result.MintStatus, result.ReportedModel = mint.status, mint.model + result.MintCompleted, result.MintStateLength = mint.complete, len(mint.state) if err != nil || !borrowShotUsable(mint) { result.Error = borrowObservationError(mint, err) return result } if borrowTargetRouteChanged(mint.cookies, candidate, time.Now()) { + result.RouteChanged = true result.Reason, result.Error = "target_route_changed", "target response changed or deleted the fixed __oailb route" return result } @@ -201,6 +208,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h } continued, err := fire(mint.state, cookies) result.ContinueStatus = continued.status + result.ContinueCompleted, result.ContinueStateLength = continued.complete, len(continued.state) if continued.model != "" { result.ReportedModel = continued.model } @@ -209,6 +217,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h return result } if borrowTargetRouteChanged(continued.cookies, candidate, time.Now()) { + result.RouteChanged = true result.Reason, result.Error = "target_route_changed", "target response changed or deleted the fixed __oailb route" return result } @@ -250,20 +259,38 @@ func borrowTargetRouteChanged(cookies []*http.Cookie, candidate *codexGatewayBor return false } -func borrowObservationPayload(model, prompt, effort string) []byte { +func borrowObservationPayload(model, prompt, effort string, serviceTier ...string) []byte { body := map[string]any{"model": model, "instructions": prompt, "input": []map[string]any{{"type": "message", "role": "user", "content": []map[string]any{{"type": "input_text", "text": prompt}}}}, "stream": true, "store": false, "parallel_tool_calls": true, "include": []string{"reasoning.encrypted_content"}} if effort != "" { body["reasoning"] = map[string]string{"effort": effort} } + if len(serviceTier) > 0 && serviceTier[0] != "" { + body["service_tier"] = serviceTier[0] + } raw, _ := json.Marshal(body) return raw } +// The native sender owns this hint and derives it from the finalized body. +// Reading its tier avoids consuming/rebuilding the potentially large business +// body and keeps the short probe's body consistent with its routing headers. +func borrowProbeServiceTier(headers http.Header) string { + for _, part := range strings.Split(headers.Get(openAICodexRoutingHintHeader), ";") { + if key, value, ok := strings.Cut(part, "="); ok && strings.TrimSpace(key) == "tier" { + switch value = strings.TrimSpace(value); value { + case OpenAIFastTierPriority, OpenAIFastTierFlex, OpenAIFastTierUltrafast: + return value + } + } + } + return "" +} + 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))) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(borrowObservationPayload(model, prompt, effort, borrowProbeServiceTier(headers)))) if err != nil { return nil, err } @@ -271,13 +298,15 @@ func borrowObservationRequest(ctx context.Context, headers http.Header, model, p if req.Header == nil { req.Header = make(http.Header) } - for _, name := range []string{"Cookie", "Content-Length", "Content-Encoding", "Content-MD5", "X-Codex-Turn-State", "X-Codex-Turn-Metadata", "X-OpenAI-Internal-Codex-Responses-Lite", "conversation_id", "session_id", "session-id", "thread-id", "x-client-request-id", "x-codex-window-id"} { + for _, name := range []string{"Cookie", "Content-Length", "Content-Encoding", "Content-MD5", "X-Codex-Turn-State", "X-Codex-Turn-Metadata", "X-OpenAI-Internal-Codex-Responses-Lite", "conversation_id"} { req.Header.Del(name) } req.Header.Set("Accept", "text/event-stream") req.Header.Set("Content-Type", "application/json") req.Header.Set("OpenAI-Beta", "responses=experimental") - ensureCodexSessionIdentityHeaders(req.Header, uuid.NewString()) + // Pinned ranxi d3e43f2: each shot gets a fresh legacy session_id, while + // native session/thread/window headers in the template remain unchanged. + req.Header.Set("session_id", uuid.NewString()) if state != "" { req.Header.Set("X-Codex-Turn-State", state) } diff --git a/backend/internal/service/codex_gateway_borrow_request_repair_test.go b/backend/internal/service/codex_gateway_borrow_request_repair_test.go new file mode 100644 index 000000000000..4b3514e25ae4 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_request_repair_test.go @@ -0,0 +1,159 @@ +package service + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestCodexBorrowNilRequestDoesNotCreateDiagnosticEvidence(t *testing.T) { + s := newBorrowCoreTest(t, nil) + wire, application, err := s.Apply(nil, borrowCoreAccount(2), "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + require.Nil(t, wire) + require.Nil(t, application) +} + +// The pinned ranxi probe changes only the legacy session_id per shot. Native +// session/thread headers belong to the template and must survive both shots. +// A synthetic upstream rotates STATE when those headers change, reproducing +// the false rejection introduced by manufacturing a new native session per shot. +func TestCodexBorrowProbePreservesNativeSessionAndTier(t *testing.T) { + for _, model := range []string{"gpt-6-astra", "gpt-6.1-sol"} { + t.Run(model, func(t *testing.T) { + var shots []*http.Request + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, proxy string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + assert.Equal(t, int64(2), id) + assert.Equal(t, "target-exit", proxy) + body := borrowCoreBody(t, req) + assert.Equal(t, model, gjson.GetBytes(body, "model").String()) + assert.Equal(t, "priority", gjson.GetBytes(body, "service_tier").String()) + assert.Equal(t, "model="+model+";tier=priority", req.Header.Get(openAICodexRoutingHintHeader)) + assert.Empty(t, req.Header.Get("Content-Encoding")) + assert.Empty(t, req.Header.Get("X-Codex-Turn-Metadata")) + assert.NotEmpty(t, req.Header.Get("session_id")) + shots = append(shots, req.Clone(context.Background())) + state := "stable-probe-state" + if len(shots) == 2 { + assert.NotEqual(t, shots[0].Header.Get("session_id"), req.Header.Get("session_id")) + assert.Equal(t, "stable-probe-state", req.Header.Get("X-Codex-Turn-State")) + assert.Contains(t, req.Header.Get("Cookie"), "__cflb=target-route") + if shots[0].Header.Get("session-id") != req.Header.Get("session-id") || shots[0].Header.Get("thread-id") != req.Header.Get("thread-id") { + state = "different-session-state" + } + } else { + assert.Empty(t, req.Header.Get("X-Codex-Turn-State")) + } + return borrowCoreResponse(model, "OK", state, "__cflb=target-route; Secure; Path=/"), nil + }, a) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + _, req, _, err := s.accountTemplate(context.Background(), 2, model) + require.NoError(t, err) + for k, v := range map[string]string{"session-id": "client-session", "thread-id": "client-thread", "x-client-request-id": "client-request", "x-codex-window-id": "client-window", "X-Codex-Turn-State": "business-state", "X-Codex-Turn-Metadata": `{"session_id":"client-session"}`, "Content-Encoding": "zstd"} { + req.Header.Set(k, v) + } + setOpenAICodexRoutingHint(req.Header, a, model, "priority") + original := req.Header.Clone() + wire, applied, err := s.Apply(req, a, model, "target-exit", nil, false) + require.NoError(t, err) + require.True(t, applied.Applied) + require.Equal(t, original, req.Header) + require.Equal(t, "business-state", wire.Header.Get("X-Codex-Turn-State")) + for _, shot := range shots { + for _, key := range []string{"session-id", "thread-id", "x-client-request-id", "x-codex-window-id"} { + require.Equal(t, original.Get(key), shot.Header.Get(key), key) + } + } + }) + } +} + +func TestCodexBorrowPreparationFailureRetainsReasonAndCounts(t *testing.T) { + var calls int + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + calls++ + state := "first-state" + if req.Header.Get("X-Codex-Turn-State") != "" { + state = "replacement-state" + } + return borrowCoreResponse("gpt-6.1-sol", "OK", state), nil + }, a) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + require.NoError(t, err) + for _, id := range []string{"first-client", "retry-client"} { + ctx := context.WithValue(context.Background(), ctxkey.ClientRequestID, id) + ctx = context.WithValue(ctx, ctxkey.RequestID, "gateway-"+id) + req.Header.Set("x-client-request-id", id) + _, _, err = s.gateway.prepareCodexGatewayBorrowHTTP(ctx, req, a, "gpt-6.1-sol", "") + var failure *CodexGatewayBorrowFailure + require.ErrorAs(t, err, &failure) + require.Equal(t, "target_validation", failure.Stage) + require.Equal(t, "target_state_changed", failure.Reason) + require.NotNil(t, failure.RetryAfter) + require.NotNil(t, failure.Verification) + require.True(t, failure.Verification.MintCompleted) + require.True(t, failure.Verification.ContinueCompleted) + require.Equal(t, len("replacement-state"), failure.Verification.ContinueStateLength) + } + require.Equal(t, 2, calls, "a different tracing ID does not bypass the failure cooldown") + usage := s.Status().RecentUsage + require.Len(t, usage, 1) + require.EqualValues(t, 2, usage[0].AttemptCount) + require.EqualValues(t, 2, usage[0].BlockedCount) + require.Zero(t, usage[0].Count) + require.Zero(t, usage[0].AppliedCount) + require.False(t, usage[0].Dispatched) + require.Equal(t, "retry-client", usage[0].ClientRequestID) + require.Equal(t, "gateway-retry-client", usage[0].GatewayRequestID) + require.Equal(t, "target_state_changed", usage[0].Reason) +} + +func TestCodexBorrowProofIncludesNativeSessionAndRoutingPolicy(t *testing.T) { + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) { + return borrowCoreResponse("gpt-6.1-sol", "OK", "stable"), nil + }, a) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + require.NoError(t, err) + _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + for _, header := range []string{"session-id", "thread-id", "x-codex-window-id", "x-codex-installation-id", "x-codex-beta-features", openAICodexRoutingHintHeader} { + changed := req.Clone(req.Context()) + changed.Header.Set(header, "changed") + _, _, err = s.Apply(changed, a, "gpt-6.1-sol", "", nil, true) + require.Error(t, err, header) + } + req.Header.Set("x-client-request-id", "another-trace") + _, application, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, true) + require.NoError(t, err) + require.True(t, application.Applied) +} + +func TestCodexBorrowAttemptCountersSurviveOutOfOrderDispatch(t *testing.T) { + s := newBorrowCoreTest(t, nil) + first := s.beginAttempt(context.Background(), 2, "gpt-6.1-sol", "http") + newer := s.beginAttempt(context.Background(), 2, "gpt-6.1-sol", "http") + first.dispatched(true) + newer.blocked(&CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: "target_state_changed"}) + newer.blocked(ErrCodexGatewayBorrowUnavailable) + first.observe([]byte(`{"type":"response.completed","response":{"status":"completed"}}`)) + first.finish(nil) + row := s.Status().RecentUsage[0] + require.Equal(t, "blocked", row.Outcome) + require.EqualValues(t, 2, row.AttemptCount) + require.EqualValues(t, 1, row.BlockedCount) + require.EqualValues(t, 1, row.Count) + require.EqualValues(t, 1, row.AppliedCount) + require.False(t, row.Dispatched) +} diff --git a/backend/internal/service/codex_gateway_borrow_usage.go b/backend/internal/service/codex_gateway_borrow_usage.go index 7cfca73202a5..01c74fe544d0 100644 --- a/backend/internal/service/codex_gateway_borrow_usage.go +++ b/backend/internal/service/codex_gateway_borrow_usage.go @@ -9,23 +9,30 @@ import ( "sync" "time" + "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" "github.com/tidwall/gjson" ) type CodexGatewayBorrowUsage struct { - AccountID int64 `json:"account_id"` - Model string `json:"model"` - Transport string `json:"transport"` - Origin string `json:"origin"` - Applied bool `json:"applied"` - Reason string `json:"reason"` - StartedAt time.Time `json:"started_at"` - FinishedAt *time.Time `json:"finished_at,omitempty"` - Outcome string `json:"outcome"` - RequestID string `json:"request_id,omitempty"` - ReportedModel string `json:"reported_model,omitempty"` - Count uint64 `json:"count"` - AppliedCount uint64 `json:"applied_count"` + Dispatched bool `json:"dispatched"` + FailureStage string `json:"failure_stage,omitempty"` + ClientRequestID string `json:"client_request_id,omitempty"` + GatewayRequestID string `json:"gateway_request_id,omitempty"` + AttemptCount uint64 `json:"attempt_count"` + BlockedCount uint64 `json:"blocked_count"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Transport string `json:"transport"` + Origin string `json:"origin"` + Applied bool `json:"applied"` + Reason string `json:"reason"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + Outcome string `json:"outcome"` + RequestID string `json:"request_id,omitempty"` + ReportedModel string `json:"reported_model,omitempty"` + Count uint64 `json:"count"` + AppliedCount uint64 `json:"applied_count"` } type codexBorrowUsageKey struct { @@ -36,6 +43,31 @@ type codexBorrowUsageKey struct { // Only configured pairs are retained, one last observation per protocol/purpose. // Payloads, credentials and routing cookies are never copied into this index. func (s *CodexGatewayBorrowService) beginUsage(ctx context.Context, accountID int64, model, transport string, applied bool) *codexBorrowUsageTracker { + t := codexBorrowUsageFromContext(ctx) + if t == nil || t.service != s || t.key.account != accountID || t.key.model != model || t.key.transport != transport || !t.pendingDispatch() { + t = s.beginAttempt(ctx, accountID, model, transport) + } + t.dispatched(applied) + return t +} + +func (t *codexBorrowUsageTracker) pendingDispatch() bool { + t.mu.Lock() + defer t.mu.Unlock() + return !t.finished && !t.row.Dispatched +} + +type codexBorrowUsageContextKey struct{} + +func codexBorrowUsageFromContext(ctx context.Context) *codexBorrowUsageTracker { + if ctx == nil { + return nil + } + t, _ := ctx.Value(codexBorrowUsageContextKey{}).(*codexBorrowUsageTracker) + return t +} + +func (s *CodexGatewayBorrowService) beginAttempt(ctx context.Context, accountID int64, model, transport string) *codexBorrowUsageTracker { if s == nil { return nil } @@ -50,12 +82,10 @@ func (s *CodexGatewayBorrowService) beginUsage(ctx context.Context, accountID in } key := codexBorrowUsageKey{accountID, model, transport, origin} previous := s.usage[key] - row := CodexGatewayBorrowUsage{AccountID: accountID, Model: model, Transport: transport, Origin: origin, Applied: applied, StartedAt: time.Now(), Outcome: "sent", Count: previous.Count + 1, AppliedCount: previous.AppliedCount} - row.Reason = "not_applied" - if applied { - row.Reason = "borrow_applied" - row.AppliedCount++ - } else if !s.config.Enabled { + row := CodexGatewayBorrowUsage{AccountID: accountID, Model: model, Transport: transport, Origin: origin, StartedAt: time.Now(), Outcome: "preparing", Reason: "preparing", Count: previous.Count, AppliedCount: previous.AppliedCount, AttemptCount: previous.AttemptCount + 1, BlockedCount: previous.BlockedCount} + row.ClientRequestID, _ = ctx.Value(ctxkey.ClientRequestID).(string) + row.GatewayRequestID, _ = ctx.Value(ctxkey.RequestID).(string) + if !s.config.Enabled { row.Reason = "disabled" } if s.usage == nil { @@ -65,6 +95,66 @@ func (s *CodexGatewayBorrowService) beginUsage(ctx context.Context, accountID in return &codexBorrowUsageTracker{service: s, key: key, row: row, revision: s.revision} } +// Counter changes apply even when a newer attempt owns the displayed row. A +// late finish may not overwrite that newer request or roll its counters back. +func (t *codexBorrowUsageTracker) publish(sent, applied, blocked uint64) { + t.service.mu.Lock() + defer t.service.mu.Unlock() + current, ok := t.service.usage[t.key] + if !ok || t.service.revision != t.revision { + return + } + current.Count += sent + current.AppliedCount += applied + current.BlockedCount += blocked + if current.AttemptCount == t.row.AttemptCount { + t.row.Count, t.row.AppliedCount, t.row.BlockedCount = current.Count, current.AppliedCount, current.BlockedCount + current = t.row + } + t.service.usage[t.key] = current +} + +func (t *codexBorrowUsageTracker) dispatched(applied bool) { + if t == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + if t.finished || t.row.Dispatched { + return + } + t.row.Dispatched, t.row.Applied, t.row.Outcome = true, applied, "sent" + if t.row.Reason != "disabled" { + t.row.Reason = "not_applied" + } + var appliedCount uint64 + if applied { + t.row.Reason = "borrow_applied" + appliedCount = 1 + } + t.publish(1, appliedCount, 0) +} + +func (t *codexBorrowUsageTracker) blocked(err error) { + if t == nil { + return + } + t.mu.Lock() + if t.finished || t.row.Dispatched { + t.mu.Unlock() + return + } + t.row.FailureStage, t.row.Reason = codexBorrowFailureStage(err) + t.row.Outcome, t.terminal = "blocked", true + // Mark finished while holding the tracker lock so repeated error owners + // cannot count the same preparation rejection twice. + t.finished = true + now := time.Now() + t.row.FinishedAt = &now + t.publish(0, 0, 1) + t.mu.Unlock() +} + type codexBorrowUsageTracker struct { mu sync.Mutex terminal bool @@ -114,6 +204,10 @@ func (t *codexBorrowUsageTracker) finish(err error) { } t.once.Do(func() { t.mu.Lock() + defer t.mu.Unlock() + if t.finished { + return + } if !t.terminal { if err != nil { t.row.Outcome = "transport_error" @@ -124,13 +218,7 @@ func (t *codexBorrowUsageTracker) finish(err error) { now := time.Now() t.row.FinishedAt = &now t.finished = true - row := t.row - t.mu.Unlock() - t.service.mu.Lock() - defer t.service.mu.Unlock() - if current, ok := t.service.usage[t.key]; ok && t.service.revision == t.revision && current.Count == row.Count { - t.service.usage[t.key] = row - } + t.publish(0, 0, 0) }) } diff --git a/backend/internal/service/codex_gateway_borrow_usage_ws.go b/backend/internal/service/codex_gateway_borrow_usage_ws.go index 080cb1d2282c..2d62505da473 100644 --- a/backend/internal/service/codex_gateway_borrow_usage_ws.go +++ b/backend/internal/service/codex_gateway_borrow_usage_ws.go @@ -18,12 +18,17 @@ type codexBorrowUsageFrameConn struct { applied bool mu sync.Mutex current *codexBorrowUsageTracker + prepared *codexBorrowUsageTracker } func (c *codexBorrowUsageFrameConn) WriteFrame(ctx context.Context, kind coderws.MessageType, payload []byte) error { c.mu.Lock() if gjson.GetBytes(payload, "type").String() == "response.create" { c.current.finish(nil) + if c.prepared != nil { + ctx = context.WithValue(ctx, codexBorrowUsageContextKey{}, c.prepared) + c.prepared = nil + } c.current = c.service.beginUsage(ctx, c.accountID, gjson.GetBytes(payload, "model").String(), "ws", c.applied) } tracker := c.current diff --git a/backend/internal/service/codex_gateway_borrow_ws_anchor.go b/backend/internal/service/codex_gateway_borrow_ws_anchor.go index baa1a98cd63b..3b56b9dc106e 100644 --- a/backend/internal/service/codex_gateway_borrow_ws_anchor.go +++ b/backend/internal/service/codex_gateway_borrow_ws_anchor.go @@ -39,6 +39,7 @@ type codexGatewayBorrowWSAnchorStore struct { } type codexGatewayBorrowWSTurn struct { + usage *codexBorrowUsageTracker store *codexGatewayBorrowWSAnchorStore key codexGatewayBorrowWSAnchorKey previousID, connID, scope, model string @@ -221,7 +222,7 @@ func (store *codexGatewayBorrowWSAnchorStore) knowsContinuation(key codexGateway } func (s *OpenAIGatewayService) prepareCodexGatewayBorrowWSTurn(ctx context.Context, c *gin.Context, account *Account, - wsURL string, headers http.Header, rawClientBody []byte, model, previousID, proxyURL string, executionScope ...string) (*codexGatewayBorrowWSTurn, error) { + wsURL string, headers http.Header, rawClientBody []byte, model, previousID, proxyURL string, executionScope ...string) (preparedTurn *codexGatewayBorrowWSTurn, preparationErr error) { if s == nil || s.gatewayBorrow == nil || account == nil { return nil, nil } @@ -229,10 +230,23 @@ func (s *OpenAIGatewayService) prepareCodexGatewayBorrowWSTurn(ctx context.Conte if err != nil { return nil, err } - if d := borrowDiagnosticFromContext(ctx); d != nil { - return d.prepareWS(ctx, s.gatewayBorrow, req, account, model, previousID, proxyURL) - } selected := s.codexGatewayBorrowWSSelected(account, model, req) + diagnostic := borrowDiagnosticFromContext(ctx) + if selected && (diagnostic == nil || diagnostic.borrow) { + tracker := s.gatewayBorrow.beginAttempt(ctx, account.ID, model, "ws") + ctx = context.WithValue(ctx, codexBorrowUsageContextKey{}, tracker) + req = req.WithContext(ctx) + defer func() { + if preparationErr != nil { + tracker.blocked(preparationErr) + } else if preparedTurn != nil { + preparedTurn.usage = tracker + } + }() + } + if diagnostic != nil { + return diagnostic.prepareWS(ctx, s.gatewayBorrow, req, account, model, previousID, proxyURL) + } if IsPelicanGeneration(ctx) { if !selected { return nil, nil @@ -358,6 +372,9 @@ func (turn *codexGatewayBorrowWSTurn) finish(pool *openAIWSConnPool, result *Ope if turn == nil { return } + if resultErr != nil { + turn.usage.blocked(resultErr) + } if turn.pelicanOneShot { turn.finished.Do(func() { if pool != nil && turn.connID != "" { diff --git a/backend/internal/service/codex_gateway_borrow_ws_phase_test.go b/backend/internal/service/codex_gateway_borrow_ws_phase_test.go index b4966a8fd839..a6a6aa80324d 100644 --- a/backend/internal/service/codex_gateway_borrow_ws_phase_test.go +++ b/backend/internal/service/codex_gateway_borrow_ws_phase_test.go @@ -23,6 +23,7 @@ func TestCodexGatewayBorrowWS_PreparationFailureUsesNeutralExistingFailoverBefor seedBorrowWSTarget(t, svc, borrow, account, borrowWSContext(), "gpt-6.1-sol") for key, check := range borrow.targets { check.result.Success = false + check.result.Reason = "target_state_changed" check.retryAfter = time.Now().Add(codexGatewayBorrowFailureWait) borrow.targets[key] = check } @@ -81,6 +82,12 @@ func TestCodexGatewayBorrowWS_PreparationFailureUsesNeutralExistingFailoverBefor require.Empty(t, dialer.headers, "no business dial occurs for a failed local preparation") require.Empty(t, probe.requests, "the cached failed qualification is not reprobed") require.Empty(t, borrow.wsAnchors.busy) + usage := borrow.Status().RecentUsage + require.Len(t, usage, 1) + require.EqualValues(t, 1, usage[0].AttemptCount) + require.EqualValues(t, 1, usage[0].BlockedCount) + require.Zero(t, usage[0].Count) + require.Equal(t, "target_state_changed", usage[0].Reason) }) } } diff --git a/backend/internal/service/openai_codex_request_compression.go b/backend/internal/service/openai_codex_request_compression.go index 4bb36b9b0e64..f2ea64b334bd 100644 --- a/backend/internal/service/openai_codex_request_compression.go +++ b/backend/internal/service/openai_codex_request_compression.go @@ -155,10 +155,12 @@ func (s *OpenAIGatewayService) doOpenAICodexUpstream(req *http.Request, account req = borrowed wire, err := prepareOpenAICodexWireRequest(req, account) if err != nil { + codexBorrowUsageFromContext(req.Context()).blocked(err) return nil, err } applied := codexBorrowHTTPApplied(wire) if err := consumeBorrowDiagnosticRequest(wire.Context()); err != nil { + codexBorrowUsageFromContext(req.Context()).blocked(err) return nil, err } tracker := s.gatewayBorrow.beginUsage(wire.Context(), account.ID, model, "http", applied) diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 5e02102b3d76..42a8a7c7d550 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -454,7 +454,11 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( } capture.mu.Unlock() } - usageTracker := s.gatewayBorrow.beginUsage(ctx, account.ID, mappedModel, "ws", borrowTurn.borrowed()) + usageContext := ctx + if borrowTurn != nil && borrowTurn.usage != nil { + usageContext = context.WithValue(ctx, codexBorrowUsageContextKey{}, borrowTurn.usage) + } + usageTracker := s.gatewayBorrow.beginUsage(usageContext, account.ID, mappedModel, "ws", borrowTurn.borrowed()) defer func() { usageTracker.finish(borrowErr) }() if err := lease.WriteJSONWithContextTimeout(ctx, payload, s.openAIWSWriteTimeout()); err != nil { lease.MarkBroken() diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 0412e0e1ee7b..def5485f4448 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -1011,6 +1011,9 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2PassthroughAttempt( return errors.New("openai ws passthrough upstream connection does not support frame relay") } usageConn := &codexBorrowUsageFrameConn{FrameConn: upstreamFrameConn, service: s.gatewayBorrow, accountID: account.ID, applied: borrowTurn.borrowed()} + if borrowTurn != nil { + usageConn.prepared = borrowTurn.usage + } defer usageConn.finish() relayUpstreamFrameConn := &openAIWSPassthroughFirstOutputFrameConn{ inner: usageConn, diff --git a/frontend/src/api/admin/codexGatewayBorrow.ts b/frontend/src/api/admin/codexGatewayBorrow.ts index efbb5e4e60d0..d38a445d7dfb 100644 --- a/frontend/src/api/admin/codexGatewayBorrow.ts +++ b/frontend/src/api/admin/codexGatewayBorrow.ts @@ -30,14 +30,25 @@ export interface BorrowTargetStatus extends BorrowSourceStatus { minted: boolean new_ticket: boolean reported_model?: string + request_shape?: string + service_tier?: string + mint_completed?: boolean + continue_completed?: boolean + mint_state_length?: number + continue_state_length?: number + route_changed?: boolean } export interface CodexGatewayBorrowUsage { account_id: number; model: string; transport: string; origin: string applied: boolean; reason: string; started_at: string; finished_at?: string outcome: string; request_id?: string; reported_model?: string; count: number; applied_count: number + dispatched?: boolean; failure_stage?: string; client_request_id?: string; gateway_request_id?: string + attempt_count?: number; blocked_count?: number } export interface CodexBorrowDiagnosticResult { + scenario?: string; failure_stage?: string; failure_reason?: string; verification?: BorrowVerification + tool_round_trip?: boolean; state_length?: number dispatched?: boolean read_error?: string mode: string; turn: number; applied: boolean; completed: boolean; response_id?: string @@ -47,7 +58,7 @@ export interface CodexBorrowDiagnosticEvent { type: 'phase' | 'request' | 'result' | 'done' | 'error'; mode?: string; requests: number; limit: number result?: CodexBorrowDiagnosticResult; error?: string } -export interface CodexBorrowDiagnosticRequest { account_id: number; model: string; transport: 'http' | 'ws'; request_limit?: number } +export interface CodexBorrowDiagnosticRequest { account_id: number; model: string; transport: 'http' | 'ws'; request_limit?: number; scenario?: 'codex_session'; mode?: 'ordinary' | 'borrowed'; service_tier?: string } export interface CodexGatewayBorrowStatus { setup?: { state: string; phase: string; account_id: number; model: string; completed: number; total: number; failed: number; started_at: string; finished_at?: string; error?: string } @@ -82,6 +93,13 @@ export interface BorrowVerification { reported_model?: string checked_at: string expires_at?: string + request_shape?: string + service_tier?: string + mint_completed?: boolean + continue_completed?: boolean + mint_state_length?: number + continue_state_length?: number + route_changed?: boolean } export type BorrowTestResultStatus = 'pending' | 'running' | 'complete' | 'failed' | 'incomplete' | 'cancelled' | 'skipped' diff --git a/frontend/src/components/admin/codex/CodexBorrowActivity.vue b/frontend/src/components/admin/codex/CodexBorrowActivity.vue index 946eeb57d188..15db98aefb0d 100644 --- a/frontend/src/components/admin/codex/CodexBorrowActivity.vue +++ b/frontend/src/components/admin/codex/CodexBorrowActivity.vue @@ -4,16 +4,17 @@

{{ text('最近实际请求', 'Recent actual requests') }}

{{ text('打开鹈鹕测试', 'Open Pelican tests') }} -

{{ text('线路验证通过不代表业务已经使用。这里记录实际发送和上游完成情况；诊断请求单独标记。', 'A passed route check does not mean a business request used it. This records actual dispatch and completion; diagnostics are labelled separately.') }}

+

{{ text('这里记录业务从借用准备到上游完成的结果。准备失败表示业务尚未发送；诊断成功不代表业务成功。', 'Business results cover preparation through upstream completion. A preparation failure means the request was not sent; a diagnostic success does not prove business success.') }}

{{ text('还没有实际发送记录。可运行下方对照，或在正常使用后刷新状态。', 'No dispatch has been observed. Run a comparison below, or refresh after normal use.') }}

- +
+ - + - - - + + +
{{ group.title }}
{{ text('账号与模型', 'Account / model') }}{{ text('实际使用', 'Actual use') }}{{ text('上游结果', 'Upstream result') }}{{ text('时间与次数', 'Time / counts') }}
{{ accountName(row.account_id) }}{{ row.model }} · {{ row.transport.toUpperCase() }} · {{ row.origin === 'business' ? text('业务', 'Business') : text('诊断', 'Diagnostic') }}{{ row.applied ? text('已使用借用', 'Borrow applied') : text('未使用借用', 'Borrow not applied') }}{{ outcome(row.outcome) }}
{{ text('请求详情', 'Request details') }}

{{ row.request_id || '—' }}

{{ row.reported_model || '—' }}

{{ row.reason }}

{{ text('发送', 'Sent') }} {{ row.count }} · {{ text('借用', 'Borrowed') }} {{ row.applied_count }}

{{ row.dispatched === false ? text('业务未发送', 'Not dispatched') : row.applied ? text('已使用借用', 'Borrow applied') : text('未使用借用', 'Borrow not applied') }}{{ outcome(row.outcome) }}
{{ text('请求详情', 'Request details') }}

{{ text('失败阶段', 'Failure stage') }}: {{ row.failure_stage }}

{{ text('网关请求', 'Gateway request') }}: {{ row.gateway_request_id || '—' }}

{{ text('客户端请求', 'Client request') }}: {{ row.client_request_id || '—' }}

{{ text('上游请求', 'Upstream request') }}: {{ row.request_id || '—' }}

{{ row.reported_model || '—' }}

{{ row.reason }}

{{ text('尝试', 'Attempts') }} {{ row.attempt_count ?? row.count }} · {{ text('阻断', 'Blocked') }} {{ row.blocked_count ?? 0 }}

{{ text('发送', 'Sent') }} {{ row.count }} · {{ text('借用', 'Borrowed') }} {{ row.applied_count }}

{{ text('当前配置的进程内统计，开始于', 'In-process counters for the current configuration, since') }} {{ time(status.observed_since) }}

@@ -24,7 +25,8 @@
- + +
@@ -34,6 +36,9 @@

{{ result.mode === 'ordinary' ? text('普通路径', 'Ordinary path') : text('借用路径', 'Borrowed path') }} · {{ text('第', 'Turn') }} {{ result.turn }} {{ result.completed ? text('轮：已完成', ': completed') : text('轮：未完成', ': not completed') }}

{{ result.dispatched === false ? text('准备阶段结束，未发出对照请求', 'Stopped before dispatching this comparison') : result.applied ? text('实际已使用借用', 'Borrow was applied') : text('未使用借用', 'Borrow was not applied') }} · {{ result.reported_model || '—' }} · {{ result.duration_ms }} ms

{{ result.answer || '—' }}

+

{{ result.tool_round_trip ? text('工具结果与续接已核对', 'Tool result and continuation verified') : text('Codex 工具调用与完整历史续接检查', 'Checking Codex tool call and full-history continuation') }}

+

{{ result.failure_stage }} · {{ result.failure_reason }}

+
{{ text('本次借用验证依据', 'Borrow verification for this request') }}
{{ JSON.stringify(result.verification, null, 2) }}
{{ result.error }}
{{ text('完整响应', 'Complete response') }}

{{ text('流读取结束信息（完成终态优先）', 'Read cleanup detail (a completed terminal takes precedence)') }}: {{ result.read_error }}

{{ result.raw_response }}
@@ -56,17 +61,21 @@ const selection = ref(''), transport = ref<'http' | 'ws'>('http'), running = ref const events = ref([]) const selected = computed(() => pairs.value.find(pair => pair.key === selection.value)) const results = computed(() => events.value.flatMap(event => event.result ? [event.result] : [])) +const usageGroups = computed(() => ['business', 'diagnostic'].map(origin => ({ + origin, title: origin === 'business' ? text('最近业务请求', 'Recent business requests') : text('诊断请求', 'Diagnostic requests'), + rows: (props.status.recent_usage || []).filter(row => row.origin === origin) +})).filter(group => group.rows.length)) watch(pairs, value => { if (!value.some(pair => pair.key === selection.value)) selection.value = value[0]?.key || '' }, { immediate: true }) let controller: AbortController | undefined let disposed = false function time(value?: string) { return value && !value.startsWith('0001') ? new Date(value).toLocaleString() : '—' } -function outcome(value: string) { return ({ unobserved: text('终态信息未完整记录', 'Terminal evidence not fully recorded'), sent: text('已发送，等待终态', 'Sent, awaiting completion'), completed: text('已完成', 'Completed'), incomplete: text('未收到完整终态', 'No complete terminal event'), upstream_error: text('上游返回错误', 'Upstream error'), transport_error: text('连接失败或中断', 'Connection failed or interrupted') } as Record)[value] || value } +function outcome(value: string) { return ({ preparing: text('正在准备借用', 'Preparing borrowing'), blocked: text('借用准备或验证失败，未发送', 'Borrow preparation or validation failed; not dispatched'), unobserved: text('终态信息未完整记录', 'Terminal evidence not fully recorded'), sent: text('已发送，等待终态', 'Sent, awaiting completion'), completed: text('已完成', 'Completed'), incomplete: text('未收到完整终态', 'No complete terminal event'), upstream_error: text('上游返回错误', 'Upstream error'), transport_error: text('连接失败或中断', 'Connection failed or interrupted') } as Record)[value] || value } function cancel() { controller?.abort() } -async function diagnose() { +async function diagnose(session: boolean) { if (!selected.value || running.value) return controller = new AbortController(); running.value = true; error.value = ''; events.value = []; requests.value = 0; phase.value = 'ordinary' try { - await codexGatewayBorrowAPI.diagnose({ account_id: selected.value.account_id, model: selected.value.model, transport: transport.value }, event => { + await codexGatewayBorrowAPI.diagnose({ account_id: selected.value.account_id, model: selected.value.model, transport: transport.value, ...(session ? { scenario: 'codex_session' as const, mode: 'borrowed' as const } : {}) }, event => { events.value.push(event); requests.value = event.requests if (event.mode) phase.value = event.mode }, controller.signal) diff --git a/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts b/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts index 6549ce6fbfca..acfcb5098b37 100644 --- a/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts +++ b/frontend/src/components/admin/codex/__tests__/CodexBorrowActivity.spec.ts @@ -19,6 +19,24 @@ function open(locale = 'zh') { beforeEach(() => { diagnose.mockReset() }) describe('Codex actual-request diagnostics', () => { + it('keeps blocked business attempts visible ahead of successful diagnostics', () => { + const wrapper = open('en') + const diagnostic = { account_id: 2, model: 'gpt-6-astra', transport: 'http', origin: 'diagnostic', applied: true, dispatched: true, reason: 'borrow_applied', started_at: new Date().toISOString(), outcome: 'completed', count: 2, applied_count: 1 } + void wrapper.setProps({ status: { ...status, recent_usage: [diagnostic, { ...diagnostic, origin: 'business', applied: false, dispatched: false, reason: 'target_state_changed', failure_stage: 'target_validation', outcome: 'blocked', count: 0, applied_count: 0, attempt_count: 24, blocked_count: 24, gateway_request_id: 'local-failed-request' }] } }) + return flushPromises().then(() => { + const tables = wrapper.findAll('table') + expect(tables[0].attributes('data-test')).toBe('borrow-usage-business') + expect(tables[0].text()).toContain('not dispatched') + expect(tables[0].text()).toContain('Attempts 24') + expect(tables[0].text()).toContain('Blocked 24') + expect(tables[0].text()).toContain('Sent 0') + expect(tables[0].text()).toContain('target_state_changed') + expect(tables[0].text()).toContain('local-failed-request') + expect(tables[1].text()).toContain('Completed') + expect(diagnose).not.toHaveBeenCalled() + wrapper.unmount() + }) + }) it('reads existing evidence without automatically issuing requests', () => { const wrapper = open() expect(diagnose).not.toHaveBeenCalled() diff --git a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts index 0707a248462e..f5ad596b630a 100644 --- a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts @@ -161,7 +161,7 @@ export default { target_probe_failed: 'Route validation failed for this account and model; see technical details', target_route_changed: 'The route changed during validation; validate it again', target_state_missing: 'The first response had no session state to validate continuation', - target_state_changed: 'The session state differed between the two responses; validate again', + target_state_changed: 'Both upstream requests completed, but continuation returned a different STATE. The route did not qualify for borrowing; inspect the verification details.', validating: 'Validating the borrow route for this account and model', candidate_ready: 'The source route has been collected', validated: 'The borrow route passed validation for this account and model', diff --git a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts index 7d4520fba7d2..2a3d254bc690 100644 --- a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts @@ -161,7 +161,7 @@ export default { target_probe_failed: '此账号与模型的线路验证失败，请查看技术详情', target_route_changed: '验证时线路发生变化，请重新验证', target_state_missing: '首轮响应缺少会话状态，无法验证续接', - target_state_changed: '两轮响应的会话状态不一致，请重新验证', + target_state_changed: '两轮上游均已完成，但续接返回了不同的 STATE，当前线路未通过借用验证。请查看本次验证依据。', validating: '正在验证此账号与模型的借用线路', candidate_ready: '来源线路已采集', validated: '此账号与模型的借用线路已验证通过', diff --git a/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue b/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue index aac93cd0a4a3..8545eb6d898e 100644 --- a/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue +++ b/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue @@ -108,8 +108,9 @@
{{ t('admin.codexGatewayBorrow.checkedAt') }}
{{ formatTime(target.checked_at) }}
-
{{ t('admin.codexGatewayBorrow.expiresAt') }}
{{ formatTime(target.expires_at) }}
-
{{ t('admin.codexGatewayBorrow.expiresAt') }}
{{ t('admin.codexGatewayBorrow.remainingSeconds', { seconds: remainingSeconds(target.expires_at) }) }}
+
{{ text('候选线路到期时间', 'Candidate route expiry') }}
{{ formatTime(target.expires_at) }}
+
{{ text('候选线路剩余时间', 'Candidate time remaining') }}
{{ t('admin.codexGatewayBorrow.remainingSeconds', { seconds: remainingSeconds(target.expires_at) }) }}
+
{{ text('验证说明', 'Validation note') }}
{{ text('候选尚未到期也不代表验证通过。', 'An unexpired candidate does not mean validation passed.') }}
{{ t('admin.codexGatewayBorrow.retryAfter') }}
{{ formatTime(target.retry_after) }}
@@ -126,6 +127,11 @@
{{ t('admin.codexGatewayBorrow.stateMinted') }}
{{ target.minted ? t('common.yes') : t('common.no') }}
{{ t('admin.codexGatewayBorrow.newTicket') }}
{{ target.new_ticket ? t('common.yes') : t('common.no') }}
{{ t('admin.codexGatewayBorrow.reportedModel') }}
{{ target.reported_model || '—' }}
+
{{ text('验证请求形态 / 实际模型', 'Validation shape / actual model') }}
{{ target.request_shape }} / {{ target.model }}
+
{{ text('验证服务档位', 'Validation service tier') }}
{{ target.service_tier || 'default' }}
+
{{ text('两轮上游完成', 'Both upstream completions') }}
{{ target.mint_completed ? t('common.yes') : t('common.no') }} / {{ target.continue_completed ? t('common.yes') : t('common.no') }}
+
{{ text('两轮 STATE 长度', 'STATE lengths') }}
{{ target.mint_state_length ?? '—' }} / {{ target.continue_state_length ?? '—' }}
+
{{ text('上游替换借用路由', 'Upstream replaced borrowed route') }}
{{ target.route_changed ? t('common.yes') : t('common.no') }}

{{ t('admin.codexGatewayBorrow.stateUnavailable') }}

{{ t('admin.codexGatewayBorrow.fullError') }}

{{ target.error }}