diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/span_utils.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/span_utils.py index 61ecb32bc8..581e921b9b 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/span_utils.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/span_utils.py @@ -392,10 +392,14 @@ def set_response_attributes(span, response): @dont_throw -def set_streaming_response_attributes(span, complete_response_events): +def set_streaming_response_attributes(span, complete_response_events, stop_reason=None): from opentelemetry.instrumentation.anthropic import set_span_attribute - if not span.is_recording() or not complete_response_events: + if not span.is_recording(): + return + + complete_response_events = complete_response_events or [] + if not complete_response_events and not stop_reason: return # Collect all parts and determine finish_reason @@ -436,6 +440,12 @@ def set_streaming_response_attributes(span, complete_response_events): "content": event.get("text"), }) + # Fallback when the stream had no content blocks: use message-level stop_reason. + if not finish_reasons and stop_reason: + mapped = _map_finish_reason(stop_reason) + if mapped: + finish_reasons.append(mapped) + if finish_reasons: span.set_attribute(GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS, finish_reasons) @@ -452,3 +462,14 @@ def set_streaming_response_attributes(span, complete_response_events): GenAIAttributes.GEN_AI_OUTPUT_MESSAGES, json.dumps(output_messages, cls=JSONEncoder), ) + elif finish_reasons and should_send_prompts(): + msg = { + "role": "assistant", + "parts": [], + "finish_reason": finish_reasons[-1], + } + set_span_attribute( + span, + GenAIAttributes.GEN_AI_OUTPUT_MESSAGES, + json.dumps([msg], cls=JSONEncoder), + ) diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py index bd2d6dc15d..2b46e263be 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py @@ -58,6 +58,9 @@ def _process_response_item(item, complete_response): if event.get("type") == "tool_use": event["input"] = event.get("input", "") + item.delta.partial_json elif item.type == "message_delta": + # Keep message-level stop_reason even when no content blocks exist + # (events stays empty and the per-event loop below is a no-op). + complete_response["stop_reason"] = item.delta.stop_reason for event in complete_response.get("events", []): event["finish_reason"] = item.delta.stop_reason if item.usage: @@ -175,7 +178,11 @@ def _handle_streaming_response(span, event_logger, complete_response): else: if not span.is_recording(): return - set_streaming_response_attributes(span, complete_response.get("events")) + set_streaming_response_attributes( + span, + complete_response.get("events"), + complete_response.get("stop_reason"), + ) class AnthropicStream(ObjectProxy): diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_semconv_span_attrs.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_semconv_span_attrs.py index f99385702f..378112e78a 100644 --- a/packages/opentelemetry-instrumentation-anthropic/tests/test_semconv_span_attrs.py +++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_semconv_span_attrs.py @@ -375,6 +375,58 @@ def test_streaming_finish_reasons_set_when_content_tracing_disabled(): assert GenAIAttributes.GEN_AI_OUTPUT_MESSAGES not in span.attributes +def _accumulate_synthetic_stream(items): + """Drive stream accumulation with synthetic items (no network / API key).""" + from opentelemetry.instrumentation.anthropic.streaming import ( + _handle_streaming_response, + _process_response_item, + ) + + complete_response = {"events": [], "model": "", "usage": {}, "id": ""} + for item in items: + _process_response_item(item, complete_response) + span = make_span() + span.is_recording = lambda: True + _handle_streaming_response(span, None, complete_response) + return span, complete_response + + +def test_streaming_empty_content_records_finish_reasons_and_output_messages(): + """Empty-content stream still records finish_reasons and output.messages. + + message_delta carries stop_reason but no content_block_* items, so events stays + empty. Both gen_ai.response.finish_reasons and gen_ai.output.messages must still + be set from the message-level stop_reason. + """ + message_delta = SimpleNamespace( + type="message_delta", + delta=SimpleNamespace(stop_reason="end_turn"), + usage=None, + ) + span, complete_response = _accumulate_synthetic_stream([message_delta]) + + assert complete_response.get("stop_reason") == "end_turn" + assert complete_response["events"] == [] + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ["stop"] + output = json.loads(span.attributes[GenAIAttributes.GEN_AI_OUTPUT_MESSAGES]) + assert output == [{"role": "assistant", "parts": [], "finish_reason": "stop"}] + + +def test_streaming_empty_content_omits_output_messages_when_content_tracing_disabled(): + """Empty-content stream still records finish_reasons when TRACELOOP_TRACE_CONTENT=false.""" + os.environ[TRACELOOP_TRACE_CONTENT] = "false" + + message_delta = SimpleNamespace( + type="message_delta", + delta=SimpleNamespace(stop_reason="end_turn"), + usage=None, + ) + span, _ = _accumulate_synthetic_stream([message_delta]) + + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ["stop"] + assert GenAIAttributes.GEN_AI_OUTPUT_MESSAGES not in span.attributes + + def test_finish_reason_empty_string_when_none(): """finish_reason must be '' (not omitted) when stop_reason is None (Bedrock convention).""" span = make_span()