Finalize open thinking blocks on messageStop for Bedrock Converse streaming

Gemfile

@@ -29,7 +29,9 @@ group :development do # rubocop:disable Metrics/BlockLength
   gem 'rubocop-performance'
   gem 'rubocop-rake', '>= 0.6'
   gem 'rubocop-rspec'
-  gem 'simplecov', '>= 0.21'
+  # Pinned below 1.0 — that release removed SimpleCov.running, which bin/rspec-queue's
+  # custom test-queue runner depends on for per-worker coverage bookkeeping.
+  gem 'simplecov', '~> 0.22'
   gem 'simplecov-cobertura'
   gem 'test-queue'
 

gemfiles/rails_7.1.gemfile

@@ -26,7 +26,7 @@ group :development do
   gem "rubocop-performance"
   gem "rubocop-rake", ">= 0.6"
   gem "rubocop-rspec"
-  gem "simplecov", ">= 0.21"
+  gem "simplecov", "~> 0.22"
   gem "simplecov-cobertura"
   gem "test-queue"
   gem "activerecord-jdbcsqlite3-adapter", platform: "jruby"

gemfiles/rails_7.2.gemfile

@@ -26,7 +26,7 @@ group :development do
   gem "rubocop-performance"
   gem "rubocop-rake", ">= 0.6"
   gem "rubocop-rspec"
-  gem "simplecov", ">= 0.21"
+  gem "simplecov", "~> 0.22"
   gem "simplecov-cobertura"
   gem "test-queue"
   gem "activerecord-jdbcsqlite3-adapter", platform: "jruby"

gemfiles/rails_8.0.gemfile

@@ -26,7 +26,7 @@ group :development do
   gem "rubocop-performance"
   gem "rubocop-rake", ">= 0.6"
   gem "rubocop-rspec"
-  gem "simplecov", ">= 0.21"
+  gem "simplecov", "~> 0.22"
   gem "simplecov-cobertura"
   gem "test-queue"
   gem "activerecord-jdbcsqlite3-adapter", platform: "jruby"

gemfiles/rails_8.1.gemfile

@@ -26,7 +26,7 @@ group :development do
   gem "rubocop-performance"
   gem "rubocop-rake", ">= 0.6"
   gem "rubocop-rspec"
-  gem "simplecov", ">= 0.21"
+  gem "simplecov", "~> 0.22"
   gem "simplecov-cobertura"
   gem "test-queue"
   gem "activerecord-jdbcsqlite3-adapter", platform: "jruby"

lib/ruby_llm/protocols/converse/streaming.rb

@@ -194,19 +194,45 @@ def thinking_delta_source(event)
           [delta_index, normalized_delta(event)['reasoningContent']]
         end
 
-        # A thinking block only finalizes here, on its content_block_stop. If the turn is
-        # truncated before this event arrives for a given index, that block is intentionally
-        # dropped rather than replayed signature-less — Anthropic rejects replay of a thinking
-        # block without a valid signature, so a half-formed block is worse than none.
+        # A thinking block finalizes on its own content_block_stop. As a safety net, messageStop
+        # (end of the event stream, on ANY stopReason including truncation) also finalizes any
+        # block still open at that point IF it is structurally complete (has a signature or
+        # redacted-content marker) — Bedrock is not guaranteed to emit a content_block_stop for
+        # every index before the stream ends (observed for at least one redacted-thinking block
+        # immediately superseded by the next index's content_block_start with no stop in
+        # between). Without this net, such a block is dropped from thinking.blocks entirely,
+        # silently falling back to format_single_thinking_block's lossy text/signature
+        # reconstruction — which Anthropic rejects on replay ("Invalid `data` in
+        # `redacted_thinking` block") since a reconstructed single block does not match the raw
+        # shape the API originally sent.
+        #
+        # A block still open with neither a signature nor redacted content (text-only, or fully
+        # empty) is genuinely half-formed — e.g. truncated mid-thought before the model ever
+        # emitted its closing signature — and stays dropped: Anthropic rejects replay of a
+        # thinking block without a valid signature, so a half-formed block is worse than none.
         def extract_finalized_thinking_blocks(event, thinking_state)
           index = event.dig('contentBlockStop', 'contentBlockIndex')
-          return nil unless index
+          if index
+            state = thinking_state.delete(index)
+            return nil unless state
 
-          state = thinking_state.delete(index)
-          return nil unless state
+            block = finalize_thinking_block(state)
+            return block ? [block] : nil
+          end
+
+          return nil unless event.key?('messageStop') && thinking_state.any?
+
+          finalize_remaining_thinking_blocks(thinking_state)
+        end
+
+        def finalize_remaining_thinking_blocks(thinking_state)
+          blocks = thinking_state.keys.sort.filter_map do |index|
+            state = thinking_state.delete(index)
+            next unless state[:redacted] || state[:signature]
 
-          block = finalize_thinking_block(state)
-          block ? [block] : nil
+            finalize_thinking_block(state)
+          end
+          blocks.empty? ? nil : blocks
         end
 
         def finalize_thinking_block(state)

spec/ruby_llm/protocols/converse/streaming_spec.rb

@@ -209,6 +209,28 @@ def accumulate(events)
       expect(message.finish_reason).to eq('max_tokens')
     end
 
+    it 'finalizes a redacted thinking block via messageStop when Bedrock never sends its own contentBlockStop' do
+      events = [
+        { 'contentBlockStart' => { 'contentBlockIndex' => 0,
+                                   'start' => { 'reasoningContent' => { 'redactedContent' => 'opaque-blob-1' } } } },
+        # No contentBlockStop for index 0 — the next block starts immediately, as Bedrock has
+        # been observed to do for a redacted-thinking block with no visible content.
+        { 'contentBlockStart' => { 'contentBlockIndex' => 1,
+                                   'start' => { 'toolUse' => { 'toolUseId' => 'call_1', 'name' => 'search' } } } },
+        { 'contentBlockDelta' => { 'contentBlockIndex' => 1,
+                                   'delta' => { 'toolUse' => { 'input' => '{}' } } } },
+        { 'contentBlockStop' => { 'contentBlockIndex' => 1 } },
+        { 'messageStop' => { 'stopReason' => 'tool_use' } }
+      ]
+
+      message = accumulate(events)
+
+      expect(message.thinking.blocks).to eq(
+        [{ 'reasoningContent' => { 'redactedContent' => 'opaque-blob-1' } }]
+      )
+      expect(message.tool_calls['call_1'].name).to eq('search')
+    end
+
     it 'round-trips a streamed multi-block thinking turn through format_thinking_blocks unmodified' do
       events = [
         { 'contentBlockStart' => { 'contentBlockIndex' => 0,