fix(stream): 识别空流/截断流，不再伪造成功

proxy/account_failover.go

@@ -1,13 +1,88 @@
 package proxy
 
 import (
+	"errors"
 	"kiro-go/config"
 	"kiro-go/logger"
 	"strings"
 )
 
 const maxAccountRetryAttempts = 3
 
+// maxSameAccountStreamRetries bounds same-account recovery of a truncated
+// stream. Kiro IDE caps its truncation retry at one (Dt3 = 1 in extension.js),
+// but that budget was set for a single-credential client. This proxy also
+// rotates accounts, and the two recover different failures: a same-account
+// retry helps when the upstream hiccupped, rotation helps when that account's
+// backend is unhealthy. Two is therefore not a copy of the IDE.
+//
+// Cost of the extra attempt is bounded and small. Truncation returns nil from
+// CallKiroAPIContext, so it never reaches the endpoint fallback or
+// maxStreamAttemptsPerEndpoint - those fire only on transport errors. Worst
+// case across maxAccountRetryAttempts accounts is 3*(1+budget) requests: 9 at
+// two retries versus 6 at one.
+//
+// The payoff lands mostly on the three fully buffered paths, whose canRetry is
+// nil: a non-stream client gets a 500 with nothing usable, so one more chance
+// is worth more there than on a stream that has already flushed partial text.
+const maxSameAccountStreamRetries = 2
+
+// errUpstreamTruncatedResponse is a soft failure raised when a transport-clean
+// stream carried content but never a terminal signal. It is retryable on the
+// same account and must not mark the account unhealthy.
+//
+// There is deliberately no empty-response error here. A stream that produced no
+// output at all is already caught one layer down: parseEventStreamTracked
+// returns errEmptyKiroStream when !sawOutput (proxy/kiro.go), and
+// CallKiroAPIContext retries it internally. Since sawOutput is set by exactly
+// the three signals classifyStreamIntegrity measures (content, reasoning,
+// toolUse), an all-zero measurement can never reach this layer with a nil
+// error.
+var errUpstreamTruncatedResponse = errors.New("upstream truncated response without stop reason")
+
+// classifyStreamIntegrity decides whether an upstream stream that returned no
+// transport error is actually complete. parseEventStream reports success on a
+// clean EOF, so a stream that died mid-answer is otherwise indistinguishable
+// from a finished one.
+//
+// Complete when a stopReason arrived, or when a tool call was delivered. Both
+// match Kiro IDE, whose empty and truncation predicates each require
+// toolCallCount === 0.
+//
+// Truncated when content arrived without any terminal signal.
+//
+// Reasoning-only with no answer is STRICTER THAN THE IDE, deliberately. The
+// IDE's truncation predicate ends in (contentChars > 0 || !reasoningSeen), so
+// reasoning with no answer and no stopReason is treated as complete there and
+// is never retried. That is the exact shape of the production symptom this
+// proxy exists to fix: thinking streams in full, then the turn dies before the
+// answer or the tool call. Handing a client reasoning with no answer as a
+// successful turn is what made the failure invisible, so it is classified as
+// truncated here.
+func classifyStreamIntegrity(contentChars, toolCallCount int, stopReason string, sawReasoning bool) error {
+	if strings.TrimSpace(stopReason) != "" {
+		return nil
+	}
+	if toolCallCount > 0 {
+		return nil
+	}
+	if contentChars > 0 || sawReasoning {
+		return errUpstreamTruncatedResponse
+	}
+	// No content, no reasoning, no tools: unreachable through the wired paths
+	// (errEmptyKiroStream fires first, see above). Treated as truncated rather
+	// than complete so a future caller that bypasses that guard still cannot
+	// ship an empty turn as a success.
+	return errUpstreamTruncatedResponse
+}
+
+// isStreamIntegrityError reports whether err is a soft integrity failure.
+// Callers may rotate accounts on these, but must not run them through
+// handleAccountFailure: an upstream blip should not mark an account unhealthy.
+func isStreamIntegrityError(err error) bool {
+	return errors.Is(err, errUpstreamTruncatedResponse)
+}
+
 func isQuotaErrorMessage(msg string) bool {
 	msg = strings.ToLower(msg)
 	return strings.Contains(msg, "429") || strings.Contains(msg, "quota")

proxy/handler.go

@@ -1277,14 +1277,44 @@ func (h *Handler) handleClaudeStream(ctx context.Context, w http.ResponseWriter,
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return rawContentBuilder.Len(), len(toolUses), upstreamStopReason, rawThinkingBuilder.Len() > 0
+		}
+
+		// A same-account retry only happens while nothing has been flushed, so
+		// SSE block indices are still at their initial values and need no
+		// rollback. What must be cleared is every accumulator plus the thinking
+		// tag parser state: processClaudeText buffers up to 50 runes before
+		// flushing, so a short truncated attempt can leave a partial tag behind
+		// that would otherwise be prefixed onto the retry's first chunk.
+		reset := func() {
+			rawContentBuilder.Reset()
+			rawThinkingBuilder.Reset()
+			toolUses = nil
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+			textBuffer = ""
+			inThinkingBlock = false
+			dropTagThinking = false
+			thinkingSource = thinkingSourceUnknown
+			thinkingStarted = false
+			eventThinkingOpen = false
+		}
+
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset,
+			func() bool { return !messageStarted })
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			lastErr = err
 			excluded[account.ID] = true
-			h.handleAccountFailure(account, err)
+			if !isStreamIntegrityError(err) {
+				h.handleAccountFailure(account, err)
+			}
 			if !messageStarted {
 				continue
 			}
@@ -1555,14 +1585,33 @@ func (h *Handler) handleClaudeNonStream(ctx context.Context, w http.ResponseWrit
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return len(content), len(toolUses), upstreamStopReason, thinkingContent != ""
+		}
+
+		reset := func() {
+			content = ""
+			thinkingContent = ""
+			toolUses = nil
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+		}
+
+		// Fully buffered: nothing reaches the client until the response is
+		// encoded, so a retry can never duplicate output.
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset, nil)
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			lastErr = err
 			excluded[account.ID] = true
-			h.handleAccountFailure(account, err)
+			if !isStreamIntegrityError(err) {
+				h.handleAccountFailure(account, err)
+			}
 			continue
 		}
 
@@ -2010,14 +2059,44 @@ func (h *Handler) handleOpenAIStream(ctx context.Context, w http.ResponseWriter,
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return rawContentBuilder.Len(), len(toolCalls), upstreamStopReason, rawReasoningBuilder.Len() > 0
+		}
+
+		// Retries only happen before anything is flushed, so chunk indices stay
+		// valid. The thinking tag parser state must be cleared too: processText
+		// holds back up to 50 runes, so a short truncated attempt would
+		// otherwise prepend its leftovers to the retry's first chunk.
+		reset := func() {
+			rawContentBuilder.Reset()
+			rawReasoningBuilder.Reset()
+			toolCalls = nil
+			toolCallIndex = 0
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+			textBuffer = ""
+			inThinkingBlock = false
+			dropTagThinking = false
+			thinkingSource = thinkingSourceUnknown
+			thinkingStarted = false
+			eventThinkingOpen = false
+		}
+
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset,
+			func() bool { return !responseStarted })
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			lastErr = err
 			excluded[account.ID] = true
-			h.handleAccountFailure(account, err)
+			// Integrity failures are upstream hiccups, not account faults.
+			if !isStreamIntegrityError(err) {
+				h.handleAccountFailure(account, err)
+			}
 			if !responseStarted {
 				continue
 			}
@@ -2133,14 +2212,34 @@ func (h *Handler) handleOpenAINonStream(ctx context.Context, w http.ResponseWrit
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return len(content), len(toolUses), upstreamStopReason, reasoningContent != ""
+		}
+
+		reset := func() {
+			content = ""
+			reasoningContent = ""
+			toolUses = nil
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+		}
+
+		// Fully buffered: nothing reaches the client until the response is
+		// encoded, so a retry can never duplicate output.
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset, nil)
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			lastErr = err
 			excluded[account.ID] = true
-			h.handleAccountFailure(account, err)
+			// Integrity failures are upstream hiccups, not account faults.
+			if !isStreamIntegrityError(err) {
+				h.handleAccountFailure(account, err)
+			}
 			continue
 		}
 

proxy/handler_test.go

@@ -70,6 +70,9 @@ func TestClaudeNonStreamRetriesNextAccountAfterPreResponseFailure(t *testing.T)
 		_, _ = w.Write(awsEventStreamFrame(t, "assistantResponseEvent", map[string]interface{}{
 			"content": "retried successfully",
 		}))
+		_, _ = w.Write(awsEventStreamFrame(t, "metadataEvent", map[string]interface{}{
+			"stopReason": "end_turn",
+		}))
 	}))
 	defer server.Close()
 

proxy/responses_handler.go

@@ -171,14 +171,34 @@ func (h *Handler) handleResponsesNonStream(
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return len(content), len(toolUses), upstreamStopReason, reasoningContent != ""
+		}
+
+		reset := func() {
+			content = ""
+			reasoningContent = ""
+			toolUses = nil
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+		}
+
+		// Fully buffered path: nothing reaches the client until the response is
+		// encoded, so a retry can never duplicate output.
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset, nil)
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			lastErr = err
 			excluded[account.ID] = true
-			h.handleAccountFailure(account, err)
+			// Integrity failures are upstream hiccups, not account faults.
+			if !isStreamIntegrityError(err) {
+				h.handleAccountFailure(account, err)
+			}
 			continue
 		}
 
@@ -500,15 +520,38 @@ func (h *Handler) handleResponsesStream(
 			},
 		}
 
-		err := CallKiroAPIContext(ctx, account, payload, callback)
+		measure := func() (int, int, string, bool) {
+			return fullText.Len(), len(toolUses), upstreamStopReason, reasoningText.Len() > 0
+		}
+
+		// Retries only run while responseStarted is false, i.e. before any
+		// content or function-call item has been sent, so the output_index /
+		// content_index cursors are still untouched. Only the accumulators need
+		// clearing.
+		reset := func() {
+			fullText.Reset()
+			reasoningText.Reset()
+			toolUses = nil
+			inputTokens = 0
+			outputTokens = 0
+			credits = 0
+			realInputTokens = 0
+			upstreamStopReason = ""
+		}
+
+		err := runKiroWithIntegrityRetry(ctx, account, payload, callback, measure, reset,
+			func() bool { return !responseStarted })
 		if err != nil {
 			if ctx.Err() != nil {
 				return
 			}
 			if !responseStarted {
 				lastErr = err
 				excluded[account.ID] = true
-				h.handleAccountFailure(account, err)
+				// Integrity failures are upstream hiccups, not account faults.
+				if !isStreamIntegrityError(err) {
+					h.handleAccountFailure(account, err)
+				}
 				continue
 			}
 			send("response.failed", map[string]interface{}{

proxy/responses_handler_test.go

@@ -303,6 +303,9 @@ func TestResponsesContinuationKeepsNewInstructions(t *testing.T) {
 		_, _ = w.Write(awsEventStreamFrame(t, "assistantResponseEvent", map[string]interface{}{
 			"content": "second reply",
 		}))
+		_, _ = w.Write(awsEventStreamFrame(t, "metadataEvent", map[string]interface{}{
+			"stopReason": "end_turn",
+		}))
 	}))
 	defer server.Close()
 	defer swapKiroEndpointsForTest(t, server)()
@@ -381,6 +384,9 @@ func TestResponsesNonStreamRoundTrip(t *testing.T) {
 		_, _ = w.Write(awsEventStreamFrame(t, "assistantResponseEvent", map[string]interface{}{
 			"content": "responses non-stream OK",
 		}))
+		_, _ = w.Write(awsEventStreamFrame(t, "metadataEvent", map[string]interface{}{
+			"stopReason": "end_turn",
+		}))
 	}))
 	defer server.Close()
 	defer swapKiroEndpointsForTest(t, server)()
@@ -467,6 +473,9 @@ func TestResponsesStreamSSE(t *testing.T) {
 		_, _ = w.Write(awsEventStreamFrame(t, "assistantResponseEvent", map[string]interface{}{
 			"content": "stream chunk",
 		}))
+		_, _ = w.Write(awsEventStreamFrame(t, "metadataEvent", map[string]interface{}{
+			"stopReason": "end_turn",
+		}))
 	}))
 	defer server.Close()
 	defer swapKiroEndpointsForTest(t, server)()