diff --git a/.downstream/upstream-review-ranxi-v2.10.3.md b/.downstream/upstream-review-ranxi-v2.10.3.md index 084ae4ad03bc..2f7b63ff7bee 100644 --- a/.downstream/upstream-review-ranxi-v2.10.3.md +++ b/.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. diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 5363b56f272a..294d78e32c74 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/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() diff --git a/backend/internal/service/codex_gateway_borrow_candidates.go b/backend/internal/service/codex_gateway_borrow_candidates.go index 1c39ab665631..303b93ab6aa9 100644 --- a/backend/internal/service/codex_gateway_borrow_candidates.go +++ b/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. } diff --git a/backend/internal/service/codex_gateway_borrow_cookie_contract_test.go b/backend/internal/service/codex_gateway_borrow_cookie_contract_test.go new file mode 100644 index 000000000000..2e2122b04b0a --- /dev/null +++ b/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) +} diff --git a/backend/internal/service/codex_gateway_borrow_coordination.go b/backend/internal/service/codex_gateway_borrow_coordination.go index 7da7e3672ed9..625b412933f6 100644 --- a/backend/internal/service/codex_gateway_borrow_coordination.go +++ b/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 } } diff --git a/backend/internal/service/codex_gateway_borrow_core_test.go b/backend/internal/service/codex_gateway_borrow_core_test.go index 488df01c05f7..78e5145f760b 100644 --- a/backend/internal/service/codex_gateway_borrow_core_test.go +++ b/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 diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go index b0fd702df58a..fb264449240b 100644 --- a/backend/internal/service/codex_gateway_borrow_probe.go +++ b/backend/internal/service/codex_gateway_borrow_probe.go @@ -25,7 +25,11 @@ import ( const codexGatewayBorrowProbeMaxBody = 1 << 20 const codexGatewayBorrowPelicanMaxBody = 32 << 20 +type codexBorrowProbePinnedRouteKey struct{} +type codexBorrowSourceObservationKey struct{} + type codexGatewayBorrowObservation struct { + routeChanged bool requestEncoding string status int state string @@ -140,6 +144,7 @@ func (s *CodexGatewayBorrowService) acquireSource(ctx context.Context, id int64) func borrowSourceObservationRequest(ctx context.Context, headers http.Header) (*http.Request, error) { ctx = WithCodexGatewayBorrowObservation(WithHTTPUpstreamRedirectsDisabled(WithHTTPUpstreamProfile(ctx, HTTPUpstreamProfileCodexBorrowSource))) ctx = context.WithValue(ctx, codexBorrowProbeConfiguredEncodingKey{}, false) + ctx = context.WithValue(ctx, codexBorrowSourceObservationKey{}, true) payload := createOpenAITestPayload(codexGatewayBorrowSourceModel, true, "Reply with OK only.") payload["reasoning"] = map[string]string{"effort": "medium"} body, err := json.Marshal(payload) @@ -164,7 +169,7 @@ func borrowSourceObservationRequest(ctx context.Context, headers http.Header) (* } func borrowCookieScoped(cookie *http.Cookie, path string) bool { - if cookie == nil || (cookie.Domain != "" && !strings.EqualFold(strings.TrimPrefix(cookie.Domain, "."), "chatgpt.com")) { + if cookie == nil || (cookie.Domain != "" && strings.TrimPrefix(cookie.Domain, ".") != "chatgpt.com") { return false } scope := cookie.Path @@ -194,7 +199,12 @@ func borrowCandidateFromCookies(cookies []*http.Cookie, path string, now time.Ti expires = now.Add(time.Duration(cookie.MaxAge) * time.Second) } } else { - if !cookie.Expires.IsZero() && cookie.Expires.Before(expires) { + // Pinned upstream requires an explicit lifetime; a session cookie + // is not a transferable bounded route lease. + if cookie.Expires.IsZero() { + continue + } + if cookie.Expires.Before(expires) { expires = cookie.Expires } } @@ -241,21 +251,22 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h fire := func(state string, cookies []*http.Cookie) (codexGatewayBorrowObservation, error) { shotCtx, cancel := context.WithTimeout(ctx, codexGatewayBorrowShotTimeout) defer cancel() + shotCtx = context.WithValue(shotCtx, codexBorrowProbePinnedRouteKey{}, candidate.cookie.Value) return s.fireObservation(shotCtx, template.Header, account, model, "Reply with OK.", "", proxy, profile, state, cookies, HTTPUpstreamProfileCodexBorrowTarget, codexGatewayBorrowProbeMaxBody) } mint, err := fire("", cookies) result.MintStatus, result.ReportedModel = mint.status, mint.model result.MintRequestEncoding = mint.requestEncoding result.MintCompleted, result.MintStateLength = mint.complete, len(mint.state) - if err != nil || !borrowShotUsable(mint) { - result.Error = borrowObservationError(mint, err) - return result - } - if borrowTargetRouteChanged(mint.cookies, candidate, time.Now()) { + if mint.routeChanged { result.RouteChanged = true result.Reason, result.Error = "target_route_changed", "target response changed or deleted the fixed __oailb route" return result } + if err != nil || !borrowShotUsable(mint) { + result.Error = borrowObservationError(mint, err) + return result + } if mint.state == "" { result.Reason, result.Error = "target_state_missing", "first response did not return x-codex-turn-state" return result @@ -264,7 +275,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h // Only the target's newly returned __cflb travels into the second shot. // __oailb remains the exact borrowed candidate, never a target replacement. for _, cookie := range mint.cookies { - if cookie.Name == "__cflb" && borrowCookieScoped(cookie, "/backend-api/codex/responses") && !borrowCookieDeleted(cookie, time.Now()) { + if cookie.Name == "__cflb" && cookie.Value != "" && cookie.MaxAge >= 0 { // The original replacement helper retains the target's __cflb first, // then appends the pinned __oailb. No business cookie header is changed. cookies = append([]*http.Cookie{{Name: cookie.Name, Value: cookie.Value}}, cookies...) @@ -278,15 +289,15 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h if continued.model != "" { result.ReportedModel = continued.model } - if err != nil || !borrowShotUsable(continued) { - result.Error = borrowObservationError(continued, err) - return result - } - if borrowTargetRouteChanged(continued.cookies, candidate, time.Now()) { + if continued.routeChanged { result.RouteChanged = true result.Reason, result.Error = "target_route_changed", "target response changed or deleted the fixed __oailb route" return result } + if err != nil || !borrowShotUsable(continued) { + result.Error = borrowObservationError(continued, err) + return result + } result.NewTicket = continued.state != "" && continued.state != mint.state if result.NewTicket { result.Reason = "target_state_changed" @@ -318,7 +329,7 @@ func borrowObservationError(shot codexGatewayBorrowObservation, err error) strin func borrowTargetRouteChanged(cookies []*http.Cookie, candidate *codexGatewayBorrowCandidate, now time.Time) bool { for _, cookie := range cookies { - if cookie.Name == "__oailb" && borrowCookieScoped(cookie, "/backend-api/codex/responses") && (cookie.Value != candidate.cookie.Value || borrowCookieDeleted(cookie, now)) { + if cookie.Name == "__oailb" && (cookie.Value != candidate.cookie.Value || cookie.MaxAge < 0 || (!cookie.Expires.IsZero() && !now.Before(cookie.Expires))) { return true } } @@ -449,6 +460,15 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account * return shot, errors.New("nil upstream observation response") } shot.status, shot.state, shot.cookies, shot.cookiesObservedAt = resp.StatusCode, strings.TrimSpace(extractOpenAICodexTurnState(resp.Header)), resp.Cookies(), time.Now() + // Match the original target wrapper: a changed route ends the probe at + // headers, even when the body would stall, fail or report a completed stream. + if pinned, _ := req.Context().Value(codexBorrowProbePinnedRouteKey{}).(string); pinned != "" && borrowTargetRouteChanged(shot.cookies, &codexGatewayBorrowCandidate{cookie: http.Cookie{Value: pinned}}, shot.cookiesObservedAt) { + shot.routeChanged, shot.errorText = true, "target_route_changed" + if resp.Body != nil { + _ = resp.Body.Close() + } + return shot, errors.New("target_route_changed") + } if resp.Body == nil { shot.errorText = "upstream observation response has no body" return shot, nil @@ -456,15 +476,39 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account * defer func() { _ = resp.Body.Close() }() var data []byte var readErr error - if shot.status == http.StatusOK { - data, readErr = readCodexGatewayBorrowSSE(resp.Body, maxBody) + pinnedProbe, _ := req.Context().Value(codexBorrowProbePinnedRouteKey{}).(string) + sourceProbe := req.Context().Value(codexBorrowSourceObservationKey{}) == true + if shot.status == http.StatusOK && pinnedProbe == "" { + data, readErr = readCodexGatewayBorrowSSE(resp.Body, maxBody, sourceProbe) } else { data, readErr = io.ReadAll(io.LimitReader(resp.Body, maxBody+1)) } shot.raw = string(data) + if sourceProbe && shot.status == http.StatusOK { + contentType := strings.ToLower(resp.Header.Get("Content-Type")) + if contentType != "" && !strings.HasPrefix(contentType, "text/event-stream") { + shot.errorText = "source response content type is not an SSE stream: " + shot.raw + return shot, nil + } + } if shot.status == http.StatusOK { stream := parseCodexGatewayBorrowSSE(data, expectedModel, maxBody) shot.model, shot.answer, shot.complete, shot.modelMismatch, shot.errorText, shot.incomplete = stream.model, stream.answer, stream.complete, stream.modelMismatch, stream.errorText, stream.incomplete + if pinnedProbe != "" { + // Original target probes read the bounded body to EOF and require a valid + // completion with a terminated SSE event. Do not substitute the ordinary + // business sender's early-terminal optimization for this criterion. + validationErr := validateCodexGatewayBorrowPinnedProbeResponse(data) + shot.complete = validationErr == nil + if validationErr == nil { + shot.errorText = "" + } else if shot.errorText == "" { + shot.errorText = validationErr.Error() + } + } + if sourceProbe && shot.complete && !bytes.HasSuffix(data, []byte("\n\n")) && !bytes.HasSuffix(data, []byte("\r\n\r\n")) { + shot.complete, shot.errorText = false, "unterminated source probe event" + } } if readErr != nil { shot.complete, shot.incomplete = false, true @@ -518,7 +562,8 @@ func acquireCodexBorrowObservation(ctx context.Context, accountID int64) (contex // remains open. Collect the original bytes, without waiting for EOF or reading // beyond that terminal payload; sendObservation closes the body immediately. // The existing pure parser still decides completion, text, models and errors. -func readCodexGatewayBorrowSSE(body io.Reader, maxBody int64) ([]byte, error) { +func readCodexGatewayBorrowSSE(body io.Reader, maxBody int64, requireDelimiter ...bool) ([]byte, error) { + strict := len(requireDelimiter) > 0 && requireDelimiter[0] reader := bufio.NewReader(io.LimitReader(body, maxBody+1)) var raw bytes.Buffer var lines []string @@ -542,10 +587,13 @@ func readCodexGatewayBorrowSSE(body io.Reader, maxBody int64) ([]byte, error) { } text := strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r") if text == "" { + if strict && terminal() { + return raw.Bytes(), nil + } lines = nil } else if strings.HasPrefix(text, "data:") { lines = append(lines, strings.TrimPrefix(strings.TrimPrefix(text, "data:"), " ")) - if terminal() { + if !strict && terminal() { // Preserve an already-received blank event delimiter byte for // byte. Never block waiting for a delimiter or subsequent EOF. if buffered := reader.Buffered(); buffered > 0 { diff --git a/backend/internal/service/codex_gateway_borrow_probe_stream_test.go b/backend/internal/service/codex_gateway_borrow_probe_stream_test.go index 701aa308666b..7d69b17fae67 100644 --- a/backend/internal/service/codex_gateway_borrow_probe_stream_test.go +++ b/backend/internal/service/codex_gateway_borrow_probe_stream_test.go @@ -68,6 +68,10 @@ func TestCodexGatewayBorrowObservationStopsAtTerminal(t *testing.T) { bodies, sent = append(bodies, body), append(sent, stream) go func() { _, _ = io.WriteString(writer, stream) + if action == "target" { + _ = writer.Close() + return + } if action == "disconnected" { _ = writer.CloseWithError(io.ErrUnexpectedEOF) return @@ -81,7 +85,7 @@ func TestCodexGatewayBorrowObservationStopsAtTerminal(t *testing.T) { }() headers := make(http.Header) if accountID == 1 { - headers.Set("Set-Cookie", "__oailb=synthetic-borrowed-cookie; Secure; Path=/") + headers.Set("Set-Cookie", "__oailb=synthetic-borrowed-cookie; Secure; Path=/; Max-Age=230") } if action == "target" && calls == 1 { headers.Set("X-Codex-Turn-State", "minted-state") diff --git a/backend/internal/service/codex_gateway_borrow_probe_terminal.go b/backend/internal/service/codex_gateway_borrow_probe_terminal.go new file mode 100644 index 000000000000..e893435535bf --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_probe_terminal.go @@ -0,0 +1,95 @@ +package service + +// Target-probe terminal contract from ranxi2001/sub2api v2.10.3, +// fd1b5ee4eeb20961fbb783fa6f136a1704271e90, LGPL-3.0. +// The original route probe calls this without an expected-model argument: +// reported labels are display evidence, not the STATE qualification criterion. + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "strings" +) + +// A clean EOF or [DONE] is not evidence of a successful Responses generation. +func validateCodexGatewayBorrowPinnedProbeResponse(body []byte) error { + completed := false + inspect := func(data []byte) error { + if bytes.Equal(bytes.TrimSpace(data), []byte("[DONE]")) { + return nil + } + var event struct { + Model string `json:"model"` + Type string `json:"type"` + Status string `json:"status"` + Error json.RawMessage `json:"error"` + Response *struct { + Model string `json:"model"` + Status string `json:"status"` + } `json:"response"` + } + if json.Unmarshal(data, &event) != nil { + return errors.New("invalid probe event") + } + + if event.Type == "error" || event.Type == "response.failed" || event.Type == "response.incomplete" || event.Status == "failed" || event.Status == "incomplete" || (len(event.Error) > 0 && string(event.Error) != "null") { + return errors.New("probe response failed") + } + if event.Response != nil && (event.Response.Status == "failed" || event.Response.Status == "incomplete") { + return errors.New("probe response failed") + } + if event.Type == "response.completed" { + if event.Response == nil || event.Response.Status != "completed" { + return errors.New("invalid probe completion") + } + completed = true + } + if event.Type == "" && event.Status == "completed" { + completed = true + } + return nil + } + trimmed := bytes.TrimSpace(body) + if len(trimmed) > 0 && trimmed[0] == '{' { + if err := inspect(trimmed); err != nil { + return err + } + } else { + scanner := bufio.NewScanner(bytes.NewReader(body)) + scanner.Buffer(make([]byte, 4096), 1<<20) + var data []string + flush := func() error { + if len(data) == 0 { + return nil + } + err := inspect([]byte(strings.Join(data, "\n"))) + data = nil + return err + } + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + if err := flush(); err != nil { + return err + } + continue + } + if strings.HasPrefix(line, "data:") { + data = append(data, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ")) + } + } + if scanner.Err() != nil { + return errors.New("invalid probe stream") + } + // A trailing unterminated event may be truncated, even if its JSON parses. + if len(data) > 0 { + return errors.New("unterminated probe event") + } + } + if !completed { + return errors.New("probe did not complete successfully") + } + return nil +}