feat: add Bedrock :invoke_anthropic protocol spike (InvokeModel)

lib/ruby_llm/protocols/converse/chat.rb

@@ -395,15 +395,35 @@ def budget_reasoning_config(thinking)
         # When the turn had more than one reasoning block (common under adaptive/interleaved
         # thinking during tool use), thinking.blocks holds the exact original blocks and must
         # be replayed verbatim instead of reconstructed from the merged text/signature.
+        #
+        # Blocks captured by the InvokeAnthropic protocol are in the Anthropic Messages shape
+        # (type: thinking / redacted_thinking); those are translated to reasoningContent so a
+        # chat that moves from InvokeModel back to Converse can still replay its reasoning.
+        # reasoningContent blocks pass through untouched.
         def format_thinking_blocks(thinking)
           return nil unless thinking
 
-          return thinking.blocks if thinking.blocks
+          return thinking.blocks.map { |block| converse_thinking_block(block) } if thinking.blocks
 
           block = format_single_thinking_block(thinking)
           block ? [block] : nil
         end
 
+        def converse_thinking_block(block)
+          return block unless block.is_a?(Hash)
+
+          case (block['type'] || block[:type]).to_s
+          when 'thinking'
+            text = { text: block['thinking'] || block[:thinking] || '',
+                     signature: block['signature'] || block[:signature] }
+            { reasoningContent: { reasoningText: text.compact } }
+          when 'redacted_thinking'
+            { reasoningContent: { redactedContent: block['data'] || block[:data] } }
+          else
+            block
+          end
+        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

lib/ruby_llm/protocols/converse/event_stream.rb

@@ -0,0 +1,149 @@
+# frozen_string_literal: true
+
+require 'base64'
+require 'faraday'
+require 'json'
+
+module RubyLLM
+  module Protocols
+    class Converse
+      # AWS Event Stream plumbing shared by every bedrock-runtime streaming endpoint: ConverseStream
+      # (Converse::Streaming) and InvokeModelWithResponseStream (InvokeAnthropic::Streaming). Both
+      # frame each event as a binary eventstream message whose JSON payload may wrap the real
+      # event as base64 `bytes`, and both report mid-stream failures as `:exception-type`
+      # messages. What each endpoint puts INSIDE the payload differs, so event interpretation
+      # stays with the including protocol.
+      module EventStream
+        ErrorResponse = Struct.new(:body, :status)
+
+        # Bedrock stream exception type => HTTP status, so ErrorMiddleware raises the same error
+        # class a synchronous call with that failure would.
+        EXCEPTION_STATUSES = {
+          'throttlingException' => 429,
+          'validationException' => 400,
+          'accessDeniedException' => 401,
+          'unrecognizedClientException' => 401,
+          'serviceUnavailableException' => 503
+        }.freeze
+
+        private
+
+        # POSTs a SigV4-signed eventstream request and hands every raw body chunk of a 200
+        # response to the block. Non-200 chunks go through handle_failed_stream.
+        def post_event_stream(url, payload, additional_headers = {}, &on_chunk)
+          body = JSON.generate(payload)
+
+          @connection.post(url, payload) do |req|
+            req.headers.merge!(@provider.sign_headers('POST', url, body))
+            req.headers.merge!(additional_headers) unless additional_headers.empty?
+            req.headers['Accept'] = 'application/vnd.amazon.eventstream'
+
+            if Faraday::VERSION.start_with?('1')
+              req.options[:on_data] = proc { |chunk, _size| on_chunk.call(chunk) }
+            else
+              req.options.on_data = proc do |chunk, _bytes, env|
+                if env&.status == 200
+                  on_chunk.call(chunk)
+                else
+                  handle_failed_stream(chunk, env)
+                end
+              end
+            end
+          end
+        end
+
+        def event_stream_decoder
+          require 'aws-eventstream'
+          Aws::EventStream::Decoder.new
+        rescue LoadError
+          raise Error,
+                'The aws-eventstream gem is required for Bedrock streaming. ' \
+                'Please add it to your Gemfile: gem "aws-eventstream"'
+        end
+
+        def handle_failed_stream(chunk, env)
+          data = JSON.parse(chunk)
+          error_response = env.merge(body: data)
+          ErrorMiddleware.parse_error(provider: self, response: error_response)
+        rescue JSON::ParserError
+          RubyLLM.logger.debug { "Failed Bedrock stream error chunk: #{chunk}" }
+        end
+
+        def handle_non_eventstream_error_chunk(raw_chunk)
+          text = raw_chunk.to_s
+
+          if text.start_with?('event: error')
+            payload = text.lines.find { |line| line.start_with?('data:') }&.delete_prefix('data:')&.strip
+            raise_streaming_chunk_error(payload) if payload
+            return
+          end
+
+          return unless text.lstrip.start_with?('{') && text.include?('"error"')
+
+          raise_streaming_chunk_error(text)
+        end
+
+        def raise_streaming_chunk_error(payload)
+          parsed = JSON.parse(payload)
+          message = parsed.dig('error', 'message') || parsed['message'] || 'Bedrock streaming error'
+          response = ErrorResponse.new({ 'message' => message }, 500)
+          ErrorMiddleware.parse_error(provider: self, response: response)
+        rescue JSON::ParserError
+          nil
+        end
+
+        # Every complete eventstream message in raw_chunk (the decoder buffers partial ones
+        # across calls).
+        def decode_event_messages(decoder, raw_chunk)
+          messages = []
+          message, eof = decoder.decode_chunk(raw_chunk)
+
+          while message
+            messages << message
+            break if eof
+
+            message, eof = decoder.decode_chunk
+          end
+
+          messages
+        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 exception_type_header(message)
+          header = message.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)
+
+          if outer['bytes'].is_a?(String)
+            JSON.parse(Base64.decode64(outer['bytes']))
+          else
+            outer
+          end
+        rescue JSON::ParserError => e
+          RubyLLM.logger.debug { "Failed to decode Bedrock stream event payload: #{e.message}" }
+          nil
+        end
+
+        def raise_event_stream_exception(exception_type, message)
+          status = EXCEPTION_STATUSES.fetch(exception_type, 500)
+          response = ErrorResponse.new({ 'message' => message || exception_type }, status)
+          ErrorMiddleware.parse_error(provider: self, response: response)
+        end
+      end
+    end
+  end
+end

lib/ruby_llm/protocols/converse/streaming.rb

@@ -1,15 +1,15 @@
 # frozen_string_literal: true
 
-require 'base64'
-require 'faraday'
 require 'json'
 
 module RubyLLM
   module Protocols
     class Converse
       # Streaming implementation for Bedrock ConverseStream (AWS Event Stream).
       module Streaming
-        ErrorResponse = Struct.new(:body, :status)
+        include EventStream
+
+        ErrorResponse = EventStream::ErrorResponse
 
         private
 
@@ -21,50 +21,16 @@ def stream_response(payload, additional_headers = {}, &block)
           accumulator = StreamAccumulator.new
           decoder = event_stream_decoder
           thinking_state = {}
-          body = JSON.generate(payload)
-
-          response = @connection.post(stream_url, payload) do |req|
-            req.headers.merge!(@provider.sign_headers('POST', stream_url, body))
-            req.headers.merge!(additional_headers) unless additional_headers.empty?
-            req.headers['Accept'] = 'application/vnd.amazon.eventstream'
-
-            if Faraday::VERSION.start_with?('1')
-              req.options[:on_data] = proc do |chunk, _size|
-                parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block)
-              end
-            else
-              req.options.on_data = proc do |chunk, _bytes, env|
-                if env&.status == 200
-                  parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block)
-                else
-                  handle_failed_stream(chunk, env)
-                end
-              end
-            end
+
+          response = post_event_stream(stream_url, payload, additional_headers) do |chunk|
+            parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block)
           end
 
           message = accumulator.to_message(response)
           RubyLLM.logger.debug { "Stream completed: #{message.content}" }
           message
         end
 
-        def event_stream_decoder
-          require 'aws-eventstream'
-          Aws::EventStream::Decoder.new
-        rescue LoadError
-          raise Error,
-                'The aws-eventstream gem is required for Bedrock streaming. ' \
-                'Please add it to your Gemfile: gem "aws-eventstream"'
-        end
-
-        def handle_failed_stream(chunk, env)
-          data = JSON.parse(chunk)
-          error_response = env.merge(body: data)
-          ErrorMiddleware.parse_error(provider: self, response: error_response)
-        rescue JSON::ParserError
-          RubyLLM.logger.debug { "Failed Bedrock stream error chunk: #{chunk}" }
-        end
-
         def parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state)
           handle_non_eventstream_error_chunk(raw_chunk)
 
@@ -77,48 +43,15 @@ def parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state)
           end
         end
 
-        def handle_non_eventstream_error_chunk(raw_chunk)
-          text = raw_chunk.to_s
-
-          if text.start_with?('event: error')
-            payload = text.lines.find { |line| line.start_with?('data:') }&.delete_prefix('data:')&.strip
-            raise_streaming_chunk_error(payload) if payload
-            return
-          end
-
-          return unless text.lstrip.start_with?('{') && text.include?('"error"')
-
-          raise_streaming_chunk_error(text)
-        end
-
-        def raise_streaming_chunk_error(payload)
-          parsed = JSON.parse(payload)
-          message = parsed.dig('error', 'message') || parsed['message'] || 'Bedrock streaming error'
-          response = ErrorResponse.new({ 'message' => message }, 500)
-          ErrorMiddleware.parse_error(provider: self, response: response)
-        rescue JSON::ParserError
-          nil
-        end
-
         def decode_events(decoder, raw_chunk)
-          events = []
-          message, eof = decoder.decode_chunk(raw_chunk)
-
-          while message
+          decode_event_messages(decoder, raw_chunk).filter_map do |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}"
-              end
-            end
-            events << event if event
-            break if eof
-
-            message, eof = decoder.decode_chunk
-          end
+            next unless event
 
-          events
+            event = nest_event_under_type(event, message)
+            RubyLLM.logger.debug { "Bedrock stream event keys: #{event.keys}" } if RubyLLM.config.log_stream_debug
+            event
+          end
         end
 
         # A ConverseStream wire message carries its event type in the eventstream
@@ -137,28 +70,6 @@ def nest_event_under_type(event, message)
           { 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)
-
-          if outer['bytes'].is_a?(String)
-            JSON.parse(Base64.decode64(outer['bytes']))
-          else
-            outer
-          end
-        rescue JSON::ParserError => e
-          RubyLLM.logger.debug { "Failed to decode Bedrock stream event payload: #{e.message}" }
-          nil
-        end
-
         def build_chunk(event, thinking_state = {})
           raise_stream_error(event) if stream_error_event?(event)
 
@@ -317,18 +228,7 @@ def raise_stream_error(event)
           end
 
           key = event.keys.find { |candidate| candidate.end_with?('Exception') }
-          payload = event[key]
-          message = payload['message'] || key
-          status = case key
-                   when 'throttlingException' then 429
-                   when 'validationException' then 400
-                   when 'accessDeniedException', 'unrecognizedClientException' then 401
-                   when 'serviceUnavailableException' then 503
-                   else 500
-                   end
-
-          response = ErrorResponse.new({ 'message' => message }, status)
-          ErrorMiddleware.parse_error(provider: self, response: response)
+          raise_event_stream_exception(key, event[key]['message'])
         end
 
         def extract_content_delta(event)