diff --git a/internal/backend/agent/model/http_error.go b/internal/backend/agent/model/http_error.go index c2bb5175..44d4ac61 100644 --- a/internal/backend/agent/model/http_error.go +++ b/internal/backend/agent/model/http_error.go @@ -2,6 +2,7 @@ package modeladapter import ( + "errors" "fmt" "io" "net/http" @@ -13,6 +14,43 @@ const ( maxErrorBodyBytes = 8192 ) +// HTTPStatusError preserves the provider HTTP status so higher layers can +// distinguish transient overloads from permanent request failures. +type HTTPStatusError struct { + Prefix string + StatusCode int + RetrySummary string + Body string + BodyReadError error +} + +func (err *HTTPStatusError) Error() string { + if err == nil { + return "provider HTTP error" + } + result := fmt.Sprintf("%s status=%d", strings.TrimSpace(err.Prefix), err.StatusCode) + if summary := strings.TrimSpace(err.RetrySummary); summary != "" { + result += " " + summary + } + if err.BodyReadError != nil { + return fmt.Sprintf("%s body_read_error=%v", result, err.BodyReadError) + } + if body := strings.TrimSpace(err.Body); body != "" { + result += " body=" + body + } + return result +} + +// ProviderHTTPStatus returns the original provider response status when the +// error came from a non-2xx HTTP response. +func ProviderHTTPStatus(err error) (int, bool) { + var statusErr *HTTPStatusError + if !errors.As(err, &statusErr) || statusErr == nil || statusErr.StatusCode <= 0 { + return 0, false + } + return statusErr.StatusCode, true +} + // buildHTTPStatusError 读取响应体摘要并生成带状态码的错误。 func buildHTTPStatusError(prefix string, resp *http.Response) error { if resp == nil { @@ -21,21 +59,17 @@ func buildHTTPStatusError(prefix string, resp *http.Response) error { limitedBody, err := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) if err != nil { - if retrySummary := ProviderRetryAttemptSummary(resp); retrySummary != "" { - return fmt.Errorf("%s status=%d %s body_read_error=%v", strings.TrimSpace(prefix), resp.StatusCode, retrySummary, err) - } - return fmt.Errorf("%s status=%d body_read_error=%v", strings.TrimSpace(prefix), resp.StatusCode, err) - } - retrySummary := ProviderRetryAttemptSummary(resp) - bodyText := strings.TrimSpace(string(limitedBody)) - if bodyText == "" { - if retrySummary != "" { - return fmt.Errorf("%s status=%d %s", strings.TrimSpace(prefix), resp.StatusCode, retrySummary) + return &HTTPStatusError{ + Prefix: prefix, + StatusCode: resp.StatusCode, + RetrySummary: ProviderRetryAttemptSummary(resp), + BodyReadError: err, } - return fmt.Errorf("%s status=%d", strings.TrimSpace(prefix), resp.StatusCode) } - if retrySummary != "" { - return fmt.Errorf("%s status=%d %s body=%s", strings.TrimSpace(prefix), resp.StatusCode, retrySummary, bodyText) + return &HTTPStatusError{ + Prefix: prefix, + StatusCode: resp.StatusCode, + RetrySummary: ProviderRetryAttemptSummary(resp), + Body: strings.TrimSpace(string(limitedBody)), } - return fmt.Errorf("%s status=%d body=%s", strings.TrimSpace(prefix), resp.StatusCode, bodyText) } diff --git a/internal/backend/agent/model/http_error_test.go b/internal/backend/agent/model/http_error_test.go new file mode 100644 index 00000000..c080d943 --- /dev/null +++ b/internal/backend/agent/model/http_error_test.go @@ -0,0 +1,29 @@ +package modeladapter + +import ( + "errors" + "io" + "net/http" + "strings" + "testing" +) + +func TestBuildHTTPStatusErrorPreservesStatus(t *testing.T) { + resp := &http.Response{ + StatusCode: http.StatusTooManyRequests, + Body: io.NopCloser(strings.NewReader(`{"error":{"type":"engine_overloaded_error"}}`)), + } + + err := buildHTTPStatusError("openai adapter", resp) + status, ok := ProviderHTTPStatus(err) + if !ok || status != http.StatusTooManyRequests { + t.Fatalf("ProviderHTTPStatus() = (%d, %t), want (%d, true)", status, ok, http.StatusTooManyRequests) + } + var statusErr *HTTPStatusError + if !errors.As(err, &statusErr) { + t.Fatalf("error type = %T, want *HTTPStatusError", err) + } + if got := err.Error(); !strings.Contains(got, "openai adapter status=429 body=") { + t.Fatalf("Error() = %q, want status and body summary", got) + } +} diff --git a/internal/backend/agent/model/retry.go b/internal/backend/agent/model/retry.go index 57ed6005..8b12f993 100644 --- a/internal/backend/agent/model/retry.go +++ b/internal/backend/agent/model/retry.go @@ -1,4 +1,5 @@ -// retry.go 保留 provider HTTP 请求入口的历史命名;provider 错误交给客户端重连链路处理。 +// retry.go 保留 provider HTTP 请求入口的历史命名;单次 HTTP 调用不在适配器内重试。 +// 对尚未产生输出的瞬时 HTTP 错误,由 forwarder 在同一 RunSSE 会话内安全重试。 package modeladapter import ( @@ -6,7 +7,7 @@ import ( "net/http" ) -// DoProviderRequestWithRetry 保留旧入口名;本地模式不在服务端重试 provider 请求。 +// DoProviderRequestWithRetry 保留旧入口名;适配器只执行一次 provider HTTP 请求。 func DoProviderRequestWithRetry( ctx context.Context, client *http.Client, diff --git a/internal/backend/forwarder/actor.go b/internal/backend/forwarder/actor.go index 851056b4..3ef62f5c 100644 --- a/internal/backend/forwarder/actor.go +++ b/internal/backend/forwarder/actor.go @@ -1,6 +1,7 @@ package forwarder import ( + "context" "encoding/json" "errors" "fmt" @@ -697,6 +698,15 @@ func (service *Service) handleProviderDoneEvent(stream *ActiveStream, payload *s usage := stream.ProviderUsage hadToolInvocation := stream.ToolInvocationCount > 0 terminalToolInvocation := stream.ProviderTerminalToolInvocation + passProducedOutput := providerPassProducedOutput( + accumulatedText, + accumulatedReasoning, + accumulatedReasoningSignature, + accumulatedReasoningItemID, + finishReason, + hadToolInvocation, + terminalToolInvocation, + ) existingCompletion := stream.PendingProviderCompletion stream.ProviderActive = false stream.ProviderCancel = nil @@ -722,12 +732,33 @@ func (service *Service) handleProviderDoneEvent(stream *ActiveStream, payload *s if payload.Err != nil { var providerErr providerTerminalError if errors.As(payload.Err, &providerErr) { + if attempt, delay, statusCode, retry := reserveTransientProviderRetry(stream, unwrapProviderTerminalError(payload.Err), passProducedOutput); retry { + service.setTurnPhase(stream, TurnPhaseWaitingExternal) + log.Printf( + "forwarder transient provider retry scheduled request_id=%s model_call_id=%s status=%d attempt=%d delay_ms=%d", + strings.TrimSpace(requestID), + strings.TrimSpace(modelCallID), + statusCode, + attempt, + delay.Milliseconds(), + ) + service.debug.LogProvider(context.Background(), requestID, conversationID, "provider_transient_retry_scheduled", map[string]any{ + "model_call_id": strings.TrimSpace(modelCallID), + "attempt": attempt, + "delay_ms": delay.Milliseconds(), + "status_code": statusCode, + "provider_error": providerErr.Error(), + }) + service.scheduleStreamTimer(stream, providerTimerKey(streamTimerProviderResume, ""), delay, streamTimerProviderResume, "", 0, "provider transient retry") + return nil + } service.setTurnPhase(stream, TurnPhaseFailed) return service.closeStreamWithProviderError(stream, conversationID, turnSeq, requestID, accumulatedText, accumulatedReasoning, accumulatedReasoningSignature, accumulatedReasoningSignatureSource, accumulatedReasoningItemID, accumulatedReasoningStatus, accumulatedReasoningSummary, usage, providerErr, !hadToolInvocation) } service.setTurnPhase(stream, TurnPhaseFailed) return service.failStream(stream, "unknown", payload.Err) } + resetTransientProviderRetry(stream) if err := service.flushAssistantText(stream, conversationID, turnSeq, requestID, accumulatedText, accumulatedReasoning, accumulatedReasoningSignature, accumulatedReasoningSignatureSource, accumulatedReasoningItemID, accumulatedReasoningStatus, accumulatedReasoningSummary, !hadToolInvocation); err != nil { return service.failStreamIfNonTerminal(stream, "unknown", err) } diff --git a/internal/backend/forwarder/provider_retry.go b/internal/backend/forwarder/provider_retry.go new file mode 100644 index 00000000..8db49443 --- /dev/null +++ b/internal/backend/forwarder/provider_retry.go @@ -0,0 +1,86 @@ +package forwarder + +import ( + "errors" + "strings" + "time" + + modeladapter "cursor/internal/backend/agent/model" +) + +const providerTransientRetryLimit = 3 + +var providerTransientRetryDelays = [...]time.Duration{ + 2 * time.Second, + 5 * time.Second, + 10 * time.Second, +} + +func retryableProviderHTTPStatus(err error) (int, bool) { + status, ok := modeladapter.ProviderHTTPStatus(err) + if !ok { + return 0, false + } + switch status { + case 429, 502, 503, 504: + return status, true + default: + return status, false + } +} + +func providerPassProducedOutput( + text string, + reasoning string, + reasoningSignature string, + reasoningItemID string, + finishReason string, + hadToolInvocation bool, + terminalToolInvocation bool, +) bool { + return strings.TrimSpace(text) != "" || + strings.TrimSpace(reasoning) != "" || + strings.TrimSpace(reasoningSignature) != "" || + strings.TrimSpace(reasoningItemID) != "" || + strings.TrimSpace(finishReason) != "" || + hadToolInvocation || + terminalToolInvocation +} + +func reserveTransientProviderRetry(stream *ActiveStream, err error, passProducedOutput bool) (int, time.Duration, int, bool) { + if stream == nil || passProducedOutput { + return 0, 0, 0, false + } + status, retryable := retryableProviderHTTPStatus(err) + if !retryable { + return 0, 0, status, false + } + + stream.mu.Lock() + defer stream.mu.Unlock() + if isTerminalStreamStatus(stream.Status) || stream.ProviderTransientRetryCount >= providerTransientRetryLimit { + return 0, 0, status, false + } + stream.ProviderTransientRetryCount++ + attempt := stream.ProviderTransientRetryCount + stream.PendingProviderAction = providerActionResume + stream.UpdatedAt = time.Now().UTC() + return attempt, providerTransientRetryDelays[attempt-1], status, true +} + +func resetTransientProviderRetry(stream *ActiveStream) { + if stream == nil { + return + } + stream.mu.Lock() + stream.ProviderTransientRetryCount = 0 + stream.mu.Unlock() +} + +func unwrapProviderTerminalError(err error) error { + var providerErr providerTerminalError + if errors.As(err, &providerErr) && providerErr.cause != nil { + return providerErr.cause + } + return err +} diff --git a/internal/backend/forwarder/provider_retry_test.go b/internal/backend/forwarder/provider_retry_test.go new file mode 100644 index 00000000..df7b9085 --- /dev/null +++ b/internal/backend/forwarder/provider_retry_test.go @@ -0,0 +1,113 @@ +package forwarder + +import ( + "fmt" + "net/http" + "testing" + "time" + + modeladapter "cursor/internal/backend/agent/model" +) + +func TestHandleProviderDoneEventKeepsStreamOpenForTransientRetry(t *testing.T) { + originalDelays := providerTransientRetryDelays + providerTransientRetryDelays[0] = time.Hour + defer func() { providerTransientRetryDelays = originalDelays }() + + service := &Service{ + broker: NewStreamBroker(), + debug: newDebugRecorder("", nil, nil), + } + stream := &ActiveStream{ + RequestID: "request-1", + ConversationID: "conversation-1", + Status: StreamStatusStreaming, + Phase: TurnPhaseProviderRunning, + ProviderActive: true, + CurrentProviderToken: 1, + CurrentModelCallID: "model-call-1", + TimerTokens: map[string]uint64{}, + } + payload := &streamProviderEvent{ + Token: 1, + Done: true, + Err: providerTerminalError{cause: &modeladapter.HTTPStatusError{ + Prefix: "openai adapter", + StatusCode: http.StatusTooManyRequests, + }}, + } + + if err := service.handleProviderDoneEvent(stream, payload); err != nil { + t.Fatalf("handleProviderDoneEvent() error = %v", err) + } + + stream.mu.Lock() + defer stream.mu.Unlock() + if stream.Status != StreamStatusStreaming { + t.Fatalf("stream status = %q, want %q", stream.Status, StreamStatusStreaming) + } + if stream.Phase != TurnPhaseWaitingExternal { + t.Fatalf("stream phase = %q, want %q", stream.Phase, TurnPhaseWaitingExternal) + } + if stream.PendingProviderAction != providerActionResume { + t.Fatalf("pending provider action = %q, want %q", stream.PendingProviderAction, providerActionResume) + } + if stream.ProviderTransientRetryCount != 1 { + t.Fatalf("retry count = %d, want 1", stream.ProviderTransientRetryCount) + } + if stream.ProviderActive { + t.Fatal("provider remained active after transient failure") + } +} + +func TestReserveTransientProviderRetry(t *testing.T) { + stream := &ActiveStream{Status: StreamStatusStreaming} + err := providerTerminalError{cause: &modeladapter.HTTPStatusError{ + Prefix: "openai adapter", + StatusCode: http.StatusTooManyRequests, + }} + + for wantAttempt := 1; wantAttempt <= providerTransientRetryLimit; wantAttempt++ { + attempt, delay, status, ok := reserveTransientProviderRetry(stream, unwrapProviderTerminalError(err), false) + if !ok || attempt != wantAttempt || delay != providerTransientRetryDelays[wantAttempt-1] || status != http.StatusTooManyRequests { + t.Fatalf("attempt %d = (%d, %s, %d, %t)", wantAttempt, attempt, delay, status, ok) + } + } + if _, _, _, ok := reserveTransientProviderRetry(stream, unwrapProviderTerminalError(err), false); ok { + t.Fatal("retry limit exceeded but retry was reserved") + } +} + +func TestReserveTransientProviderRetryRejectsUnsafeCases(t *testing.T) { + tests := []struct { + name string + err error + hadOutput bool + wantStatus int + }{ + {name: "partial output", err: &modeladapter.HTTPStatusError{StatusCode: 429}, hadOutput: true, wantStatus: 0}, + {name: "bad request", err: &modeladapter.HTTPStatusError{StatusCode: 400}, wantStatus: 400}, + {name: "untyped error", err: fmt.Errorf("openai adapter status=429"), wantStatus: 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + stream := &ActiveStream{Status: StreamStatusStreaming} + _, _, status, ok := reserveTransientProviderRetry(stream, tt.err, tt.hadOutput) + if ok || status != tt.wantStatus { + t.Fatalf("reserveTransientProviderRetry() = status %d, retry %t; want status %d, retry false", status, ok, tt.wantStatus) + } + }) + } +} + +func TestProviderPassProducedOutput(t *testing.T) { + if providerPassProducedOutput("", "", "", "", "", false, false) { + t.Fatal("empty provider pass reported output") + } + if !providerPassProducedOutput("partial", "", "", "", "", false, false) { + t.Fatal("text output was not detected") + } + if !providerPassProducedOutput("", "", "", "", "", true, false) { + t.Fatal("tool invocation was not detected") + } +} diff --git a/internal/backend/forwarder/types.go b/internal/backend/forwarder/types.go index 4f7111e4..7b78a010 100644 --- a/internal/backend/forwarder/types.go +++ b/internal/backend/forwarder/types.go @@ -134,6 +134,7 @@ type ActiveStream struct { ProviderActive bool ProviderCancel func() ProviderPassCount int + ProviderTransientRetryCount int ToolInvocationCount int ActorMailbox chan streamCommandEnvelope ActorDone chan struct{}