Fix Bedrock Converse streaming losing reasoning blocks on replay

Converse::Chat (non-streaming) already collects every reasoning
(thinking/redacted_thinking) content block from a turn into
Thinking#blocks and replays them verbatim, since Anthropic requires the
full original block set to be replayed byte-for-byte on the next
request. Converse::Streaming#build_chunk never did the same — it only
ever built Thinking from merged text/signature per chunk, so
StreamAccumulator's blocks-collecting logic had nothing to collect and
the final streamed message lost the raw blocks. Replaying that lossy
reconstruction on the next turn gets rejected by Anthropic with
"Invalid data in redacted_thinking block" (surfaced downstream as
appfolio/agents_app DUX-7838, in an app that streams every Bedrock
Converse completion).

Reassemble each reasoning block from its contentBlockStart/Delta/Stop
events (they always stream fully sequentially, so a single running
accumulator suffices) and emit it once complete, in the same
{'reasoningContent' => {...}} shape the non-streaming parser produces.

Author: mattwebbio

lib/ruby_llm/protocols/converse/streaming.rb

@@ -143,7 +143,8 @@ def build_chunk(event)
             content: extract_content_delta(event),
             thinking: Thinking.build(
               text: extract_thinking_delta(event),
-              signature: extract_thinking_signature(event)
+              signature: extract_thinking_signature(event),
+              blocks: extract_finished_reasoning_block(event)
             ),
             tool_calls: extract_tool_calls(event),
             input_tokens: extract_input_tokens(metadata_usage, usage),
@@ -252,6 +253,55 @@ def extract_signature_from_start(event)
           nil
         end
 
+        # Anthropic (and Bedrock Converse, which proxies to it) requires every reasoning
+        # block from a turn to be replayed byte-for-byte on the next request (see
+        # Thinking's doc comment) — merging text+signature loses any additional blocks.
+        # Content blocks always stream fully sequentially (start -> delta* -> stop for one
+        # index before the next begins), so a single running accumulator is enough; no
+        # per-index bookkeeping is needed. Returns the finished block, in the same
+        # {'reasoningContent' => {...}} shape Converse::Chat's non-streaming parser
+        # produces, once its contentBlockStop arrives — nil for every other event,
+        # including every event belonging to a non-reasoning content block (nothing was
+        # accumulated for it).
+        def extract_finished_reasoning_block(event)
+          accumulate_reasoning_content(event)
+          return nil unless event.key?('contentBlockStop')
+
+          finalize_reasoning_block
+        end
+
+        def accumulate_reasoning_content(event)
+          start = event.dig('contentBlockStart', 'start', 'reasoningContent')
+          accumulate_reasoning_fragment(start) if start
+
+          delta = event.dig('contentBlockDelta', 'delta', 'reasoningContent')
+          accumulate_reasoning_fragment(delta) if delta
+        end
+
+        def reasoning_block
+          @reasoning_block ||= { text: +'', signature: nil, redacted: nil }
+        end
+
+        def accumulate_reasoning_fragment(fragment)
+          reasoning_text = fragment['reasoningText'] || {}
+          reasoning_block[:text] << reasoning_text['text'] if reasoning_text['text'].is_a?(String)
+          reasoning_block[:signature] ||= reasoning_text['signature']
+          reasoning_block[:redacted] ||= fragment['redactedContent']
+        end
+
+        def finalize_reasoning_block
+          block = @reasoning_block
+          @reasoning_block = nil
+          return nil unless block
+          return [{ 'reasoningContent' => { 'redactedContent' => block[:redacted] } }] if block[:redacted]
+
+          text = block[:text].empty? ? nil : block[:text]
+          return nil if text.nil? && block[:signature].nil?
+
+          [{ 'reasoningContent' => { 'reasoningText' => { 'text' => text,
+                                                          'signature' => block[:signature] }.compact } }]
+        end
+
         def extract_tool_calls(event)
           return extract_tool_call_start(event) if tool_call_start_event?(event)
           return extract_tool_call_delta(event) if tool_call_delta_event?(event)

spec/ruby_llm/protocols/converse/streaming_spec.rb

@@ -98,4 +98,85 @@
     expect(message.thinking.text).to eq('thinking text')
     expect(message.thinking.signature).to eq('thinking-signature')
   end
+
+  describe 'reasoning block replay (DUX-7838)' do
+    # Anthropic requires every thinking/redacted_thinking block from a turn to be replayed
+    # unmodified on the next request. Converse::Chat (non-streaming) already collects these
+    # into thinking.blocks; build_chunk must do the same by reassembling each block from its
+    # contentBlockStart/Delta/Stop events, or the accumulator has nothing to collect and the
+    # next request gets rejected with "Invalid data in redacted_thinking block".
+
+    def stop_event(index: 0)
+      { 'contentBlockStop' => { 'contentBlockIndex' => index } }
+    end
+
+    def start_event(index: 0, reasoning_content: {})
+      { 'contentBlockStart' => { 'contentBlockIndex' => index,
+                                 'start' => { 'reasoningContent' => reasoning_content } } }
+    end
+
+    def reasoning_delta_event(index: 0, text: nil, signature: nil)
+      reasoning_text = { 'text' => text, 'signature' => signature }.compact
+      delta = { 'reasoningContent' => { 'reasoningText' => reasoning_text } }
+      { 'contentBlockDelta' => { 'contentBlockIndex' => index, 'delta' => delta } }
+    end
+
+    def accumulate(events)
+      accumulator = RubyLLM::StreamAccumulator.new
+      events.each { |event| accumulator.add(streaming.send(:build_chunk, event)) }
+      accumulator.to_message(nil)
+    end
+
+    it 'captures a redacted block that arrives fully at contentBlockStart' do
+      message = accumulate([
+                             start_event(reasoning_content: { 'redactedContent' => 'opaque-blob-1' }),
+                             stop_event
+                           ])
+
+      expect(message.thinking.blocks).to eq([{ 'reasoningContent' => { 'redactedContent' => 'opaque-blob-1' } }])
+    end
+
+    it 'assembles a normal reasoning block streamed as incremental text+signature deltas' do
+      message = accumulate([
+                             reasoning_delta_event(text: 'step one. '),
+                             reasoning_delta_event(text: 'step two.'),
+                             reasoning_delta_event(signature: 'sig-1'),
+                             stop_event
+                           ])
+
+      expect(message.thinking.blocks).to eq(
+        [{ 'reasoningContent' => { 'reasoningText' => { 'text' => 'step one. step two.', 'signature' => 'sig-1' } } }]
+      )
+      expect(message.thinking.text).to eq('step one. step two.')
+    end
+
+    it 'preserves block order across a redacted block followed by a normal reasoning block' do
+      message = accumulate([
+                             start_event(index: 0, reasoning_content: { 'redactedContent' => 'opaque-blob-1' }),
+                             stop_event(index: 0),
+                             start_event(index: 1),
+                             reasoning_delta_event(index: 1, text: 'step two', signature: 'sig-2'),
+                             stop_event(index: 1)
+                           ])
+
+      expect(message.thinking.blocks).to eq(
+        [
+          { 'reasoningContent' => { 'redactedContent' => 'opaque-blob-1' } },
+          { 'reasoningContent' => { 'reasoningText' => { 'text' => 'step two', 'signature' => 'sig-2' } } }
+        ]
+      )
+    end
+
+    it 'leaves a stream with no reasoning content unaffected' do
+      text_start = { 'contentBlockStart' => { 'contentBlockIndex' => 0, 'start' => {} } }
+      text_delta = { 'contentBlockDelta' => { 'contentBlockIndex' => 0, 'delta' => { 'text' => 'Hello' } } }
+
+      accumulator = RubyLLM::StreamAccumulator.new
+      [text_start, text_delta, stop_event].each { |event| accumulator.add(streaming.send(:build_chunk, event)) }
+      message = accumulator.to_message(nil)
+
+      expect(message.thinking).to be_nil
+      expect(message.content).to eq('Hello')
+    end
+  end
 end