补齐 Codex 借用的原版 Cookie 与流完成契约

.downstream/upstream-review-ranxi-v2.10.3.md

@@ -29,6 +29,15 @@ Official Wei-Shaw v0.2.15 is already integrated; do not merge it again.
 - Extra cache keys and business-tier injection into the fixed target probe.
 - Target rejection forcing a fresh source candidate and retry in the non-rotation path.
 - Separate source/target HTTP manager configuration copies and custom TLS policy.
+- Displacing a live active source with a backup or failing refresh despite a
+  still-valid primary/backup. Original active-source preference is retained.
+- Stopping target probes at an early terminal instead of reading their bounded
+  body to EOF and applying the original completion/event-delimiter validator.
+  Ordinary business/Pelican streaming retains its own completion behavior.
+- Accepting a source session cookie without expiry, or a non-SSE source response.
+- Scope-filtering target route replacement and waiting for its body before rejection.
+  The target probe now uses the original header-time route check and first usable
+  `__cflb` rule; only sources enforce the original Secure/domain/path/expiry filter.
 
 Source alignment and offline regression do not prove successful live borrowing.
 Production and live acceptance are recorded separately in the deployment evidence.

backend/internal/service/codex_gateway_borrow.go

@@ -765,6 +765,14 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6
 				}
 			}
 			s.sources[id] = row
+			// Original non-rotation preparation may still have a usable primary or
+			// backup after a failed refresh. Preserve it without extending its lease.
+			if current := s.currentCandidateLocked(); current != nil && time.Now().Before(current.expires) {
+				s.acquisition.Phase, s.acquisition.Error = "candidate_retained", lastErr.Error()
+				s.prepareFailedUntil, s.prepareFailure = time.Time{}, ""
+				s.mu.Unlock()
+				return nil, nil
+			}
 			s.mu.Unlock()
 		}
 		s.mu.Lock()

backend/internal/service/codex_gateway_borrow_candidates.go

@@ -35,7 +35,10 @@ func (s *CodexGatewayBorrowService) storeCandidateLocked(candidate *codexGateway
 		clamp(old)
 	}
 	s.candidates[candidate.sourceID] = candidate
-	s.candidate = candidate
+	// A backup does not displace a live primary from a different source.
+	if s.candidate == nil || !now.Before(s.candidate.expires) || s.candidate.sourceID == candidate.sourceID {
+		s.candidate = candidate
+	}
 	// Other source proofs remain usable. Cookie, model and identity are already
 	// in their keys, so no target can accidentally inherit another qualification.
 }

backend/internal/service/codex_gateway_borrow_cookie_contract_test.go

@@ -0,0 +1,174 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	"github.com/stretchr/testify/require"
+)
+
+type borrowHeaderRejectedBody struct{ reads, closes int }
+
+func (b *borrowHeaderRejectedBody) Read([]byte) (int, error) {
+	b.reads++
+	return 0, io.ErrUnexpectedEOF
+}
+func (b *borrowHeaderRejectedBody) Close() error { b.closes++; return nil }
+
+// Fixed contract: ranxi fd1b5ee4 ProbeOpenAICodexStateRoute checks every
+// replacement __oailb at response headers, before consuming a potentially
+// stalled/error body. Cookie scope does not hide a target route replacement.
+func TestCodexBorrowPinnedRouteRejectsAtHeaders(t *testing.T) {
+	for _, cookie := range []string{
+		"__oailb=changed; Path=/unrelated; Domain=other.invalid",
+		"__oailb=synthetic-borrowed-cookie; Max-Age=230; Expires=Thu, 01 Jan 1970 00:00:00 GMT",
+		"__oailb=synthetic-borrowed-cookie; Max-Age=0",
+	} {
+		body := &borrowHeaderRejectedBody{}
+		calls := 0
+		account := borrowCoreAccount(2)
+		s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+			calls++
+			return &http.Response{StatusCode: 200, Header: http.Header{"Set-Cookie": []string{cookie}}, Body: body}, nil
+		}, account)
+		borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+		_, req, _, err := s.accountTemplate(context.Background(), account.ID, "gpt-6-astra")
+		require.NoError(t, err)
+		_, applied, err := s.Apply(req, account, "gpt-6-astra", "", nil, false)
+		require.Nil(t, applied)
+		var failure *CodexGatewayBorrowFailure
+		require.ErrorAs(t, err, &failure)
+		require.Equal(t, "target_route_changed", failure.Reason)
+		require.True(t, failure.Verification.RouteChanged)
+		require.False(t, failure.Verification.MintCompleted)
+		require.Zero(t, body.reads)
+		require.Equal(t, 1, body.closes)
+		require.Equal(t, 1, calls)
+	}
+}
+
+func TestCodexBorrowPinnedContinuationUsesFirstUsableCflb(t *testing.T) {
+	account := borrowCoreAccount(2)
+	calls := 0
+	s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+		calls++
+		if calls == 1 {
+			return borrowCoreResponse("gpt-6-astra", "OK", "state",
+				"__cflb=deleted; Max-Age=0", "__cflb=first; Path=/other; Expires=Thu, 01 Jan 1970 00:00:00 GMT", "__cflb=later"), nil
+		}
+		require.Equal(t, "__cflb=first; __oailb=synthetic-borrowed-cookie", req.Header.Get("Cookie"))
+		return borrowCoreResponse("gpt-6-astra", "OK", "state"), nil
+	}, account)
+	borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+	_, req, _, err := s.accountTemplate(context.Background(), account.ID, "gpt-6-astra")
+	require.NoError(t, err)
+	_, applied, err := s.Apply(req, account, "gpt-6-astra", "", nil, false)
+	require.NoError(t, err)
+	require.True(t, applied.Applied)
+	require.Equal(t, 2, calls)
+}
+
+func TestCodexBorrowSourceRequiresSSEAndExplicitCookieLifetime(t *testing.T) {
+	for _, test := range []struct{ contentType, cookie string }{
+		{"application/json", "__oailb=value; Secure; Path=/; Max-Age=230"},
+		{"text/event-stream", "__oailb=value; Secure; Path=/"},
+	} {
+		source := borrowCoreAccount(1)
+		s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+			response := borrowCoreResponse("gpt-6-astra", "OK", "state", test.cookie)
+			response.Header.Set("Content-Type", test.contentType)
+			return response, nil
+		}, source)
+		candidate, _, err := s.acquireSource(context.Background(), 1)
+		require.Error(t, err)
+		require.Nil(t, candidate)
+	}
+}
+
+func TestCodexBorrowSourceRefreshFailureKeepsLivePrimary(t *testing.T) {
+	calls := 0
+	s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+		calls++
+		return nil, errors.New("synthetic refresh failure")
+	}, borrowCoreAccount(1))
+	borrowCoreCandidate(s, time.Now().Add(50*time.Second))
+	primary := s.candidate
+	backup := *primary
+	backup.sourceID = 4
+	backup.cookie.Value = "backup"
+	s.storeCandidateLocked(&backup)
+	require.Same(t, primary, s.currentCandidateLocked(), "a backup does not replace a live primary")
+	require.NoError(t, s.prepareSource(context.Background(), s.revision, s.revisionCtx))
+	require.Same(t, primary, s.currentCandidateLocked())
+	require.Equal(t, 1, calls)
+	require.True(t, s.prepareFailedUntil.IsZero())
+	require.Equal(t, "candidate_retained", s.Status().Acquisition.Phase)
+}
+
+func TestCodexBorrowPinnedTargetRequiresOriginalCompletedStream(t *testing.T) {
+	for _, kind := range []string{"valid", "unterminated", "late_error", "done_alias"} {
+		t.Run(kind, func(t *testing.T) {
+			account := borrowCoreAccount(2)
+			s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+				body := borrowCoreSSE("gpt-6.1-sol", "OK")
+				switch kind {
+				case "unterminated":
+					body = strings.TrimSuffix(body, "\n")
+				case "late_error":
+					body += "data: {\"type\":\"error\",\"error\":{\"message\":\"retained failure\"}}\n\n"
+				case "done_alias":
+					body = strings.ReplaceAll(body, "response.completed", "response.done")
+				}
+				return &http.Response{StatusCode: 200, Header: http.Header{"X-Codex-Turn-State": []string{"state"}}, Body: io.NopCloser(strings.NewReader(body))}, nil
+			}, account)
+			borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+			_, req, _, err := s.accountTemplate(context.Background(), account.ID, "gpt-6.1-sol")
+			require.NoError(t, err)
+			_, applied, err := s.Apply(req, account, "gpt-6.1-sol", "", nil, false)
+			if kind == "valid" {
+				require.NoError(t, err)
+				require.True(t, applied.Applied)
+			} else {
+				require.Error(t, err)
+				require.Nil(t, applied)
+			}
+		})
+	}
+}
+
+type borrowProbeWaitForEOF struct {
+	io.Reader
+	ctx context.Context
+}
+
+func (b *borrowProbeWaitForEOF) Read(data []byte) (int, error) {
+	n, err := b.Reader.Read(data)
+	if err == io.EOF {
+		<-b.ctx.Done()
+		return n, b.ctx.Err()
+	}
+	return n, err
+}
+func (b *borrowProbeWaitForEOF) Close() error { return nil }
+func TestCodexBorrowPinnedTargetDoesNotCertifyBeforeEOF(t *testing.T) {
+	account := borrowCoreAccount(2)
+	s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+		return &http.Response{StatusCode: 200, Header: http.Header{"X-Codex-Turn-State": []string{"state"}}, Body: &borrowProbeWaitForEOF{Reader: strings.NewReader(borrowCoreSSE("gpt-6-astra", "OK")), ctx: req.Context()}}, nil
+	}, account)
+	borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+	_, req, _, err := s.accountTemplate(context.Background(), account.ID, "gpt-6-astra")
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	result := s.probeTarget(ctx, req, account, "gpt-6-astra", "", nil, s.candidate)
+	require.False(t, result.Success)
+	require.False(t, result.MintCompleted)
+	require.Zero(t, result.ContinueStatus)
+	require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded)
+}

backend/internal/service/codex_gateway_borrow_coordination.go

@@ -64,6 +64,10 @@ func (g *codexBorrowFlights) do(ctx, revision context.Context, key string, timeo
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	case <-f.done:
+		// A concurrently completed flight cannot outlive this waiter's deadline.
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		return f.value, f.err
 	}
 }

backend/internal/service/codex_gateway_borrow_core_test.go

@@ -141,23 +141,23 @@ func TestCodexGatewayBorrowConfigDefaultsAndLoadNeverProbe(t *testing.T) {
 
 func TestCodexGatewayBorrowCookieScopeAndExpiry(t *testing.T) {
 	now := time.Now()
-	valid := http.Cookie{Name: "__oailb", Value: "route", Secure: true, Path: "/backend-api/codex"}
+	valid := http.Cookie{Name: "__oailb", Value: "route", Secure: true, Path: "/backend-api/codex", MaxAge: 230}
 	for _, tc := range []struct {
 		name   string
 		change func(*http.Cookie)
 		ttl    time.Duration
 		valid  bool
 	}{
-		{"session", func(*http.Cookie) {}, 230 * time.Second, true},
+		{"session without expiry", func(c *http.Cookie) { c.MaxAge = 0 }, 0, false},
 		{"actual max age", func(c *http.Cookie) { c.MaxAge = 30 }, 30 * time.Second, true},
 		{"local cap", func(c *http.Cookie) { c.MaxAge = 1000000 }, 230 * time.Second, true},
-		{"actual expires", func(c *http.Cookie) { c.Expires = now.Add(20 * time.Second) }, 20 * time.Second, true},
+		{"actual expires", func(c *http.Cookie) { c.MaxAge = 0; c.Expires = now.Add(20 * time.Second) }, 20 * time.Second, true},
 		{"default path", func(c *http.Cookie) { c.Path = "" }, 230 * time.Second, true},
 		{"foreign domain", func(c *http.Cookie) { c.Domain = "example.invalid" }, 0, false},
 		{"wrong path", func(c *http.Cookie) { c.Path = "/backend-api/codex/responses-other" }, 0, false},
 		{"not secure", func(c *http.Cookie) { c.Secure = false }, 0, false},
 		{"deleted", func(c *http.Cookie) { c.MaxAge = -1 }, 0, false},
-		{"expired", func(c *http.Cookie) { c.Expires = now.Add(-time.Second) }, 0, false},
+		{"expired", func(c *http.Cookie) { c.MaxAge = 0; c.Expires = now.Add(-time.Second) }, 0, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			cookie := valid