Type Converse stream events from the :event-type header

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)

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)
 

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

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

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