diff --git a/packages/bedrock/README.md b/packages/bedrock/README.md index f256813..3d16ca3 100644 --- a/packages/bedrock/README.md +++ b/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. diff --git a/packages/bedrock/src/internal.ts b/packages/bedrock/src/internal.ts index 6860343..534c2ee 100644 --- a/packages/bedrock/src/internal.ts +++ b/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(value: T): JsonValue { if (!isRecord(value)) return undefined; @@ -559,7 +551,11 @@ export function wrapAsyncIterable( 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( } } - // 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; diff --git a/packages/bedrock/tests/bedrock.test.ts b/packages/bedrock/tests/bedrock.test.ts index 925006e..0c12a86 100644 --- a/packages/bedrock/tests/bedrock.test.ts +++ b/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", diff --git a/sdks/python-bedrock/README.md b/sdks/python-bedrock/README.md index 8df0a28..778bc94 100644 --- a/sdks/python-bedrock/README.md +++ b/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 diff --git a/sdks/python-bedrock/src/telemetry_dev_bedrock/_instrument.py b/sdks/python-bedrock/src/telemetry_dev_bedrock/_instrument.py index e8f0b78..be5c07a 100644 --- a/sdks/python-bedrock/src/telemetry_dev_bedrock/_instrument.py +++ b/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)) diff --git a/sdks/python-bedrock/src/telemetry_dev_bedrock/_streams.py b/sdks/python-bedrock/src/telemetry_dev_bedrock/_streams.py index ae39c2d..e113d7e 100644 --- a/sdks/python-bedrock/src/telemetry_dev_bedrock/_streams.py +++ b/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: diff --git a/sdks/python-bedrock/tests/test_bedrock.py b/sdks/python-bedrock/tests/test_bedrock.py index 48e2410..764f3dc 100644 --- a/sdks/python-bedrock/tests/test_bedrock.py +++ b/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] = []