From b4339e822452fa18a72363afe8e8d38a0429ac7f Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 10 Oct 2026 03:13:47 +0800 Subject: [PATCH 1/5] Replace rejected Codex route candidates within a bounded request --- .downstream/native-domains.md | 11 +++ .../internal/service/codex_gateway_borrow.go | 54 ++++++++--- .../codex_gateway_borrow_candidate_retry.go | 80 +++++++++++++++++ ...dex_gateway_borrow_candidate_retry_test.go | 90 +++++++++++++++++++ .../codex_gateway_borrow_diagnose_session.go | 33 +++++++ ...ex_gateway_borrow_diagnose_session_test.go | 37 ++++++++ .../locales/en/admin/codexGatewayBorrow.ts | 2 + .../locales/zh/admin/codexGatewayBorrow.ts | 2 + 8 files changed, 295 insertions(+), 14 deletions(-) create mode 100644 backend/internal/service/codex_gateway_borrow_candidate_retry.go create mode 100644 backend/internal/service/codex_gateway_borrow_candidate_retry_test.go create mode 100644 backend/internal/service/codex_gateway_borrow_diagnose_session_test.go diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 1f74540d1c04..51f07c60b868 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -109,6 +109,13 @@ 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. +When both completed target shots reject a candidate's STATE or the upstream +replaces its route, a request may acquire and validate one replacement through +the existing source/proxy configuration. Replacement is capped at 90 seconds and +never retires a candidate with another live successful proof or validation in +flight. Another rejection cools acquisition for 15 seconds. Receiving the same +rejected cookie cannot renew or qualify it. This remains demand-driven, with no +exit rotation, renewal timer, ordinary fallback or business replay. Source and target probes use separate HTTP/TLS pools. Business WS connections stay in the ordinary account pool, with fixed one-hour continuation anchors. @@ -163,6 +170,10 @@ Optional `service_tier` exercises the effective routing tier. Results include th 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. +Complete `response.output_item.done` events rebuild tool calls and reasoning +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`. Ranxi reference: v2.10.2 (`d3e43f2de33af9e987cffa511d76dfabbcd749da`). Its five fingerprint, gateway-cookie, target-probe and automatic-setup core files are diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 176e1b8b427a..a0fd060c4efa 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -47,12 +47,13 @@ var ( // 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 - Stage string - Reason string - Detail string - RetryAfter *time.Time - Verification *CodexGatewayBorrowVerification + CookieFingerprint string + Cause error + Stage string + Reason string + Detail string + RetryAfter *time.Time + Verification *CodexGatewayBorrowVerification } func (e *CodexGatewayBorrowFailure) Error() string { @@ -81,7 +82,7 @@ func IsCodexGatewayBorrowFailure(err error) bool { func borrowVerificationFailure(check codexGatewayBorrowTargetCheck) error { result := check.result - failure := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: result.Reason, Detail: result.Error, Verification: &result} + failure := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: result.Reason, Detail: result.Error, Verification: &result, CookieFingerprint: check.cookieKey} if !check.retryAfter.IsZero() { retry := check.retryAfter failure.RetryAfter = &retry @@ -274,6 +275,8 @@ type codexGatewayBorrowTargetCheck struct { } type CodexGatewayBorrowService struct { + rejectedCookie string + rejectedCause error validationSequence uint64 sourceSequence uint64 qualifications map[string]codexGatewayBorrowTargetCheck @@ -410,6 +413,7 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.revision++ s.config, s.loaded = cloneCodexGatewayBorrowConfig(cfg), true s.candidate = nil + s.rejectedCookie, s.rejectedCause = "", nil s.setup = CodexGatewayBorrowSetup{} if saved && cfg.Enabled { s.setup = CodexGatewayBorrowSetup{State: "queued", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} @@ -514,6 +518,9 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { row.State, row.Reason = "expired", "route_expired" } } + if s.config.Enabled && s.candidate != nil && s.candidate.sourceID == id && s.candidateRejectedLocked() { + row.State, row.Reason = "candidate", "source_passed_target_failed" + } out.Sources = append(out.Sources, row) } for _, id := range s.config.TargetAccountIDs { @@ -521,7 +528,7 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { row := CodexGatewayBorrowTargetStatus{CodexGatewayBorrowVerification: CodexGatewayBorrowVerification{AccountID: id, Model: model, Reason: "not_verified"}, State: "waiting"} if c, ok := s.targets[codexGatewayBorrowTargetKey{id, model}]; ok { row.CodexGatewayBorrowVerification = c.result - row.CacheValid = c.policyRevision == currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success && s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) + row.CacheValid = c.policyRevision == currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success && s.candidate != nil && !s.candidateRejectedLocked() && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) expires := c.expires if s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && s.candidate.expires.Before(expires) { expires = s.candidate.expires @@ -552,6 +559,10 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { if !s.config.Enabled { row.State, row.Reason, row.CacheValid = "disabled", "disabled", false } + if s.config.Enabled && !row.CacheValid && s.candidateRejectedLocked() && now.Before(s.prepareFailedUntil) { + retry := s.prepareFailedUntil + row.RetryAfter = &retry + } out.Targets = append(out.Targets, row) } } @@ -677,13 +688,17 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Unlock() return nil, ErrCodexGatewayBorrowChanged } - if s.candidate != nil && time.Until(s.candidate.expires) > codexGatewayBorrowTimeout { + if s.candidate != nil && !s.candidateRejectedLocked() && time.Until(s.candidate.expires) > codexGatewayBorrowTimeout { s.mu.Unlock() return nil, nil } if time.Now().Before(s.prepareFailedUntil) { detail := s.prepareFailure + cause := s.rejectedCause s.mu.Unlock() + if cause != nil { + return nil, cause + } if detail != "" { return nil, errors.New(detail) } @@ -717,6 +732,12 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 return nil, ErrCodexGatewayBorrowChanged } row := CodexGatewayBorrowSourceStatus{AccountID: id, State: "rejected", Reason: "source_probe_failed", CheckedAt: &now, Error: detail} + if err == nil && candidate != nil && s.rejectedCookie != "" && borrowHash(candidate.cookie.Value) == s.rejectedCookie { + // Keep the old candidate and expiry: reacquiring the identical + // rejected credential must not renew its lease or qualify it. + err = errors.New("source returned the same rejected route") + row.Reason, row.Error = "source_returned_rejected_route", err.Error() + } if err == nil && candidate != nil && now.Before(candidate.expires) { // Seeing the same value while its original lease is live never // renews it, even when a different source returns that value. @@ -727,6 +748,7 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.qualifications = map[string]codexGatewayBorrowTargetCheck{} } s.candidate = candidate + s.rejectedCookie, s.rejectedCause = "", nil row.State, row.Reason, row.Error, row.ExpiresAt = "ready", "source_qualified", "", &candidate.expires s.sources[id] = row s.preparing = false @@ -744,7 +766,7 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Unlock() } s.mu.Lock() - if s.revision == rev && s.sourceSequence == sequence && !errors.Is(probeCtx.Err(), context.Canceled) { + if s.revision == rev && s.sourceSequence == sequence && !errors.Is(probeCtx.Err(), context.Canceled) && !errors.Is(lastErr, ErrCodexBorrowDiagnosticBudget) { s.preparing = false s.prepareFailedUntil = time.Now().Add(codexGatewayBorrowFailureWait) s.prepareFailure = lastErr.Error() @@ -776,7 +798,7 @@ func borrowTargetFingerprint(req *http.Request, account *Account, model, proxy s // __cflb, other cookies, body and proxy stay exactly as supplied by its caller. // cachedOnly is used by manual generation: it never acquires or validates. func (s *CodexGatewayBorrowService) Apply(req *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, cachedOnly bool) (*http.Request, *CodexGatewayBorrowApplication, error) { - wire, application, err := s.apply(req, account, model, proxy, profile, cachedOnly, false) + wire, application, err := s.applyWithCandidateRetry(req, account, model, proxy, profile, cachedOnly, false) if err != nil && !IsCodexGatewayBorrowFailure(err) { err = &CodexGatewayBorrowFailure{Cause: err} } @@ -812,15 +834,19 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m } rev, revisionCtx := s.revision, s.revisionCtx candidate := s.candidate + rejected := s.candidateRejectedLocked() s.mu.Unlock() if !codexGatewayBorrowAccountSupported(account) { return req, nil, ErrCodexGatewayBorrowUnavailable } - if candidate == nil || !time.Now().Before(candidate.expires) { + if candidate == nil || rejected || !time.Now().Before(candidate.expires) { if cachedOnly { return req, nil, ErrCodexGatewayBorrowUnavailable } if err := s.prepareSource(req.Context(), rev, revisionCtx); err != nil { + if IsCodexGatewayBorrowFailure(err) { + return req, nil, err + } return req, nil, &CodexGatewayBorrowFailure{Cause: err, Stage: "source_preparation", Reason: "source_prepare_failed"} } s.mu.Lock() @@ -929,7 +955,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m } func (s *CodexGatewayBorrowService) candidateCurrentLocked(rev uint64, candidate *codexGatewayBorrowCandidate) bool { - return s.revision == rev && s.config.Enabled && s.revisionCtx.Err() == nil && s.candidate != nil && s.candidate.sourceID == candidate.sourceID && s.candidate.cookie.Value == candidate.cookie.Value && s.candidate.cookie.Path == candidate.cookie.Path && s.candidate.expires.Equal(candidate.expires) && time.Now().Before(s.candidate.expires) && time.Now().Before(candidate.expires) + return s.revision == rev && s.config.Enabled && s.revisionCtx.Err() == nil && s.candidate != nil && !s.candidateRejectedLocked() && s.candidate.sourceID == candidate.sourceID && s.candidate.cookie.Value == candidate.cookie.Value && s.candidate.cookie.Path == candidate.cookie.Path && s.candidate.expires.Equal(candidate.expires) && time.Now().Before(s.candidate.expires) && time.Now().Before(candidate.expires) } func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBorrowTargetKey, fingerprint string, candidate *codexGatewayBorrowCandidate) (codexGatewayBorrowTargetCheck, bool, error) { @@ -1007,7 +1033,7 @@ func (s *CodexGatewayBorrowService) Verify(ctx context.Context, accountID int64, if err != nil { return result, err } - _, _, err = s.apply(req, a, model, proxy, nil, false, true) + _, _, err = s.applyWithCandidateRetry(req, a, model, proxy, nil, false, true) s.mu.Lock() if check, ok := s.targets[codexGatewayBorrowTargetKey{accountID, model}]; ok { result = check.result diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry.go b/backend/internal/service/codex_gateway_borrow_candidate_retry.go new file mode 100644 index 000000000000..c89a2edcd217 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry.go @@ -0,0 +1,80 @@ +package service + +import ( + "context" + "errors" + "net/http" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" +) + +func (s *CodexGatewayBorrowService) candidateRejectedLocked() bool { + return s.candidate != nil && s.rejectedCookie != "" && borrowHash(s.candidate.cookie.Value) == s.rejectedCookie +} + +// Source completion produces a candidate, not proof that a target can use it. +// Retire a definitively rejected candidate only when it has no live successful +// proof and no other validation in flight. Other targets' usable routes survive. +func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling bool) bool { + var failure *CodexGatewayBorrowFailure + if s == nil || !errors.As(err, &failure) || failure.CookieFingerprint == "" || + (failure.Reason != "target_state_changed" && failure.Reason != "target_route_changed") { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if !s.config.Enabled || s.candidate == nil || s.revisionCtx.Err() != nil { + return false + } + if borrowHash(s.candidate.cookie.Value) != failure.CookieFingerprint { + // A concurrent waiter has already replaced this candidate. Retry the + // published replacement; never retire it using the older observation. + return !cooling + } + now := time.Now() + if now.Before(s.prepareFailedUntil) { + return false + } + for _, check := range s.qualifications { + if check.cookieKey == failure.CookieFingerprint && check.result.Success && now.Before(check.expires) { + return false + } + } + for _, check := range s.targets { + if check.cookieKey == failure.CookieFingerprint && check.validating { + return false + } + } + s.rejectedCookie, s.rejectedCause = failure.CookieFingerprint, err + if cooling { + s.prepareFailedUntil = now.Add(codexGatewayBorrowFailureWait) + s.prepareFailure = err.Error() + } + return true +} + +// One request may try at most one replacement through the same configured +// sources/proxies. There is no background renewal or business replay. A second +// rejection cools acquisition for 15 seconds; per-shot and caller deadlines, +// shared-flight cancellation and the diagnostic's eight-request cap still apply. +func (s *CodexGatewayBorrowService) applyWithCandidateRetry(req *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, cachedOnly, force bool) (*http.Request, *CodexGatewayBorrowApplication, error) { + wire, application, firstErr := s.apply(req, account, model, proxy, profile, cachedOnly, force) + if firstErr == nil || cachedOnly || req == nil || req.Context().Err() != nil || !s.rejectCandidateForRetry(firstErr, false) { + return wire, application, firstErr + } + retryCtx, cancel := context.WithTimeout(req.Context(), codexGatewayBorrowTimeout) + defer cancel() + wire, application, retryErr := s.apply(req.WithContext(retryCtx), account, model, proxy, profile, false, force) + if retryErr != nil { + s.rejectCandidateForRetry(retryErr, true) + // Preserve the original target verdict as well as the replacement's + // complete cause, including cancellation or an exhausted diagnostic cap. + return wire, nil, errors.Join(firstErr, retryErr) + } + if wire != nil && application != nil && application.Applied { + // The bounded preparation context must not cancel the business send. + wire = wire.WithContext(context.WithValue(req.Context(), codexBorrowAppliedContextKey{}, true)) + } + return wire, application, nil +} diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go new file mode 100644 index 000000000000..aa1b34fa676a --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -0,0 +1,90 @@ +package service + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestCodexBorrowRejectedCandidateGetsOneReplacement(t *testing.T) { + for _, replacement := range []string{"usable", "rejected", "identical"} { + t.Run(replacement, func(t *testing.T) { + var sourceCalls, targetCalls int + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 1 { + sourceCalls++ + cookie := "replacement-cookie" + if replacement == "identical" { + cookie = "synthetic-borrowed-cookie" + } + return borrowCoreResponse("gpt-6-astra", "OK", "source-state", "__oailb="+cookie+"; Secure; Path=/; Max-Age=230"), nil + } + targetCalls++ + cookie, err := req.Cookie("__oailb") + assert.NoError(t, err) + model := gjson.GetBytes(borrowCoreBody(t, req), "model").String() + state := "first" + if req.Header.Get("X-Codex-Turn-State") != "" { + state = "changed" + if cookie != nil && cookie.Value == "replacement-cookie" && replacement == "usable" { + state = "first" + } + } + return borrowCoreResponse(model, "OK", state), nil + }, borrowCoreAccount(1), a) + originalExpiry := time.Now().Add(120 * time.Second) + borrowCoreCandidate(s, originalExpiry) + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + require.NoError(t, err) + wire, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.Equal(t, 1, sourceCalls, "at most one replacement source acquisition") + if replacement == "usable" { + require.NoError(t, err) + require.True(t, applied.Applied) + require.NoError(t, wire.Context().Err(), "retry cleanup must not cancel the business request") + require.Contains(t, wire.Header.Get("Cookie"), "__oailb=replacement-cookie") + require.Equal(t, 4, targetCalls) + return + } + require.Error(t, err) + require.Nil(t, applied) + before := targetCalls + _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.Error(t, err) + require.Equal(t, 1, sourceCalls, "exhausted replacement cools acquisition") + require.Equal(t, before, targetCalls, "cooldown does not revalidate a known rejected candidate") + if replacement == "identical" { + require.Equal(t, originalExpiry, s.candidate.expires, "the same rejected credential never gains lifetime") + require.Equal(t, 2, targetCalls) + } else { + require.Equal(t, 4, targetCalls) + } + }) + } +} + +func TestCodexBorrowCandidateReplacementPreservesOtherProofsAndWaiters(t *testing.T) { + for _, other := range []string{"passed", "validating"} { + t.Run(other, func(t *testing.T) { + s := newBorrowCoreTest(t, nil) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + key := borrowHash(s.candidate.cookie.Value) + check := codexGatewayBorrowTargetCheck{cookieKey: key, expires: s.candidate.expires, validating: other == "validating", result: CodexGatewayBorrowVerification{AccountID: 3, Model: "gpt-6-astra", Success: other == "passed"}} + if other == "passed" { + s.qualifications["other"] = check + } else { + s.targets[codexGatewayBorrowTargetKey{3, "gpt-6-astra"}] = check + } + err := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} + require.False(t, s.rejectCandidateForRetry(err, false)) + require.Empty(t, s.rejectedCookie) + }) + } +} diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session.go b/backend/internal/service/codex_gateway_borrow_diagnose_session.go index a82850d1c3e2..b0d497614b95 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_session.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session.go @@ -4,10 +4,12 @@ import ( "encoding/json" "errors" "net/http" + "sort" "strings" "github.com/google/uuid" "github.com/tidwall/gjson" + "github.com/tidwall/sjson" ) const codexBorrowEchoTool = "codex_borrow_echo" @@ -45,15 +47,46 @@ func (d *codexBorrowDiagnostic) prepareHTTPSessionTurn(headers http.Header, payl } func codexBorrowSessionTerminal(raw string) (gjson.Result, error) { + // Native Codex streams may omit output from the final envelope. Preserve + // completed output items (including encrypted reasoning) by their stream + // index; deltas alone never prove a complete item or a completed response. + items := make(map[int]json.RawMessage) 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.output_item.done": + index := int(event.Get("output_index").Int()) + item := event.Get("item") + if index < 0 || index >= 128 || !item.IsObject() { + return gjson.Result{}, errors.New("invalid diagnostic output item") + } + items[index] = json.RawMessage(item.Raw) case "response.completed", "response.done": response := event.Get("response") if response.Get("status").String() == "completed" && !response.Get("error").IsObject() { + if len(response.Get("output").Array()) == 0 && len(items) > 0 { + indexes := make([]int, 0, len(items)) + for index := range items { + indexes = append(indexes, index) + } + sort.Ints(indexes) + output := make([]json.RawMessage, 0, len(items)) + for _, index := range indexes { + output = append(output, items[index]) + } + encoded, err := json.Marshal(output) + if err != nil { + return gjson.Result{}, err + } + materialized, err := sjson.SetRaw(response.Raw, "output", string(encoded)) + if err != nil { + return gjson.Result{}, err + } + response = gjson.Parse(materialized) + } return response, nil } return response, errors.New("upstream terminal was not completed") diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go new file mode 100644 index 000000000000..5a14c30137a3 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go @@ -0,0 +1,37 @@ +package service + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCodexBorrowSessionRebuildsCompletedStreamItems(t *testing.T) { + // Captures the production shape without retaining any real response data: + // complete items are emitted in SSE, while response.completed has output:[]. + first := "data: " + `{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","name":"codex_borrow_echo","call_id":"synthetic-call","arguments":"{\"value\":\"borrow-continuation-check\"}"}}` + "\n\n" + + "data: " + `{"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","encrypted_content":"synthetic-encrypted-item"}}` + "\n\n" + + "data: " + `{"type":"response.completed","response":{"id":"synthetic-1","model":"gpt-6.1-sol","status":"completed","output":[]}}` + "\n\n" + diagnostic := &codexBorrowDiagnostic{history: []any{map[string]string{"role": "user"}}} + result := CodexBorrowDiagnosticResult{RawResponse: first} + diagnostic.completeHTTPSessionTurn(&result, nil, 200) + require.True(t, result.Completed, result.Error) + require.Len(t, diagnostic.history, 4, "preserve reasoning, tool call and its linked output") + require.Contains(t, string(diagnostic.history[1].(json.RawMessage)), "synthetic-encrypted-item") + second := "data: " + `{"type":"response.output_item.done","output_index":0,"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"borrow-continuation-check"}]}}` + "\n\n" + + "data: " + `{"type":"response.completed","response":{"id":"synthetic-2","model":"gpt-6.1-sol","status":"completed","output":[]}}` + "\n\n" + result = CodexBorrowDiagnosticResult{RawResponse: second} + diagnostic.completeHTTPSessionTurn(&result, nil, 200) + require.True(t, result.Completed, result.Error) + require.True(t, result.ToolRoundTrip) + require.Equal(t, codexBorrowEchoValue, result.Answer) +} + +func TestCodexBorrowSessionItemsStillRequireSuccessfulTerminal(t *testing.T) { + item := "data: " + `{"type":"response.output_item.done","output_index":0,"item":{"type":"message","content":[{"type":"output_text","text":"borrow-continuation-check"}]}}` + "\n\n" + for _, ending := range []string{"", "data: " + `{"type":"response.failed","response":{"status":"failed"}}` + "\n\n"} { + _, err := codexBorrowSessionTerminal(item + ending) + require.Error(t, err) + } +} diff --git a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts index f5ad596b630a..14115d8fcc68 100644 --- a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts @@ -153,6 +153,8 @@ export default { not_configured: 'This account and model are absent from the saved configuration', disabled: 'Borrowing is currently disabled', route_expired: 'The borrow route has expired; prepare or validate it again', + source_passed_target_failed: 'The source returned a candidate but target validation failed. Replacement is bounded and demand-driven, with an acquisition cooldown.', + source_returned_rejected_route: 'The source returned the same rejected route. Its lease was not extended; acquisition can resume after cooldown.', source_qualified: 'The source account can provide a borrow route', source_probe_failed: 'Source route collection failed; see technical details', source_response_not_qualified: 'The source response did not meet the borrowing conditions', diff --git a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts index 2a3d254bc690..2b62348d6521 100644 --- a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts @@ -153,6 +153,8 @@ export default { not_configured: '此账号与模型未包含在已保存配置中', disabled: '借用功能当前关闭', route_expired: '借用线路已过期，请重新准备或验证', + source_passed_target_failed: '来源返回了候选，但目标验证失败；按需限次重新获取，冷却期间不重复采集。', + source_returned_rejected_route: '来源再次返回同一条失败线路，未延长租期；冷却后再按需尝试。', source_qualified: '来源账号可以提供借用线路', source_probe_failed: '采集来源线路失败，请查看技术详情', source_response_not_qualified: '来源响应未满足借用条件', From 2453ad4f9e1affc849d72bc920ac943dc6d5cf05 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 10 Oct 2026 03:26:20 +0800 Subject: [PATCH 2/5] Bind candidate recovery to config and original rejection leases --- .../internal/service/codex_gateway_borrow.go | 26 +++- .../codex_gateway_borrow_candidate_retry.go | 21 +++- ...dex_gateway_borrow_candidate_retry_test.go | 111 +++++++++++++++++- 3 files changed, 151 insertions(+), 7 deletions(-) diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index a0fd060c4efa..2dc3e699094b 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -47,6 +47,7 @@ var ( // request failing upstream. Callers must surface it before ordinary account // failover, stream-timeout or health hooks. Unwrap preserves the specific cause. type CodexGatewayBorrowFailure struct { + Revision uint64 CookieFingerprint string Cause error Stage string @@ -80,9 +81,9 @@ func IsCodexGatewayBorrowFailure(err error) bool { return errors.As(err, &failure) } -func borrowVerificationFailure(check codexGatewayBorrowTargetCheck) error { +func borrowVerificationFailure(check codexGatewayBorrowTargetCheck, revision uint64) error { result := check.result - failure := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: result.Reason, Detail: result.Error, Verification: &result, CookieFingerprint: check.cookieKey} + failure := &CodexGatewayBorrowFailure{Revision: revision, Cause: ErrCodexGatewayBorrowUnavailable, Stage: "target_validation", Reason: result.Reason, Detail: result.Error, Verification: &result, CookieFingerprint: check.cookieKey} if !check.retryAfter.IsZero() { retry := check.retryAfter failure.RetryAfter = &retry @@ -277,6 +278,7 @@ type codexGatewayBorrowTargetCheck struct { type CodexGatewayBorrowService struct { rejectedCookie string rejectedCause error + rejectedRoutes map[string]time.Time validationSequence uint64 sourceSequence uint64 qualifications map[string]codexGatewayBorrowTargetCheck @@ -414,6 +416,7 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.config, s.loaded = cloneCodexGatewayBorrowConfig(cfg), true s.candidate = nil s.rejectedCookie, s.rejectedCause = "", nil + s.rejectedRoutes = nil s.setup = CodexGatewayBorrowSetup{} if saved && cfg.Enabled { s.setup = CodexGatewayBorrowSetup{State: "queued", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} @@ -732,7 +735,17 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 return nil, ErrCodexGatewayBorrowChanged } row := CodexGatewayBorrowSourceStatus{AccountID: id, State: "rejected", Reason: "source_probe_failed", CheckedAt: &now, Error: detail} - if err == nil && candidate != nil && s.rejectedCookie != "" && borrowHash(candidate.cookie.Value) == s.rejectedCookie { + previouslyRejected := false + if candidate != nil { + key := borrowHash(candidate.cookie.Value) + if until, found := s.rejectedRoutes[key]; found { + previouslyRejected = now.Before(until) + if !previouslyRejected { + delete(s.rejectedRoutes, key) + } + } + } + if err == nil && candidate != nil && previouslyRejected { // Keep the old candidate and expiry: reacquiring the identical // rejected credential must not renew its lease or qualify it. err = errors.New("source returned the same rejected route") @@ -940,7 +953,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m s.qualifications[qualificationKey] = check s.mu.Unlock() if !result.Success { - return nil, borrowVerificationFailure(check) + return nil, borrowVerificationFailure(check, rev) } return check, nil }) @@ -972,6 +985,9 @@ func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBor return old, false, nil } now := time.Now() + if !now.Before(old.expires) { + return old, false, nil + } if old.result.Success && now.Before(old.expires) { if candidate.expires.Before(old.expires) { old.expires = candidate.expires @@ -979,7 +995,7 @@ func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBor return old, true, nil } if !old.result.Success && now.Before(old.retryAfter) { - return old, false, borrowVerificationFailure(old) + return old, false, borrowVerificationFailure(old, rev) } return old, false, nil } diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry.go b/backend/internal/service/codex_gateway_borrow_candidate_retry.go index c89a2edcd217..493b64b9623b 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry.go @@ -24,7 +24,7 @@ func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling b } s.mu.Lock() defer s.mu.Unlock() - if !s.config.Enabled || s.candidate == nil || s.revisionCtx.Err() != nil { + if s.revision != failure.Revision || !s.config.Enabled || s.candidate == nil || s.revisionCtx.Err() != nil { return false } if borrowHash(s.candidate.cookie.Value) != failure.CookieFingerprint { @@ -32,6 +32,9 @@ func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling b // published replacement; never retire it using the older observation. return !cooling } + if failure.Verification != nil && failure.Verification.ExpiresAt != nil && !failure.Verification.ExpiresAt.Equal(s.candidate.expires) { + return !cooling // A fresh lease cannot be retired by its predecessor's result. + } now := time.Now() if now.Before(s.prepareFailedUntil) { return false @@ -46,6 +49,22 @@ func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling b return false } } + if s.rejectedRoutes == nil { + s.rejectedRoutes = make(map[string]time.Time) + } + // Remember retired credentials through their original lease, including + // A -> B -> A source responses. Never forget a live rejection to make room. + if len(s.rejectedRoutes) >= 1024 { + for key, expires := range s.rejectedRoutes { + if !now.Before(expires) { + delete(s.rejectedRoutes, key) + } + } + if _, known := s.rejectedRoutes[failure.CookieFingerprint]; !known && len(s.rejectedRoutes) >= 1024 { + return false + } + } + s.rejectedRoutes[failure.CookieFingerprint] = s.candidate.expires s.rejectedCookie, s.rejectedCause = failure.CookieFingerprint, err if cooling { s.prepareFailedUntil = now.Add(codexGatewayBorrowFailureWait) diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go index aa1b34fa676a..be94fcb6e374 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -3,6 +3,7 @@ package service import ( "context" "net/http" + "sync/atomic" "testing" "time" @@ -82,9 +83,117 @@ func TestCodexBorrowCandidateReplacementPreservesOtherProofsAndWaiters(t *testin } else { s.targets[codexGatewayBorrowTargetKey{3, "gpt-6-astra"}] = check } - err := &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} + err := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} require.False(t, s.rejectCandidateForRetry(err, false)) require.Empty(t, s.rejectedCookie) }) } } + +func TestCodexBorrowReplacementCannotRetireNewConfiguration(t *testing.T) { + s := newBorrowCoreTest(t, nil) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + err := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: borrowHash(s.candidate.cookie.Value)} + s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{4}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6-astra"}}, false) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + require.False(t, s.rejectCandidateForRetry(err, false), "even an identical cookie belongs to the new config revision") + require.Empty(t, s.rejectedCookie) +} + +func TestCodexBorrowReplacementBudgetDoesNotCoolOtherCallers(t *testing.T) { + var sourceCalls int + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 1 { + sourceCalls++ + return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=replacement-cookie; Secure; Path=/; Max-Age=230"), nil + } + state := "first" + cookie, _ := req.Cookie("__oailb") + if req.Header.Get("X-Codex-Turn-State") != "" && cookie != nil && cookie.Value != "replacement-cookie" { + state = "changed" + } + return borrowCoreResponse("gpt-6.1-sol", "OK", state), nil + }, borrowCoreAccount(1), a) + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + require.NoError(t, err) + var used atomic.Int32 + limited := context.WithValue(req.Context(), codexBorrowDiagnosticContextKey{}, &codexBorrowDiagnostic{limit: 2, requests: &used, borrow: true}) + _, _, err = s.Apply(req.WithContext(limited), a, "gpt-6.1-sol", "", nil, false) + require.ErrorIs(t, err, ErrCodexBorrowDiagnosticBudget) + require.EqualValues(t, 2, used.Load()) + require.Zero(t, sourceCalls) + require.True(t, s.prepareFailedUntil.IsZero(), "a caller's cap is not a source failure") + _, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + require.True(t, applied.Applied) + require.Equal(t, 1, sourceCalls) +} + +func TestCodexBorrowReplacementCannotReplayEarlierRejectedCredential(t *testing.T) { + var sources, targets int + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 1 { + sources++ + cookie := "replacement-cookie" + if sources == 2 { + cookie = "synthetic-borrowed-cookie" + } + return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb="+cookie+"; Secure; Path=/; Max-Age=230"), nil + } + targets++ + state := "first" + if req.Header.Get("X-Codex-Turn-State") != "" { + state = "changed" + } + return borrowCoreResponse("gpt-6.1-sol", "OK", state), nil + }, borrowCoreAccount(1), 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.Error(t, err) + require.Equal(t, 1, sources) + require.Equal(t, 4, targets) + expires := s.candidate.expires + s.prepareFailedUntil = time.Time{} + _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.Error(t, err) + require.Equal(t, 2, sources) + require.Equal(t, 4, targets, "an earlier rejected cookie is not qualified again") + require.Equal(t, "replacement-cookie", s.candidate.cookie.Value) + require.Equal(t, expires, s.candidate.expires) +} + +func TestCodexBorrowExpiredRejectionRequiresFreshSourceAndProof(t *testing.T) { + var sources, targets int + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(_ *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if id == 1 { + sources++ + return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=synthetic-borrowed-cookie; Secure; Path=/; Max-Age=230"), nil + } + targets++ + return borrowCoreResponse("gpt-6.1-sol", "OK", "fresh-stable-state"), nil + }, borrowCoreAccount(1), a) + oldExpiry := time.Now().Add(-time.Second) + borrowCoreCandidate(s, oldExpiry) + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + require.NoError(t, err) + key := borrowHash(s.candidate.cookie.Value) + s.rejectedCookie = key + s.rejectedRoutes = map[string]time.Time{key: oldExpiry} + s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{ + key: borrowTargetFingerprint(req, a, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value), cookieKey: key, + expires: oldExpiry, retryAfter: time.Now().Add(time.Minute), + result: CodexGatewayBorrowVerification{Reason: "target_state_changed", ExpiresAt: &oldExpiry}, + } + _, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + require.True(t, applied.Applied) + require.Equal(t, 1, sources, "expiration requires a new source observation") + require.Equal(t, 2, targets, "an old failure cooldown cannot certify or reject the new lease") + require.Empty(t, s.rejectedRoutes) +} From ab5fbbdecf61ffbac15a044cd65da8a8209e1d94 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 10 Oct 2026 03:34:02 +0800 Subject: [PATCH 3/5] Check streamed diagnostic test assertions explicitly --- .../service/codex_gateway_borrow_candidate_retry_test.go | 3 ++- .../service/codex_gateway_borrow_diagnose_session_test.go | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go index be94fcb6e374..21c5ced07492 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -109,7 +109,8 @@ func TestCodexBorrowReplacementBudgetDoesNotCoolOtherCallers(t *testing.T) { return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=replacement-cookie; Secure; Path=/; Max-Age=230"), nil } state := "first" - cookie, _ := req.Cookie("__oailb") + cookie, cookieErr := req.Cookie("__oailb") + assert.NoError(t, cookieErr) if req.Header.Get("X-Codex-Turn-State") != "" && cookie != nil && cookie.Value != "replacement-cookie" { state = "changed" } diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go index 5a14c30137a3..850771748a74 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go @@ -18,7 +18,9 @@ func TestCodexBorrowSessionRebuildsCompletedStreamItems(t *testing.T) { diagnostic.completeHTTPSessionTurn(&result, nil, 200) require.True(t, result.Completed, result.Error) require.Len(t, diagnostic.history, 4, "preserve reasoning, tool call and its linked output") - require.Contains(t, string(diagnostic.history[1].(json.RawMessage)), "synthetic-encrypted-item") + reasoning, ok := diagnostic.history[1].(json.RawMessage) + require.True(t, ok) + require.Contains(t, string(reasoning), "synthetic-encrypted-item") second := "data: " + `{"type":"response.output_item.done","output_index":0,"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"borrow-continuation-check"}]}}` + "\n\n" + "data: " + `{"type":"response.completed","response":{"id":"synthetic-2","model":"gpt-6.1-sol","status":"completed","output":[]}}` + "\n\n" result = CodexBorrowDiagnosticResult{RawResponse: second} From 111e31d5cc77f1b4e1dc282121237cbb21c4f873 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 10 Oct 2026 03:44:30 +0800 Subject: [PATCH 4/5] Report upstream completion separately from diagnostic answer checks --- .../internal/service/codex_gateway_borrow_diagnose.go | 2 +- .../service/codex_gateway_borrow_diagnose_session.go | 6 +++--- .../codex_gateway_borrow_diagnose_session_test.go | 9 +++++++++ .../src/components/admin/codex/CodexBorrowActivity.vue | 3 ++- 4 files changed, 15 insertions(+), 5 deletions(-) diff --git a/backend/internal/service/codex_gateway_borrow_diagnose.go b/backend/internal/service/codex_gateway_borrow_diagnose.go index 6eb5c5ac3a19..ee81bfe86f4b 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose.go @@ -184,7 +184,7 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB result := s.diagnoseTurn(operation, account, request.Model, previous) result.Mode, result.Turn = mode, turn emit(CodexBorrowDiagnosticEvent{Type: "result", Mode: mode, Result: &result, Requests: requests.Load(), Limit: limit}) - if !result.Completed { + if !result.Completed || result.Error != "" { break } previous = result.ResponseID diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session.go b/backend/internal/service/codex_gateway_borrow_diagnose_session.go index b0d497614b95..775b218629c6 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_session.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session.go @@ -99,6 +99,7 @@ func codexBorrowSessionTerminal(raw string) (gjson.Result, error) { func (d *codexBorrowDiagnostic) completeHTTPSessionTurn(result *CodexBorrowDiagnosticResult, sendErr error, status int) { response, err := codexBorrowSessionTerminal(result.RawResponse) + result.Completed = err == nil && status >= 200 && status < 300 if sendErr != nil { result.Error = sendErr.Error() return @@ -137,9 +138,8 @@ func (d *codexBorrowDiagnostic) completeHTTPSessionTurn(result *CodexBorrowDiagn } } result.Answer = strings.TrimSpace(answer.String()) - result.Completed = result.Answer == codexBorrowEchoValue - result.ToolRoundTrip = result.Completed - if !result.Completed { + result.ToolRoundTrip = result.Answer == codexBorrowEchoValue + if !result.ToolRoundTrip { result.Error = "continuation did not return the fixed tool result" } } diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go index 850771748a74..821257e16e4b 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go @@ -37,3 +37,12 @@ func TestCodexBorrowSessionItemsStillRequireSuccessfulTerminal(t *testing.T) { require.Error(t, err) } } + +func TestCodexBorrowSessionSeparatesCompletionFromTaskResult(t *testing.T) { + diagnostic := &codexBorrowDiagnostic{history: []any{"user", "tool-output"}} + result := CodexBorrowDiagnosticResult{RawResponse: "data: " + `{"type":"response.completed","response":{"status":"completed","output":[{"type":"message","content":[{"type":"output_text","text":"different answer"}]}]}}` + "\n\n"} + diagnostic.completeHTTPSessionTurn(&result, nil, 200) + require.True(t, result.Completed, "upstream completion is independent of the fixed-answer check") + require.False(t, result.ToolRoundTrip) + require.NotEmpty(t, result.Error) +} diff --git a/frontend/src/components/admin/codex/CodexBorrowActivity.vue b/frontend/src/components/admin/codex/CodexBorrowActivity.vue index 15db98aefb0d..e67a238adbf0 100644 --- a/frontend/src/components/admin/codex/CodexBorrowActivity.vue +++ b/frontend/src/components/admin/codex/CodexBorrowActivity.vue @@ -33,10 +33,11 @@

{{ phase === 'ordinary' ? text('正在检查普通路径…', 'Checking ordinary path…') : text('正在准备并检查借用路径…', 'Preparing and checking borrowed path…') }} {{ text('已发送', 'Requests sent') }} {{ requests }} / 8

-

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

+

{{ result.mode === 'ordinary' ? text('普通路径', 'Ordinary path') : text('借用路径', 'Borrowed path') }} · {{ text('第', 'Turn') }} {{ result.turn }} {{ result.completed ? text('轮：上游已完成', ': upstream 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') }}

+

{{ text('上游已完成，但本轮检查未通过。', 'The upstream completed, but this diagnostic check did not pass.') }}

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

{{ text('本次借用验证依据', 'Borrow verification for this request') }}
{{ JSON.stringify(result.verification, null, 2) }}
{{ result.error }}
From a171c96a9fa56c6adc05535d06c24b47e9fe36bb Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 10 Oct 2026 04:12:07 +0800 Subject: [PATCH 5/5] Preserve active validation identities and fail missing source dependencies --- .../internal/service/codex_gateway_borrow.go | 11 ++++++ .../codex_gateway_borrow_candidate_retry.go | 26 +++++++++++-- ...dex_gateway_borrow_candidate_retry_test.go | 38 +++++++++++++++++++ 3 files changed, 71 insertions(+), 4 deletions(-) diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 2dc3e699094b..0445f484a66f 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -279,6 +279,7 @@ type CodexGatewayBorrowService struct { rejectedCookie string rejectedCause error rejectedRoutes map[string]time.Time + validatingRoutes map[string]int validationSequence uint64 sourceSequence uint64 qualifications map[string]codexGatewayBorrowTargetCheck @@ -417,6 +418,7 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.candidate = nil s.rejectedCookie, s.rejectedCause = "", nil s.rejectedRoutes = nil + s.validatingRoutes = nil s.setup = CodexGatewayBorrowSetup{} if saved && cfg.Enabled { s.setup = CodexGatewayBorrowSetup{State: "queued", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} @@ -695,6 +697,10 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Unlock() return nil, nil } + if s.accounts == nil || s.gateway == nil { + s.mu.Unlock() + return nil, ErrCodexGatewayBorrowUnavailable + } if time.Now().Before(s.prepareFailedUntil) { detail := s.prepareFailure cause := s.rejectedCause @@ -708,6 +714,9 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 return nil, ErrCodexGatewayBorrowUnavailable } ids := append([]int64{}, s.config.SourceAccountIDs...) + if s.sources == nil { + s.sources = make(map[int64]CodexGatewayBorrowSourceStatus) + } s.preparing = true s.sourceSequence++ sequence := s.sourceSequence @@ -903,9 +912,11 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m } s.validationSequence++ validationID := s.validationSequence + validationDone := s.trackCandidateValidationLocked(rev, borrowHash(candidate.cookie.Value)) s.targets[targetKey] = codexGatewayBorrowTargetCheck{validationID: validationID, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), validating: true, result: CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, Reason: "validating", CheckedAt: time.Now()}} s.mu.Unlock() + defer validationDone() probeCtx, cancel := borrowRevisionContext(operation, revisionCtx, codexGatewayBorrowTimeout) defer cancel() credential, credentialErr := resolveCredentialAccount(probeCtx, s.accounts, account) diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry.go b/backend/internal/service/codex_gateway_borrow_candidate_retry.go index 493b64b9623b..9b422db36b1f 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry.go @@ -13,6 +13,24 @@ func (s *CodexGatewayBorrowService) candidateRejectedLocked() bool { return s.candidate != nil && s.rejectedCookie != "" && borrowHash(s.candidate.cookie.Value) == s.rejectedCookie } +func (s *CodexGatewayBorrowService) trackCandidateValidationLocked(revision uint64, cookie string) func() { + if s.validatingRoutes == nil { + s.validatingRoutes = make(map[string]int) + } + s.validatingRoutes[cookie]++ + return func() { + s.mu.Lock() + defer s.mu.Unlock() + if s.revision != revision { + return + } + s.validatingRoutes[cookie]-- + if s.validatingRoutes[cookie] == 0 { + delete(s.validatingRoutes, cookie) + } + } +} + // Source completion produces a candidate, not proof that a target can use it. // Retire a definitively rejected candidate only when it has no live successful // proof and no other validation in flight. Other targets' usable routes survive. @@ -44,10 +62,10 @@ func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling b return false } } - for _, check := range s.targets { - if check.cookieKey == failure.CookieFingerprint && check.validating { - return false - } + // The presentation index holds only the latest check per account/model; + // another request identity can still be validating the same candidate. + if s.validatingRoutes[failure.CookieFingerprint] > 0 { + return false } if s.rejectedRoutes == nil { s.rejectedRoutes = make(map[string]time.Time) diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go index 21c5ced07492..581754969c1a 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -82,6 +82,7 @@ func TestCodexBorrowCandidateReplacementPreservesOtherProofsAndWaiters(t *testin s.qualifications["other"] = check } else { s.targets[codexGatewayBorrowTargetKey{3, "gpt-6-astra"}] = check + s.validatingRoutes = map[string]int{key: 1} } err := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} require.False(t, s.rejectCandidateForRetry(err, false)) @@ -90,6 +91,43 @@ func TestCodexBorrowCandidateReplacementPreservesOtherProofsAndWaiters(t *testin } } +func TestCodexBorrowReplacementWaitsForHiddenRequestIdentity(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var first atomic.Bool + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + if first.CompareAndSwap(false, true) { + close(started) + select { + case <-release: + case <-req.Context().Done(): + return nil, req.Context().Err() + } + } + 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) + done := make(chan error, 1) + go func() { _, _, runErr := s.Apply(req, a, "gpt-6.1-sol", "", nil, false); done <- runErr }() + <-started + s.mu.Lock() + key := borrowHash(s.candidate.cookie.Value) + failure := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} + // Another identity completed later and owns the presentation row, while + // the first identity's real validation is still using this candidate. + s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{cookieKey: key, validationID: 999} + s.mu.Unlock() + retired := s.rejectCandidateForRetry(failure, false) + close(release) + require.False(t, retired) + require.NoError(t, <-done) + s.mu.Lock() + defer s.mu.Unlock() + require.Empty(t, s.validatingRoutes) +} + func TestCodexBorrowReplacementCannotRetireNewConfiguration(t *testing.T) { s := newBorrowCoreTest(t, nil) borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))