From a2dc057af7baba758ce21caf052d2c0c51ed0e3e Mon Sep 17 00:00:00 2001 From: Atkins Chang Date: Thu, 17 Sep 2026 11:42:28 +0800 Subject: [PATCH] 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. --- CHANGELOG.md | 4 +++ internal/api/chat.go | 10 +++++- internal/api/chat_usage_test.go | 57 +++++++++++++++++++++++++++++++++ internal/api/compat.go | 2 +- internal/api/compat_test.go | 25 +++++++++++++++ 5 files changed, 96 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d804c1b..9092b09 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/internal/api/chat.go b/internal/api/chat.go index 8fa860b..84b8f54 100644 --- a/internal/api/chat.go +++ b/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 diff --git a/internal/api/chat_usage_test.go b/internal/api/chat_usage_test.go index afe5d97..061913e 100644 --- a/internal/api/chat_usage_test.go +++ b/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 { diff --git a/internal/api/compat.go b/internal/api/compat.go index c74f1be..5600c4a 100644 --- a/internal/api/compat.go +++ b/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 diff --git a/internal/api/compat_test.go b/internal/api/compat_test.go index c5501c6..43f43ba 100644 --- a/internal/api/compat_test.go +++ b/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)