fix: retry transient provider overloads in stream

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)
 }

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)
+	}
+}

internal/backend/agent/model/retry.go

@@ -1,12 +1,13 @@
-// retry.go 保留 provider HTTP 请求入口的历史命名；provider 错误交给客户端重连链路处理。
+// retry.go 保留 provider HTTP 请求入口的历史命名；单次 HTTP 调用不在适配器内重试。
+// 对尚未产生输出的瞬时 HTTP 错误，由 forwarder 在同一 RunSSE 会话内安全重试。
 package modeladapter
 
 import (
 	"context"
 	"net/http"
 )
 
-// DoProviderRequestWithRetry 保留旧入口名；本地模式不在服务端重试 provider 请求。
+// DoProviderRequestWithRetry 保留旧入口名；适配器只执行一次 provider HTTP 请求。
 func DoProviderRequestWithRetry(
 	ctx context.Context,
 	client *http.Client,

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)
 	}

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
+}

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")
+	}
+}

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{}