From c22b381a9d5f1d23eec209ddc8985aa13bb7e2b8 Mon Sep 17 00:00:00 2001 From: Matt Webb Date: Mon, 13 Jul 2026 20:09:45 -0700 Subject: [PATCH 1/2] Type Converse stream events from the :event-type header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConverseStream wire messages carry the event type in the eventstream :event-type header; the JSON payload is the bare member struct ({"contentBlockIndex":0,"delta":{...}}, {"stopReason":"tool_use"}) — verified against a live Bedrock ConverseStream response. The streaming extractors instead dig by a nested event name ({'contentBlockDelta' => {...}}), a shape that only this repo's specs ever produced. Extractors with flat fallbacks (text/signature deltas, finish_reason, tool calls) worked by accident; the ones without — thinking_state index tracking, contentBlockStop finalization, and the messageStop safety net from #14 — never matched a real event, so thinking.blocks was never populated from a stream and every text-less thinking turn fell back to format_single_thinking_block's lossy reconstruction on replay. decode_events now re-nests each payload under its :event-type (or :exception-type) header, so the nested-shape paths engage on real traffic; already-nested payloads pass through unchanged for compatibility. Also fix that lossy fallback: a persisted signature is always a real reasoningText signature (redacted blobs are only captured into blocks), so replay it as an empty-text reasoningText block. Wrapping it in redactedContent made Anthropic reject the next request with "Invalid `data` in `redacted_thinking` block" — the failure that broke every Supernova MCP tool-call turn. Co-Authored-By: Claude Fable 5 --- lib/ruby_llm/protocols/converse/chat.rb | 32 +++-- lib/ruby_llm/protocols/converse/streaming.rb | 26 ++++ spec/ruby_llm/protocols/converse/chat_spec.rb | 13 ++ .../protocols/converse/streaming_spec.rb | 129 ++++++++++++++++++ 4 files changed, 185 insertions(+), 15 deletions(-) diff --git a/lib/ruby_llm/protocols/converse/chat.rb b/lib/ruby_llm/protocols/converse/chat.rb index 092f868b8..64a5869d1 100644 --- a/lib/ruby_llm/protocols/converse/chat.rb +++ b/lib/ruby_llm/protocols/converse/chat.rb @@ -376,23 +376,25 @@ def format_thinking_blocks(thinking) block ? [block] : nil end + # Lossy fallback for messages persisted without raw thinking blocks. `signature` + # here is always a real reasoningText signature (streaming captures it from + # reasoningContent signature deltas; redacted content is only ever captured into + # `blocks`), so a signature with no text means the model emitted a thinking block + # whose text was empty — replay it as such. Wrapping the signature in + # redactedContent instead makes Anthropic reject the request with + # "Invalid `data` in `redacted_thinking` block": redactedContent must carry the + # opaque encrypted blob, never a signature. def format_single_thinking_block(thinking) - if thinking.text - { - reasoningContent: { - reasoningText: { - text: thinking.text, - signature: thinking.signature - }.compact - } - } - elsif thinking.signature - { - reasoningContent: { - redactedContent: thinking.signature - } + return nil unless thinking.text || thinking.signature + + { + reasoningContent: { + reasoningText: { + text: thinking.text || '', + signature: thinking.signature + }.compact } - end + } end def parse_text_content(content_blocks) diff --git a/lib/ruby_llm/protocols/converse/streaming.rb b/lib/ruby_llm/protocols/converse/streaming.rb index 01a0a42f1..431b1d7e4 100644 --- a/lib/ruby_llm/protocols/converse/streaming.rb +++ b/lib/ruby_llm/protocols/converse/streaming.rb @@ -106,6 +106,7 @@ def decode_events(decoder, raw_chunk) while message event = decode_event_payload(message.payload.read) + event = nest_event_under_type(event, message) if event if event && RubyLLM.config.log_stream_debug RubyLLM.logger.debug do "Bedrock stream event keys: #{event.keys}" @@ -120,6 +121,31 @@ def decode_events(decoder, raw_chunk) events end + # A ConverseStream wire message carries its event type in the eventstream + # :event-type header (":exception-type" for error events); the JSON payload is the + # bare member struct — e.g. {"contentBlockIndex":0,"delta":{...}} for a + # contentBlockDelta, {"stopReason":"tool_use"} for a messageStop. The extractors in + # this module dig by event name ({'contentBlockDelta' => {...}}), so re-nest the + # payload under its header type here. A payload already nested under its type (the + # shape older specs fabricate) or a message with no type header passes through + # unchanged. + def nest_event_under_type(event, message) + type = event_type_header(message) + return event unless type + return event if event.key?(type) + + { type => event } + end + + def event_type_header(message) + headers = message.headers + header = headers[':event-type'] || headers[':exception-type'] + value = header.respond_to?(:value) ? header.value : header + value.is_a?(String) && !value.empty? ? value : nil + rescue StandardError + nil + end + def decode_event_payload(payload) outer = JSON.parse(payload) diff --git a/spec/ruby_llm/protocols/converse/chat_spec.rb b/spec/ruby_llm/protocols/converse/chat_spec.rb index 518278ac6..389996c89 100644 --- a/spec/ruby_llm/protocols/converse/chat_spec.rb +++ b/spec/ruby_llm/protocols/converse/chat_spec.rb @@ -243,5 +243,18 @@ def render_payload(messages = [], **overrides) reasoning_blocks = payload[:messages].first[:content].select { |block| block[:reasoningContent] } expect(reasoning_blocks).to eq([{ reasoningContent: { reasoningText: { text: 'thought', signature: 'sig' } } }]) end + + it 'reconstructs a signature-only thinking turn as an empty reasoningText, never redactedContent' do + # A persisted signature is always a real reasoningText signature (redacted blobs are + # only ever captured into thinking.blocks). Wrapping it in redactedContent makes + # Anthropic reject the next request with "Invalid `data` in `redacted_thinking` block". + thinking = RubyLLM::Thinking.build(text: nil, signature: 'sig-only') + message = RubyLLM::Message.new(role: :assistant, content: 'Done', thinking: thinking) + + payload = render_payload([message], schema: nil) + + reasoning_blocks = payload[:messages].first[:content].select { |block| block[:reasoningContent] } + expect(reasoning_blocks).to eq([{ reasoningContent: { reasoningText: { text: '', signature: 'sig-only' } } }]) + end end end diff --git a/spec/ruby_llm/protocols/converse/streaming_spec.rb b/spec/ruby_llm/protocols/converse/streaming_spec.rb index 043a72883..1fdda8d71 100644 --- a/spec/ruby_llm/protocols/converse/streaming_spec.rb +++ b/spec/ruby_llm/protocols/converse/streaming_spec.rb @@ -231,6 +231,20 @@ def accumulate(events) expect(message.tool_calls['call_1'].name).to eq('search') end + it 'finalizes a signature-only thinking block as an empty reasoningText' do + events = [ + { 'contentBlockDelta' => { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'signature' => 'sig-only' } } } }, + { 'contentBlockStop' => { 'contentBlockIndex' => 0 } } + ] + + message = accumulate(events) + + expect(message.thinking.blocks).to eq( + [{ 'reasoningContent' => { 'reasoningText' => { 'text' => '', 'signature' => 'sig-only' } } }] + ) + end + it 'round-trips a streamed multi-block thinking turn through format_thinking_blocks unmodified' do events = [ { 'contentBlockStart' => { 'contentBlockIndex' => 0, @@ -253,4 +267,119 @@ def accumulate(events) expect(thinking_blocks).to eq(message.thinking.blocks) end end + + # ConverseStream's real wire format carries the event type in the eventstream + # :event-type HEADER; the JSON payload is the bare member struct — e.g. + # {"contentBlockIndex":0,"delta":{...}}, {"stopReason":"tool_use"} — never nested + # under the event name like the hand-built hashes above. These specs frame events + # exactly as Bedrock does (captured from a live ConverseStream response) and drive + # them through the same decode path stream_response uses. + describe 'real wire framing (:event-type header + flat payload)' do + require 'aws-eventstream' + + def wire_chunk(*typed_payloads) + encoder = Aws::EventStream::Encoder.new + typed_payloads.map do |(type, payload)| + message = Aws::EventStream::Message.new( + headers: { + ':event-type' => Aws::EventStream::HeaderValue.new(value: type, type: 'string'), + ':content-type' => Aws::EventStream::HeaderValue.new(value: 'application/json', type: 'string'), + ':message-type' => Aws::EventStream::HeaderValue.new(value: 'event', type: 'string') + }, + payload: StringIO.new(JSON.generate(payload)) + ) + encoder.encode(message) + end.join + end + + def stream_to_message(*typed_payloads) + accumulator = RubyLLM::StreamAccumulator.new + decoder = Aws::EventStream::Decoder.new + chunks = [] + streaming.send(:parse_stream_chunk, decoder, wire_chunk(*typed_payloads), accumulator, {}) do |chunk| + chunks << chunk + end + accumulator.to_message(nil) + end + + it 'captures thinking blocks from a flat-framed tool-call turn' do + message = stream_to_message( + ['messageStart', { 'role' => 'assistant' }], + ['contentBlockDelta', { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'text' => 'need the weather' } } }], + ['contentBlockDelta', { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'signature' => 'sig-1' } } }], + ['contentBlockStop', { 'contentBlockIndex' => 0 }], + ['contentBlockStart', { 'contentBlockIndex' => 1, + 'start' => { 'toolUse' => { 'toolUseId' => 'call_1', 'name' => 'weather' } } }], + ['contentBlockDelta', { 'contentBlockIndex' => 1, 'delta' => { 'toolUse' => { 'input' => '{}' } } }], + ['contentBlockStop', { 'contentBlockIndex' => 1 }], + ['messageStop', { 'stopReason' => 'tool_use' }], + ['metadata', { 'usage' => { 'inputTokens' => 10, 'outputTokens' => 5 }, 'metrics' => {} }] + ) + + expect(message.thinking.blocks).to eq( + [{ 'reasoningContent' => { 'reasoningText' => { 'text' => 'need the weather', + 'signature' => 'sig-1' } } }] + ) + expect(message.thinking.text).to eq('need the weather') + expect(message.thinking.signature).to eq('sig-1') + expect(message.tool_calls['call_1'].name).to eq('weather') + expect(message.finish_reason).to eq('tool_use') + expect(message.output_tokens).to eq(5) + end + + it 'finalizes a flat-framed signature-only thinking block left open at messageStop' do + message = stream_to_message( + ['messageStart', { 'role' => 'assistant' }], + ['contentBlockDelta', { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'signature' => 'sig-only' } } }], + ['messageStop', { 'stopReason' => 'tool_use' }] + ) + + expect(message.thinking.blocks).to eq( + [{ 'reasoningContent' => { 'reasoningText' => { 'text' => '', 'signature' => 'sig-only' } } }] + ) + end + + it 'accumulates flat-framed redacted content deltas into a redacted block' do + message = stream_to_message( + ['contentBlockDelta', { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'redactedContent' => 'blob-part-1' } } }], + ['contentBlockDelta', { 'contentBlockIndex' => 0, + 'delta' => { 'reasoningContent' => { 'redactedContent' => 'blob-part-2' } } }], + ['contentBlockStop', { 'contentBlockIndex' => 0 }], + ['messageStop', { 'stopReason' => 'end_turn' }] + ) + + expect(message.thinking.blocks).to eq( + [{ 'reasoningContent' => { 'redactedContent' => 'blob-part-1blob-part-2' } }] + ) + end + + it 'nests an exception payload under its :exception-type header' do + message = Aws::EventStream::Message.new( + headers: { + ':exception-type' => Aws::EventStream::HeaderValue.new(value: 'throttlingException', type: 'string'), + ':message-type' => Aws::EventStream::HeaderValue.new(value: 'exception', type: 'string') + }, + payload: StringIO.new(JSON.generate({ 'message' => 'Too many requests' })) + ) + + event = streaming.send(:nest_event_under_type, { 'message' => 'Too many requests' }, message) + + expect(event).to eq({ 'throttlingException' => { 'message' => 'Too many requests' } }) + expect(streaming.send(:stream_error_event?, event)).to be(true) + end + + it 'passes an already-nested payload through unchanged' do + message = Aws::EventStream::Message.new( + headers: { ':event-type' => Aws::EventStream::HeaderValue.new(value: 'messageStop', type: 'string') }, + payload: StringIO.new('{}') + ) + nested = { 'messageStop' => { 'stopReason' => 'end_turn' } } + + expect(streaming.send(:nest_event_under_type, nested, message)).to equal(nested) + end + end end From 3a6c20dd6c9ec98e4ef2f4ec13f67f9f667850c0 Mon Sep 17 00:00:00 2001 From: Matt Webb Date: Mon, 13 Jul 2026 20:35:36 -0700 Subject: [PATCH 2/2] Restore RubyLLM config after logger specs instead of nilling it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The logger specs left @config nil for the rest of the worker process. The next lazy RubyLLM.config builds a bare Configuration whose model_registry_source is nil — ActsAs installs its ActiveRecordSource only once, when the module is included at load time — so acts_as_model's registry integration specs fail with a fallback to the JSON registry whenever test-queue happens to schedule ruby_llm_spec.rb earlier in the same worker. Reproducible with: rspec spec/ruby_llm_spec.rb spec/ruby_llm/active_record/acts_as_model_spec.rb --order defined Save and restore the original config/logger around each example instead. Co-Authored-By: Claude Fable 5 --- spec/ruby_llm_spec.rb | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/spec/ruby_llm_spec.rb b/spec/ruby_llm_spec.rb index 7615d3995..98d1d5f70 100644 --- a/spec/ruby_llm_spec.rb +++ b/spec/ruby_llm_spec.rb @@ -8,14 +8,21 @@ let(:log_file) { double } let(:log_level) { double } - before do - described_class.instance_variable_set(:@config, nil) - described_class.instance_variable_set(:@logger, nil) - end - - after do + # Swap in a fresh config/logger for these examples, then restore the originals. + # Leaving @config nil here leaks across the whole worker process: the next lazy + # RubyLLM.config builds a bare Configuration whose model_registry_source is nil, + # which breaks acts_as_model's registry integration (its ActiveRecordSource is + # installed once, when ActsAs is included at load time) in any spec file that + # happens to run later in the same process. + around do |example| + original_config = described_class.instance_variable_get(:@config) + original_logger = described_class.instance_variable_get(:@logger) described_class.instance_variable_set(:@config, nil) described_class.instance_variable_set(:@logger, nil) + example.run + ensure + described_class.instance_variable_set(:@config, original_config) + described_class.instance_variable_set(:@logger, original_logger) end context 'with configuration options' do