Recover rejected Codex candidates and streamed tool diagnostics

.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

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,14 +523,17 @@ 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 {
 		for _, model := range s.config.Models {
 			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,14 +996,17 @@ 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
 		}
 		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

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
+}