fix(sdk-bedrock): report a failure in stream chunk timing and the first-chunk update

packages/bedrock/README.md

@@ -162,7 +162,7 @@ InvokeModel captures provider-native JSON bodies verbatim. A body whose content
 - Streams are pull-through. The wrapper does not pre-buffer chunks and therefore preserves backpressure.
 - Streams end spans exactly once. Exhaustion records full output and finish reason; early `break`, `return()`, `close()`, abort, or stream errors record partial output and set `telemetry.dev.capture.truncated`.
 - Stream reconstruction (output, Agent trace events, and Agent output-file metadata) is kept up to 16 MiB, and all streams in flight share the `batch.maxStreamBytes` bound of the core (64 MiB by default); events past the bound still pass through to the caller. The configured `maxAttributeLength` does not shrink this bound; the core applies it to the exported attribute after the mask runs.
-- A stream cut at the bound, stopped early, or failed sets `telemetry.dev.capture.truncated`. An event that fails to map is reported to `onError`; only the first mapping failure of a stream is reported. A failure before the event's output is kept also sets the attribute, and no later text or content is kept; so does an `InvokeModel` stream chunk that is not UTF-8 or not JSON (a blank chunk holds no output and is skipped). Agent text is decoded across chunks, so a character split between two chunks is kept whole; Agent text that is not UTF-8, or a stream that ends inside a character, loses that text like a failed event and is reported; an Agent `returnControl` that arrives later still replaces the text output whole, also with a mask configured. A failure in the event's usage, metrics, Agent citations, or Agent trace summaries loses only that metadata and does not set the attribute. A failure while Agent trace content (`captureAgentTrace`) or output-file metadata is captured loses that metadata and sets the attribute, but later text is still kept, also with a mask configured. A stream whose output fails to map when it ends records no output and sets the attribute; the stream still ends normally for the caller. With a mask configured, its incomplete reconstructed output is omitted because the mask cannot inspect the complete value. Agent trace events and output-file metadata share the bound, but dropping them only sets the attribute; Agent text is omitted only when a text chunk was dropped or failed to map, or the stream stopped early or failed. A normally exhausted Agent stream that returned control keeps its complete `returnControl` output even if earlier text or metadata passed the bound; an early stop or failure still omits the output when a mask is configured.
+- A stream cut at the bound, stopped early, or failed sets `telemetry.dev.capture.truncated`. An event that fails to map, or a failure while the time to first chunk or an output-chunk timestamp is recorded, is reported to `onError`; only the first such failure of a stream is reported. A failure before the event's output is kept also sets the attribute, and no later text or content is kept; so does an `InvokeModel` stream chunk that is not UTF-8 or not JSON (a blank chunk holds no output and is skipped). Agent text is decoded across chunks, so a character split between two chunks is kept whole; Agent text that is not UTF-8, or a stream that ends inside a character, loses that text like a failed event and is reported; an Agent `returnControl` that arrives later still replaces the text output whole, also with a mask configured. A failure in the event's usage, metrics, Agent citations, or Agent trace summaries loses only that metadata and does not set the attribute, and so does a timing failure. A failure while Agent trace content (`captureAgentTrace`) or output-file metadata is captured loses that metadata and sets the attribute, but later text is still kept, also with a mask configured. A stream whose output fails to map when it ends records no output and sets the attribute; the stream still ends normally for the caller. With a mask configured, its incomplete reconstructed output is omitted because the mask cannot inspect the complete value. Agent trace events and output-file metadata share the bound, but dropping them only sets the attribute; Agent text is omitted only when a text chunk was dropped or failed to map, or the stream stopped early or failed. A normally exhausted Agent stream that returned control keeps its complete `returnControl` output even if earlier text or metadata passed the bound; an early stop or failure still omits the output when a mask is configured.
 - `AbortSignal` passed to `send(command, { abortSignal })` is forwarded untouched. Pre-response aborts end the span with the thrown abort error. Mid-stream aborts end the span with partial output and the stream error.
 - `Rerank` input and results are bounded to 16 MiB each. A value over the bound sets `telemetry.dev.capture.truncated`. Without a mask, the bounded part is kept. With a mask configured, that value is omitted because the mask cannot inspect the complete value. Input is omitted before the span starts, so a custom `sampler` sees the bounded input when no mask is configured.
 - Callback-style `send(command, callback)` and `send(command, options, callback)` calls are instrumented; callbacks receive the original data/error while the span is ended before callback invocation.

packages/bedrock/src/internal.ts

@@ -277,14 +277,6 @@ export function endSpan(span: SpanHandle, fields?: SpanFields): void {
   }
 }
 
-export function updateSpan(span: SpanHandle, fields: SpanFields): void {
-  try {
-    span.update(fields);
-  } catch {
-    // Instrumentation must never throw into caller code.
-  }
-}
-
 export function metadataOf<T>(value: T): JsonValue {
   if (!isRecord(value)) return undefined;
 
@@ -559,7 +551,11 @@ export function wrapAsyncIterable<T>(
 
         if (!first) {
           first = true;
-          updateSpan(span, { timeToFirstChunkMs: receivedAt - t0 });
+          try {
+            span.update({ timeToFirstChunkMs: receivedAt - t0 });
+          } catch (cause) {
+            reportError(cause);
+          }
         }
 
         state.outputRecorded = false;
@@ -578,14 +574,14 @@ export function wrapAsyncIterable<T>(
           }
         }
 
-        // Chunk timing must not reach the caller. When feed() failed, the fallback reads the
-        // event again and can fail the same way, which feed() reported.
+        // Chunk timing must not reach the caller. A failure loses only the timing; when feed()
+        // already failed, the fallback can fail the same way and the reporter drops the repeat.
         try {
           if (trackOutputChunks && (stateHasOutput ?? bedrockEventHasOutput(event))) {
             span.recordOutputChunk?.(receivedAt);
           }
-        } catch {
-          // Instrumentation must never throw into caller code.
+        } catch (cause) {
+          reportError(cause);
         }
 
         yield event;

packages/bedrock/tests/bedrock.test.ts

@@ -1073,6 +1073,27 @@ const rawChunk = (raw: string | number[]) => ({
   chunk: { bytes: typeof raw === "string" ? new TextEncoder().encode(raw) : new Uint8Array(raw) },
 });
 
+// Makes a method of each started span throw on the calls `fails` selects.
+function failSpanMethod(method: "update" | "recordOutputChunk", fails: (call: number) => boolean) {
+  const startSpan = telemetrySdk.startSpan;
+
+  vi.spyOn(telemetrySdk, "startSpan").mockImplementation((...args) => {
+    const span = startSpan(...args);
+    const original = span[method].bind(span) as (...rest: unknown[]) => unknown;
+    let calls = 0;
+
+    vi.spyOn(span, method).mockImplementation(((...rest: unknown[]) => {
+      calls += 1;
+
+      if (fails(calls)) return mapperBroke();
+
+      return original(...rest);
+    }) as never);
+
+    return span;
+  });
+}
+
 test.each([
   {
     name: "a Converse delta",
@@ -1199,6 +1220,36 @@ test.each([
     cut: false,
     reported: ["MapperError"],
   },
+  {
+    // A failed chunk timing loses only that timing.
+    name: "the Converse chunk timing",
+    operation: "converse",
+    fail: () => failSpanMethod("recordOutputChunk", () => true),
+    events: [
+      { messageStart: { role: "assistant" } },
+      converseDelta("Hello"),
+      converseDelta(" world"),
+      { messageStop: { stopReason: "end_turn" } },
+    ],
+    kept: "Hello world",
+    cut: false,
+    reported: ["MapperError"],
+  },
+  {
+    // The stream's only update is the time to first chunk.
+    name: "the Converse first-chunk update",
+    operation: "converse",
+    fail: () => failSpanMethod("update", () => true),
+    events: [
+      { messageStart: { role: "assistant" } },
+      converseDelta("Hello"),
+      converseDelta(" world"),
+      { messageStop: { stopReason: "end_turn" } },
+    ],
+    kept: "Hello world",
+    cut: false,
+    reported: ["MapperError"],
+  },
   {
     // A blank chunk carries no output, so it is neither lost nor flagged.
     name: "a blank InvokeModel chunk",

sdks/python-bedrock/README.md

@@ -146,7 +146,7 @@ InvokeModel captures provider-native JSON bodies verbatim. A body whose content
 - Requests are never modified.
 - Streams end spans exactly once. Exhaustion records full output and finish reason; early `break`, `close()`, `GeneratorExit`, or stream errors record partial output and set `telemetry.dev.capture.truncated`.
 - Event-stream reconstruction and streamed `InvokeModel` bodies are kept up to 16 MiB, and all streams in flight share the `max_stream_bytes` bound of the core (64 MiB by default). These bounds need a `telemetry-dev` core with `max_stream_bytes`; with an older core, each stream keeps up to 64 KiB and 1,024 items. `InvokeModel` bodies and `Rerank` inputs and results are retained up to 16 MiB each; a body over that bound records no output and sets `telemetry.dev.capture.truncated`. The configured `max_attribute_length` does not shrink these retention bounds; the core applies it to the exported attribute after the mask runs.
-- A stream cut at its retention bound, closed early, or failed sets `telemetry.dev.capture.truncated`. An event that fails to map is reported to `on_error`; only the first such failure of a stream is reported. A failure before the event's output is kept also sets the attribute, and no later text or content is kept; so does an `InvokeModel` stream chunk that is not UTF-8 or not JSON (a blank chunk holds no output and is skipped). Agent text is decoded across chunks, so a character split between two chunks is kept whole; Agent text that is not UTF-8, or a stream that ends inside a character, loses that text like a failed event and is reported; an Agent `returnControl` that arrives later still replaces the text output whole, also with a mask configured. A failure in the event's usage, metrics, or Agent trace summaries loses only that metadata and does not set the attribute. A failure while Agent trace content is captured (`capture_agent_trace`) loses that trace and sets the attribute, but later text is still kept, also with a mask configured. A stream whose output fails to map when it ends records no output and sets the attribute. An `InvokeModel` body that is closed before it is fully read, whose read fails, or that fails to map (for example, a body that is not UTF-8 or not JSON) records no output and sets the attribute too; a mapping failure is reported to `on_error`. With a mask configured, incomplete reconstructed output is omitted because the mask cannot inspect the complete value. Agent trace events share the bound, but dropping them only sets the attribute; Agent text is omitted only when a text chunk was dropped or failed to map, or the stream was closed early or failed. A normally exhausted Agent stream that returned control keeps its complete `returnControl` output even if earlier text or trace metadata passed the bound; an early close or failure still omits the output when a mask is configured.
+- A stream cut at its retention bound, closed early, or failed sets `telemetry.dev.capture.truncated`. An event that fails to map, or a failure while the time to first chunk or an output-chunk timestamp is recorded, is reported to `on_error`; only the first such failure of a stream is reported. A failure before the event's output is kept also sets the attribute, and no later text or content is kept; so does an `InvokeModel` stream chunk that is not UTF-8 or not JSON (a blank chunk holds no output and is skipped). Agent text is decoded across chunks, so a character split between two chunks is kept whole; Agent text that is not UTF-8, or a stream that ends inside a character, loses that text like a failed event and is reported; an Agent `returnControl` that arrives later still replaces the text output whole, also with a mask configured. A failure in the event's usage, metrics, or Agent trace summaries loses only that metadata and does not set the attribute, and so does a timing failure. A failure while Agent trace content is captured (`capture_agent_trace`) loses that trace and sets the attribute, but later text is still kept, also with a mask configured. A stream whose output fails to map when it ends records no output and sets the attribute. An `InvokeModel` body that is closed before it is fully read, whose read fails, or that fails to map (for example, a body that is not UTF-8 or not JSON) records no output and sets the attribute too; a mapping failure is reported to `on_error`. With a mask configured, incomplete reconstructed output is omitted because the mask cannot inspect the complete value. Agent trace events share the bound, but dropping them only sets the attribute; Agent text is omitted only when a text chunk was dropped or failed to map, or the stream was closed early or failed. A normally exhausted Agent stream that returned control keeps its complete `returnControl` output even if earlier text or trace metadata passed the bound; an early close or failure still omits the output when a mask is configured.
 - boto3 has no async client. `aiobotocore` is a separate package and is out of scope.
 
 ## Coverage gaps

sdks/python-bedrock/src/telemetry_dev_bedrock/_instrument.py

@@ -181,8 +181,8 @@ def finish(fields: dict[str, Any]) -> None:
             if fields.keys() == {"time_to_first_chunk_ms"}:
                 try:
                     handle.update(**fields)
-                except Exception:
-                    pass
+                except Exception as exc:
+                    report_error(exc)
                 return
             end(**merge_fields(base_fields, fields))
 

sdks/python-bedrock/src/telemetry_dev_bedrock/_streams.py

@@ -832,8 +832,9 @@ def __iter__(self) -> Iterator[Any]:
                             self._state.text_rejected = True
                     state_has_output = None
                 if self._track_output_chunks:
-                    # Chunk timing must not reach the caller. When feed() failed, the fallback
-                    # reads the event again and can fail the same way, which feed() reported.
+                    # Chunk timing must not reach the caller. A failure loses only the timing;
+                    # when feed() already failed, the fallback can fail the same way and the
+                    # reporter drops the repeat.
                     try:
                         if (
                             state_has_output
@@ -843,8 +844,8 @@ def __iter__(self) -> Iterator[Any]:
                             record_output_chunk = getattr(self._handle, "record_output_chunk", None)
                             if callable(record_output_chunk):
                                 record_output_chunk(received_at * 1000)
-                    except Exception:
-                        pass
+                    except Exception as exc:
+                        self._report_error(exc)
                 yield event
             completed = True
         except GeneratorExit:

sdks/python-bedrock/tests/test_bedrock.py

@@ -2324,6 +2324,33 @@ def _invoke_chunk(text: str) -> dict[str, Any]:
             True,
             json.JSONDecodeError,
         ),
+        # A chunk timing or first-chunk update that fails loses only that timing.
+        (
+            "converse_stream",
+            (telemetry_dev.SpanHandle, "record_output_chunk", None),
+            [
+                {"messageStart": {"role": "assistant"}},
+                _converse_delta("Hello"),
+                _converse_delta(" world"),
+                {"messageStop": {"stopReason": "end_turn"}},
+            ],
+            "Hello world",
+            False,
+            ValueError,
+        ),
+        (
+            "converse_stream",
+            (telemetry_dev.SpanHandle, "update", 1),
+            [
+                {"messageStart": {"role": "assistant"}},
+                _converse_delta("Hello"),
+                _converse_delta(" world"),
+                {"messageStop": {"stopReason": "end_turn"}},
+            ],
+            "Hello world",
+            False,
+            ValueError,
+        ),
         (
             "invoke_model_with_response_stream",
             None,
@@ -2341,6 +2368,8 @@ def _invoke_chunk(text: str) -> dict[str, Any]:
         "invoke-undecodable",
         "invoke-unparsable",
         "invoke-not-json",
+        "converse-chunk-timing",
+        "converse-first-chunk",
         "invoke-blank",
     ],
 )
@@ -2359,12 +2388,12 @@ def test_a_runtime_stream_mapping_failure_is_reported_and_flags_only_lost_output
         original = getattr(owner, name)
         calls = 0
 
-        def fail(*args: Any) -> Any:
+        def fail(*args: Any, **kwargs: Any) -> Any:
             nonlocal calls
             calls += 1
             if failing_call is None or calls == failing_call:
                 raise ValueError("mapper broke")
-            return original(*args)
+            return original(*args, **kwargs)
 
         monkeypatch.setattr(owner, name, fail)
     errors: list[BaseException] = []