diff --git a/CHANGELOG.md b/CHANGELOG.md index ee42850..224997d 100644 --- a/CHANGELOG.md +++ b/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 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)