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.

<img width="875" height="311" alt="image"
src="https://github.com/user-attachments/assets/69c6f7b5-c885-46d2-a88b-b7a2b111bd55"
/>

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 {

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"),

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:

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()
 

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",

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},
 	}

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

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

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=