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..0445f484a66f 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -47,12 +47,14 @@ 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 + Revision uint64 + CookieFingerprint string + Cause error + Stage string + Reason string + Detail string + RetryAfter *time.Time + Verification *CodexGatewayBorrowVerification } func (e *CodexGatewayBorrowFailure) Error() string { @@ -79,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} + 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 @@ -274,6 +276,10 @@ type codexGatewayBorrowTargetCheck struct { } type CodexGatewayBorrowService struct { + rejectedCookie string + rejectedCause error + rejectedRoutes map[string]time.Time + validatingRoutes map[string]int validationSequence uint64 sourceSequence uint64 qualifications map[string]codexGatewayBorrowTargetCheck @@ -410,6 +416,9 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.revision++ s.config, s.loaded = cloneCodexGatewayBorrowConfig(cfg), true 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()} @@ -514,6 +523,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 +533,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 +564,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,19 +693,30 @@ 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 s.accounts == nil || s.gateway == nil { + s.mu.Unlock() + return nil, ErrCodexGatewayBorrowUnavailable + } 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) } 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 @@ -717,6 +744,22 @@ 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} + 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") + 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 +770,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 +788,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 +820,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 +856,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() @@ -864,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) @@ -914,7 +964,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 }) @@ -929,7 +979,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) { @@ -946,6 +996,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 @@ -953,7 +1006,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 } @@ -1007,7 +1060,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..9b422db36b1f --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry.go @@ -0,0 +1,117 @@ +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 +} + +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. +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.revision != failure.Revision || !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 + } + 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 + } + for _, check := range s.qualifications { + if check.cookieKey == failure.CookieFingerprint && check.result.Success && now.Before(check.expires) { + 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) + } + // 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) + 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..581754969c1a --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -0,0 +1,238 @@ +package service + +import ( + "context" + "net/http" + "sync/atomic" + "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 + 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)) + require.Empty(t, s.rejectedCookie) + }) + } +} + +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)) + 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, cookieErr := req.Cookie("__oailb") + assert.NoError(t, cookieErr) + 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) +} 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 a82850d1c3e2..775b218629c6 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") @@ -66,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 @@ -104,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 new file mode 100644 index 000000000000..821257e16e4b --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_diagnose_session_test.go @@ -0,0 +1,48 @@ +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") + 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} + 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) + } +} + +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 }}
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: '来源响应未满足借用条件',