fix(model): classify empty provider streams as retryable (#183)

Bedrock intermittently terminates ConverseStream responses without ever
starting an assistant message when a model emits an empty completion
(observed in production on Haiku 4.5 as messageStop with no prior
messageStart and zero output tokens). The strict stream state machine
correctly rejects the shape but surfaced it as an opaque protocol error
that retry middleware could not detect.

Introduce model.ErrEmptyStream plus model.NewEmptyStreamError, which pair
the sentinel with a retryable unavailable ProviderError (code
empty_stream). The Bedrock and Anthropic adapters classify message stop
without an active message and event streams that close before message
start through it, and classify streams that close mid-message as
retryable truncated_stream provider errors. Duplicate message stops stay
hard protocol errors. Adapters still never fabricate responses; retry
policy remains with the integrating application.

Author: raphael

DESIGN.md

@@ -261,6 +261,32 @@ that UIs and stream bridges can consume without heuristics.
 
 This keeps consumers simple: render `error`, gate “Retry” on `retryable`, and treat `canceled` as non-error.
 
+## Provider Stream Integrity Contract
+
+Provider adapters (Bedrock, Anthropic) validate the streaming event protocol
+with a strict state machine: a message must start before content blocks flow
+and must stop exactly once before metadata. Violations never produce a
+fabricated response; they fail the stream with a precise error. Two terminal
+shapes are classified instead of surfaced as opaque protocol errors:
+
+- **Empty stream** — the stream terminates before any message starts (a
+  `messageStop` with no prior `messageStart`, or a stream that closes with no
+  events at all). Providers intermittently do this when a model emits an
+  empty completion. Adapters build the error with `model.NewEmptyStreamError`,
+  which carries the `model.ErrEmptyStream` sentinel plus a retryable
+  `unavailable` ProviderError (code `empty_stream`). Callers detect it with
+  `errors.Is(err, model.ErrEmptyStream)` and may retry the request a bounded
+  number of times before surfacing the failure.
+- **Truncated stream** — the stream closes cleanly after a message started but
+  before `messageStop`. Adapters classify it as a retryable `unavailable`
+  ProviderError (code `truncated_stream`) without the empty-stream sentinel:
+  output was partially produced, so blind pre-output retry policies must not
+  match it.
+
+Adapters never retry internally; retry policy belongs to the integrating
+application (for example, a guard that retries only before the first visible
+output chunk).
+
 ## Runtime Tracing Error Contract
 
 The runtime uses one generic rule for span failures across model clients and

docs/runtime.md

@@ -2361,8 +2361,15 @@ var ErrNotFound = errors.New("run not found")  // run.ErrNotFound
 ```go
 var ErrStreamingUnsupported = errors.New("model: streaming not supported")
 var ErrRateLimited = errors.New("model: rate limited")
+var ErrEmptyStream = errors.New("model: provider returned an empty stream")
 ```
 
+`ErrEmptyStream` is joined (via `model.NewEmptyStreamError`) with a retryable
+`unavailable` ProviderError when a provider terminates a stream before any
+assistant message starts — the wire shape produced by intermittent empty model
+completions. Detect it with `errors.Is(err, model.ErrEmptyStream)` and retry
+the request a bounded number of times before surfacing the failure.
+
 ---
 
 ## Best Practices

features/model/anthropic/client.go

@@ -86,6 +86,9 @@ type (
 	}
 )
 
+// anthropicProviderName identifies this adapter in model.ProviderError values.
+const anthropicProviderName = "anthropic"
+
 // New builds an Anthropic-backed model client from the provided Anthropic
 // Messages client and configuration options.
 func New(msg MessagesClient, opts Options) (*Client, error) {
@@ -625,7 +628,7 @@ func wrapAnthropicError(operation string, err error) error {
 	} else {
 		message = err.Error()
 	}
-	return model.ClassifyHTTPStatus("anthropic", operation, status, message, err)
+	return model.ClassifyHTTPStatus(anthropicProviderName, operation, status, message, err)
 }
 
 // anthropicErrorMessage safely renders an *sdk.Error's message.

features/model/anthropic/stream.go

@@ -113,7 +113,7 @@ func (s *anthropicStreamer) run() {
 			} else if err := s.ctx.Err(); err != nil {
 				s.setErr(err)
 			} else if !processor.complete {
-				s.setErr(errors.New("anthropic: stream ended before message_stop"))
+				s.setErr(streamEndedEarlyError(processor.started))
 			} else {
 				translated, err := translateResponse(&response, s.toolNameMap)
 				if err != nil {
@@ -410,8 +410,18 @@ func (p *anthropicChunkProcessor) Handle(event sdk.MessageStreamEventUnion) erro
 		}
 		return p.emit(model.UsageChunk{Usage: usage})
 	case sdk.MessageStopEvent:
-		if !p.started || p.complete {
-			return errors.New("anthropic stream: message stop received without an active message")
+		if !p.started {
+			// Anthropic models intermittently emit an empty completion whose
+			// stream stops a message that never started. Classify as a
+			// retryable empty stream instead of an opaque protocol error.
+			return model.NewEmptyStreamError(
+				anthropicProviderName,
+				"stream_recv",
+				"message stop received without an active message",
+			)
+		}
+		if p.complete {
+			return errors.New("anthropic stream: duplicate message stop")
 		}
 		if len(p.openBlocks) > 0 {
 			return fmt.Errorf("anthropic stream: message stopped with %d open content blocks", len(p.openBlocks))
@@ -490,3 +500,30 @@ func decodeToolPayload(raw string) (rawjson.Message, error) {
 	}
 	return rawjson.Message(data), nil
 }
+
+// streamEndedEarlyError classifies an Anthropic event stream that closed
+// cleanly before message stop. When no message ever started, the provider
+// produced an empty completion and callers may retry (model.ErrEmptyStream).
+// When a message was underway, the stream was truncated mid-generation: a
+// fresh request regenerates the full response, so the failure is a retryable
+// provider fault but not an empty stream.
+func streamEndedEarlyError(started bool) error {
+	if !started {
+		return model.NewEmptyStreamError(
+			anthropicProviderName,
+			"stream_recv",
+			"stream ended before message start",
+		)
+	}
+	return model.NewProviderError(
+		anthropicProviderName,
+		"stream_recv",
+		0,
+		model.ProviderErrorKindUnavailable,
+		"truncated_stream",
+		"stream ended before message stop",
+		"",
+		true,
+		nil,
+	)
+}

features/model/anthropic/stream_test.go

@@ -223,19 +223,44 @@ func TestAnthropicStreamer_ContextCancelPassthrough(t *testing.T) {
 	assert.False(t, ok)
 }
 
-// TestAnthropicStreamerRejectsEOFBeforeCanonicalResponse verifies that a stream
-// cannot terminate successfully without the provider's message_stop boundary.
-func TestAnthropicStreamerRejectsEOFBeforeCanonicalResponse(t *testing.T) {
+// TestAnthropicStreamerClassifiesEventlessStreamAsEmptyStream verifies that a
+// stream closing before any message starts is classified as a retryable empty
+// stream (model.ErrEmptyStream) instead of an opaque protocol error, so retry
+// middleware can safely reissue the request.
+func TestAnthropicStreamerClassifiesEventlessStreamAsEmptyStream(t *testing.T) {
 	dec := &testDecoder{events: nil}
 	stream := ssestream.NewStream[sdk.MessageStreamEventUnion](dec, nil)
 
 	s := newAnthropicStreamer(context.Background(), stream, nil)
 	defer func() { _ = s.Close() }()
 
 	_, err := s.Recv()
-	require.EqualError(t, err, "anthropic: stream ended before message_stop")
-	_, ok := model.AsProviderError(err)
-	assert.False(t, ok)
+	require.ErrorIs(t, err, model.ErrEmptyStream)
+	pe, ok := model.AsProviderError(err)
+	require.True(t, ok)
+	assert.Equal(t, model.ProviderErrorKindUnavailable, pe.Kind())
+	assert.True(t, pe.Retryable())
+}
+
+// TestAnthropicStreamerClassifiesMessageStopWithoutStartAsEmptyStream verifies
+// that a message_stop arriving before message_start carries the empty-stream
+// classification: this is the wire shape Anthropic-family models produce when
+// they emit an empty completion.
+func TestAnthropicStreamerClassifiesMessageStopWithoutStartAsEmptyStream(t *testing.T) {
+	var stop sdk.MessageStreamEventUnion
+	require.NoError(t, json.Unmarshal([]byte(`{"type":"message_stop"}`), &stop))
+	events := []ssestream.Event{{Type: "message_stop", Data: mustJSON(stop)}}
+	stream := ssestream.NewStream[sdk.MessageStreamEventUnion](&testDecoder{events: events}, nil)
+
+	s := newAnthropicStreamer(context.Background(), stream, nil)
+	defer func() { _ = s.Close() }()
+
+	_, err := s.Recv()
+	require.ErrorIs(t, err, model.ErrEmptyStream)
+	pe, ok := model.AsProviderError(err)
+	require.True(t, ok)
+	assert.Equal(t, model.ProviderErrorKindUnavailable, pe.Kind())
+	assert.True(t, pe.Retryable())
 }
 
 func TestAnthropicStreamerRejectsMessageStopWithOpenContentBlock(t *testing.T) {

features/model/bedrock/stream.go

@@ -127,7 +127,7 @@ func (s *bedrockStreamer) run() {
 				} else if err := s.ctx.Err(); err != nil {
 					s.setErr(err)
 				} else if !processor.complete {
-					s.setErr(errors.New("bedrock: stream ended before message stop"))
+					s.setErr(streamEndedEarlyError(processor.started))
 				} else if err := processor.finishStream(); err != nil {
 					s.setErr(err)
 				} else {
@@ -489,8 +489,18 @@ func (p *chunkProcessor) Handle(event any) error {
 		}
 		return nil
 	case *brtypes.ConverseStreamOutputMemberMessageStop:
-		if !p.started || p.complete {
-			return errors.New("bedrock stream: message stop received without an active message")
+		if !p.started {
+			// Bedrock intermittently stops a message it never started when the
+			// model produces an empty completion (observed on Haiku). Classify
+			// as a retryable empty stream instead of an opaque protocol error.
+			return model.NewEmptyStreamError(
+				bedrockProviderName,
+				"converse_stream",
+				"message stop received without an active message",
+			)
+		}
+		if p.complete {
+			return errors.New("bedrock stream: duplicate message stop")
 		}
 		if len(p.openBlocks) > 0 {
 			return fmt.Errorf("bedrock stream: message stopped with %d open content blocks", len(p.openBlocks))
@@ -579,6 +589,33 @@ func (p *chunkProcessor) finishStream() error {
 	return p.emit(model.StopChunk{Reason: p.canonical.StopReason})
 }
 
+// streamEndedEarlyError classifies a Bedrock event stream that closed cleanly
+// before message stop. When no message ever started, the provider produced an
+// empty completion and callers may retry (model.ErrEmptyStream). When a
+// message was underway, the stream was truncated mid-generation: a fresh
+// request regenerates the full response, so the failure is a retryable
+// provider fault but not an empty stream.
+func streamEndedEarlyError(started bool) error {
+	if !started {
+		return model.NewEmptyStreamError(
+			bedrockProviderName,
+			"converse_stream",
+			"stream ended before message start",
+		)
+	}
+	return model.NewProviderError(
+		bedrockProviderName,
+		"converse_stream",
+		0,
+		model.ProviderErrorKindUnavailable,
+		"truncated_stream",
+		"stream ended before message stop",
+		"",
+		true,
+		nil,
+	)
+}
+
 type toolBuffer struct {
 	name      string
 	id        string

features/model/bedrock/stream_usage_test.go

@@ -319,3 +319,85 @@ func TestChunkProcessorRejectsMessageStopWithOpenContentBlock(t *testing.T) {
 
 	require.EqualError(t, err, "bedrock stream: message stopped with 1 open content blocks")
 }
+
+// TestChunkProcessorClassifiesMessageStopWithoutStartAsEmptyStream verifies
+// that a messageStop arriving before messageStart is classified as a
+// retryable empty stream (model.ErrEmptyStream). Bedrock intermittently
+// produces this wire shape when the model emits an empty completion, so retry
+// middleware must be able to detect it without string matching.
+func TestChunkProcessorClassifiesMessageStopWithoutStartAsEmptyStream(t *testing.T) {
+	cp := newChunkProcessor(
+		func(model.Chunk) error { return nil },
+		map[string]string{},
+		"test-model-id",
+		model.ModelClassDefault,
+		nil,
+	)
+
+	err := cp.Handle(&brtypes.ConverseStreamOutputMemberMessageStop{
+		Value: brtypes.MessageStopEvent{StopReason: brtypes.StopReasonEndTurn},
+	})
+
+	require.ErrorIs(t, err, model.ErrEmptyStream)
+	pe, ok := model.AsProviderError(err)
+	require.True(t, ok)
+	require.Equal(t, model.ProviderErrorKindUnavailable, pe.Kind())
+	require.Equal(t, "empty_stream", pe.Code())
+	require.True(t, pe.Retryable())
+}
+
+// TestChunkProcessorRejectsDuplicateMessageStop verifies that a second
+// messageStop after a completed message stays a hard protocol error and is
+// not mistaken for an empty stream.
+func TestChunkProcessorRejectsDuplicateMessageStop(t *testing.T) {
+	cp := newChunkProcessor(
+		func(model.Chunk) error { return nil },
+		map[string]string{},
+		"test-model-id",
+		model.ModelClassDefault,
+		nil,
+	)
+
+	require.NoError(t, cp.Handle(&brtypes.ConverseStreamOutputMemberMessageStart{}))
+	require.NoError(t, cp.Handle(&brtypes.ConverseStreamOutputMemberMessageStop{
+		Value: brtypes.MessageStopEvent{StopReason: brtypes.StopReasonEndTurn},
+	}))
+	err := cp.Handle(&brtypes.ConverseStreamOutputMemberMessageStop{
+		Value: brtypes.MessageStopEvent{StopReason: brtypes.StopReasonEndTurn},
+	})
+
+	require.EqualError(t, err, "bedrock stream: duplicate message stop")
+	require.NotErrorIs(t, err, model.ErrEmptyStream)
+}
+
+// TestStreamEndedEarlyErrorClassification verifies the two terminal shapes of
+// a Bedrock event stream that closes before messageStop: never-started
+// streams are retryable empty streams, mid-message closes are retryable
+// truncations that must not carry the empty-stream sentinel.
+func TestStreamEndedEarlyErrorClassification(t *testing.T) {
+	tests := []struct {
+		name      string
+		started   bool
+		wantEmpty bool
+		wantCode  string
+	}{
+		{name: "never started", started: false, wantEmpty: true, wantCode: "empty_stream"},
+		{name: "truncated mid message", started: true, wantEmpty: false, wantCode: "truncated_stream"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := streamEndedEarlyError(tt.started)
+
+			if tt.wantEmpty {
+				require.ErrorIs(t, err, model.ErrEmptyStream)
+			} else {
+				require.NotErrorIs(t, err, model.ErrEmptyStream)
+			}
+			pe, ok := model.AsProviderError(err)
+			require.True(t, ok)
+			require.Equal(t, model.ProviderErrorKindUnavailable, pe.Kind())
+			require.Equal(t, tt.wantCode, pe.Code())
+			require.True(t, pe.Retryable())
+		})
+	}
+}

runtime/agent/model/model.go

@@ -926,6 +926,15 @@ var ErrStructuredOutputUnsupported = errors.New("model: structured output not su
 // failure that is safe to surface to higher layers.
 var ErrRateLimited = errors.New("model: rate limited")
 
+// ErrEmptyStream indicates the provider terminated a streaming response
+// without ever starting an assistant message, so no output was produced.
+// Providers intermittently do this when a model emits an empty completion
+// (observed on Bedrock as messageStop with no prior messageStart). Adapters
+// classify the condition with NewEmptyStreamError instead of fabricating an
+// empty response; callers detect it with errors.Is and may retry the request
+// a bounded number of times before surfacing the failure.
+var ErrEmptyStream = errors.New("model: provider returned an empty stream")
+
 // ToolCalls derives model tool invocations from their ordered ToolUsePart
 // entries. Response.Content remains the single source of provider output.
 func (r *Response) ToolCalls() []ToolCall {

runtime/agent/model/provider_error.go

@@ -29,6 +29,11 @@ const (
 	ProviderErrorKindUnknown ProviderErrorKind = "unknown"
 )
 
+// providerErrorCodeEmptyStream is the stable code carried by empty-stream
+// provider errors. Detection goes through errors.Is(err, ErrEmptyStream);
+// the code exists for observability (span/error attributes), not matching.
+const providerErrorCodeEmptyStream = "empty_stream"
+
 // ProviderError describes a failure returned by a model provider (e.g. Bedrock).
 // It is intended to cross package boundaries so runtimes can surface stable,
 // structured information to callers.
@@ -125,6 +130,30 @@ func AsProviderError(err error) (*ProviderError, bool) {
 	return nil, false
 }
 
+// NewEmptyStreamError classifies a streaming response that the provider
+// terminated without ever starting an assistant message. Adapters call it
+// from their stream event loops when the terminal event (or the end of the
+// event stream) arrives before any message started, which providers
+// intermittently produce for empty model completions. The result carries
+// ErrEmptyStream for errors.Is detection plus a retryable unavailable
+// ProviderError so retry middleware and observability see one consistent
+// classification. message describes the protocol shape observed (for
+// example, "message stop received without an active message").
+func NewEmptyStreamError(provider, operation, message string) error {
+	pe := NewProviderError(
+		provider,
+		operation,
+		0,
+		ProviderErrorKindUnavailable,
+		providerErrorCodeEmptyStream,
+		message,
+		"",
+		true,
+		nil,
+	)
+	return errors.Join(ErrEmptyStream, pe)
+}
+
 // ClassifyHTTPStatus maps an HTTP status code returned by a model provider to
 // the goa-ai provider error contract. It is the single status-to-kind table
 // for adapters that need only status-based classification (Vertex Gemini,