修复 Codex 探针断流后的连续拒绝和状态页陈旧显示

.downstream/native-domains.md

@@ -82,8 +82,10 @@ request usage and upstream completion. Pelican stays at `/admin/pelican-tests`;
 its legacy comparison URL redirects there. Page reads and navigation only read
 local data. Unsaved configuration stays in browser-session storage. Saving an
 enabled configuration starts exactly one finite preparation, shown as queued
-before the worker starts. Status polls every five seconds only while work is
-active; countdowns expire routes without renewing them.
+before the worker starts. While enabled and visible, status polls every five
+seconds, including idle periods so later business failures become visible. Hidden pages stop polling and
+refresh on return. Failed or stale reads cannot display current readiness;
+countdowns expire routes without renewing them.
 
 `GET/PUT /admin/codex-gateway-borrow/config` owns one JSON setting,
 `codex_gateway_borrow_config`: `enabled`, `source_account_ids`,
@@ -126,9 +128,14 @@ The original cache identity includes route cookie, proxy, Authorization,
 ChatGPT-Account-ID, UA, Originator, Version, business STATE and TLS profile.
 The host adds actual model and fingerprint policy revision. Trace IDs, native
 session headers and business tier are not additional cache rules. STATE changes
-therefore require fresh validation. A failed target cools that qualification for
-15 seconds; it does not discard a source candidate or force a replacement attempt.
-After cooldown the next request may validate again; missing or expired candidates
+therefore require fresh validation. A completed rejection or protocol failure
+cools that qualification for 15 seconds. A transport interruption instead retains failed evidence with
+`retry_on_next_request`: the next independent request starts a fresh complete
+two-shot validation through the existing shared flight, without a negative-cache
+rejection. It never retries an uncertain observation within the failed request,
+admits an incomplete probe or replays a business generation. Neither failure
+discards a source candidate or extends its lease. After a rejection cooldown
+the next request may validate again; missing or expired candidates
 are acquired on demand. There is no renewal timer, exit rotation, ordinary fallback
 or business replay. Business WS keeps its existing continuation anchors.
 
@@ -168,7 +175,9 @@ separate from existing `count` (dispatches) and `applied_count`. `dispatched`,
 `failure_stage`, client and gateway request IDs distinguish a local rejection
 from an inference failure. Original target reasons survive failure cooldowns;
 Ops classifies local borrowing rejection as gateway/routing with upstream status
-zero, retaining the compatible client 503 code. Business evidence is displayed
+zero, retaining the compatible client 503 code. The event also retains probe
+shot/transport evidence, next-request recovery policy and whether a rejection
+came from the qualification cache. Business evidence is displayed
 ahead of diagnostics. Candidate expiry never implies a failed check is usable.
 
 `POST /admin/codex-gateway-borrow/diagnose` streams an explicit fixed-account

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 {
+	CacheHit          bool
 	Revision          uint64
 	CookieFingerprint string
 	Cause             error
@@ -81,7 +82,7 @@ func IsCodexGatewayBorrowFailure(err error) bool {
 	return errors.As(err, &failure)
 }
 
-func borrowVerificationFailure(check codexGatewayBorrowTargetCheck, revision uint64) error {
+func borrowVerificationFailure(check codexGatewayBorrowTargetCheck, revision uint64) *CodexGatewayBorrowFailure {
 	result := check.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() {
@@ -199,6 +200,9 @@ type CodexGatewayBorrowSourceStatus struct {
 }
 
 type CodexGatewayBorrowVerification struct {
+	RetryOnNextRequest        bool       `json:"retry_on_next_request,omitempty"`
+	FailedShot                string     `json:"failed_shot,omitempty"`
+	FailureKind               string     `json:"failure_kind,omitempty"`
 	MintTransportPhase        string     `json:"mint_transport_phase,omitempty"`
 	ContinueTransportPhase    string     `json:"continue_transport_phase,omitempty"`
 	MintTransportAttempts     int        `json:"mint_transport_attempts,omitempty"`
@@ -555,6 +559,8 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus {
 					row.State = "ready"
 				} else if c.validating {
 					row.State = "validating"
+				} else if c.result.RetryOnNextRequest {
+					row.State = "retry_pending"
 				} else if c.result.Success {
 					row.State, row.Reason = "expired", "route_expired"
 				}
@@ -940,7 +946,7 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m
 			acquisition.Phase = "failed"
 		}
 		check := codexGatewayBorrowTargetCheck{acquisition: acquisition, validationID: validationID, identityRevision: identityRevision, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: result}
-		if !result.Success {
+		if !result.Success && !result.RetryOnNextRequest {
 			check.retryAfter = time.Now().Add(codexGatewayBorrowFailureWait)
 			check.acquisition.RetryAfter = &check.retryAfter
 		}
@@ -1006,8 +1012,10 @@ 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, rev)
+	if !old.result.Success && !old.result.RetryOnNextRequest && now.Before(old.retryAfter) {
+		failure := borrowVerificationFailure(old, rev)
+		failure.CacheHit = true
+		return old, false, failure
 	}
 	return old, false, nil
 }

backend/internal/service/codex_gateway_borrow_core_test.go

@@ -284,7 +284,7 @@ func TestCodexGatewayBorrowTargetCacheModelAndFingerprintIsolation(t *testing.T)
 }
 
 func TestCodexGatewayBorrowTargetChangedStateRouteAndFailureCooldown(t *testing.T) {
-	for _, mode := range []string{"state", "cookie", "missing", "incomplete"} {
+	for _, mode := range []string{"state", "cookie", "missing", "incomplete", "upstream_failed"} {
 		t.Run(mode, func(t *testing.T) {
 			var calls int
 			a := borrowCoreAccount(2)
@@ -300,6 +300,9 @@ func TestCodexGatewayBorrowTargetChangedStateRouteAndFailureCooldown(t *testing.
 				if mode == "incomplete" {
 					return &http.Response{StatusCode: 200, Header: http.Header{"X-Codex-Turn-State": []string{"first"}}, Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n"))}, nil
 				}
+				if mode == "upstream_failed" {
+					return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("data: {\"type\":\"response.failed\"}\n\n"))}, nil
+				}
 				state := "first"
 				if calls == 2 {
 					state = "changed"
@@ -314,10 +317,16 @@ func TestCodexGatewayBorrowTargetChangedStateRouteAndFailureCooldown(t *testing.
 			initial := calls
 			_, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false)
 			require.Error(t, err)
-			require.Equal(t, initial, calls, "failure stays in 15-second cooldown")
 			status := s.Status()
 			require.False(t, status.Targets[1].CacheValid)
-			require.NotNil(t, status.Targets[1].RetryAfter)
+			if mode == "incomplete" {
+				require.Equal(t, initial+1, calls, "each independent request revalidates the incomplete stream")
+				require.Nil(t, status.Targets[1].RetryAfter)
+				require.True(t, status.Targets[1].RetryOnNextRequest)
+			} else {
+				require.Equal(t, initial, calls, "rejection stays in 15-second cooldown")
+				require.NotNil(t, status.Targets[1].RetryAfter)
+			}
 		})
 	}
 }

backend/internal/service/codex_gateway_borrow_http.go

@@ -12,6 +12,12 @@ import (
 
 const CodexGatewayBorrowPreparationFailureReason GatewayFailureReason = "codex_gateway_borrow_preparation_failed"
 
+type CodexGatewayBorrowFailureEvidence struct {
+	CacheHit     bool                            `json:"cache_hit"`
+	RetryAfter   *time.Time                      `json:"retry_after,omitempty"`
+	Verification *CodexGatewayBorrowVerification `json:"verification,omitempty"`
+}
+
 // The existing failover owner may select another account. A local preparation
 // failure is not an inference attempt or evidence against the selected account.
 func NewCodexGatewayBorrowRequestFailure(_ error) *UpstreamFailoverError {
@@ -61,6 +67,11 @@ func RecordCodexGatewayBorrowPreparationFailure(c *gin.Context, account *Account
 		event.ProxyID, event.ProxyName = opsUpstreamProxyID(account), opsUpstreamProxyName(account)
 		event.Platform, event.AccountID, event.AccountName = account.Platform, account.ID, account.Name
 	}
+	var failure *CodexGatewayBorrowFailure
+	if errors.As(err, &failure) && failure.Verification != nil {
+		verification := *failure.Verification
+		event.BorrowQualification = &CodexGatewayBorrowFailureEvidence{CacheHit: failure.CacheHit, RetryAfter: failure.RetryAfter, Verification: &verification}
+	}
 	appendOpsUpstreamError(c, event)
 }
 

backend/internal/service/codex_gateway_borrow_probe.go

@@ -13,6 +13,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"strings"
 	"time"
@@ -41,6 +42,7 @@ type codexGatewayBorrowObservation struct {
 	errorText         string
 	complete          bool
 	incomplete        bool
+	interrupted       bool
 	modelMismatch     bool
 	cookies           []*http.Cookie
 	cookiesObservedAt time.Time
@@ -267,7 +269,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h
 		return result
 	}
 	if err != nil || !borrowShotUsable(mint) {
-		result.Error = borrowObservationError(mint, err)
+		borrowTargetObservationFailure(ctx, &result, "mint", mint, err)
 		return result
 	}
 	if mint.state == "" {
@@ -299,7 +301,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h
 		return result
 	}
 	if err != nil || !borrowShotUsable(continued) {
-		result.Error = borrowObservationError(continued, err)
+		borrowTargetObservationFailure(ctx, &result, "continue", continued, err)
 		return result
 	}
 	result.NewTicket = continued.state != "" && continued.state != mint.state
@@ -315,6 +317,28 @@ func borrowShotUsable(shot codexGatewayBorrowObservation) bool {
 	return shot.status == http.StatusOK && shot.complete && shot.errorText == ""
 }
 
+// A transport interruption is inconclusive, not a reusable rejection. Keep its
+// evidence, but let the next independent request run a fresh complete pair.
+// Never retry this observation or replay an uncertain business generation here.
+func borrowTargetObservationFailure(ctx context.Context, result *CodexGatewayBorrowVerification, phase string, shot codexGatewayBorrowObservation, err error) {
+	result.Error = borrowObservationError(shot, err)
+	result.FailedShot, result.FailureKind = phase, "response"
+	if ctx.Err() != nil || errors.Is(err, context.Canceled) {
+		result.FailureKind = "cancelled"
+		return
+	}
+	if shot.interrupted {
+		result.FailureKind = "stream_incomplete"
+		result.RetryOnNextRequest = true
+	}
+	var transport *codexBorrowTransportError
+	var network net.Error
+	if err != nil && (errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || errors.As(err, &transport) || errors.As(err, &network)) {
+		result.FailureKind = "transport"
+		result.RetryOnNextRequest = true
+	}
+}
+
 func borrowObservationError(shot codexGatewayBorrowObservation, err error) string {
 	if shot.errorText != "" {
 		return shot.errorText
@@ -505,6 +529,7 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account *
 			// business sender's early-terminal optimization for this criterion.
 			validationErr := validateCodexGatewayBorrowPinnedProbeResponse(data)
 			shot.complete = validationErr == nil
+			shot.interrupted = errors.Is(validationErr, errCodexBorrowProbeNoCompletion) || errors.Is(validationErr, errCodexBorrowProbeUnterminated)
 			if validationErr == nil {
 				shot.errorText = ""
 			} else if shot.errorText == "" {

backend/internal/service/codex_gateway_borrow_probe_terminal.go

@@ -13,6 +13,11 @@ import (
 	"strings"
 )
 
+var (
+	errCodexBorrowProbeNoCompletion = errors.New("probe did not complete successfully")
+	errCodexBorrowProbeUnterminated = errors.New("unterminated probe event")
+)
+
 // A clean EOF or [DONE] is not evidence of a successful Responses generation.
 func validateCodexGatewayBorrowPinnedProbeResponse(body []byte) error {
 	completed := false
@@ -85,11 +90,11 @@ func validateCodexGatewayBorrowPinnedProbeResponse(body []byte) error {
 		}
 		// A trailing unterminated event may be truncated, even if its JSON parses.
 		if len(data) > 0 {
-			return errors.New("unterminated probe event")
+			return errCodexBorrowProbeUnterminated
 		}
 	}
 	if !completed {
-		return errors.New("probe did not complete successfully")
+		return errCodexBorrowProbeNoCompletion
 	}
 	return nil
 }