fix(api): preserve typed stream read errors

CHANGELOG.md

@@ -7,6 +7,7 @@ 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
 - Report Devin cache reads and writes in OpenAI-compatible usage, with prompt
   totals including all upstream input tokens
 - Generate Devin chat and account-status protobuf types from extracted descriptors
@@ -15,6 +16,7 @@ Write each change in both `### English` and `### 中文` under `## Unreleased`.
 
 ### 中文
 
+- OpenAI、Anthropic 与 Responses 流式转发会保留上游的类型化错误，避免无效的 Devin 请求被错误地冷却账号，同时传输中断仍可重试
 - Devin 的缓存读取与写入会显示在 OpenAI 兼容 usage 中，prompt 总数包含全部上游输入 token
 - 新增手动更新命令，提取 descriptor 并生成 Devin 聊天与账号状态 protobuf 类型；构建与 CI 直接使用已提交的 Go 文件，不下载发行包或重新生成 schema
 

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)