From 7b1f36684c6705272cb4a118af0665904eeb2278 Mon Sep 17 00:00:00 2001 From: RKS Date: Sun, 13 Sep 2026 08:56:20 -0400 Subject: [PATCH 1/3] Python: record gen_ai.client.operation.duration for failed chat and embedding calls --- .../core/agent_framework/observability.py | 50 ++++- .../core/tests/core/test_observability.py | 182 ++++++++++++++++++ 2 files changed, 231 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 5dc9fa8828c..9d3f27eb2fd 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -2064,6 +2064,12 @@ def _record_duration() -> None: ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time, + ) _close_span() raise @@ -2080,6 +2086,12 @@ async def _finalize_stream() -> None: exception=result_stream._stream_error, # type: ignore timestamp=time_ns(), ) + _capture_operation_error( + attributes=attributes, + exception=result_stream._stream_error, # type: ignore + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration_state.get("duration", perf_counter() - start_time), + ) return response: ChatResponse[Any] = await result_stream.get_final_response() duration = duration_state.get("duration") @@ -2171,6 +2183,12 @@ async def _get_response() -> ChatResponse: ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time_stamp, + ) raise duration = perf_counter() - start_time_stamp response_attributes = _get_response_attributes(attributes, response) @@ -2256,6 +2274,12 @@ async def get_embeddings( ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time_stamp, + ) raise duration = perf_counter() - start_time_stamp response_attributes: dict[str, Any] = {**attributes} @@ -3517,6 +3541,30 @@ def _get_response_attributes( ) +def _filter_metric_attributes(attributes: dict[str, Any]) -> dict[str, Any]: + """Filter attributes to only the GenAI metric attribute set.""" + return {key: attributes[key] for key in GEN_AI_METRIC_ATTRIBUTES if key in attributes} + + +def _capture_operation_error( + attributes: dict[str, Any], + exception: BaseException, + operation_duration_histogram: metrics.Histogram | None = None, + duration: float | None = None, +) -> None: + """Record the operation duration metric for a call that failed. + + The GenAI semantic conventions define ``gen_ai.client.operation.duration`` for failed + operations as well as successful ones, with ``error.type`` set to the class of the error. + Recording only successes leaves error latency out of the metric and gives no error rate. + """ + if operation_duration_histogram is None or duration is None: + return + attrs = _filter_metric_attributes(attributes) + attrs[OtelAttr.ERROR_TYPE] = type(exception).__name__ + operation_duration_histogram.record(duration, attributes=attrs) + + def _capture_response( span: trace.Span, attributes: dict[str, Any], @@ -3526,7 +3574,7 @@ def _capture_response( ) -> None: """Set the response for a given span.""" span.set_attributes(attributes) - attrs: dict[str, Any] = {k: v for k, v in attributes.items() if k in GEN_AI_METRIC_ATTRIBUTES} + attrs = _filter_metric_attributes(attributes) if token_usage_histogram and (input_tokens := attributes.get(OtelAttr.INPUT_TOKENS)) is not None: token_usage_histogram.record(input_tokens, attributes={**attrs, OtelAttr.T_TYPE: OtelAttr.T_TYPE_INPUT}) if token_usage_histogram and (output_tokens := attributes.get(OtelAttr.OUTPUT_TOKENS)) is not None: diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index f496e44d07f..3f858d09de0 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -5723,6 +5723,188 @@ def test_capture_response_with_error_type(span_exporter: InMemorySpanExporter): assert spans[0].attributes.get(OtelAttr.ERROR_TYPE) == "ValueError" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] +def test_filter_metric_attributes() -> None: + """_filter_metric_attributes preserves only the allowed GenAI metric attributes.""" + from agent_framework.observability import OtelAttr, _filter_metric_attributes + + attributes = { + OtelAttr.OPERATION: "chat", + OtelAttr.PROVIDER_NAME: "test_provider", + OtelAttr.SYSTEM: "test_system", + OtelAttr.REQUEST_MODEL: "gpt-4o", + OtelAttr.RESPONSE_MODEL: "gpt-4o-mini", + OtelAttr.ADDRESS: "127.0.0.1", + OtelAttr.PORT: 8000, + "custom_attribute": "ignored", + OtelAttr.CONVERSATION_ID: "conv-123", + } + filtered = _filter_metric_attributes(attributes) + assert filtered == { + OtelAttr.OPERATION: "chat", + OtelAttr.PROVIDER_NAME: "test_provider", + OtelAttr.SYSTEM: "test_system", + OtelAttr.REQUEST_MODEL: "gpt-4o", + OtelAttr.RESPONSE_MODEL: "gpt-4o-mini", + OtelAttr.ADDRESS: "127.0.0.1", + OtelAttr.PORT: 8000, + } + + +def test_capture_operation_error_keeps_only_metric_attributes() -> None: + """The error record carries the metric attribute set plus error.type, nothing else.""" + from agent_framework.observability import _capture_operation_error + + histogram = Mock() + _capture_operation_error( + attributes={ + OtelAttr.OPERATION: OtelAttr.CHAT_COMPLETION_OPERATION, + OtelAttr.REQUEST_MODEL: "test-model", + OtelAttr.CONVERSATION_ID: "conv-1", + }, + exception=TimeoutError("slow"), + operation_duration_histogram=histogram, + duration=0.25, + ) + + histogram.record.assert_called_once_with( + 0.25, + attributes={ + OtelAttr.OPERATION: OtelAttr.CHAT_COMPLETION_OPERATION, + OtelAttr.REQUEST_MODEL: "test-model", + OtelAttr.ERROR_TYPE: "TimeoutError", + }, + ) + + +def test_capture_operation_error_without_histogram_or_duration() -> None: + """No histogram or no duration means nothing is recorded.""" + from agent_framework.observability import _capture_operation_error + + histogram = Mock() + _capture_operation_error(attributes={}, exception=ValueError("x"), operation_duration_histogram=histogram) + _capture_operation_error(attributes={}, exception=ValueError("x"), duration=1.0) + histogram.record.assert_not_called() + + +async def test_chat_client_records_duration_on_error( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A failed non-streaming call records gen_ai.client.operation.duration with error.type.""" + + class FailingChatClient(mock_chat_client): # type: ignore[misc, valid-type] + async def _get_non_streaming_response(self, **kwargs: Any) -> ChatResponse: + raise ValueError("boom") + + client = FailingChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="boom"): + await client.get_response(messages=[Message(role="user", contents=["hi"])], options={"model": "Test"}) + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_chat_client_records_duration_on_streaming_error( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream that fails mid-iteration also records the duration metric with error.type.""" + + class FailingStreamChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=[Content.from_text("Hello")], role="assistant") + raise RuntimeError("stream broke") + + return ResponseStream(_stream(), finalizer=lambda updates: ChatResponse.from_updates(updates)) + + client = FailingStreamChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="stream broke"): + async for _ in client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ): + pass + + histogram.record.assert_called_once() + attributes = histogram.record.call_args[1]["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + + +async def test_chat_client_records_duration_when_stream_setup_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream that fails before it is built records the duration metric with error.type.""" + + class FailingSetupChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + raise RuntimeError("setup broke") + + client = FailingSetupChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="setup broke"): + async for _ in client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ): + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_embedding_client_records_duration_on_error(span_exporter: InMemorySpanExporter) -> None: + """A failed embedding call records gen_ai.client.operation.duration with error.type.""" + from agent_framework import BaseEmbeddingClient, GeneratedEmbeddings + from agent_framework.observability import EmbeddingTelemetryLayer + + class RawFailingEmbeddingClient(BaseEmbeddingClient[str, list[float], Any]): # type: ignore[type-arg] + async def get_embeddings( + self, values: Sequence[str], *, options: Any = None + ) -> GeneratedEmbeddings[list[float], Any]: + raise ValueError("embed boom") + + class FailingEmbeddingClient(EmbeddingTelemetryLayer, RawFailingEmbeddingClient): # type: ignore[misc] + OTEL_PROVIDER_NAME = "test" + + def service_url(self) -> str: + return "https://test.example.com" + + client = FailingEmbeddingClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="embed boom"): + await client.get_embeddings(["hi"], options={"model": "test-embed"}) + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "test-embed" + assert attributes[OtelAttr.OPERATION] == OtelAttr.EMBEDDING_OPERATION + + def test_backfill_request_model_when_unknown(span_exporter: InMemorySpanExporter): """_backfill_request_model updates the span name and REQUEST_MODEL attribute when unknown.""" from agent_framework.observability import OtelAttr, get_tracer From 1ba927b9dc7755681489eb3d758420586ace9364 Mon Sep 17 00:00:00 2001 From: RKS Date: Sun, 13 Sep 2026 09:02:00 -0400 Subject: [PATCH 2/3] Python: record duration metric when stream finalizer raises --- .../core/agent_framework/observability.py | 6 ++++ .../core/tests/core/test_observability.py | 36 +++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 9d3f27eb2fd..3a7b7d9d427 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -2129,6 +2129,12 @@ async def _finalize_stream() -> None: ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration_state.get("duration", perf_counter() - start_time), + ) finally: _close_span() diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index 3f858d09de0..bad4d02370f 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -5871,6 +5871,42 @@ def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseU assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION +async def test_chat_client_records_duration_when_stream_finalizer_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream whose finalizer raises records the duration metric with error.type.""" + + class FailingFinalizerChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=[Content.from_text("Hello")], role="assistant") + + def _broken_finalizer(updates: Any) -> ChatResponse: + raise ValueError("finalizer failed") + + return ResponseStream(_stream(), finalizer=_broken_finalizer) + + client = FailingFinalizerChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="finalizer failed"): + stream = client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ) + async for _ in stream: + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + async def test_embedding_client_records_duration_on_error(span_exporter: InMemorySpanExporter) -> None: """A failed embedding call records gen_ai.client.operation.duration with error.type.""" from agent_framework import BaseEmbeddingClient, GeneratedEmbeddings From 465198aa0ee754055af18fd535b068464c1b40e1 Mon Sep 17 00:00:00 2001 From: RKS Date: Mon, 14 Sep 2026 08:05:41 -0400 Subject: [PATCH 3/3] fix(python): re-raise finalizer and result-hook exceptions in _finalize_stream Narrow error handling to get_final_response and re-raise exceptions while independently guarding telemetry capture, ensuring caller outcomes and duration error metrics remain aligned. --- .../core/agent_framework/observability.py | 106 ++++++++++-------- .../core/tests/core/test_observability.py | 30 +++++ 2 files changed, 87 insertions(+), 49 deletions(-) diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 3a7b7d9d427..9ecc90d06c8 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -2076,65 +2076,73 @@ def _record_duration() -> None: async def _finalize_stream() -> None: from ._types import ChatResponse + if result_stream._stream_error is not None: # pyright: ignore[reportPrivateUsage] + # Stream errored; skip get_final_response() to avoid firing + # result hooks such as after_run context providers on error + # paths. Capture the error on the span before returning. + capture_exception( + span=span, + exception=result_stream._stream_error, # type: ignore + timestamp=time_ns(), + ) + _capture_operation_error( + attributes=attributes, + exception=result_stream._stream_error, # type: ignore + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration_state.get("duration", perf_counter() - start_time), + ) + _close_span() + return + try: - if result_stream._stream_error is not None: # pyright: ignore[reportPrivateUsage] - # Stream errored; skip get_final_response() to avoid firing - # result hooks such as after_run context providers on error - # paths. Capture the error on the span before returning. - capture_exception( - span=span, - exception=result_stream._stream_error, # type: ignore - timestamp=time_ns(), - ) + try: + response: ChatResponse[Any] = await result_stream.get_final_response() + except Exception as exception: + capture_exception(span=span, exception=exception, timestamp=time_ns()) _capture_operation_error( attributes=attributes, - exception=result_stream._stream_error, # type: ignore + exception=exception, operation_duration_histogram=getattr(self, "duration_histogram", None), duration=duration_state.get("duration", perf_counter() - start_time), ) - return - response: ChatResponse[Any] = await result_stream.get_final_response() - duration = duration_state.get("duration") - response_attributes = _get_response_attributes(attributes, response) - self._backfill_request_model(span, response_attributes) - _capture_response( - span=span, - attributes=response_attributes, - token_usage_histogram=getattr(self, "token_usage_histogram", None), - operation_duration_histogram=getattr(self, "duration_histogram", None), - duration=duration, - ) - _mark_inner_response_telemetry_captured(response) - if ( - OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED - and isinstance(response, ChatResponse) - and response.messages - and span.is_recording() - ): - finish_reason = _get_response_finish_reason(response) - # Activate the span: this cleanup hook runs after the final pull has - # exited its _activate_span context, so it wouldn't otherwise be current. - with _activate_span(span): - _capture_message_events_v1_36( - provider_name=provider_name, + raise + + try: + duration = duration_state.get("duration") + response_attributes = _get_response_attributes(attributes, response) + self._backfill_request_model(span, response_attributes) + _capture_response( + span=span, + attributes=response_attributes, + token_usage_histogram=getattr(self, "token_usage_histogram", None), + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration, + ) + _mark_inner_response_telemetry_captured(response) + if ( + OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED + and isinstance(response, ChatResponse) + and response.messages + and span.is_recording() + ): + finish_reason = _get_response_finish_reason(response) + # Activate the span: this cleanup hook runs after the final pull has + # exited its _activate_span context, so it wouldn't otherwise be current. + with _activate_span(span): + _capture_message_events_v1_36( + provider_name=provider_name, + messages=response.messages, + finish_reason=finish_reason, + output=True, + ) + _capture_message_span_attributes_latest_experimental( + span=span, messages=response.messages, finish_reason=finish_reason, output=True, ) - _capture_message_span_attributes_latest_experimental( - span=span, - messages=response.messages, - finish_reason=finish_reason, - output=True, - ) - except Exception as exception: - capture_exception(span=span, exception=exception, timestamp=time_ns()) - _capture_operation_error( - attributes=attributes, - exception=exception, - operation_duration_histogram=getattr(self, "duration_histogram", None), - duration=duration_state.get("duration", perf_counter() - start_time), - ) + except Exception as telemetry_exception: + logger.debug("Failed to capture telemetry for stream: %s", telemetry_exception) finally: _close_span() diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index bad4d02370f..0dcafce37f8 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -5907,6 +5907,36 @@ def _broken_finalizer(updates: Any) -> ChatResponse: assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION +async def test_chat_client_records_duration_when_stream_result_hook_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream whose result hook raises records the duration metric with error.type.""" + + def _broken_result_hook(response: ChatResponse) -> ChatResponse: + raise RuntimeError("result hook failed") + + client = mock_chat_client() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="result hook failed"): + stream = client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ) + stream.with_result_hook(_broken_result_hook) + async for _ in stream: + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + async def test_embedding_client_records_duration_on_error(span_exporter: InMemorySpanExporter) -> None: """A failed embedding call records gen_ai.client.operation.duration with error.type.""" from agent_framework import BaseEmbeddingClient, GeneratedEmbeddings