From b6dbc5614c347a78ea9121bb4606860f5eb9e3fe Mon Sep 17 00:00:00 2001 From: Ethan <39577870+ethanndickson@users.noreply.github.com> Date: Fri, 8 May 2026 15:52:42 +1000 Subject: [PATCH] fix(coderd/x/chatd): handle truncated provider streams (#25074) coder/fantasy now fails closed when Anthropic or OpenAI Responses streams close before their provider terminal events instead of yielding a successful finish. This bumps the fantasy replacement to coder/fantasy#33 and teaches chat error classification to treat those failures as retryable timeout errors with explicit stream-closed messages. image --- coderd/x/chatd/chaterror/classify.go | 58 +++++++++++++++++++ coderd/x/chatd/chaterror/classify_test.go | 29 ++++++++++ coderd/x/chatd/chaterror/message.go | 4 ++ coderd/x/chatd/chaterror/payload_test.go | 20 +++++++ .../x/chatd/chatprovider/chatprovider_test.go | 52 ++++++++++++++++- coderd/x/chatd/chatretry/chatretry_test.go | 17 ++++++ enterprise/coderd/x/chatd/chatd_test.go | 38 +++++++++--- go.mod | 8 ++- go.sum | 4 +- 9 files changed, 215 insertions(+), 15 deletions(-) diff --git a/coderd/x/chatd/chaterror/classify.go b/coderd/x/chatd/chaterror/classify.go index 3fae9ed344..e426a55fb4 100644 --- a/coderd/x/chatd/chaterror/classify.go +++ b/coderd/x/chatd/chaterror/classify.go @@ -32,6 +32,11 @@ type responsesAPIDiagnosticMatch struct { detail string } +type streamIncompleteMatch struct { + pattern string + provider string +} + // responsesAPIDiagnosticMatches maps provider error fragments to safe // diagnostics. Details must not include provider item IDs because they are // returned to clients and used by operators for grepping. @@ -46,6 +51,20 @@ var responsesAPIDiagnosticMatches = []responsesAPIDiagnosticMatch{ }, } +// streamIncompleteMatches maps provider stream-truncation errors from +// fantasy to clearer user-facing messages before broad EOF handling +// classifies them as generic transport timeouts. +var streamIncompleteMatches = []streamIncompleteMatch{ + { + pattern: "anthropic stream closed before message_stop", + provider: "anthropic", + }, + { + pattern: "openai responses stream closed before terminal event", + provider: "openai", + }, +} + // WithProvider returns a copy of the classification using an explicit // provider hint. Explicit provider hints are trusted over provider names // heuristically parsed from the error text. @@ -137,6 +156,15 @@ func Classify(err error) ClassifiedError { }) } + if classified, ok := streamIncompleteClassification( + lower, + provider, + statusCode, + structured, + ); ok { + return classified + } + deadline := errors.Is(err, context.DeadlineExceeded) || strings.Contains(lower, "context deadline exceeded") overloadedMatch := statusCode == 529 || containsAny(lower, overloadedPatterns...) authStrong := statusCode == 401 || containsAny(lower, authStrongPatterns...) @@ -218,6 +246,36 @@ func Classify(err error) ClassifiedError { }) } +func streamIncompleteClassification( + lowerMessage string, + provider string, + statusCode int, + structured providerErrorDetails, +) (ClassifiedError, bool) { + for _, match := range streamIncompleteMatches { + if !strings.Contains(lowerMessage, match.pattern) { + continue + } + if provider == "" { + provider = match.provider + } + return normalizeClassification(ClassifiedError{ + Message: streamIncompleteMessage(provider), + Detail: structured.detail, + Kind: codersdk.ChatErrorKindTimeout, + Provider: provider, + Retryable: true, + StatusCode: statusCode, + RetryAfter: structured.retryAfter, + }), true + } + return ClassifiedError{}, false +} + +func streamIncompleteMessage(provider string) string { + return providerSubject(provider) + " stream closed unexpectedly before the response completed." +} + func responsesAPIDiagnostic(lowerMessage, detail string) (string, bool) { lowerDetail := strings.ToLower(detail) for _, match := range responsesAPIDiagnosticMatches { diff --git a/coderd/x/chatd/chaterror/classify_test.go b/coderd/x/chatd/chaterror/classify_test.go index d987685f29..c73eac709b 100644 --- a/coderd/x/chatd/chaterror/classify_test.go +++ b/coderd/x/chatd/chaterror/classify_test.go @@ -2,6 +2,7 @@ package chaterror_test import ( "context" + "io" "net/http" "strings" "testing" @@ -45,6 +46,34 @@ func TestClassify(t *testing.T) { StatusCode: 0, }, }, + { + name: "AnthropicMissingMessageStop", + err: xerrors.Errorf( + "anthropic stream closed before message_stop: %w", + io.EOF, + ), + want: chaterror.ClassifiedError{ + Message: "Anthropic stream closed unexpectedly before the response completed.", + Kind: codersdk.ChatErrorKindTimeout, + Provider: "anthropic", + Retryable: true, + StatusCode: 0, + }, + }, + { + name: "OpenAIResponsesMissingTerminalEvent", + err: xerrors.Errorf( + "openai responses stream closed before terminal event: %w", + io.EOF, + ), + want: chaterror.ClassifiedError{ + Message: "OpenAI stream closed unexpectedly before the response completed.", + Kind: codersdk.ChatErrorKindTimeout, + Provider: "openai", + Retryable: true, + StatusCode: 0, + }, + }, { name: "AuthBeatsConfig", err: xerrors.New("authentication failed: invalid model"), diff --git a/coderd/x/chatd/chaterror/message.go b/coderd/x/chatd/chaterror/message.go index 4cd311908c..3bdb4c1482 100644 --- a/coderd/x/chatd/chaterror/message.go +++ b/coderd/x/chatd/chaterror/message.go @@ -63,6 +63,10 @@ func terminalMessage(classified ClassifiedError) string { // codes (surfaced separately in the payload) and remediation // guidance (not actionable while auto-retrying). func retryMessage(classified ClassifiedError) string { + if classified.Retryable && classified.Message != "" { + return classified.Message + } + subject := providerSubject(classified.Provider) switch classified.Kind { case codersdk.ChatErrorKindOverloaded: diff --git a/coderd/x/chatd/chaterror/payload_test.go b/coderd/x/chatd/chaterror/payload_test.go index be7cd14ace..2843e37430 100644 --- a/coderd/x/chatd/chaterror/payload_test.go +++ b/coderd/x/chatd/chaterror/payload_test.go @@ -1,6 +1,7 @@ package chaterror_test import ( + "io" "testing" "time" @@ -47,6 +48,25 @@ func TestTerminalErrorPayloadNilForEmptyClassification(t *testing.T) { require.Nil(t, chaterror.TerminalErrorPayload(chaterror.ClassifiedError{})) } +func TestStreamRetryPayloadPreservesRetryableMessage(t *testing.T) { + t.Parallel() + + delay := 3 * time.Second + classified := chaterror.Classify(xerrors.Errorf( + "anthropic stream closed before message_stop: %w", + io.EOF, + )) + payload := chaterror.StreamRetryPayload(2, delay, classified) + + require.NotNil(t, payload) + require.Equal(t, + "Anthropic stream closed unexpectedly before the response completed.", + payload.Error, + ) + require.Equal(t, codersdk.ChatErrorKindTimeout, payload.Kind) + require.Equal(t, "anthropic", payload.Provider) +} + func TestStreamRetryPayloadUsesNormalizedClassification(t *testing.T) { t.Parallel() diff --git a/coderd/x/chatd/chatprovider/chatprovider_test.go b/coderd/x/chatd/chatprovider/chatprovider_test.go index 01e403be6f..234d50857b 100644 --- a/coderd/x/chatd/chatprovider/chatprovider_test.go +++ b/coderd/x/chatd/chatprovider/chatprovider_test.go @@ -1,6 +1,7 @@ package chatprovider_test import ( + "encoding/base64" "encoding/json" "io" "net/http" @@ -13,6 +14,8 @@ import ( fantasyopenai "charm.land/fantasy/providers/openai" fantasyopenrouter "charm.land/fantasy/providers/openrouter" fantasyvercel "charm.land/fantasy/providers/vercel" + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream" + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream/eventstreamapi" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -1080,8 +1083,12 @@ func TestModelFromConfig_BedrockStreamingHeaders(t *testing.T) { ReadError: err, } - w.Header().Set("Content-Type", "application/vnd.amazon.eventstream") - w.WriteHeader(http.StatusOK) + if err := writeBedrockAnthropicStream(w, + `{"type":"message_start","message":{}}`, + `{"type":"message_stop"}`, + ); err != nil { + t.Errorf("write bedrock stream: %v", err) + } })) defer server.Close() @@ -1130,6 +1137,47 @@ func TestModelFromConfig_BedrockStreamingHeaders(t *testing.T) { require.Contains(t, got.Body, `"anthropic_version":"bedrock-2023-05-31"`) } +func writeBedrockAnthropicStream(w http.ResponseWriter, events ...string) error { + w.Header().Set("Content-Type", "application/vnd.amazon.eventstream") + w.WriteHeader(http.StatusOK) + + encoder := eventstream.NewEncoder() + for _, event := range events { + payload, err := json.Marshal(map[string]string{ + "bytes": base64.StdEncoding.EncodeToString([]byte(event)), + }) + if err != nil { + return err + } + + err = encoder.Encode(w, eventstream.Message{ + Headers: eventstream.Headers{ + { + Name: eventstreamapi.MessageTypeHeader, + Value: eventstream.StringValue(eventstreamapi.EventMessageType), + }, + { + Name: eventstreamapi.EventTypeHeader, + Value: eventstream.StringValue("chunk"), + }, + { + Name: eventstreamapi.ContentTypeHeader, + Value: eventstream.StringValue("application/json"), + }, + }, + Payload: payload, + }) + if err != nil { + return err + } + } + + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + return nil +} + func bedrockNonStreamingResponse() map[string]any { return map[string]any{ "id": "msg_01Test", diff --git a/coderd/x/chatd/chatretry/chatretry_test.go b/coderd/x/chatd/chatretry/chatretry_test.go index 640ffdf4e0..d17774d2f4 100644 --- a/coderd/x/chatd/chatretry/chatretry_test.go +++ b/coderd/x/chatd/chatretry/chatretry_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "sync/atomic" "testing" "time" @@ -27,6 +28,22 @@ func TestIsRetryableDelegatesToClassification(t *testing.T) { {name: "Nil", err: nil, retryable: false}, {name: "RetryableExplicitStatus429", err: xerrors.New("received status 429 from upstream"), retryable: true}, {name: "RetryableTimeout", err: xerrors.New("service unavailable"), retryable: true}, + { + name: "RetryableAnthropicMissingMessageStop", + err: xerrors.Errorf( + "anthropic stream closed before message_stop: %w", + io.EOF, + ), + retryable: true, + }, + { + name: "RetryableOpenAIResponsesMissingTerminalEvent", + err: xerrors.Errorf( + "openai responses stream closed before terminal event: %w", + io.EOF, + ), + retryable: true, + }, {name: "NonRetryableAuth", err: xerrors.New("invalid api key"), retryable: false}, {name: "NonRetryableGeneric", err: xerrors.New("boom"), retryable: false}, } diff --git a/enterprise/coderd/x/chatd/chatd_test.go b/enterprise/coderd/x/chatd/chatd_test.go index ad8d0867a1..6d66cc9179 100644 --- a/enterprise/coderd/x/chatd/chatd_test.go +++ b/enterprise/coderd/x/chatd/chatd_test.go @@ -1444,6 +1444,7 @@ func TestSubscribeRelayDrainWithinGraceLeavesBufferRetained(t *testing.T) { db, ps := dbtestutil.NewDB(t) workerID := uuid.New() subscriberID := uuid.New() + ctx := testutil.Context(t, testutil.WaitLong) openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse { if !req.Stream { @@ -1458,16 +1459,19 @@ func TestSubscribeRelayDrainWithinGraceLeavesBufferRetained(t *testing.T) { // Freeze the worker's clock so streamJanitorLoop cannot race the // buffer-retained assertion on slow CI. workerClock := quartz.NewMock(t) + trapAcquire := workerClock.Trap().NewTicker("chatd", "acquire") + defer trapAcquire.Close() worker := osschatd.New(osschatd.Config{ Logger: workerLogger, Database: db, ReplicaID: workerID, Pubsub: ps, - PendingChatAcquireInterval: time.Hour, + PendingChatAcquireInterval: time.Millisecond, InFlightChatStaleAfter: testutil.WaitSuperLong, Clock: workerClock, }) worker.Start() + trapAcquire.MustWait(ctx).MustRelease(ctx) t.Cleanup(func() { require.NoError(t, worker.Close()) }) @@ -1501,23 +1505,41 @@ func TestSubscribeRelayDrainWithinGraceLeavesBufferRetained(t *testing.T) { return snapshot, relayEvents, cancel, nil }, subscriberClock) - ctx := testutil.Context(t, testutil.WaitLong) user, org, model := seedChatDependencies(t, db) setOpenAIProviderBaseURL(ctx, t, db, openAIURL) chat := seedWaitingChat(t, db, org.ID, user, model, "relay-drain-characterization") + // Seed the pending turn directly instead of using SendMessage. + // SendMessage publishes a pending control notification that is + // irrelevant to this relay-retention case. Under CI that + // notification can arrive after processChat arms its control + // subscription and interrupt the worker before it emits parts. + dbgen.ChatMessage(t, db, database.ChatMessage{ + ChatID: chat.ID, + CreatedBy: uuid.NullUUID{UUID: user.ID, Valid: true}, + ModelConfigID: uuid.NullUUID{UUID: model.ID, Valid: true}, + Role: database.ChatMessageRoleUser, + Content: pqtype.NullRawMessage{ + RawMessage: json.RawMessage(`[{"type":"text","text":"hello"}]`), + Valid: true, + }, + }) + _, err := db.UpdateChatStatus(ctx, database.UpdateChatStatusParams{ + ID: chat.ID, + Status: database.ChatStatusPending, + }) + require.NoError(t, err) + // Attach before processing so the relay opens as soon as // status=running arrives. _, events, subCancel, ok := subscriber.Subscribe(ctx, chat.ID, nil, 0) require.True(t, ok) - _, err := worker.SendMessage(ctx, osschatd.SendMessageOptions{ - ChatID: chat.ID, - CreatedBy: user.ID, - Content: []codersdk.ChatMessagePart{codersdk.ChatMessageText("hello")}, - }) - require.NoError(t, err) + // Wake the worker with the acquire ticker. This keeps the + // setup free of pending control notifications while still + // exercising the normal processing loop. + workerClock.Advance(time.Millisecond).MustWait(ctx) // Drain events until processing has clearly completed: we need // the assistant message and at least one message_part so we know diff --git a/go.mod b/go.mod index 59da642853..5677bd9054 100644 --- a/go.mod +++ b/go.mod @@ -86,8 +86,10 @@ replace github.com/spf13/afero => github.com/aslilac/afero v0.0.0-20250403163713 // 7) coder/fantasy#mike/openai-responses-continuity, OpenAI Responses replay safety: // replay stored reasoning item references, only replay web_search references // when paired with reasoning, and validate function_call output pairing. -// See: https://github.com/coder/fantasy/commits/f83367a4a205 -replace charm.land/fantasy => github.com/coder/fantasy v0.0.0-20260427164812-d0e6ce2243af +// 8) coder/fantasy#33, fail closed when Anthropic or OpenAI Responses +// streams close before their terminal events. +// See: https://github.com/coder/fantasy/commits/246c4ae7aff9e +replace charm.land/fantasy => github.com/coder/fantasy v0.0.0-20260507124503-246c4ae7aff9 // coder/coder uses a fork of charmbracelet's fork of the Anthropic Go SDK // with performance improvements and Bedrock header cleanup. @@ -500,6 +502,7 @@ require ( require ( charm.land/fantasy v0.8.1 github.com/anthropics/anthropic-sdk-go v1.19.0 + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 github.com/aymanbagabas/go-udiff v0.4.1 github.com/brianvoe/gofakeit/v7 v7.14.0 github.com/coder/agentapi-sdk-go v0.0.0-20250505131810-560d1d88d225 @@ -544,7 +547,6 @@ require ( github.com/aquasecurity/jfather v0.0.8 // indirect github.com/aquasecurity/trivy v0.61.1-0.20250407075540-f1329c7ea1aa // indirect github.com/aquasecurity/trivy-checks v1.12.2-0.20251219190323-79d27547baf5 // indirect - github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 // indirect github.com/aws/aws-sdk-go-v2/service/s3 v1.97.3 // indirect diff --git a/go.sum b/go.sum index 6a9bf16d5d..88a1370db3 100644 --- a/go.sum +++ b/go.sum @@ -322,8 +322,8 @@ github.com/coder/bubbletea v1.2.2-0.20241212190825-007a1cdb2c41 h1:SBN/DA63+ZHwu github.com/coder/bubbletea v1.2.2-0.20241212190825-007a1cdb2c41/go.mod h1:I9ULxr64UaOSUv7hcb3nX4kowodJCVS7vt7VVJk/kW4= github.com/coder/clistat v1.2.1 h1:P9/10njXMyj5cWzIU5wkRsSy5LVQH49+tcGMsAgWX0w= github.com/coder/clistat v1.2.1/go.mod h1:m7SC0uj88eEERgvF8Kn6+w6XF21BeSr+15f7GoLAw0A= -github.com/coder/fantasy v0.0.0-20260427164812-d0e6ce2243af h1:5X38dLzIc5FSgVm9EuKkuKgtXt4fNV5iSCraxfgQXns= -github.com/coder/fantasy v0.0.0-20260427164812-d0e6ce2243af/go.mod h1:wZ0e3lEPqrM0XiIdAUQLvMKCLYhc3gi96MRX2wjbX44= +github.com/coder/fantasy v0.0.0-20260507124503-246c4ae7aff9 h1:Tj9Gq45h0zdDz3o1Un7ESGXkxO39dg+lRpWN7lks28A= +github.com/coder/fantasy v0.0.0-20260507124503-246c4ae7aff9/go.mod h1:wZ0e3lEPqrM0XiIdAUQLvMKCLYhc3gi96MRX2wjbX44= github.com/coder/flog v1.1.0 h1:kbAes1ai8fIS5OeV+QAnKBQE22ty1jRF/mcAwHpLBa4= github.com/coder/flog v1.1.0/go.mod h1:UQlQvrkJBvnRGo69Le8E24Tcl5SJleAAR7gYEHzAmdQ= github.com/coder/go-httpstat v0.0.0-20230801153223-321c88088322 h1:m0lPZjlQ7vdVpRBPKfYIFlmgevoTkBxB10wv6l2gOaU=