From 8eb3e4f879d7b3ddabff88938e9bb663d9cc443d Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 20 Aug 2026 15:50:38 +1000 Subject: [PATCH] fix: stop emitting an empty contentBlockStart union for text in ConverseStream (#844) Real Bedrock emits contentBlockStart only for toolUse blocks; text flows messageStart -> contentBlockDelta -> contentBlockStop -> messageStop. An empty start:{} union violates Smithy's exactly-one-member rule and breaks strict SDK stream consumers (boto3 errors, the AWS Go SDK stalls). --- spinifex/gateway/bedrock/anthropic.go | 22 ++++++++++--- spinifex/gateway/bedrock/anthropic_test.go | 19 ++++++----- spinifex/gateway/bedrock/contract_test.go | 4 +-- .../gateway/bedrock/converse_stream_test.go | 11 +++---- .../gateway/bedrock/guardrail_inference.go | 1 - .../bedrock/guardrail_inference_test.go | 28 ++++++++-------- spinifex/gateway/bedrock/vllm.go | 9 ++---- spinifex/gateway/bedrock/vllm_test.go | 32 +++++++++---------- 8 files changed, 65 insertions(+), 61 deletions(-) diff --git a/spinifex/gateway/bedrock/anthropic.go b/spinifex/gateway/bedrock/anthropic.go index 90d2e1d1e..ee107f1d8 100644 --- a/spinifex/gateway/bedrock/anthropic.go +++ b/spinifex/gateway/bedrock/anthropic.go @@ -363,7 +363,14 @@ func (s *anthropicConverseStreamSource) Next(_ context.Context) (ConverseStreamE case "message_start": return s.handleMessageStart(ev.Data) case "content_block_start": - return s.handleContentBlockStart(ev.Data) + event, emit, err := s.handleContentBlockStart(ev.Data) + if err != nil { + return ConverseStreamEvent{}, false, err + } + if !emit { + continue + } + return event, true, nil case "content_block_delta": return s.handleContentBlockDelta(ev.Data) case "content_block_stop": @@ -403,6 +410,9 @@ func (s *anthropicConverseStreamSource) handleMessageStart(data string) (Convers }, true, nil } +// handleContentBlockStart emits contentBlockStart only for tool_use blocks, +// matching real Bedrock: a text block's start carries no union member, which +// violates Smithy's exactly-one-of rule, so text blocks emit nothing here. func (s *anthropicConverseStreamSource) handleContentBlockStart(data string) (ConverseStreamEvent, bool, error) { var payload struct { Index int64 `json:"index"` @@ -415,12 +425,14 @@ func (s *anthropicConverseStreamSource) handleContentBlockStart(data string) (Co if err := json.Unmarshal([]byte(data), &payload); err != nil { return ConverseStreamEvent{}, false, newStreamFault(fmt.Errorf("anthropic stream: decode content_block_start: %w", err)) } - start := &bedrockruntime.ContentBlockStart{} - if payload.ContentBlock.Type == "tool_use" { - start.ToolUse = &bedrockruntime.ToolUseBlockStart{ + if payload.ContentBlock.Type != "tool_use" { + return ConverseStreamEvent{}, false, nil + } + start := &bedrockruntime.ContentBlockStart{ + ToolUse: &bedrockruntime.ToolUseBlockStart{ ToolUseId: aws.String(payload.ContentBlock.ID), Name: aws.String(payload.ContentBlock.Name), - } + }, } return ConverseStreamEvent{ Kind: converseStreamEventContentBlockStart, diff --git a/spinifex/gateway/bedrock/anthropic_test.go b/spinifex/gateway/bedrock/anthropic_test.go index 427c22ad1..9f24b61bd 100644 --- a/spinifex/gateway/bedrock/anthropic_test.go +++ b/spinifex/gateway/bedrock/anthropic_test.go @@ -197,7 +197,7 @@ func TestAnthropicProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { assert.True(t, capturedBody.Stream) events := drainConverseStream(t, src) - require.Len(t, events, 10) + require.Len(t, events, 9) kinds := make([]converseStreamEventKind, len(events)) for i, ev := range events { @@ -205,7 +205,6 @@ func TestAnthropicProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { } assert.Equal(t, []converseStreamEventKind{ converseStreamEventMessageStart, - converseStreamEventContentBlockStart, converseStreamEventContentBlockDelta, converseStreamEventContentBlockDelta, converseStreamEventContentBlockStop, @@ -217,22 +216,22 @@ func TestAnthropicProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { }, kinds) assert.Equal(t, bedrockruntime.ConversationRoleAssistant, *events[0].MessageStart.Role) - assert.Equal(t, "Hello", *events[2].ContentBlockDelta.Delta.Text) - assert.Equal(t, " there", *events[3].ContentBlockDelta.Delta.Text) + assert.Equal(t, "Hello", *events[1].ContentBlockDelta.Delta.Text) + assert.Equal(t, " there", *events[2].ContentBlockDelta.Delta.Text) - toolStart := events[5].ContentBlockStart + toolStart := events[4].ContentBlockStart require.NotNil(t, toolStart.Start.ToolUse) assert.Equal(t, "toolu_1", *toolStart.Start.ToolUse.ToolUseId) assert.Equal(t, "get_weather", *toolStart.Start.ToolUse.Name) - toolDelta := events[6].ContentBlockDelta + toolDelta := events[5].ContentBlockDelta require.NotNil(t, toolDelta.Delta.ToolUse) assert.Equal(t, `{"city":`, *toolDelta.Delta.ToolUse.Input) - assert.Equal(t, bedrockruntime.StopReasonToolUse, *events[8].MessageStop.StopReason) - assert.Equal(t, int64(12), *events[9].Metadata.Usage.InputTokens) - assert.Equal(t, int64(5), *events[9].Metadata.Usage.OutputTokens) - assert.Equal(t, int64(17), *events[9].Metadata.Usage.TotalTokens) + assert.Equal(t, bedrockruntime.StopReasonToolUse, *events[7].MessageStop.StopReason) + assert.Equal(t, int64(12), *events[8].Metadata.Usage.InputTokens) + assert.Equal(t, int64(5), *events[8].Metadata.Usage.OutputTokens) + assert.Equal(t, int64(17), *events[8].Metadata.Usage.TotalTokens) } func TestAnthropicConverseStreamSource_MessageStopEmitsMetadata(t *testing.T) { diff --git a/spinifex/gateway/bedrock/contract_test.go b/spinifex/gateway/bedrock/contract_test.go index 06b466de2..ca0e5423c 100644 --- a/spinifex/gateway/bedrock/contract_test.go +++ b/spinifex/gateway/bedrock/contract_test.go @@ -97,7 +97,7 @@ func TestContract_ConverseStream_SelfHost_RealSDKConsumerDecodesFrames(t *testin require.NoError(t, resp.GetStream().Err()) assert.Equal(t, []string{ - "messageStart", "contentBlockStart", "contentBlockDelta", "contentBlockDelta", + "messageStart", "contentBlockDelta", "contentBlockDelta", "contentBlockStop", "messageStop", "metadata", }, kinds) assert.Equal(t, []string{"Hello", " world"}, deltas) @@ -174,7 +174,7 @@ func TestContract_ConverseStream_Anthropic_RealSDKConsumerDecodesFrames(t *testi require.NoError(t, resp.GetStream().Err()) assert.Equal(t, []string{ - "messageStart", "contentBlockStart", "contentBlockDelta", "contentBlockDelta", "contentBlockStop", + "messageStart", "contentBlockDelta", "contentBlockDelta", "contentBlockStop", "contentBlockStart", "contentBlockDelta", "contentBlockStop", "messageStop", "metadata", }, kinds) assert.True(t, sawToolUse) diff --git a/spinifex/gateway/bedrock/converse_stream_test.go b/spinifex/gateway/bedrock/converse_stream_test.go index f1e419434..2d6f2b4d4 100644 --- a/spinifex/gateway/bedrock/converse_stream_test.go +++ b/spinifex/gateway/bedrock/converse_stream_test.go @@ -193,14 +193,13 @@ func TestConverseStream_SelfHostHappyPath_WritesFramedTaxonomy(t *testing.T) { assert.Equal(t, eventStreamContentType, rec.Header().Get("Content-Type")) frames := decodeAllFrames(t, rec.Body.Bytes()) - require.Len(t, frames, 7) + require.Len(t, frames, 6) assert.Equal(t, "messageStart", frames[0].Type) - assert.Equal(t, "contentBlockStart", frames[1].Type) + assert.Equal(t, "contentBlockDelta", frames[1].Type) assert.Equal(t, "contentBlockDelta", frames[2].Type) - assert.Equal(t, "contentBlockDelta", frames[3].Type) - assert.Equal(t, "contentBlockStop", frames[4].Type) - assert.Equal(t, "messageStop", frames[5].Type) - assert.Equal(t, "metadata", frames[6].Type) + assert.Equal(t, "contentBlockStop", frames[3].Type) + assert.Equal(t, "messageStop", frames[4].Type) + assert.Equal(t, "metadata", frames[5].Type) } func TestPumpConverseStream_RecordsInvocationOnCleanEnd(t *testing.T) { diff --git a/spinifex/gateway/bedrock/guardrail_inference.go b/spinifex/gateway/bedrock/guardrail_inference.go index da082c635..695fb8adb 100644 --- a/spinifex/gateway/bedrock/guardrail_inference.go +++ b/spinifex/gateway/bedrock/guardrail_inference.go @@ -207,7 +207,6 @@ func writeBlockedConverseStream(ctx context.Context, w http.ResponseWriter, mode events := []ConverseStreamEvent{ {Kind: converseStreamEventMessageStart, MessageStart: &bedrockruntime.MessageStartEvent{Role: aws.String(bedrockruntime.ConversationRoleAssistant)}}, - {Kind: converseStreamEventContentBlockStart, ContentBlockStart: &bedrockruntime.ContentBlockStartEvent{ContentBlockIndex: aws.Int64(0), Start: &bedrockruntime.ContentBlockStart{}}}, {Kind: converseStreamEventContentBlockDelta, ContentBlockDelta: &bedrockruntime.ContentBlockDeltaEvent{ContentBlockIndex: aws.Int64(0), Delta: &bedrockruntime.ContentBlockDelta{Text: aws.String(message)}}}, {Kind: converseStreamEventContentBlockStop, ContentBlockStop: &bedrockruntime.ContentBlockStopEvent{ContentBlockIndex: aws.Int64(0)}}, {Kind: converseStreamEventMessageStop, MessageStop: &bedrockruntime.MessageStopEvent{StopReason: aws.String(bedrockruntime.StopReasonGuardrailIntervened)}}, diff --git a/spinifex/gateway/bedrock/guardrail_inference_test.go b/spinifex/gateway/bedrock/guardrail_inference_test.go index cefb32185..ce40476d2 100644 --- a/spinifex/gateway/bedrock/guardrail_inference_test.go +++ b/spinifex/gateway/bedrock/guardrail_inference_test.go @@ -238,16 +238,16 @@ func TestConverseStream_GuardrailInputBlock_SkipsBackend(t *testing.T) { assert.Equal(t, int32(0), hits.Load(), "the backend stream must never be started on an INPUT block") frames := decodeAllFrames(t, rec.Body.Bytes()) - require.Len(t, frames, 6) - assert.Equal(t, []string{"messageStart", "contentBlockStart", "contentBlockDelta", "contentBlockStop", "messageStop", "metadata"}, - []string{frames[0].Type, frames[1].Type, frames[2].Type, frames[3].Type, frames[4].Type, frames[5].Type}) + require.Len(t, frames, 5) + assert.Equal(t, []string{"messageStart", "contentBlockDelta", "contentBlockStop", "messageStop", "metadata"}, + []string{frames[0].Type, frames[1].Type, frames[2].Type, frames[3].Type, frames[4].Type}) var delta decodedDelta - require.NoError(t, json.Unmarshal(frames[2].Payload, &delta)) + require.NoError(t, json.Unmarshal(frames[1].Payload, &delta)) assert.Equal(t, "Your input violates our policy.", delta.Delta.Text) var stop decodedMessageStop - require.NoError(t, json.Unmarshal(frames[4].Payload, &stop)) + require.NoError(t, json.Unmarshal(frames[3].Payload, &stop)) assert.Equal(t, bedrockruntime.StopReasonGuardrailIntervened, stop.StopReason) } @@ -284,10 +284,10 @@ func TestConverseStream_NoGuardrailConfig_Regression(t *testing.T) { require.NoError(t, err) frames := decodeAllFrames(t, rec.Body.Bytes()) - require.Len(t, frames, 7) - assert.Equal(t, "contentBlockDelta", frames[2].Type) + require.Len(t, frames, 6) + assert.Equal(t, "contentBlockDelta", frames[1].Type) var delta decodedDelta - require.NoError(t, json.Unmarshal(frames[2].Payload, &delta)) + require.NoError(t, json.Unmarshal(frames[1].Payload, &delta)) assert.Equal(t, "Hello", delta.Delta.Text) } @@ -297,7 +297,6 @@ func TestConverseStream_NoGuardrailConfig_Regression(t *testing.T) { func guardrailStreamFixtureEvents(chunks ...string) []ConverseStreamEvent { events := []ConverseStreamEvent{ {Kind: converseStreamEventMessageStart, MessageStart: &bedrockruntime.MessageStartEvent{Role: aws.String(bedrockruntime.ConversationRoleAssistant)}}, - {Kind: converseStreamEventContentBlockStart, ContentBlockStart: &bedrockruntime.ContentBlockStartEvent{ContentBlockIndex: aws.Int64(0), Start: &bedrockruntime.ContentBlockStart{}}}, } for _, c := range chunks { events = append(events, ConverseStreamEvent{ @@ -336,16 +335,15 @@ func TestGuardrailStreamSource_OutputBlock_BuffersAndReplacesText(t *testing.T) } assert.Equal(t, []converseStreamEventKind{ converseStreamEventMessageStart, - converseStreamEventContentBlockStart, converseStreamEventContentBlockDelta, converseStreamEventContentBlockStop, converseStreamEventMessageStop, converseStreamEventMetadata, }, kinds) - deltaEvent := events[2] + deltaEvent := events[1] assert.Equal(t, "The model response violates our policy.", aws.StringValue(deltaEvent.ContentBlockDelta.Delta.Text)) - assert.Equal(t, bedrockruntime.StopReasonGuardrailIntervened, aws.StringValue(events[4].MessageStop.StopReason)) + assert.Equal(t, bedrockruntime.StopReasonGuardrailIntervened, aws.StringValue(events[3].MessageStop.StopReason)) } func TestGuardrailStreamSource_OutputAnonymize_RedactsAccumulatedText(t *testing.T) { @@ -358,10 +356,10 @@ func TestGuardrailStreamSource_OutputAnonymize_RedactsAccumulatedText(t *testing src := newGuardrailStreamSource(inner, store, grCallerAccount, aws.StringValue(createOut.GuardrailId), guardrailDraftVersion, false, nil) events := drainConverseStream(t, src) - require.Len(t, events, 6) - assert.Equal(t, "contact {EMAIL} for support", aws.StringValue(events[2].ContentBlockDelta.Delta.Text)) + require.Len(t, events, 5) + assert.Equal(t, "contact {EMAIL} for support", aws.StringValue(events[1].ContentBlockDelta.Delta.Text)) // Redaction alone leaves the model's own stop reason intact. - assert.Equal(t, bedrockruntime.StopReasonEndTurn, aws.StringValue(events[4].MessageStop.StopReason)) + assert.Equal(t, bedrockruntime.StopReasonEndTurn, aws.StringValue(events[3].MessageStop.StopReason)) } func TestGuardrailStreamSource_TraceEnabledSurfacesAssessment(t *testing.T) { diff --git a/spinifex/gateway/bedrock/vllm.go b/spinifex/gateway/bedrock/vllm.go index 745cedf96..0a5708c5c 100644 --- a/spinifex/gateway/bedrock/vllm.go +++ b/spinifex/gateway/bedrock/vllm.go @@ -455,8 +455,9 @@ func (s *vllmConverseStreamSource) consume(chunk vllmStreamChunk) { } } -// ensureStarted queues messageStart+contentBlockStart exactly once, so the stop -// events can never appear without a preceding start on a zero-content stream. +// ensureStarted queues messageStart exactly once. Real Bedrock text streams +// carry no contentBlockStart, so contentBlockStop needs no preceding start +// event on a zero-content stream. func (s *vllmConverseStreamSource) ensureStarted() { if s.started { return @@ -467,10 +468,6 @@ func (s *vllmConverseStreamSource) ensureStarted() { Kind: converseStreamEventMessageStart, MessageStart: &bedrockruntime.MessageStartEvent{Role: aws.String(bedrockruntime.ConversationRoleAssistant)}, }, - ConverseStreamEvent{ - Kind: converseStreamEventContentBlockStart, - ContentBlockStart: &bedrockruntime.ContentBlockStartEvent{ContentBlockIndex: aws.Int64(0), Start: &bedrockruntime.ContentBlockStart{}}, - }, ) } diff --git a/spinifex/gateway/bedrock/vllm_test.go b/spinifex/gateway/bedrock/vllm_test.go index 5d7e17788..8250813eb 100644 --- a/spinifex/gateway/bedrock/vllm_test.go +++ b/spinifex/gateway/bedrock/vllm_test.go @@ -177,7 +177,7 @@ func TestVLLMProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { defer func() { _ = src.Close() }() events := drainConverseStream(t, src) - require.Len(t, events, 7) + require.Len(t, events, 6) kinds := make([]converseStreamEventKind, len(events)) for i, ev := range events { @@ -185,7 +185,6 @@ func TestVLLMProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { } assert.Equal(t, []converseStreamEventKind{ converseStreamEventMessageStart, - converseStreamEventContentBlockStart, converseStreamEventContentBlockDelta, converseStreamEventContentBlockDelta, converseStreamEventContentBlockStop, @@ -194,15 +193,18 @@ func TestVLLMProvider_ConverseStream_MapsSSEToTaxonomy(t *testing.T) { }, kinds) assert.Equal(t, bedrockruntime.ConversationRoleAssistant, *events[0].MessageStart.Role) - assert.Equal(t, int64(0), *events[1].ContentBlockStart.ContentBlockIndex) - assert.Equal(t, "Hello", *events[2].ContentBlockDelta.Delta.Text) - assert.Equal(t, " world", *events[3].ContentBlockDelta.Delta.Text) - assert.Equal(t, int64(0), *events[4].ContentBlockStop.ContentBlockIndex) - assert.Equal(t, bedrockruntime.StopReasonEndTurn, *events[5].MessageStop.StopReason) - assert.Equal(t, int64(8), *events[6].Metadata.Usage.InputTokens) - assert.Equal(t, int64(3), *events[6].Metadata.Usage.OutputTokens) - assert.Equal(t, int64(11), *events[6].Metadata.Usage.TotalTokens) - assert.GreaterOrEqual(t, *events[6].Metadata.Metrics.LatencyMs, int64(0)) + assert.Equal(t, "Hello", *events[1].ContentBlockDelta.Delta.Text) + assert.Equal(t, " world", *events[2].ContentBlockDelta.Delta.Text) + assert.Equal(t, int64(0), *events[3].ContentBlockStop.ContentBlockIndex) + assert.Equal(t, bedrockruntime.StopReasonEndTurn, *events[4].MessageStop.StopReason) + assert.Equal(t, int64(8), *events[5].Metadata.Usage.InputTokens) + assert.Equal(t, int64(3), *events[5].Metadata.Usage.OutputTokens) + assert.Equal(t, int64(11), *events[5].Metadata.Usage.TotalTokens) + assert.GreaterOrEqual(t, *events[5].Metadata.Metrics.LatencyMs, int64(0)) + + for _, ev := range events { + assert.NotEqual(t, converseStreamEventContentBlockStart, ev.Kind, "real Bedrock text streams carry no contentBlockStart") + } } func TestVLLMProvider_ConverseStream_UnresolvedEndpointReturnsModelNotReady(t *testing.T) { @@ -302,7 +304,6 @@ func TestVLLMProvider_ConverseStream_UsageWithoutFinishReasonKeepsOrder(t *testi } assert.Equal(t, []converseStreamEventKind{ converseStreamEventMessageStart, - converseStreamEventContentBlockStart, converseStreamEventContentBlockDelta, converseStreamEventContentBlockStop, converseStreamEventMessageStop, @@ -336,7 +337,6 @@ func TestVLLMProvider_ConverseStream_EmptyStreamIsWellFormed(t *testing.T) { } assert.Equal(t, []converseStreamEventKind{ converseStreamEventMessageStart, - converseStreamEventContentBlockStart, converseStreamEventContentBlockStop, converseStreamEventMessageStop, converseStreamEventMetadata, @@ -369,9 +369,9 @@ func TestVLLMConverseStreamSource_EstimatesOutputTokensBeforeUsageChunk(t *testi require.NoError(t, err) defer func() { _ = src.Close() }() - // Drain exactly the 4 events both chunks queue (messageStart, - // contentBlockStart, and two content deltas) without reaching EOF. - for range 4 { + // Drain exactly the 3 events both chunks queue (messageStart and two + // content deltas) without reaching EOF. + for range 3 { _, ok, nerr := src.Next(context.Background()) require.NoError(t, nerr) require.True(t, ok)