fix(api): preserve typed stream read errors

Devin can finish the HTTP exchange and then deliver a typed Connect error
through the stream body. Reclassifying it as an interruption turns malformed
requests into account cooldowns.

Preserve wrapped provider errors at each SSE reader and retain their
classification, failover decision, and retry hint. Continue classifying unknown
transport errors as unavailable so account routing can retry them.

Author: AtkinsChang

CHANGELOG.md

@@ -7,8 +7,12 @@ Write each change in both `### English` and `### 中文` under `## Unreleased`.
 
 ### English
 
+- Preserve typed upstream stream errors through the OpenAI, Anthropic, and Responses relays so invalid Devin requests do not falsely cool accounts, while transport interruptions remain retryable
+
 ### 中文
 
+- OpenAI、Anthropic 与 Responses 流式转发会保留上游的类型化错误，避免无效的 Devin 请求被错误地冷却账号，同时传输中断仍可重试
+
 ## 0.5.6 - 2026-09-17
 
 ### English

internal/api/chat.go

@@ -153,7 +153,7 @@ func relayOpenAIStream(w http.ResponseWriter, body io.Reader) (stats streamRelay
 		if streamErr != nil {
 			return stats, streamErr
 		}
-		streamErr := newStreamProviderError("upstream_stream_interrupted", "stream read error: "+err.Error(), http.StatusBadGateway)
+		streamErr := streamReadProviderError(err)
 		if writeErr := writeStructuredStreamError(writer, streamErr); writeErr != nil {
 			return stats, writeErr
 		}
@@ -376,6 +376,14 @@ func newStreamProviderError(code, message string, status int) *providers.Error {
 	return providerErrorFromClassified(classified)
 }
 
+func streamReadProviderError(err error) *providers.Error {
+	var providerErr *providers.Error
+	if errors.As(err, &providerErr) && providerErr != nil {
+		return providerErr
+	}
+	return newStreamProviderError("upstream_stream_interrupted", "stream read error: "+err.Error(), http.StatusBadGateway)
+}
+
 func providerErrorFromClassified(classified accounts.Classified) *providers.Error {
 	failover := classified.Failover
 	retryAfter := classified.RetryAfter

internal/api/chat_usage_test.go

@@ -2,6 +2,8 @@ package api
 
 import (
 	"errors"
+	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -17,6 +19,12 @@ import (
 
 func intPtr(value int) *int { return &value }
 
+func closedStreamPipe(err error) *io.PipeReader {
+	reader, writer := io.Pipe()
+	_ = writer.CloseWithError(err)
+	return reader
+}
+
 func TestWriteClassifiedErrKeepsTraeQuotaKind(t *testing.T) {
 	recorder := httptest.NewRecorder()
 	failover := true
@@ -247,6 +255,55 @@ func TestRelayOpenAIStreamReportsIncompleteStreamStructurally(t *testing.T) {
 	}
 }
 
+func TestRelayOpenAIStreamPreservesTypedReadError(t *testing.T) {
+	recorder := httptest.NewRecorder()
+	failover := false
+	want := &providers.Error{
+		Kind: accounts.KindInvalidRequest, Status: http.StatusBadRequest,
+		Code: "invalid_argument", Type: "invalid_request_error", Message: "upstream rejected request",
+		RetryAfter: 45 * time.Second, Failover: &failover,
+	}
+	_, err := relayOpenAIStream(recorder, closedStreamPipe(fmt.Errorf("Connect trailer: %w", want)))
+	var got *providers.Error
+	if !errors.As(err, &got) || got != want {
+		t.Fatalf("error=%T %+v want pointer=%p", err, err, want)
+	}
+	if got.Kind != accounts.KindInvalidRequest || got.Status != http.StatusBadRequest || got.Code != "invalid_argument" ||
+		got.Type != "invalid_request_error" || got.RetryAfter != 45*time.Second || got.Failover == nil || *got.Failover {
+		t.Fatalf("provider error=%+v", got)
+	}
+	output := recorder.Body.String()
+	if !strings.Contains(output, `"code":"invalid_argument"`) || !strings.Contains(output, `"retry_after":45`) || strings.Contains(output, "upstream_stream_interrupted") {
+		t.Fatalf("structured error=%s", output)
+	}
+	pool := accounts.NewPool(nil, nil)
+	pool.Upsert(accounts.Item{ID: "devin-account"})
+	executor.NewChatExecutor(pool, "").ObserveStreamFailure("devin-account", got, "swe-2")
+	item, _ := pool.ByID("devin-account")
+	if item.LastKind != "" || !item.DownUntil.IsZero() {
+		t.Fatalf("invalid request cooled account: kind=%q down=%v", item.LastKind, item.DownUntil)
+	}
+}
+
+func TestRelayOpenAIStreamWrapsUnknownReadError(t *testing.T) {
+	recorder := httptest.NewRecorder()
+	_, err := relayOpenAIStream(recorder, closedStreamPipe(errors.New("socket closed")))
+	var got *providers.Error
+	if !errors.As(err, &got) || got.Kind != accounts.KindUnavailable || got.Code != "upstream_stream_interrupted" || got.Status != http.StatusBadGateway {
+		t.Fatalf("error=%T %+v", err, err)
+	}
+	if !strings.Contains(got.Message, "stream read error: socket closed") {
+		t.Fatalf("message=%q", got.Message)
+	}
+	pool := accounts.NewPool(nil, nil)
+	pool.Upsert(accounts.Item{ID: "devin-account"})
+	executor.NewChatExecutor(pool, "").ObserveStreamFailure("devin-account", got, "swe-2")
+	item, _ := pool.ByID("devin-account")
+	if item.LastKind != accounts.KindUnavailable || item.DownUntil.IsZero() {
+		t.Fatalf("transport interruption was not unavailable: kind=%q down=%v", item.LastKind, item.DownUntil)
+	}
+}
+
 func TestParseStreamUsageLineReadsWorkBuddyCredit(t *testing.T) {
 	stats, ok := parseStreamUsageLine(`data: {"model":"hy3","usage":{"prompt_tokens":16,"completion_tokens":2,"credit":0.75}}`)
 	if !ok || stats.Credits == nil || *stats.Credits != 0.75 {

internal/api/compat.go

@@ -524,7 +524,7 @@ func consumeOpenAIStream(body io.Reader, handle func(json.RawMessage, *streamedC
 		frame = append(frame, line)
 	}
 	if err := scanner.Err(); err != nil {
-		return stats, output, newStreamProviderError("upstream_stream_interrupted", "stream read error: "+err.Error(), http.StatusBadGateway)
+		return stats, output, streamReadProviderError(err)
 	}
 	if err := flush(); err != nil {
 		return stats, output, err

internal/api/compat_test.go

@@ -2,19 +2,44 @@ package api
 
 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/caigee-cmd/cli2api/internal/accounts"
 	"github.com/caigee-cmd/cli2api/internal/auth"
 	"github.com/caigee-cmd/cli2api/internal/executor"
+	"github.com/caigee-cmd/cli2api/internal/providers"
 	"github.com/caigee-cmd/cli2api/internal/translate"
 )
 
+func TestCompatibilityStreamsPreserveTypedReadError(t *testing.T) {
+	failover := false
+	want := &providers.Error{
+		Kind: accounts.KindInvalidRequest, Status: http.StatusBadRequest,
+		Code: "invalid_argument", Type: "invalid_request_error", Message: "upstream rejected request",
+		RetryAfter: 45 * time.Second, Failover: &failover,
+	}
+	for name, relay := range map[string]func(io.Writer, io.Reader, string, string) (streamRelayStats, error){
+		"anthropic": relayAnthropicStream,
+		"responses": relayResponsesStream,
+	} {
+		t.Run(name, func(t *testing.T) {
+			_, err := relay(httptest.NewRecorder(), closedStreamPipe(fmt.Errorf("Connect trailer: %w", want)), "req_1", "devin/swe-2")
+			var got *providers.Error
+			if !errors.As(err, &got) || got != want {
+				t.Fatalf("error=%T %+v want pointer=%p", err, err, want)
+			}
+		})
+	}
+}
+
 func newCompatibilityServer(t *testing.T, worker http.HandlerFunc) (*Server, func()) {
 	t.Helper()
 	upstream := httptest.NewServer(worker)