fix(llm): 提交前安全窗口内对同一 provider 重试,减少 model_error

SDK 的 doStream 只重试建连阶段(拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接冒泡成 model_error,零重试。表现为 worker 常出现「无总结,终态 model_error」,甚至只跑了一两个工具就中断。在 streamTaskLLM(所有任务 agent 的统一出口)里补一层同 provider 退避重试: - 仅在 committed 之前(还没向调用方交付任何输出)触发,重放相同请求不会重复模型输出或工具副作用,绝对安全 - 最多 2 次,退避 0.5s→1s(封顶 4s),ctx 取消立即让位 - 只重试瞬时故障(传输 reset/EOF/timeout、408/429/5xx、流内 overloaded); 额度耗尽交给 profile 故障转移、上下文过长交给 harness reactive 压缩、 4xx 确定性拒绝一律不重试 - 给不上时原样落到既有透传/故障转移逻辑,行为不变新增 3 个回归测试(重试到成功 / 重试耗尽如实上报 / 4xx 不重试),退避以变量暴露便于测试置零。go test -race 全绿。
2026-08-23 13:02:54 -04:00
1 parent d8f34cc79d
commit 9d8e0ba988
@@ -183,27 +183,46 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest
 			committed := false
 			var pending []llm.StreamEvent
 			var streamErr error
-			for event, err := range selection.provider.Stream(ctx, req) {
-				if err != nil {
-					streamErr = err
-					break
+			// 同 provider 安全窗口重试:committed 之前(还没向调用方交付任何输出)
+			// 的瞬时失败可以原样重放,不会重复模型输出或工具执行。committed 之后、
+			// ctx 取消、或确定性/额度错误则跳出,交给下方原有的透传/故障转移逻辑。
+			for attempt := 0; ; attempt++ {
+				committed = false
+				pending = nil
+				streamErr = nil
+				for event, err := range selection.provider.Stream(ctx, req) {
+					if err != nil {
+						streamErr = err
+						break
+					}
+					if !committed && !streamEventCommitsOutput(event) {
+						pending = append(pending, event)
+						continue
+					}
+					if !committed {
+						for _, buffered := range pending {
+							if !yield(buffered, nil) {
+								return
+							}
+						}
+						pending = nil
+						committed = true
+					}
+					if !yield(event, nil) {
+						return
+					}
 				}
-				if !committed && !streamEventCommitsOutput(event) {
-					pending = append(pending, event)
+				if streamErr != nil && !committed && ctx.Err() == nil &&
+					attempt < sameProviderStreamRetries && isRetryableStreamError(streamErr) {
+					backoff := sameProviderRetryBackoff(attempt)
+					log.Printf("[task-llm] task %s 提交前流失败,%v 后同 provider 重试 (%d/%d): %v",
+						taskID, backoff, attempt+1, sameProviderStreamRetries, streamErr)
+					if sleepCtx(ctx, backoff) {
+						break // 退避期间 ctx 取消 → 停止重试
+					}
 					continue
 				}
-				if !committed {
-					for _, buffered := range pending {
-						if !yield(buffered, nil) {
-							return
-						}
-					}
-					pending = nil
-					committed = true
-				}
-				if !yield(event, nil) {
-					return
-				}
+				break
 			}
 			if streamErr == nil {
 				for _, buffered := range pending {
@@ -262,6 +281,48 @@ func streamEventCommitsOutput(event llm.StreamEvent) bool {
 	}
 }

+// 提交前安全窗口内、对同一 provider 的重试次数。SDK 的 doStream 只重试建连阶段
+// (拿到 200 之前);流一旦开始,中途断流 / overloaded / 流内 429 等瞬时故障会直接
+// 冒泡成 model_error,零重试。只要一个 token 都还没交给调用方(!committed),重放
+// 完全相同的请求就不会重复模型输出或工具副作用,因此这里补一层同 provider 退避重试,
+// 把这类抖动挡在意图整体重跑之前。
+const sameProviderStreamRetries = 2
+
+// sameProviderRetryBackoff 是第 attempt 次重试前的退避(0.5s、1s…,上限 4s),
+// 与 SDK backoffSleep 同风格但封顶更小,避免拖住 worker 的收尾/取消响应。
+// 以变量形式暴露,便于测试将退避置零。
+var sameProviderRetryBackoff = func(attempt int) time.Duration {
+	return min(500*time.Millisecond*(1<<attempt), 4*time.Second)
+}
+
+// isRetryableStreamError 判断「提交前的流失败」是否值得在同一 provider 上重放。
+// 瞬时的传输中断 / 供应商过载 / 限流会自行恢复,可安全重试;而以下三类不重试:
+//   - 额度耗尽:交给 profile 故障转移处理,别在这里白烧重试
+//   - 上下文过长:相同请求重放也没用,交给 harness 的 reactive 压缩兜底
+//   - 4xx 确定性拒绝(400/401/403/404/422):到哪个 provider 都一样会失败
+func isRetryableStreamError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if isQuotaExhaustedError(err) {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	if strings.Contains(s, "too long") || strings.Contains(s, "context length") ||
+		strings.Contains(s, "context_length") || strings.Contains(s, "maximum context") ||
+		strings.Contains(s, "status 413") {
+		return false
+	}
+	for _, code := range []string{"status 400", "status 401", "status 403", "status 404", "status 422"} {
+		if strings.Contains(s, code) {
+			return false
+		}
+	}
+	// 其余(传输 reset/EOF/timeout、408/429/5xx、流内 error 事件如 anthropic
+	// overloaded_error 等)一律视为瞬时,允许重试。
+	return true
+}
+
 // CompactionWindow mirrors current()'s precedence so the context window always
 // matches the provider the role will actually stream on.
 func (r *taskLLMRuntime) CompactionWindow() int {
@@ -36,6 +36,42 @@ func (p *scriptedLLMProvider) Stream(context.Context, llm.CompletionRequest) ite
 	}
 }

+// withZeroRetryBackoff sets the same-provider retry backoff to zero for the
+// duration of a (serial) test and returns a restore func for defer.
+func withZeroRetryBackoff() func() {
+	prev := sameProviderRetryBackoff
+	sameProviderRetryBackoff = func(int) time.Duration { return 0 }
+	return func() { sameProviderRetryBackoff = prev }
+}
+
+// flakyThenOKProvider fails its first failCount stream attempts pre-commit
+// (emitting only a non-committing SEMessageStart before the error, mirroring a
+// gateway that returns 200 then drops), then serves okEvents.
+type flakyThenOKProvider struct {
+	failCount int
+	failErr   error
+	okEvents  []llm.StreamEvent
+	calls     int
+}
+
+func (p *flakyThenOKProvider) Stream(context.Context, llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] {
+	return func(yield func(llm.StreamEvent, error) bool) {
+		p.calls++
+		if p.calls <= p.failCount {
+			if !yield(llm.StreamEvent{Type: llm.SEMessageStart}, nil) {
+				return
+			}
+			yield(llm.StreamEvent{}, p.failErr)
+			return
+		}
+		for _, ev := range p.okEvents {
+			if !yield(ev, nil) {
+				return
+			}
+		}
+	}
+}
+
 func collectTaskLLMStream(seq iter.Seq2[llm.StreamEvent, error]) ([]llm.StreamEvent, error) {
 	var events []llm.StreamEvent
 	var streamErr error
@@ -265,7 +301,9 @@ func TestTaskLLMStreamRetriesStalePreStreamFailureAgainstReplacementChain(t *tes
 }

 func TestTaskLLMStreamDoesNotSwitchForOrdinaryErrors(t *testing.T) {
-	t.Parallel()
+	// Not parallel: overrides the package-level backoff so the retryable cases
+	// don't sleep. Serial tests never overlap the parallel batch, so this is safe.
+	defer withZeroRetryBackoff()()
 	tests := []string{
 		"openai: status 429: rate limit exceeded",
 		"openai: status 401: invalid api key",
@@ -379,6 +417,91 @@ func TestTaskLLMRuntimeErrorSkipsWorkerReplay(t *testing.T) {
 	}
 }

+// A transient pre-commit stream failure (200-then-drop / overloaded / 5xx) is
+// retried on the SAME provider, without switching profiles, and succeeds once the
+// blip clears — instead of surfacing as a model_error.
+func TestTaskLLMStreamRetriesTransientPreCommitOnSameProvider(t *testing.T) {
+	defer withZeroRetryBackoff()()
+	provider := &flakyThenOKProvider{
+		failCount: 2, // fail twice, succeed on the 3rd attempt (initial + 2 retries)
+		failErr:   errors.New("anthropic: overloaded_error"),
+		okEvents:  []llm.StreamEvent{{Type: llm.SEMessageStart}, {Type: llm.SETextDelta, Text: "ok"}},
+	}
+	exhausted := 0
+	hooks := taskLLMStreamHooks{
+		current: func() (taskLLMSelection, error) {
+			return taskLLMSelection{profileID: 11, provider: provider}, nil
+		},
+		exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
+			exhausted++
+			return db.TaskLLMTransition{}, nil
+		},
+	}
+	events, err := collectTaskLLMStream(streamTaskLLM(context.Background(), "7", llm.CompletionRequest{}, hooks))
+	if err != nil {
+		t.Fatalf("transient pre-commit failure should have been retried to success, got %v", err)
+	}
+	if provider.calls != 3 {
+		t.Fatalf("provider calls=%d, want 3 (1 initial + 2 retries)", provider.calls)
+	}
+	if exhausted != 0 {
+		t.Fatalf("same-provider retry must not change failover state (exhausted=%d)", exhausted)
+	}
+	if len(events) == 0 || events[len(events)-1].Text != "ok" {
+		t.Fatalf("recovered stream missing its committed output: %+v", events)
+	}
+}
+
+// Once retries are exhausted, the transient error is surfaced (becoming a
+// model_error upstream) after exactly sameProviderStreamRetries+1 attempts.
+func TestTaskLLMStreamSurfacesTransientAfterRetriesExhausted(t *testing.T) {
+	defer withZeroRetryBackoff()()
+	provider := &flakyThenOKProvider{
+		failCount: 99, // never recovers
+		failErr:   errors.New("dial tcp: connection reset by peer"),
+	}
+	hooks := taskLLMStreamHooks{
+		current: func() (taskLLMSelection, error) {
+			return taskLLMSelection{profileID: 11, provider: provider}, nil
+		},
+		exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
+			return db.TaskLLMTransition{}, nil
+		},
+	}
+	_, err := collectTaskLLMStream(streamTaskLLM(context.Background(), "7", llm.CompletionRequest{}, hooks))
+	if err == nil {
+		t.Fatal("persistent transient failure must still surface an error")
+	}
+	if provider.calls != sameProviderStreamRetries+1 {
+		t.Fatalf("provider calls=%d, want %d", provider.calls, sameProviderStreamRetries+1)
+	}
+}
+
+// A deterministic 4xx rejection is NOT retried on the same provider — replaying
+// the identical request everywhere fails the same way.
+func TestTaskLLMStreamDoesNotRetryDeterministicRejection(t *testing.T) {
+	defer withZeroRetryBackoff()()
+	provider := &flakyThenOKProvider{
+		failCount: 99,
+		failErr:   errors.New("anthropic: status 400: messages.1: invalid request"),
+	}
+	hooks := taskLLMStreamHooks{
+		current: func() (taskLLMSelection, error) {
+			return taskLLMSelection{profileID: 11, provider: provider}, nil
+		},
+		exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
+			return db.TaskLLMTransition{}, nil
+		},
+	}
+	_, err := collectTaskLLMStream(streamTaskLLM(context.Background(), "7", llm.CompletionRequest{}, hooks))
+	if err == nil {
+		t.Fatal("expected the deterministic rejection to surface")
+	}
+	if provider.calls != 1 {
+		t.Fatalf("deterministic 4xx must not be retried: provider calls=%d, want 1", provider.calls)
+	}
+}
+
 func TestEngineReadyForTaskOverride(t *testing.T) {
 	t.Parallel()
 	e := &Engine{}