fix(boto3): Skip malformed Bedrock stream chunks and gate stream outputs once

Malformed InvokeModelWithResponseStream chunks are skipped quietly instead
of reporting an SDK internal error, like the non-streamed response path.

The buffered stream content is already gated on recording outputs, so
serialize it directly rather than checking the gate a second time. That
makes the buffering gate the only gate, so tests cover it.

Co-Authored-By: Sergiy Dybskiy <3760543+sergical@users.noreply.github.com>

Author: sentry-junior[bot]

sentry_sdk/integrations/boto3/_services/bedrock_runtime.py

@@ -810,8 +810,12 @@ def get_attributes(self) -> "Attributes":
                         tool_call["arguments"] = content
                 parts.append(tool_call)
 
+        # `record_outputs` already gated the buffered content blocks.
         if parts:
-            attributes.update(_get_output_attributes([_to_output_message(parts)]))
+            with capture_internal_exceptions():
+                attributes[SPANDATA.GEN_AI_OUTPUT_MESSAGES] = json.dumps(
+                    [_to_output_message(parts)]
+                )
 
         return attributes
 
@@ -882,7 +886,11 @@ def _record_invoke_model_stream_event(
     if not payload:
         return
 
-    chunk = json.loads(payload)
+    try:
+        chunk = json.loads(payload)
+    except ValueError:
+        return
+
     if not isinstance(chunk, Mapping):
         return
 

tests/integrations/boto3/test_bedrock_runtime.py

@@ -2150,6 +2150,43 @@ def call():
     )
 
 
+@pytest.mark.parametrize("trace_mode", TRACE_MODES)
+def test_invoke_model_with_response_stream_skips_malformed_chunks(
+    init_bedrock, capture_items, trace_mode
+):
+    client = init_bedrock(trace_mode, send_default_pii=True)
+    malformed_chunk = ("chunk", {"bytes": base64.b64encode(b"not json").decode()})
+    events = [malformed_chunk] + _invoke_model_stream_chunks(
+        {
+            "outputs": [{"text": "Hi!", "stop_reason": "stop"}],
+            "amazon-bedrock-invocationMetrics": {
+                "inputTokenCount": 7,
+                "outputTokenCount": 3,
+            },
+        }
+    )
+    received_chunks = []
+
+    def call():
+        response = client.invoke_model_with_response_stream(
+            modelId="mistral.mistral-7b-instruct-v0:2",
+            body=json.dumps({"prompt": "Hello!"}),
+        )
+        received_chunks.extend(event["chunk"]["bytes"] for event in response["body"])
+
+    with _event_stream_response(client, events):
+        spans = _capture_spans(capture_items, trace_mode, call)
+
+    assert received_chunks[0] == b"not json"
+    assert len(received_chunks) == 2
+
+    attributes = _get_attributes(_get_bedrock_span(spans))
+    assert attributes[SPANDATA.GEN_AI_OUTPUT_MESSAGES] == [
+        {"role": "assistant", "parts": [{"type": "text", "content": "Hi!"}]}
+    ]
+    assert attributes[SPANDATA.GEN_AI_USAGE_OUTPUT_TOKENS] == 3
+
+
 @pytest.mark.parametrize("trace_mode", TRACE_MODES)
 def test_invoke_model_with_response_stream_error_is_unchanged(
     init_bedrock, capture_items, trace_mode