refactor(model): centralize early stream termination classification (#184)

Move the empty-vs-truncated classification of streams that close before
message stop into model.NewStreamEndedEarlyError, deleting the duplicated
streamEndedEarlyError helpers in the Bedrock and Anthropic adapters. The
model package owns the stream integrity error contract; adapters now only
report the provider, operation, and whether a message had started.

Author: raphael

DESIGN.md

@@ -278,11 +278,16 @@ shapes are classified instead of surfaced as opaque protocol errors:
   `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`
+  before `messageStop`. The classification is 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.
 
+`model.NewStreamEndedEarlyError(provider, operation, started)` is the single
+classifier for streams that close before message stop; adapters pass whether a
+message had started and the model package owns which of the two shapes
+applies.
+
 Adapters never retry internally; retry policy belongs to the integrating
 application (for example, a guard that retries only before the first visible
 output chunk).

features/model/anthropic/stream.go

@@ -113,7 +113,11 @@ func (s *anthropicStreamer) run() {
 			} else if err := s.ctx.Err(); err != nil {
 				s.setErr(err)
 			} else if !processor.complete {
-				s.setErr(streamEndedEarlyError(processor.started))
+				s.setErr(model.NewStreamEndedEarlyError(
+					anthropicProviderName,
+					"stream_recv",
+					processor.started,
+				))
 			} else {
 				translated, err := translateResponse(&response, s.toolNameMap)
 				if err != nil {
@@ -500,30 +504,3 @@ 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/bedrock/stream.go

@@ -127,7 +127,11 @@ func (s *bedrockStreamer) run() {
 				} else if err := s.ctx.Err(); err != nil {
 					s.setErr(err)
 				} else if !processor.complete {
-					s.setErr(streamEndedEarlyError(processor.started))
+					s.setErr(model.NewStreamEndedEarlyError(
+						bedrockProviderName,
+						"converse_stream",
+						processor.started,
+					))
 				} else if err := processor.finishStream(); err != nil {
 					s.setErr(err)
 				} else {
@@ -589,33 +593,6 @@ 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

@@ -369,35 +369,3 @@ func TestChunkProcessorRejectsDuplicateMessageStop(t *testing.T) {
 	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/provider_error.go

@@ -29,10 +29,17 @@ 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"
+const (
+	// 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.
+	providerErrorCodeEmptyStream = "empty_stream"
+
+	// providerErrorCodeTruncatedStream is the stable code carried by provider
+	// errors for streams that closed mid-message. Like the empty-stream code,
+	// it exists for observability, not matching.
+	providerErrorCodeTruncatedStream = "truncated_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,
@@ -154,6 +161,36 @@ func NewEmptyStreamError(provider, operation, message string) error {
 	return errors.Join(ErrEmptyStream, pe)
 }
 
+// NewStreamEndedEarlyError classifies an event stream that terminated cleanly
+// before the provider's message-stop boundary. Adapters call it from their
+// stream event loops when the provider closes the connection without error
+// but the message protocol is unfinished. started reports whether an
+// assistant message had begun:
+//
+//   - started=false means the provider produced an empty completion; the
+//     result is an empty-stream error (see NewEmptyStreamError) that callers
+//     may retry via errors.Is(err, ErrEmptyStream).
+//   - started=true means the stream was truncated mid-generation; the result
+//     is a retryable unavailable ProviderError (code truncated_stream)
+//     without the empty-stream sentinel, so pre-output retry policies never
+//     match partially delivered responses.
+func NewStreamEndedEarlyError(provider, operation string, started bool) error {
+	if !started {
+		return NewEmptyStreamError(provider, operation, "stream ended before message start")
+	}
+	return NewProviderError(
+		provider,
+		operation,
+		0,
+		ProviderErrorKindUnavailable,
+		providerErrorCodeTruncatedStream,
+		"stream ended before message stop",
+		"",
+		true,
+		nil,
+	)
+}
+
 // 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,

runtime/agent/model/provider_error_test.go

@@ -65,6 +65,40 @@ func TestNewEmptyStreamError(t *testing.T) {
 	assert.Equal(t, "message stop received without an active message", pe.Message())
 }
 
+// TestNewStreamEndedEarlyError verifies the two terminal shapes of an event
+// stream that closes before message stop: never-started streams are retryable
+// empty streams, mid-message closes are retryable truncations that must not
+// carry the empty-stream sentinel.
+func TestNewStreamEndedEarlyError(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 := NewStreamEndedEarlyError("test-provider", "converse_stream", tt.started)
+
+			if tt.wantEmpty {
+				require.ErrorIs(t, err, ErrEmptyStream)
+			} else {
+				require.NotErrorIs(t, err, ErrEmptyStream)
+			}
+			pe, ok := AsProviderError(err)
+			require.True(t, ok)
+			assert.Equal(t, ProviderErrorKindUnavailable, pe.Kind())
+			assert.Equal(t, tt.wantCode, pe.Code())
+			assert.True(t, pe.Retryable())
+			assert.Equal(t, "test-provider", pe.Provider())
+			assert.Equal(t, "converse_stream", pe.Operation())
+		})
+	}
+}
+
 func TestClassifyHTTPStatusPreservesRateLimitedCause(t *testing.T) {
 	// A pre-classified sentinel (status 0) must still satisfy errors.Is via
 	// the Unwrap chain even though the status alone does not select the