gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 17:11:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: stop emitting an empty contentBlockStart union for text in ConverseStream (#844)

Browse Source 
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).
This commit is contained in:
Josh Siviour
2026-08-20 15:50:38 +10:00
committed by GitHub
parent d4d8fefae6
commit 8eb3e4f879
8 changed files with 65 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/gateway/bedrock/anthropic.go
+17 -5
View File
@@ -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,
spinifex/gateway/bedrock/anthropic_test.go
+9 -10
View File
@@ -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) {
spinifex/gateway/bedrock/contract_test.go
+2 -2
View File
@@ -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)
spinifex/gateway/bedrock/converse_stream_test.go
+5 -6
View File
@@ -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) {
spinifex/gateway/bedrock/guardrail_inference.go
-1
View File
@@ -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)}},
spinifex/gateway/bedrock/guardrail_inference_test.go
+13 -15
View File
@@ -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) {
spinifex/gateway/bedrock/vllm.go
+3 -6
View File
@@ -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{}},
},
)
}
spinifex/gateway/bedrock/vllm_test.go
+16 -16
View File
@@ -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)