diff --git a/backend/internal/service/pelican_account_generator.go b/backend/internal/service/pelican_account_generator.go index 13764daa5cb..0ac2318bb06 100644 --- a/backend/internal/service/pelican_account_generator.go +++ b/backend/internal/service/pelican_account_generator.go @@ -447,6 +447,14 @@ func (s *AccountTestService) GeneratePelican(ctx context.Context, accountID int6 // This represents the server's internal generation operation. No client // UA, protocol handshake, API-key subject or group policy is synthesized. sendErr := s.sendPelicanWithAccount(ctx, c, account, model, option.UpstreamModel, effort) + // The gateway keeps client-facing failover text generic, but records the + // complete local borrow cause on this invocation's Gin context. Preserve it + // in administrator task history even when no generation request was sent. + if sendErr != nil && IsCodexGatewayBorrowRequestFailure(sendErr) { + if detail := c.GetString(OpsUpstreamErrorDetailKey); detail != "" && !strings.Contains(sendErr.Error(), detail) { + sendErr = fmt.Errorf("%w\n%s", sendErr, detail) + } + } last := snapshotPelicanAttempt(capture) if last == nil { result.RawResponse = w.Body.String() @@ -491,7 +499,11 @@ func (s *AccountTestService) GeneratePelican(ctx context.Context, accountID int6 if responseModel != "" { result.ModelID = responseModel } - if last.err != nil { + // The native sender cancels its read-ahead goroutine after committing a + // semantic terminal event. That cleanup can be captured before this snapshot; + // it must not replace a successfully parsed, successfully forwarded result. + // Without both completion signals the same cancellation remains incomplete. + if last.err != nil && (parseErr != nil || sendErr != nil || !errors.Is(last.err, context.Canceled)) { parseErr = fmt.Errorf("%w: %v", ErrAccountTestIncomplete, last.err) } if limited { diff --git a/backend/internal/service/pelican_account_generator_test.go b/backend/internal/service/pelican_account_generator_test.go index ef1dae338f1..6c76c5257ce 100644 --- a/backend/internal/service/pelican_account_generator_test.go +++ b/backend/internal/service/pelican_account_generator_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "strings" @@ -43,6 +44,7 @@ type pelicanGeneratorUpstream struct { proxies []string profiles []*tlsfingerprint.Profile beforeSend func(*http.Request) + readErr error } func (s *pelicanGeneratorUpstream) Do(req *http.Request, proxy string, id int64, concurrency int) (*http.Response, error) { @@ -75,7 +77,26 @@ func (s *pelicanGeneratorUpstream) DoWithTLS(req *http.Request, proxy string, id if contentType == "" { contentType = "text/event-stream" } - return finish(&http.Response{StatusCode: status, Proto: "HTTP/1.1", Header: http.Header{"Content-Type": {contentType}}, Body: io.NopCloser(strings.NewReader(s.body))}, nil) + var reader io.Reader = strings.NewReader(s.body) + if s.readErr != nil { + reader = &pelicanTerminalReadError{Reader: strings.NewReader(s.body), err: s.readErr} + } + return finish(&http.Response{StatusCode: status, Proto: "HTTP/1.1", Header: http.Header{"Content-Type": {contentType}}, Body: io.NopCloser(reader)}, nil) +} + +// Return the final bytes and cleanup error together so the capture always sees +// the error before the native sender has a chance to consume the terminal event. +type pelicanTerminalReadError struct { + *strings.Reader + err error +} + +func (r *pelicanTerminalReadError) Read(p []byte) (int, error) { + n, err := r.Reader.Read(p) + if r.Len() == 0 { + err = r.err + } + return n, err } func TestPelicanAccountGeneratorQueuePreservesSharedFirstOutputPolicy(t *testing.T) { @@ -158,6 +179,32 @@ func TestPelicanAccountGeneratorSharedOpenAISettings(t *testing.T) { } } +func TestPelicanAccountGeneratorKeepsBorrowPreparationCause(t *testing.T) { + account := borrowCoreAccount(2) + upstream := &pelicanGeneratorUpstream{} + svc, _ := newPelicanGeneratorForTest(account, upstream) + shots := 0 + borrow := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + shots++ + state := "first" + if req.Header.Get("X-Codex-Turn-State") != "" { + state = "changed" + } + return borrowCoreResponse("gpt-6-astra", "OK", state), nil + }, account) + borrowCoreCandidate(borrow, time.Now().Add(codexGatewayBorrowTTL)) + svc.openaiGatewayService.SetCodexGatewayBorrowService(borrow) + result, err := svc.GeneratePelican(context.Background(), account.ID, "gpt-6-astra", "low") + require.Error(t, err) + require.True(t, IsCodexGatewayBorrowRequestFailure(err)) + require.Contains(t, result.Error, "target_validation/target_state_changed") + require.Contains(t, result.Error, "CODEX_GATEWAY_BORROW_UNAVAILABLE") + require.Equal(t, "failed", result.Status) + require.Empty(t, result.RawAnswer) + require.Empty(t, upstream.requests, "a rejected qualification never reaches generation") + require.Equal(t, 2, shots) +} + func TestPelicanAccountGeneratorHTTPFailureKeepsSelectedAccountAndRawBody(t *testing.T) { raw := `{"error":{"type":"rate_limit_error","message":"` + strings.Repeat("complete original detail ", 2000) + `"}}` account := &Account{ID: 28, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "local-key", "base_url": "http://fake.example"}} @@ -184,6 +231,33 @@ func TestPelicanAccountGeneratorTruncatedWorkIsIncomplete(t *testing.T) { require.Len(t, upstream.requests, 1) } +func TestPelicanAccountGeneratorTerminalBeforeReadCancellation(t *testing.T) { + for _, completed := range []bool{true, false} { + t.Run(fmt.Sprintf("completed=%t", completed), func(t *testing.T) { + account := &Account{ID: 32, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "local-key", "base_url": "http://fake.example"}} + raw := pelicanResponsesSSE("gpt-6-astra", "completed", "") + if !completed { + raw, _, _ = strings.Cut(raw, "\n\ndata: ") + raw += "\n\n" + } + upstream := &pelicanGeneratorUpstream{body: raw, readErr: context.Canceled} + svc, _ := newPelicanGeneratorForTest(account, upstream) + result, err := svc.GeneratePelican(context.Background(), account.ID, "gpt-6-astra", "low") + require.Equal(t, raw, result.RawResponse) + require.Equal(t, "", result.RawAnswer) + require.Len(t, upstream.requests, 1, "cleanup never triggers a generation replay") + if completed { + require.NoError(t, err) + require.Equal(t, "complete", result.Status, result.Error) + require.Empty(t, result.Error) + } else { + require.Equal(t, "incomplete", result.Status) + require.Contains(t, result.Error, "context canceled") + } + }) + } +} + func TestPelicanAccountGeneratorNativeTextDispatch(t *testing.T) { claudeSSE := "data: {\"type\":\"content_block_delta\",\"delta\":{\"text\":\"work\"}}\n\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}}\n\ndata: {\"type\":\"message_stop\"}\n\n" chatSSE := "data: {\"model\":\"text-model\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"work\"},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n\n"